[core] Support REST catalog version management

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -755,6 +755,27 @@ class CatalogError(ValueError):
     resource_name: Optional[str]
     message: str
 
+class Instant:
+    @staticmethod
+    def snapshot(snapshot_id: int) -> "Instant": ...
+    @staticmethod
+    def tag(tag_name: str) -> "Instant": ...
+
+class GetTagResponse:
+    def tag_name(self) -> str: ...
+    def snapshot(self) -> Snapshot: ...
+    def tag_create_time(self) -> Optional[int]: ...
+    def tag_time_retained(self) -> Optional[str]: ...
+    def to_json(self) -> str: ...
+
+class TableSnapshot:
+    def snapshot(self) -> Snapshot: ...
+    def record_count(self) -> Optional[int]: ...
+    def file_size_in_bytes(self) -> Optional[int]: ...
+    def file_count(self) -> Optional[int]: ...
+    def last_file_creation_time(self) -> Optional[int]: ...
+    def to_json(self) -> str: ...
+
 class PaimonCatalog:
     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."""
@@ -772,6 +793,17 @@ class PaimonCatalog:
         """Replace the identifier's current branch metadata through the REST server."""
         ...
     def list_branches(self, identifier: Union[str, Tuple[str, str]]) -> List[str]: ...
+    def create_tag(self, identifier: Union[str, Tuple[str, str]], tag_name: str,
+                   snapshot_id: Optional[int] = None, time_retained: Optional[str] = None,
+                   ignore_if_exists: bool = False) -> None: ...
+    def get_tag(self, identifier: Union[str, Tuple[str, str]], tag_name: str) -> GetTagResponse: ...
+    def list_tags_paged(self, identifier: Union[str, Tuple[str, str]], max_results: Optional[int] = None,
+                       page_token: Optional[str] = None,
+                       tag_name_prefix: Optional[str] = None) -> Tuple[List[str], Optional[str]]: ...
+    def delete_tag(self, identifier: Union[str, Tuple[str, str]], tag_name: str) -> None: ...
+    def load_snapshot(self, identifier: Union[str, Tuple[str, str]]) -> Optional[TableSnapshot]: ...
+    def rollback_to(self, identifier: Union[str, Tuple[str, str]], instant: Instant,
+                    from_snapshot: Optional[int] = None, *, table_uuid: Optional[str] = None) -> None: ...
 
 class PythonScalarUDF:
     def __init__(

bindings/python/src/catalog_version.rs

@@ -0,0 +1,97 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use paimon::api::{GetTagResponse, TableSnapshot};
+use paimon::table::Instant;
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::*;
+
+use crate::snapshot::PySnapshot;
+
+/// Typed rollback target; serialization and rollback semantics belong to core.
+#[pyclass(name = "Instant", module = "pypaimon_rust.datafusion")]
+pub(crate) struct PyInstant {
+    pub(crate) inner: Instant,
+}
+
+#[pymethods]
+impl PyInstant {
+    #[staticmethod]
+    fn snapshot(snapshot_id: i64) -> Self {
+        Self {
+            inner: Instant::snapshot(snapshot_id),
+        }
+    }
+
+    #[staticmethod]
+    fn tag(tag_name: String) -> Self {
+        Self {
+            inner: Instant::tag(tag_name),
+        }
+    }
+}
+
+#[pyclass(name = "GetTagResponse", module = "pypaimon_rust.datafusion")]
+pub(crate) struct PyGetTagResponse {
+    pub(crate) inner: GetTagResponse,
+}
+
+#[pymethods]
+impl PyGetTagResponse {
+    fn tag_name(&self) -> &str {
+        &self.inner.tag_name
+    }
+    fn snapshot(&self) -> PySnapshot {
+        PySnapshot::new(self.inner.snapshot.clone())
+    }
+    fn tag_create_time(&self) -> Option<i64> {
+        self.inner.tag_create_time
+    }
+    fn tag_time_retained(&self) -> Option<&str> {
+        self.inner.tag_time_retained.as_deref()
+    }
+    fn to_json(&self) -> PyResult<String> {
+        serde_json::to_string(&self.inner).map_err(|error| PyValueError::new_err(error.to_string()))
+    }
+}
+
+#[pyclass(name = "TableSnapshot", module = "pypaimon_rust.datafusion")]
+pub(crate) struct PyTableSnapshot {
+    pub(crate) inner: TableSnapshot,
+}
+
+#[pymethods]
+impl PyTableSnapshot {
+    fn snapshot(&self) -> PySnapshot {
+        PySnapshot::new(self.inner.snapshot.clone())
+    }
+    fn record_count(&self) -> Option<i64> {
+        self.inner.record_count
+    }
+    fn file_size_in_bytes(&self) -> Option<i64> {
+        self.inner.file_size_in_bytes
+    }
+    fn file_count(&self) -> Option<i64> {
+        self.inner.file_count
+    }
+    fn last_file_creation_time(&self) -> Option<i64> {
+        self.inner.last_file_creation_time
+    }
+    fn to_json(&self) -> PyResult<String> {
+        serde_json::to_string(&self.inner).map_err(|error| PyValueError::new_err(error.to_string()))
+    }
+}

bindings/python/src/context.rs

@@ -41,6 +41,7 @@ use pyo3::prelude::*;
 use pyo3::types::PyCapsule;
 
 use crate::blob::PyBlobReaderRegistry;
+use crate::catalog_version::{PyGetTagResponse, PyInstant, PyTableSnapshot};
 use crate::error::{df_to_py_err, to_catalog_py_err, to_py_err};
 use crate::read::PyRecordBatchReader;
 use crate::table::PyTable;
@@ -292,7 +293,7 @@ pub struct PaimonCatalog {
     provider: Arc<PaimonCatalogProvider>,
 }
 
-fn catalog_table_identifier(identifier: &Bound<'_, PyAny>) -> PyResult<Identifier> {
+fn catalog_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!(
@@ -308,7 +309,11 @@ fn catalog_table_identifier(identifier: &Bound<'_, PyAny>) -> PyResult<Identifie
             "database and table names must not be empty",
         ));
     }
-    let identifier = Identifier::new(&database, &object_name);
+    Ok(Identifier::new(&database, &object_name))
+}
+
+fn catalog_table_identifier(identifier: &Bound<'_, PyAny>) -> PyResult<Identifier> {
+    let identifier = catalog_identifier(identifier)?;
     if identifier
         .parsed_object_name()
         .map_err(to_py_err)?
@@ -455,6 +460,109 @@ impl PaimonCatalog {
         py.detach(|| runtime().block_on(self.catalog.list_branches(&identifier)))
             .map_err(to_catalog_py_err)
     }
+
+    /// Create a tag through core catalog handling, including optional retention.
+    #[pyo3(signature = (identifier, tag_name, snapshot_id=None, time_retained=None, ignore_if_exists=false))]
+    fn create_tag(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+        tag_name: &str,
+        snapshot_id: Option<i64>,
+        time_retained: Option<&str>,
+        ignore_if_exists: bool,
+    ) -> PyResult<()> {
+        let identifier = catalog_identifier(identifier)?;
+        py.detach(|| {
+            runtime().block_on(self.catalog.create_tag_with_retention(
+                &identifier,
+                tag_name,
+                snapshot_id,
+                time_retained,
+                ignore_if_exists,
+            ))
+        })
+        .map_err(to_catalog_py_err)
+    }
+
+    fn get_tag(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+        tag_name: &str,
+    ) -> PyResult<PyGetTagResponse> {
+        let identifier = catalog_identifier(identifier)?;
+        let inner = py
+            .detach(|| runtime().block_on(self.catalog.get_tag(&identifier, tag_name)))
+            .map_err(to_catalog_py_err)?;
+        Ok(PyGetTagResponse { inner })
+    }
+
+    #[pyo3(signature = (identifier, max_results=None, page_token=None, tag_name_prefix=None))]
+    fn list_tags_paged(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+        max_results: Option<i32>,
+        page_token: Option<&str>,
+        tag_name_prefix: Option<&str>,
+    ) -> PyResult<(Vec<String>, Option<String>)> {
+        let identifier = catalog_identifier(identifier)?;
+        let page = py
+            .detach(|| {
+                runtime().block_on(self.catalog.list_tags_paged(
+                    &identifier,
+                    max_results,
+                    page_token,
+                    tag_name_prefix,
+                ))
+            })
+            .map_err(to_catalog_py_err)?;
+        Ok((page.elements, page.next_page_token))
+    }
+
+    fn delete_tag(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+        tag_name: &str,
+    ) -> PyResult<()> {
+        let identifier = catalog_identifier(identifier)?;
+        py.detach(|| runtime().block_on(self.catalog.delete_tag(&identifier, tag_name, false)))
+            .map_err(to_catalog_py_err)
+    }
+
+    fn load_snapshot(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+    ) -> PyResult<Option<PyTableSnapshot>> {
+        let identifier = catalog_identifier(identifier)?;
+        py.detach(|| runtime().block_on(self.catalog.load_snapshot(&identifier)))
+            .map(|snapshot| snapshot.map(|inner| PyTableSnapshot { inner }))
+            .map_err(to_catalog_py_err)
+    }
+
+    #[pyo3(signature = (identifier, instant, from_snapshot=None, *, table_uuid=None))]
+    fn rollback_to(
+        &self,
+        py: Python<'_>,
+        identifier: &Bound<'_, PyAny>,
+        instant: &PyInstant,
+        from_snapshot: Option<i64>,
+        table_uuid: Option<&str>,
+    ) -> PyResult<()> {
+        let identifier = catalog_identifier(identifier)?;
+        py.detach(|| {
+            runtime().block_on(self.catalog.rollback_to(
+                &identifier,
+                &instant.inner,
+                from_snapshot,
+                table_uuid,
+            ))
+        })
+        .map_err(to_catalog_py_err)
+    }
 }
 
 #[pyclass(name = "SQLOptions", module = "pypaimon_rust.datafusion")]
