[core] Support REST catalog branch management

bindings/python/README.md

@@ -359,6 +359,26 @@ without reading a branch schema file. Omit it to retain the original branch.
 Cached time-travel resolution is discarded so the supplied options select the
 snapshot, with the externally resolved fields preserved.
 
+REST catalogs also expose `create_branch(identifier, branch, from_tag=None)`,
+`drop_branch(identifier, branch)`, `fast_forward(identifier, branch)` and
+`list_branches(identifier)`. These delegate to core REST catalog operations and
+release the GIL while waiting for requests. Creation without a tag creates an
+empty branch, as in Java. REST branch rename is unsupported.
+
+REST authentication also supports explicit `dlf.signing-algorithm=openapi-v4`.
+Rust core signs ACS4-HMAC-SHA256 requests with the same canonical request,
+credential scope and action headers as Java. GET requests retry HTTP 429/503 up
+to five times and honor Retry-After; branch mutations are sent once and ignore
+successful response payloads.
+
+`PaimonCatalog(options, config_required=False)` reuses already resolved REST
+options without fetching `/config` again. Branch operation failures raise
+`CatalogError`, a `ValueError` subclass with `code`, `resource_type`,
+`resource_name` and `message` attributes. `code=0` identifies transport or
+response-decoding failures without an HTTP error status. Malformed identifiers
+remain `ValueError`, unsupported catalogs raise `NotImplementedError`, and
+inherited Native state still raises `ForkSafetyError`.
+
 REST tables load the latest snapshot through the catalog, including empty
 results and branch-scoped requests. Permission and service failures (including
 HTTP 501) are propagated as in Java, and the

bindings/python/project-description.md