@@ -879,6 +987,9 @@ 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::<PyInstant>()?;
+    this.add_class::<PyGetTagResponse>()?;
+    this.add_class::<PyTableSnapshot>()?;
     this.add_class::<crate::table::PyTable>()?;
     this.add_class::<crate::system_table::PySystemRead>()?;
     this.add_class::<crate::index_build::PyGlobalIndexBuildBuilder>()?;

bindings/python/src/error.rs

@@ -45,6 +45,10 @@ pub(crate) fn to_catalog_py_err(error: paimon::Error) -> PyErr {
     if error.is_process_fork_unsupported() {
         return to_py_err(error);
     }
+    let snapshot_name = match &error {
+        Error::SnapshotNotExist { snapshot_id } => Some(snapshot_id.to_string()),
+        _ => None,
+    };
     let (code, resource_type, resource_name, message) = match &error {
         Error::BranchAlreadyExist { branch_name, .. } => (
             409,
@@ -67,6 +71,15 @@ pub(crate) fn to_catalog_py_err(error: paimon::Error) -> PyErr {
         Error::TagNotExist { tag_name } => {
             (404, Some("TAG"), Some(tag_name.as_str()), error.to_string())
         }
+        Error::TagAlreadyExist { tag_name } => {
+            (409, Some("TAG"), Some(tag_name.as_str()), error.to_string())
+        }
+        Error::SnapshotNotExist { .. } => (
+            404,
+            Some("SNAPSHOT"),
+            snapshot_name.as_deref(),
+            error.to_string(),
+        ),
         Error::DataInvalid { message, .. } => (400, None, None, message.clone()),
         Error::RestApi { source } => {
             let (code, resource_type, resource_name, message) = match source {

bindings/python/src/lib.rs

@@ -19,6 +19,7 @@ use pyo3::prelude::*;
 
 mod blob;
 mod blob_uri_reader;
+mod catalog_version;
 mod commit_callback;
 mod context;
 mod error;