@@ -148,4 +148,21 @@ ctx = SessionContext()
 ctx.register_catalog_provider("paimon", catalog)
 ```
 
+REST catalogs also support branch management through Rust core:
+
+```python
+identifier = ("db", "table")
+catalog.create_branch(identifier, "empty-branch")
+catalog.create_branch(identifier, "release-branch", from_tag="release-tag")
+branches = catalog.list_branches(identifier)
+release = catalog.get_table(("db", "table$branch_release-branch"))
+catalog.fast_forward(identifier, "release-branch")
+catalog.drop_branch(identifier, "empty-branch")
+```
+
+The server validates branch names and publishes metadata changes. Fast-forward
+updates the identifier's current branch (main by default); it does not change
+the source branch. These catalog methods currently require a REST catalog.
+Other catalog implementations raise `NotImplementedError`.
+
 Time travel queries are not supported in the Python binding at this time.

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -749,12 +749,29 @@ class StreamWriteBuilder:
     def new_update(self) -> StreamTableUpdate: ...
     def new_commit(self) -> StreamTableCommit: ...
 
+class CatalogError(ValueError):
+    code: int
+    resource_type: Optional[str]
+    resource_name: Optional[str]
+    message: str
+
 class PaimonCatalog:
-    def __init__(self, catalog_options: Dict[str, str]) -> None: ...
+    def __init__(self, catalog_options: Dict[str, str], *, config_required: bool = True) -> None:
+        """Set config_required=False when the REST configuration has already been resolved."""
+        ...
     def __datafusion_catalog_provider__(self, session: Any) -> object: ...
     def list_databases(self) -> List[str]: ...
     def list_tables(self, database_name: str) -> List[str]: ...
     def get_table(self, identifier: Union[str, Tuple[str, str]]) -> Table: ...
+    def create_branch(self, identifier: Union[str, Tuple[str, str]], branch_name: str,
+                      from_tag: Optional[str] = None) -> None:
+        """Create a REST branch, empty unless from_tag is supplied."""
+        ...
+    def drop_branch(self, identifier: Union[str, Tuple[str, str]], branch_name: str) -> None: ...
+    def fast_forward(self, identifier: Union[str, Tuple[str, str]], branch_name: str) -> None:
+        """Replace the identifier's current branch metadata through the REST server."""
+        ...
+    def list_branches(self, identifier: Union[str, Tuple[str, str]]) -> List[str]: ...
 
 class PythonScalarUDF:
     def __init__(

bindings/python/src/context.rs

@@ -41,7 +41,7 @@ use pyo3::prelude::*;
 use pyo3::types::PyCapsule;
 
 use crate::blob::PyBlobReaderRegistry;
-use crate::error::{df_to_py_err, to_py_err};
+use crate::error::{df_to_py_err, to_catalog_py_err, to_py_err};
 use crate::read::PyRecordBatchReader;
 use crate::table::PyTable;
 use crate::udf::{build_python_scalar_udf, udf, PyPythonScalarUDFObject};
@@ -189,12 +189,26 @@ pub(crate) fn complete_storage_options(
     )
 }
 
-fn build_paimon_catalog(catalog_options: HashMap<String, String>) -> PyResult<Arc<dyn Catalog>> {
+fn build_paimon_catalog(
+    catalog_options: HashMap<String, String>,
+    config_required: bool,
+) -> PyResult<Arc<dyn Catalog>> {
     let catalog_options = complete_storage_options(catalog_options);
     let rt = runtime();
     rt.block_on(async {
         let options = Options::from_map(catalog_options);
-        let catalog = CatalogFactory::create(options).await.map_err(to_py_err)?;
+        let catalog: Arc<dyn Catalog> =
+            if !config_required && options.get("metastore").map(String::as_str) == Some("rest") {
+                Arc::new(
+                    paimon::catalog::RESTCatalog::new(options, false)
+                        .await
+                        .map_err(to_catalog_py_err)?,
+                )
+            } else {
+                CatalogFactory::create(options)
+                    .await
+                    .map_err(to_catalog_py_err)?
+            };
         Ok::<_, PyErr>(catalog)
     })
 }
@@ -278,12 +292,47 @@ pub struct PaimonCatalog {
     provider: Arc<PaimonCatalogProvider>,
 }
 
+fn catalog_table_identifier(identifier: &Bound<'_, PyAny>) -> PyResult<Identifier> {
+    let (database, object_name) = if let Ok(name) = identifier.extract::<String>() {
+        let (database, table) = name.split_once('.').ok_or_else(|| {
+            PyValueError::new_err(format!(
+                "expected identifier in 'db.table' format, got '{name}'"
+            ))
+        })?;
+        (database.to_string(), table.to_string())
+    } else {
+        identifier.extract::<(String, String)>()?
+    };
+    if database.is_empty() || object_name.is_empty() {
+        return Err(PyValueError::new_err(
+            "database and table names must not be empty",
+        ));
+    }
+    let identifier = Identifier::new(&database, &object_name);
+    if identifier
+        .parsed_object_name()
+        .map_err(to_py_err)?
+        .system_table()
+        .is_some()
+    {
+        return Err(PyValueError::new_err(
+            "Catalog table operations do not support system-table identifiers",
+        ));
+    }
+    Ok(identifier)
+}
+
 #[pymethods]
 impl PaimonCatalog {
     /// Create a Paimon catalog that can be registered into a DataFusion session.
     #[new]
-    fn new(py: Python<'_>, catalog_options: HashMap<String, String>) -> PyResult<Self> {
-        let catalog = py.detach(|| build_paimon_catalog(catalog_options))?;
+    #[pyo3(signature = (catalog_options, *, config_required=true))]
+    fn new(
+        py: Python<'_>,
+        catalog_options: HashMap<String, String>,
+        config_required: bool,
+    ) -> PyResult<Self> {
+        let catalog = py.detach(|| build_paimon_catalog(catalog_options, config_required))?;
         let provider = Arc::new(
             PaimonCatalogProvider::new(
                 None,
@@ -334,29 +383,9 @@ impl PaimonCatalog {
     /// Get a table by "db.table" or a (database, table) tuple.
     /// Tuple components preserve dots; the table can include a branch suffix.
     fn get_table(&self, py: Python<'_>, identifier: &Bound<'_, PyAny>) -> PyResult<PyTable> {
-        let (database, object_name) = if let Ok(name) = identifier.extract::<String>() {
-            let (database, table) = name.split_once('.').ok_or_else(|| {
-                PyValueError::new_err(format!(
-                    "expected identifier in 'db.table' format, got '{name}'"
-                ))
-            })?;
-            (database.to_string(), table.to_string())
-        } else {
-            identifier.extract::<(String, String)>()?
-        };
-        if database.is_empty() || object_name.is_empty() {
-            return Err(PyValueError::new_err(
-                "database and table names must not be empty",
-            ));
-        }
-        let id = Identifier::new(&database, &object_name);
+        let id = catalog_table_identifier(identifier)?;
         let parsed = id.parsed_object_name().map_err(to_py_err)?;
-        if parsed.system_table().is_some() {
-            return Err(PyValueError::new_err(
-                "get_table() does not support system-table identifiers",
-            ));
-        }
-        let base_id = Identifier::new(&database, parsed.table());
+        let base_id = Identifier::new(id.database(), parsed.table());
         let branch = parsed.branch().map(str::to_string);
         let catalog = Arc::clone(&self.catalog);
         let table = py.detach(|| {
@@ -372,6 +401,60 @@ impl PaimonCatalog {
         })?;
         Ok(PyTable::new(Arc::new(table)))
     }
+
+    /// Create an empty branch or one initialized from a tag through the REST catalog.
+    #[pyo3(signature = (identifier, branch_name, from_tag=None))]
+    fn create_branch(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+        branch_name: &str,
+        from_tag: Option<&str>,
+    ) -> PyResult<()> {
+        let identifier = catalog_table_identifier(identifier)?;
+        py.detach(|| {
+            runtime().block_on(
+                self.catalog
+                    .create_branch(&identifier, branch_name, from_tag),
+            )
+        })
+        .map_err(to_catalog_py_err)
+    }
+
+    /// Delete branch metadata through the REST catalog.
+    fn drop_branch(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+        branch_name: &str,
+    ) -> PyResult<()> {
+        let identifier = catalog_table_identifier(identifier)?;
+        py.detach(|| runtime().block_on(self.catalog.drop_branch(&identifier, branch_name)))
+            .map_err(to_catalog_py_err)
+    }
+
+    /// Fast-forward the identifier's current branch through the REST catalog.
+    fn fast_forward(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+        branch_name: &str,
+    ) -> PyResult<()> {
+        let identifier = catalog_table_identifier(identifier)?;
+        py.detach(|| runtime().block_on(self.catalog.fast_forward(&identifier, branch_name)))
+            .map_err(to_catalog_py_err)
+    }
+
+    /// List branch names in REST server order.
+    fn list_branches(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+    ) -> PyResult<Vec<String>> {
+        let identifier = catalog_table_identifier(identifier)?;
+        py.detach(|| runtime().block_on(self.catalog.list_branches(&identifier)))
+            .map_err(to_catalog_py_err)
+    }
 }
 
 #[pyclass(name = "SQLOptions", module = "pypaimon_rust.datafusion")]
@@ -794,6 +877,7 @@ impl PySQLContext {
 
 pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
     let this = PyModule::new(py, "datafusion")?;
+    this.add("CatalogError", py.get_type::<crate::error::CatalogError>())?;
     this.add_class::<PaimonCatalog>()?;
     this.add_class::<crate::table::PyTable>()?;
     this.add_class::<crate::system_table::PySystemRead>()?;

bindings/python/src/error.rs

@@ -25,8 +25,96 @@ pyo3::create_exception!(
     "Raised when native state inherited across process fork cannot be used safely."
 );
 
+pyo3::create_exception!(
+    pypaimon_rust,
+    CatalogError,
+    PyValueError,
+    "A catalog failure with code, resource_type, resource_name and message attributes."
+);
+
 pub fn register_module(py: Python<'_>, module: &Bound<'_, PyModule>) -> PyResult<()> {
-    module.add("ForkSafetyError", py.get_type::<ForkSafetyError>())
+    module.add("ForkSafetyError", py.get_type::<ForkSafetyError>())?;
+    module.add("CatalogError", py.get_type::<CatalogError>())
+}
+
+/// Preserve the catalog error contract across the FFI without parsing display strings.
+pub(crate) fn to_catalog_py_err(error: paimon::Error) -> PyErr {
+    use paimon::api::RestError;
+    use paimon::Error;
+
+    if error.is_process_fork_unsupported() {
+        return to_py_err(error);
+    }
+    let (code, resource_type, resource_name, message) = match &error {
+        Error::BranchAlreadyExist { branch_name, .. } => (
+            409,
+            Some("BRANCH"),
+            Some(branch_name.as_str()),
+            error.to_string(),
+        ),
+        Error::BranchNotExist { branch_name, .. } => (
+            404,
+            Some("BRANCH"),
+            Some(branch_name.as_str()),
+            error.to_string(),
+        ),
+        Error::TableNotExist { full_name } => (
+            404,
+            Some("TABLE"),
+            Some(full_name.as_str()),
+            error.to_string(),
+        ),
+        Error::TagNotExist { tag_name } => {
+            (404, Some("TAG"), Some(tag_name.as_str()), error.to_string())
+        }
+        Error::DataInvalid { message, .. } => (400, None, None, message.clone()),
+        Error::RestApi { source } => {
+            let (code, resource_type, resource_name, message) = match source {
+                RestError::BadRequest { message } => (400, None, None, message),
+                RestError::NotAuthorized { message } => (401, None, None, message),
+                RestError::Forbidden { message } => (403, None, None, message),
+                RestError::NoSuchResource {
+                    resource_type,
+                    resource_name,
+                    message,
+                } => (
+                    404,
+                    resource_type.as_deref(),
+                    resource_name.as_deref(),
+                    message,
+                ),
+                RestError::AlreadyExists {
+                    resource_type,
+                    resource_name,
+                    message,
+                } => (
+                    409,
+                    resource_type.as_deref(),
+                    resource_name.as_deref(),
+                    message,
+                ),
+                RestError::ServiceFailure { message } => (500, None, None, message),
+                RestError::NotImplemented { message } => (501, None, None, message),
+                RestError::ServiceUnavailable { message } => (503, None, None, message),
+                RestError::Unexpected { code, message } => (*code, None, None, message),
+            };
+            (code, resource_type, resource_name, message.clone())
+        }
+        Error::UnexpectedError { .. } | Error::IoUnexpected { .. } => {
+            (0, None, None, error.to_string())
+        }
+        _ => return to_py_err(error),
+    };
+    Python::attach(|py| {
+        let exception = CatalogError::new_err(error.to_string());
+        let value = exception.value(py);
+        value.setattr("code", code)?;
+        value.setattr("resource_type", resource_type)?;
+        value.setattr("resource_name", resource_name)?;
+        value.setattr("message", message)?;
+        Ok::<_, PyErr>(exception)
+    })
+    .unwrap_or_else(|error| error)
 }
 
 pub fn to_py_err(err: paimon::Error) -> PyErr {