This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 53dfa2cb [core] Support REST catalog version management (#1142)
53dfa2cb is described below
commit 53dfa2cb5e18141cb5f4bf1b37c1494a950ec00b
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 17 13:32:46 2026 +0800
[core] Support REST catalog version management (#1142)
---
.../python/python/pypaimon_rust/datafusion.pyi | 32 ++
bindings/python/src/catalog_version.rs | 97 ++++++
bindings/python/src/context.rs | 115 ++++++-
bindings/python/src/error.rs | 13 +
bindings/python/src/lib.rs | 1 +
bindings/python/tests/test_catalog_versions.py | 172 ++++++++++
crates/integrations/datafusion/src/procedures.rs | 21 +-
crates/paimon/src/api/api_request.rs | 11 +
crates/paimon/src/api/api_response.rs | 8 +
crates/paimon/src/api/mod.rs | 4 +-
crates/paimon/src/api/rest_api.rs | 117 ++++---
crates/paimon/src/catalog/mod.rs | 36 +++
crates/paimon/src/catalog/rest/rest_catalog.rs | 181 +++++++++--
crates/paimon/src/spec/snapshot.rs | 16 +
.../lib.rs => crates/paimon/src/table/instant.rs | 64 ++--
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/tests/mock_server.rs | 8 +-
crates/paimon/tests/rest_branch_test.rs | 356 ++++++++++++++++++++-
18 files changed, 1131 insertions(+), 123 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 44cee590..8c1bc75b 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/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__(
diff --git a/bindings/python/src/catalog_version.rs
b/bindings/python/src/catalog_version.rs
new file mode 100644
index 00000000..ba92f1fe
--- /dev/null
+++ b/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()))
+ }
+}
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index d7223350..170b3eab 100644
--- a/bindings/python/src/context.rs
+++ b/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>()?;
diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs
index b94999cb..4e9b277a 100644
--- a/bindings/python/src/error.rs
+++ b/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 {
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 0b4508c8..72d17d42 100644
--- a/bindings/python/src/lib.rs
+++ b/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;
diff --git a/bindings/python/tests/test_catalog_versions.py
b/bindings/python/tests/test_catalog_versions.py
new file mode 100644
index 00000000..47760084
--- /dev/null
+++ b/bindings/python/tests/test_catalog_versions.py
@@ -0,0 +1,172 @@
+# 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.
+
+"""REST version management exposes typed core operations without Python
work."""
+
+from http.server import BaseHTTPRequestHandler, HTTPServer
+import json
+import threading
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
+
+from pypaimon_rust.datafusion import CatalogError, Instant, PaimonCatalog
+
+
+SNAPSHOT = dict(version=3, id=7, schemaId=2, baseManifestList="base",
deltaManifestList="delta",
+ commitUser="writer", commitIdentifier=1, commitKind="APPEND",
timeMillis=123,
+ uuid="snapshot-uuid", totalRecordCount=8,
properties={"source": "test"},
+ baseManifestListSize=321, deltaManifestListSize=0)
+
+
[email protected]
+def transport():
+ class Handler(BaseHTTPRequestHandler):
+ def operation(self):
+ size = int(self.headers.get("Content-Length", 0))
+ body = json.loads(self.rfile.read(size)) if size else None
+ self.server.requests.append((self.command, self.path, body))
+ assert self.headers["Authorization"] == "Bearer token"
+ data = self.server.body
+ if data is None and self.server.status == 200:
+ path = urlsplit(self.path).path
+ if path.endswith("/snapshot"):
+ data = {"snapshot": dict(snapshot=SNAPSHOT, recordCount=0,
fileCount=2,
+ fileSizeInBytes=123,
lastFileCreationTime=456)}
+ elif path.endswith("/tags"):
+ data = {"tags": ["z", "a", "z"], "nextPageToken": "next
+/%一"}
+ elif "/tags/" in path:
+ data = dict(tagName="release", snapshot=SNAPSHOT,
tagCreateTime=123,
+ tagTimeRetained="2 d")
+ output = data.encode() if isinstance(data, str) else
json.dumps(data).encode()
+ self.send_response(self.server.status)
+ self.send_header("Content-Length", str(len(output)))
+ self.end_headers()
+ self.wfile.write(output)
+
+ do_GET = do_POST = do_DELETE = operation
+
+ def log_message(self, *args):
+ pass
+
+ server = HTTPServer(("127.0.0.1", 0), Handler)
+ server.requests, server.status, server.body = [], 200, None
+ thread = threading.Thread(target=server.serve_forever)
+ thread.start()
+ try:
+ yield server, PaimonCatalog({
+ "metastore": "rest", "uri": "http://127.0.0.1:%d"; %
server.server_port,
+ "warehouse": "w", "prefix": "tenant", "token.provider": "bear",
"token": "token",
+ }, config_required=False)
+ finally:
+ server.shutdown()
+ server.server_close()
+ thread.join()
+
+
+def test_typed_metadata_retains_snapshot_and_catalog_statistics(transport):
+ server, native = transport
+ snapshot = native.load_snapshot(("db.name", "table$branch_dev /一"))
+ assert snapshot.snapshot().id() == 7
+ assert snapshot.snapshot().uuid() == "snapshot-uuid"
+ assert snapshot.snapshot().properties() == {"source": "test"}
+ assert (snapshot.record_count(), snapshot.file_count(),
snapshot.file_size_in_bytes(),
+ snapshot.last_file_creation_time()) == (0, 2, 123, 456)
+ assert json.loads(snapshot.to_json())["snapshot"]["baseManifestList"] ==
"base"
+ assert json.loads(snapshot.to_json())["snapshot"]["baseManifestListSize"]
== 321
+ assert json.loads(snapshot.to_json())["snapshot"]["deltaManifestListSize"]
== 0
+ tag = native.get_tag("db.table", "release")
+ assert tag.tag_name() == "release"
+ assert tag.snapshot().id() == 7
+ assert tag.tag_create_time() == 123
+ assert tag.tag_time_retained() == "2 d"
+ assert json.loads(tag.to_json())["snapshot"]["properties"] == {"source":
"test"}
+ assert json.loads(tag.to_json())["snapshot"]["baseManifestListSize"] == 321
+ assert json.loads(tag.to_json())["snapshot"]["deltaManifestListSize"] == 0
+ assert len(server.requests) == 2
+
+
+def test_tag_page_uses_tuple_and_preserves_opaque_parameters(transport):
+ server, native = transport
+ page = native.list_tags_paged("db.table", 3, "next +/%一", "release +/%一")
+ assert page == (["z", "a", "z"], "next +/%一")
+ query = parse_qs(urlsplit(server.requests[0][1]).query)
+ assert query == {"maxResults": ["3"], "pageToken": ["next +/%一"],
"tagNamePrefix": ["release +/%一"]}
+
+
[email protected]("instant,payload", [
+ (Instant.snapshot(7), {"type": "snapshot", "snapshotId": 7}),
+ (Instant.tag("release /一"), {"type": "tag", "tagName": "release /一"}),
+])
[email protected]("guard", [False, True])
+def test_rollback_instants_and_guards_are_forwarded(transport, instant,
payload, guard):
+ server, native = transport
+ server.body = "OK"
+ kwargs = dict(from_snapshot=9, table_uuid="table-incarnation") if guard
else {}
+ native.rollback_to(("db.name", "table$branch_dev /一"), instant, **kwargs)
+ expected = {"instant": payload}
+ if guard:
+ expected.update(fromSnapshot=9, tableId="table-incarnation")
+ assert len(server.requests) == 1
+ method, path, body = server.requests[0]
+ assert method == "POST"
+ assert path.endswith("/tables/table%24branch_dev+%2F%E4%B8%80/rollback")
+ assert body == expected
+
+
[email protected]("operation", ["create_tag", "delete_tag",
"rollback_to"])
[email protected]("status", [400, 403, 409, 429, 500, 501, 503])
+def test_failed_mutations_are_not_replayed(transport, operation, status):
+ server, native = transport
+ server.status, server.body = status, {"message": "rejected"}
+ args = (Instant.snapshot(7),) if operation == "rollback_to" else
("release",)
+ with pytest.raises(CatalogError) as caught:
+ getattr(native, operation)("db.table", *args)
+ assert caught.value.code == status
+ assert len(server.requests) == 1
+
+
[email protected]("operation,resource,name", [
+ ("create_tag", "SNAPSHOT", "7"), ("create_tag", "TAG", "release"),
+ ("get_tag", "TAG", "release"), ("delete_tag", "TAG", "release"),
+ ("rollback_to", "SNAPSHOT", "7"), ("rollback_to", "TAG", "release"),
+ ("rollback_to", "TABLE", "db.table"), ("list_tags_paged", "TABLE",
"db.table"),
+])
+def test_structured_version_errors_preserve_missing_resource(transport,
operation, resource, name):
+ server, native = transport
+ server.status = 409 if operation == "create_tag" and resource == "TAG"
else 404
+ server.body = dict(resourceType=resource, resourceName=name,
message="missing")
+ if operation == "rollback_to":
+ args = (Instant.snapshot(7),)
+ elif operation == "list_tags_paged":
+ args = ()
+ else:
+ args = ("release",)
+ if operation == "create_tag":
+ args += (7,)
+ with pytest.raises(CatalogError) as caught:
+ getattr(native, operation)("db.table", *args)
+ assert (caught.value.code, caught.value.resource_type,
caught.value.resource_name) == (
+ server.status, resource, name)
+ assert len(server.requests) == 1
+
+
+def test_native_tag_ignore_existing_is_the_core_catalog_contract(transport):
+ server, native = transport
+ server.status, server.body = 409, {"resourceType": "TAG", "resourceName":
"release"}
+ native.create_tag("db.table", "release", ignore_if_exists=True)
+ assert len(server.requests) == 1
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 0161536c..3124a863 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -635,13 +635,23 @@ async fn proc_rollback_to(
.map_err(|_| DataFusionError::Plan(format!("Invalid
snapshot_id: '{id_str}'")))?;
rest_env
.api()
- .rollback_to_snapshot(rest_env.identifier(), id)
+ .rollback_to(
+ rest_env.identifier(),
+ &paimon::table::Instant::snapshot(id),
+ None,
+ Some(rest_env.uuid()),
+ )
.await
.map_err(to_datafusion_error)?;
} else if let Some(tag_name) = args.get("tag") {
rest_env
.api()
- .rollback_to_tag(rest_env.identifier(), tag_name)
+ .rollback_to(
+ rest_env.identifier(),
+ &paimon::table::Instant::tag(tag_name),
+ None,
+ Some(rest_env.uuid()),
+ )
.await
.map_err(to_datafusion_error)?;
} else {
@@ -723,7 +733,12 @@ async fn proc_rollback_to_timestamp(
if let Some(rest_env) = table.rest_env() {
rest_env
.api()
- .rollback_to_snapshot(rest_env.identifier(), snapshot.id())
+ .rollback_to(
+ rest_env.identifier(),
+ &paimon::table::Instant::snapshot(snapshot.id()),
+ None,
+ Some(rest_env.uuid()),
+ )
.await
.map_err(to_datafusion_error)?;
} else {
diff --git a/crates/paimon/src/api/api_request.rs
b/crates/paimon/src/api/api_request.rs
index b46300c3..132b11bd 100644
--- a/crates/paimon/src/api/api_request.rs
+++ b/crates/paimon/src/api/api_request.rs
@@ -87,6 +87,17 @@ pub struct CreateBranchRequest {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ForwardBranchRequest {}
+/// A rollback guarded by the observed table incarnation and latest snapshot.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RollbackTableRequest {
+ pub instant: crate::table::Instant,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub from_snapshot: Option<i64>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub table_id: Option<String>,
+}
+
/// Request to create a table tag.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
diff --git a/crates/paimon/src/api/api_response.rs
b/crates/paimon/src/api/api_response.rs
index aa489f62..8863c597 100644
--- a/crates/paimon/src/api/api_response.rs
+++ b/crates/paimon/src/api/api_response.rs
@@ -170,6 +170,14 @@ pub struct GetTagResponse {
pub tag_time_retained: Option<String>,
}
+/// Tag names and an opaque continuation token supplied by the REST server.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListTagsResponse {
+ pub tags: Option<Vec<String>>,
+ pub next_page_token: Option<String>,
+}
+
/// Response for getting a persistent view.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
diff --git a/crates/paimon/src/api/mod.rs b/crates/paimon/src/api/mod.rs
index 55527ae0..f21a1e6a 100644
--- a/crates/paimon/src/api/mod.rs
+++ b/crates/paimon/src/api/mod.rs
@@ -37,7 +37,7 @@ pub use api_request::{
CreateDatabaseRequest, CreateFunctionRequest, CreatePartitionsRequest,
CreateTableRequest,
CreateTagRequest, CreateViewRequest, DropPartitionsRequest,
DropPolicyRequest,
ForwardBranchRequest, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
- PolicyRequest, RenameTableRequest, RevokePermissionRequest,
+ PolicyRequest, RenameTableRequest, RevokePermissionRequest,
RollbackTableRequest,
};
// Re-export response types
@@ -46,7 +46,7 @@ pub use api_response::{
GetFunctionResponse, GetTableResponse, GetTableSnapshotResponse,
GetTableTokenResponse,
GetTagResponse, GetViewResponse, ListBranchesResponse,
ListDatabasesResponse,
ListFunctionsResponse, ListPartitionsResponse, ListPermissionsResponse,
ListPoliciesResponse,
- ListTablesResponse, ListViewsResponse, PagedList, TableSnapshot,
+ ListTablesResponse, ListTagsResponse, ListViewsResponse, PagedList,
TableSnapshot,
};
// Re-export management types
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index 45ec5a86..c7329a3b 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -33,13 +33,13 @@ use super::api_request::{
CreateDatabaseRequest, CreateFunctionRequest, CreatePartitionsRequest,
CreateTableRequest,
CreateTagRequest, CreateViewRequest, DropPartitionsRequest,
DropPolicyRequest,
ForwardBranchRequest, ListPartitionsByFilterRequest,
ListPartitionsByNamesRequest,
- PolicyRequest, RenameTableRequest, RevokePermissionRequest,
+ PolicyRequest, RenameTableRequest, RevokePermissionRequest,
RollbackTableRequest,
};
use super::api_response::{
AuthTableQueryResponse, ConfigResponse, ErrorResponse, GetDatabaseResponse,
GetFunctionResponse, GetTableResponse, GetTagResponse, GetViewResponse,
ListBranchesResponse,
ListDatabasesResponse, ListFunctionsResponse, ListPartitionsResponse,
ListPermissionsResponse,
- ListPoliciesResponse, ListTablesResponse, ListViewsResponse, PagedList,
+ ListPoliciesResponse, ListTablesResponse, ListTagsResponse,
ListViewsResponse, PagedList,
};
use super::auth::{AuthProviderFactory, RESTAuthFunction};
use super::management::{
@@ -420,16 +420,16 @@ impl RESTApi {
&self,
identifier: &Identifier,
) -> Result<Option<super::TableSnapshot>> {
- validate_non_empty_multi(&[
+ validate_resource_path(&[
(identifier.database(), "database name"),
(identifier.object(), "table name"),
])?;
let path = self
.resource_paths
.table_snapshot(identifier.database(), identifier.object());
- let response: super::GetTableSnapshotResponse =
+ let response: Option<super::GetTableSnapshotResponse> =
self.client.get(&path, None::<&[(&str, &str)]>).await?;
- Ok(response.snapshot)
+ Ok(response.and_then(|response| response.snapshot))
}
/// Rename a table.
@@ -1067,22 +1067,18 @@ impl RESTApi {
) -> Result<()> {
let database = identifier.database();
let table = identifier.object();
- validate_non_empty_multi(&[
- (database, "database name"),
- (table, "table name"),
- (tag_name, "tag name"),
- ])?;
+ validate_resource_path(&[(database, "database name"), (table, "table
name")])?;
+ validate_non_empty(tag_name, "tag name")?;
let path = self.resource_paths.tags(database, table);
let mut request = CreateTagRequest::new(tag_name.to_string(),
snapshot_id);
request.time_retained = time_retained.map(str::to_string);
- let _response: serde_json::Value = self.client.post(&path,
&request).await?;
- Ok(())
+ self.client.post_without_response(&path, &request).await
}
pub async fn get_tag(&self, identifier: &Identifier, tag_name: &str) ->
Result<GetTagResponse> {
let database = identifier.database();
let table = identifier.object();
- validate_non_empty_multi(&[
+ validate_resource_path(&[
(database, "database name"),
(table, "table name"),
(tag_name, "tag name"),
@@ -1091,18 +1087,50 @@ impl RESTApi {
self.client.get(&path, None::<&[(&str, &str)]>).await
}
+ /// Fetch one tag-name page, preserving server ordering and opaque tokens.
+ pub async fn list_tags_paged(
+ &self,
+ identifier: &Identifier,
+ max_results: Option<i32>,
+ page_token: Option<&str>,
+ tag_name_prefix: Option<&str>,
+ ) -> Result<PagedList<String>> {
+ validate_resource_path(&[
+ (identifier.database(), "database name"),
+ (identifier.object(), "table name"),
+ ])?;
+ let mut params = Vec::new();
+ if let Some(max_results) = max_results.filter(|value| *value > 0) {
+ params.push((Self::MAX_RESULTS, max_results.to_string()));
+ }
+ if let Some(page_token) = page_token {
+ params.push((Self::PAGE_TOKEN, page_token.to_string()));
+ }
+ if let Some(prefix) = tag_name_prefix.filter(|value|
!value.is_empty()) {
+ params.push(("tagNamePrefix", prefix.to_string()));
+ }
+ let path = self
+ .resource_paths
+ .tags(identifier.database(), identifier.object());
+ let response: ListTagsResponse = self.client.get(&path,
Some(¶ms)).await?;
+ // Java ends pagination when the response has no tags field, including
+ // explicit null, even if a server also returned a continuation token.
+ Ok(match response.tags {
+ Some(tags) => PagedList::new(tags, response.next_page_token),
+ None => PagedList::new(Vec::new(), None),
+ })
+ }
+
pub async fn delete_tag(&self, identifier: &Identifier, tag_name: &str) ->
Result<()> {
let database = identifier.database();
let table = identifier.object();
- validate_non_empty_multi(&[
+ validate_resource_path(&[
(database, "database name"),
(table, "table name"),
(tag_name, "tag name"),
])?;
let path = self.resource_paths.tag(database, table, tag_name);
- let _response: serde_json::Value =
- self.client.delete(&path, None::<&[(&str, &str)]>).await?;
- Ok(())
+ self.client.delete_without_response(&path).await
}
// ==================== Policy Management ====================
@@ -1196,43 +1224,30 @@ impl RESTApi {
.unwrap_or(false))
}
- /// Rollback a table to a specific snapshot.
- pub async fn rollback_to_snapshot(
+ /// Roll back a table through the server, optionally comparing its identity
+ /// and current head. Omitted guards stay omitted on the wire, as in Java.
+ pub async fn rollback_to(
&self,
identifier: &Identifier,
- snapshot_id: i64,
+ instant: &crate::table::Instant,
+ from_snapshot: Option<i64>,
+ table_uuid: Option<&str>,
) -> Result<()> {
- let database = identifier.database();
- let table = identifier.object();
- validate_non_empty_multi(&[(database, "database name"), (table, "table
name")])?;
- let path = self.resource_paths.rollback(database, table);
- let request = serde_json::json!({
- "instant": {
- "type": "snapshot",
- "snapshotId": snapshot_id,
- }
- });
- let _resp: serde_json::Value = self.client.post(&path,
&request).await?;
- Ok(())
- }
-
- /// Rollback a table to a specific tag.
- pub async fn rollback_to_tag(&self, identifier: &Identifier, tag_name:
&str) -> Result<()> {
- let database = identifier.database();
- let table = identifier.object();
- validate_non_empty_multi(&[
- (database, "database name"),
- (table, "table name"),
- (tag_name, "tag name"),
+ validate_resource_path(&[
+ (identifier.database(), "database name"),
+ (identifier.object(), "table name"),
])?;
- let path = self.resource_paths.rollback(database, table);
- let request = serde_json::json!({
- "instant": {
- "type": "tag",
- "tagName": tag_name,
- }
- });
- let _resp: serde_json::Value = self.client.post(&path,
&request).await?;
- Ok(())
+ if let crate::table::Instant::Tag { tag_name } = instant {
+ validate_non_empty(tag_name, "tag name")?;
+ }
+ let request = RollbackTableRequest {
+ instant: instant.clone(),
+ from_snapshot,
+ table_id: table_uuid.map(str::to_string),
+ };
+ let path = self
+ .resource_paths
+ .rollback(identifier.database(), identifier.object());
+ self.client.post_without_response(&path, &request).await
}
}
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index 12e73a84..3c5923b1 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -570,6 +570,42 @@ pub trait Catalog: Send + Sync {
})
}
+ /// Load the catalog's current snapshot and statistics, or None for an
empty table.
+ async fn load_snapshot(
+ &self,
+ _identifier: &Identifier,
+ ) -> Result<Option<crate::api::TableSnapshot>> {
+ Err(Error::Unsupported {
+ message: "catalog snapshot loading is not supported by this
catalog".to_string(),
+ })
+ }
+
+ /// Roll back through the catalog, forwarding optional expected head and
table identity.
+ async fn rollback_to(
+ &self,
+ _identifier: &Identifier,
+ _instant: &crate::table::Instant,
+ _from_snapshot: Option<i64>,
+ _table_uuid: Option<&str>,
+ ) -> Result<()> {
+ Err(Error::Unsupported {
+ message: "catalog rollback is not supported by this
catalog".to_string(),
+ })
+ }
+
+ /// Fetch a page of tag names without scanning tag files on storage.
+ async fn list_tags_paged(
+ &self,
+ _identifier: &Identifier,
+ _max_results: Option<i32>,
+ _page_token: Option<&str>,
+ _tag_name_prefix: Option<&str>,
+ ) -> Result<PagedList<String>> {
+ Err(Error::Unsupported {
+ message: "tag pagination is not supported by this
catalog".to_string(),
+ })
+ }
+
// ======================= branch methods =============================
/// Create an empty branch, or create one from a tag when `from_tag` is
set.
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 3f4f623d..93875d28 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -408,6 +408,44 @@ impl Catalog for RESTCatalog {
})
}
+ async fn load_snapshot(
+ &self,
+ identifier: &Identifier,
+ ) -> Result<Option<crate::api::TableSnapshot>> {
+ match self.api.load_snapshot(identifier).await {
+ Err(Error::RestApi {
+ source: RestError::NoSuchResource { resource_type, .. },
+ }) if resource_type.as_deref() == Some("SNAPSHOT") => Ok(None),
+ result => result.map_err(|error| map_table_not_exist(error,
identifier)),
+ }
+ }
+
+ async fn rollback_to(
+ &self,
+ identifier: &Identifier,
+ instant: &crate::table::Instant,
+ from_snapshot: Option<i64>,
+ table_uuid: Option<&str>,
+ ) -> Result<()> {
+ self.api
+ .rollback_to(identifier, instant, from_snapshot, table_uuid)
+ .await
+ .map_err(|error| map_rest_error_for_rollback(error, identifier,
instant))
+ }
+
+ async fn list_tags_paged(
+ &self,
+ identifier: &Identifier,
+ max_results: Option<i32>,
+ page_token: Option<&str>,
+ tag_name_prefix: Option<&str>,
+ ) -> Result<PagedList<String>> {
+ self.api
+ .list_tags_paged(identifier, max_results, page_token,
tag_name_prefix)
+ .await
+ .map_err(|error| map_table_not_exist(error, identifier))
+ }
+
async fn create_tag(
&self,
identifier: &Identifier,
@@ -874,6 +912,60 @@ fn map_rest_error_for_table(err: Error, identifier:
&Identifier) -> Error {
}
}
+fn map_table_not_exist(error: Error, identifier: &Identifier) -> Error {
+ match error {
+ Error::RestApi {
+ source: RestError::NoSuchResource { .. },
+ } => Error::TableNotExist {
+ full_name: identifier.full_name(),
+ },
+ other => other,
+ }
+}
+
+fn map_rest_error_for_rollback(
+ error: Error,
+ identifier: &Identifier,
+ instant: &crate::table::Instant,
+) -> Error {
+ if let Error::RestApi {
+ source:
+ RestError::NoSuchResource {
+ resource_type,
+ resource_name,
+ ..
+ },
+ } = &error
+ {
+ if resource_type.as_deref() == Some("SNAPSHOT") {
+ let snapshot_id = resource_name
+ .as_deref()
+ .and_then(|name| name.parse().ok())
+ .or(match instant {
+ crate::table::Instant::Snapshot { snapshot_id } =>
Some(*snapshot_id),
+ _ => None,
+ });
+ return snapshot_id
+ .map(|snapshot_id| Error::SnapshotNotExist { snapshot_id })
+ .unwrap_or(error);
+ }
+ if resource_type.as_deref() == Some("TAG") {
+ let tag_name = resource_name
+ .clone()
+ .or_else(|| match instant {
+ crate::table::Instant::Tag { tag_name } =>
Some(tag_name.clone()),
+ _ => None,
+ })
+ .unwrap_or_default();
+ return Error::TagNotExist { tag_name };
+ }
+ return Error::TableNotExist {
+ full_name: identifier.full_name(),
+ };
+ }
+ error
+}
+
fn map_rest_error_for_tag(
error: Error,
identifier: &Identifier,
@@ -891,32 +983,29 @@ fn map_rest_error_for_tag(
RestError::NoSuchResource {
resource_type,
resource_name,
- ..
+ message,
},
- } if resource_type
- .as_deref()
- .is_some_and(|kind| kind.eq_ignore_ascii_case("snapshot")) =>
- {
- match snapshot_id.or_else(|| resource_name.and_then(|value|
value.parse().ok())) {
+ } if resource_type.as_deref() == Some("SNAPSHOT") => {
+ match snapshot_id.or_else(|| {
+ resource_name
+ .as_deref()
+ .and_then(|value| value.parse().ok())
+ }) {
Some(snapshot_id) => Error::SnapshotNotExist { snapshot_id },
- None => Error::DataInvalid {
- message: format!(
- "Cannot create tag '{tag_name}' because the latest
snapshot does not exist"
- ),
- source: None,
+ None => Error::RestApi {
+ source: RestError::NoSuchResource {
+ resource_type,
+ resource_name,
+ message,
+ },
},
}
}
Error::RestApi {
source: RestError::NoSuchResource { resource_type, .. },
- } if resource_type
- .as_deref()
- .is_some_and(|kind| kind.eq_ignore_ascii_case("tag")) =>
- {
- Error::TagNotExist {
- tag_name: tag_name.to_string(),
- }
- }
+ } if resource_type.as_deref() == Some("TAG") => Error::TagNotExist {
+ tag_name: tag_name.to_string(),
+ },
Error::RestApi {
source: RestError::NoSuchResource { .. },
} => Error::TableNotExist {
@@ -928,7 +1017,9 @@ fn map_rest_error_for_tag(
message,
source: None,
},
- other => map_unsupported_endpoint(other, "tag"),
+ // Preserve HTTP status and resource metadata for catalog clients. A
+ // remote 501 is not a locally unsupported operation eligible for
fallback.
+ other => other,
}
}
@@ -1144,16 +1235,58 @@ mod tests {
};
assert!(matches!(
- map_rest_error_for_tag(error("tag"), &identifier, "release", None),
+ map_rest_error_for_tag(error("TAG"), &identifier, "release", None),
Error::TagNotExist { .. }
));
assert!(matches!(
- map_rest_error_for_tag(error("snapshot"), &identifier, "release",
Some(7)),
+ map_rest_error_for_tag(error("SNAPSHOT"), &identifier, "release",
Some(7)),
Error::SnapshotNotExist { snapshot_id: 7 }
));
+ // Java compares resource types exactly. Other 404 resources describe
+ // a missing table, including noncanonical lowercase names.
+ for resource_type in ["TABLE", "OTHER", "tag", "snapshot"] {
+ assert!(matches!(
+ map_rest_error_for_tag(error(resource_type), &identifier,
"release", None),
+ Error::TableNotExist { full_name } if full_name == "db.table"
+ ));
+ }
+ }
+
+ #[test]
+ fn test_tag_error_mapping_preserves_missing_latest_snapshot() {
+ let identifier = Identifier::new("db", "table");
+ for name in [None, Some("null"), Some("7")] {
+ let error = Error::RestApi {
+ source: RestError::NoSuchResource {
+ resource_type: Some("SNAPSHOT".to_string()),
+ resource_name: name.map(str::to_string),
+ message: "missing latest snapshot".to_string(),
+ },
+ };
+ let mapped = map_rest_error_for_tag(error, &identifier, "release",
None);
+ if name == Some("7") {
+ assert!(matches!(mapped, Error::SnapshotNotExist {
snapshot_id: 7 }));
+ } else {
+ assert!(matches!(mapped, Error::RestApi {
+ source: RestError::NoSuchResource { resource_type,
resource_name, message }
+ } if resource_type.as_deref() == Some("SNAPSHOT")
+ && resource_name.as_deref() == name
+ && message == "missing latest snapshot"));
+ }
+ }
+ }
+
+ #[test]
+ fn test_tag_error_mapping_preserves_unimplemented_endpoint() {
+ let error = Error::RestApi {
+ source: RestError::NotImplemented {
+ message: "tag endpoint unavailable".to_string(),
+ },
+ };
assert!(matches!(
- map_rest_error_for_tag(error("table"), &identifier, "release",
None),
- Error::TableNotExist { .. }
+ map_rest_error_for_tag(error, &Identifier::new("db", "table"),
"release", None),
+ Error::RestApi { source: RestError::NotImplemented { message } }
+ if message == "tag endpoint unavailable"
));
}
diff --git a/crates/paimon/src/spec/snapshot.rs
b/crates/paimon/src/spec/snapshot.rs
index 1af888b0..ffb67462 100644
--- a/crates/paimon/src/spec/snapshot.rs
+++ b/crates/paimon/src/spec/snapshot.rs
@@ -62,8 +62,16 @@ pub struct Snapshot {
schema_id: i64,
/// a manifest list recording all changes from the previous snapshots
base_manifest_list: String,
+ /// byte size of the base manifest list, absent in older snapshots
+ #[builder(default = None)]
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ base_manifest_list_size: Option<i64>,
/// a manifest list recording all new changes occurred in this snapshot
delta_manifest_list: String,
+ /// byte size of the delta manifest list, absent in older snapshots
+ #[builder(default = None)]
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ delta_manifest_list_size: Option<i64>,
/// a manifest list recording all changelog produced in this snapshot
#[builder(default = None)]
#[serde(skip_serializing_if = "Option::is_none")]
@@ -154,12 +162,20 @@ impl Snapshot {
&self.base_manifest_list
}
+ pub fn base_manifest_list_size(&self) -> Option<i64> {
+ self.base_manifest_list_size
+ }
+
/// Get the delta manifest list of this snapshot.
#[inline]
pub fn delta_manifest_list(&self) -> &str {
&self.delta_manifest_list
}
+ pub fn delta_manifest_list_size(&self) -> Option<i64> {
+ self.delta_manifest_list_size
+ }
+
/// Get the changelog manifest list of this snapshot.
#[inline]
pub fn changelog_manifest_list(&self) -> Option<&str> {
diff --git a/bindings/python/src/lib.rs b/crates/paimon/src/table/instant.rs
similarity index 51%
copy from bindings/python/src/lib.rs
copy to crates/paimon/src/table/instant.rs
index 0b4508c8..c6e5f0ac 100644
--- a/bindings/python/src/lib.rs
+++ b/crates/paimon/src/table/instant.rs
@@ -15,42 +15,34 @@
// specific language governing permissions and limitations
// under the License.
-use pyo3::prelude::*;
+//! Table rollback targets, matching Java's snapshot and tag instants.
-mod blob;
-mod blob_uri_reader;
-mod commit_callback;
-mod context;
-mod error;
-mod evolution;
-mod full_text_search;
-mod global_index_result;
-mod hybrid_search;
-mod index_build;
-mod merge_into;
-mod oss_cpp;
-mod predicate;
-mod read;
-mod schema;
-mod shard_update;
-mod system_table;
-mod table;
-mod udf;
-mod update_assignment;
-mod variant;
-mod vector_search;
-mod vindex;
-mod write;
-// ---- #285: observability ----
-mod partition;
-mod snapshot;
-mod tag;
+use serde::{Deserialize, Serialize};
-#[pymodule]
-fn pypaimon_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
- error::register_module(py, m)?;
- context::register_module(py, m)?;
- variant::register_module(py, m)?;
- evolution::register_module(m)?;
- Ok(())
+/// A snapshot or tag to which a catalog rolls back a table.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum Instant {
+ #[serde(rename = "snapshot")]
+ Snapshot {
+ #[serde(rename = "snapshotId")]
+ snapshot_id: i64,
+ },
+ #[serde(rename = "tag")]
+ Tag {
+ #[serde(rename = "tagName")]
+ tag_name: String,
+ },
+}
+
+impl Instant {
+ pub fn snapshot(snapshot_id: i64) -> Self {
+ Self::Snapshot { snapshot_id }
+ }
+
+ pub fn tag(tag_name: impl Into<String>) -> Self {
+ Self::Tag {
+ tag_name: tag_name.into(),
+ }
+ }
}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 08a84743..33382b11 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -86,6 +86,7 @@ mod hybrid_search_builder;
mod incremental_scan;
pub(crate) mod index_file_path;
mod inline_blob;
+mod instant;
mod kv_file_reader;
mod kv_file_writer;
mod lumina_index_build_builder;
@@ -216,6 +217,7 @@ pub use hybrid_search_builder::{
pub use incremental_scan::{
IncrementalPlan, IncrementalScan, IncrementalScanMode, IncrementalSplit,
};
+pub use instant::Instant;
pub use object_table::{ObjectEntry, ObjectTable};
pub use partition_row_count::PartitionRowCount;
pub use partition_stat::PartitionStat;
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index eeb2ad47..e70828c8 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -1055,11 +1055,11 @@ impl RESTServer {
let key = format!("{db}.{table}.{}", request.tag_name);
if state.tags.contains_key(&key) {
- return resource_error(StatusCode::CONFLICT, "tag",
&request.tag_name);
+ return resource_error(StatusCode::CONFLICT, "TAG",
&request.tag_name);
}
let snapshot_id = request.snapshot_id.unwrap_or(1);
if snapshot_id != 1 {
- return resource_error(StatusCode::NOT_FOUND, "snapshot",
&snapshot_id.to_string());
+ return resource_error(StatusCode::NOT_FOUND, "SNAPSHOT",
&snapshot_id.to_string());
}
state.tags.insert(
key,
@@ -1083,7 +1083,7 @@ impl RESTServer {
}
match state.tags.get(&format!("{db}.{table}.{tag}")) {
Some(response) => (StatusCode::OK,
Json(response.clone())).into_response(),
- None => resource_error(StatusCode::NOT_FOUND, "tag", &tag),
+ None => resource_error(StatusCode::NOT_FOUND, "TAG", &tag),
}
}
@@ -1098,7 +1098,7 @@ impl RESTServer {
if state.tags.remove(&format!("{db}.{table}.{tag}")).is_some() {
(StatusCode::OK, Json(json!(""))).into_response()
} else {
- resource_error(StatusCode::NOT_FOUND, "tag", &tag)
+ resource_error(StatusCode::NOT_FOUND, "TAG", &tag)
}
}
diff --git a/crates/paimon/tests/rest_branch_test.rs
b/crates/paimon/tests/rest_branch_test.rs
index 1b5a4f00..7dbfde17 100644
--- a/crates/paimon/tests/rest_branch_test.rs
+++ b/crates/paimon/tests/rest_branch_test.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! Java REST branch protocol, exercised through the public Catalog contract.
+//! Java REST branch and version protocols, exercised through the public
Catalog contract.
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
@@ -31,6 +31,360 @@ use paimon::common::Options;
use paimon::Error;
use serde_json::{json, Value};
+fn version_snapshot(id: i64) -> Value {
+ json!({
+ "version": 3, "id": id, "schemaId": 0,
+ "baseManifestList": "base", "deltaManifestList": "delta",
+ "baseManifestListSize": 321, "deltaManifestListSize": 0,
+ "commitUser": "writer", "commitIdentifier": 1, "commitKind": "APPEND",
+ "timeMillis": 123, "uuid": "snapshot-uuid", "totalRecordCount": 8,
+ })
+}
+
+#[tokio::test]
+async fn rollback_instants_and_optional_guards_match_java() {
+ use paimon::table::Instant;
+ for instant in [Instant::snapshot(7), Instant::tag("release /一")] {
+ for from_snapshot in [None, Some(0), Some(9)] {
+ for table_uuid in [None, Some("table-incarnation")] {
+ let ctx = TestCatalog::new_raw(vec![(StatusCode::OK,
"OK".to_string())]).await;
+ let id = Identifier::new("db /一", "table$branch_dev /一");
+ ctx.catalog
+ .rollback_to(&id, &instant, from_snapshot, table_uuid)
+ .await
+ .unwrap();
+ let requests = ctx.requests();
+ assert_eq!(requests.len(), 1);
+ assert_eq!(requests[0].method, Method::POST);
+ assert!(requests[0]
+ .uri
+
.ends_with("/tables/table%24branch_dev+%2F%E4%B8%80/rollback"));
+ let mut expected = json!({"instant": instant});
+ if let Some(from_snapshot) = from_snapshot {
+ expected["fromSnapshot"] = json!(from_snapshot);
+ }
+ if let Some(uuid) = table_uuid {
+ expected["tableId"] = json!(uuid);
+ }
+ assert_eq!(requests[0].body, Some(expected));
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn rejected_rollback_preserves_guards_without_reload_or_retry() {
+ use paimon::table::Instant;
+ for status in [400, 403, 409, 429, 500, 501, 503] {
+ let ctx = TestCatalog::new(vec![(
+ StatusCode::from_u16(status).unwrap(),
+ json!({"message": "rejected"}),
+ )])
+ .await;
+ let id = Identifier::new("db", "t");
+ let error = ctx
+ .catalog
+ .rollback_to(&id, &Instant::snapshot(1), Some(2),
Some("old-table"))
+ .await
+ .unwrap_err();
+ assert!(matches!(error, Error::RestApi { .. }), "{error:?}");
+ let requests = ctx.requests();
+ assert_eq!(requests.len(), 1);
+ assert_eq!(
+ requests[0].body,
+ Some(json!({
+ "instant": {"type":"snapshot", "snapshotId":1},
+ "fromSnapshot":2, "tableId":"old-table",
+ }))
+ );
+ }
+}
+
+#[tokio::test]
+async fn rollback_missing_targets_remain_distinct_from_missing_tables() {
+ use paimon::table::Instant;
+ for resource in ["SNAPSHOT", "TAG", "TABLE", "OTHER"] {
+ let name = if resource == "SNAPSHOT" {
+ "7"
+ } else {
+ "release"
+ };
+ let ctx = TestCatalog::new(vec![(
+ StatusCode::NOT_FOUND,
+ json!({
+ "resourceType":resource, "resourceName":name,
"message":"missing",
+ }),
+ )])
+ .await;
+ let error = ctx
+ .catalog
+ .rollback_to(
+ &Identifier::new("db", "t"),
+ &Instant::tag("release"),
+ None,
+ None,
+ )
+ .await
+ .unwrap_err();
+ match resource {
+ "SNAPSHOT" => assert!(matches!(error, Error::SnapshotNotExist {
snapshot_id: 7 })),
+ "TAG" => {
+ assert!(matches!(error, Error::TagNotExist { tag_name } if
tag_name == "release"))
+ }
+ _ => {
+ assert!(matches!(error, Error::TableNotExist { full_name } if
full_name == "db.t"))
+ }
+ }
+ assert_eq!(ctx.requests().len(), 1);
+ }
+}
+
+#[tokio::test]
+async fn missing_latest_snapshot_is_not_an_invalid_tag_request() {
+ for name in [Some("null"), None] {
+ let ctx = TestCatalog::new(vec![(
+ StatusCode::NOT_FOUND,
+ json!({"resourceType":"SNAPSHOT", "resourceName":name,
"message":"empty table"}),
+ )])
+ .await;
+ let error = ctx
+ .catalog
+ .create_tag(&Identifier::new("db", "t"), "release", None, false)
+ .await
+ .unwrap_err();
+ assert!(matches!(error, Error::RestApi {
+ source: RestError::NoSuchResource { resource_type, resource_name,
.. }
+ } if resource_type.as_deref() == Some("SNAPSHOT") &&
resource_name.as_deref() == name));
+ assert_eq!(ctx.requests().len(), 1);
+ }
+}
+
+#[tokio::test]
+async fn tag_pages_preserve_query_encoding_order_duplicates_and_tokens() {
+ let token = "next +/%一";
+ let prefix = "release +/%一";
+ let ctx = TestCatalog::new(vec![(
+ StatusCode::OK,
+ json!({
+ "tags":["z","a","z","一"], "nextPageToken":token,
+ }),
+ )])
+ .await;
+ let page = ctx
+ .catalog
+ .list_tags_paged(
+ &Identifier::new("db", "t$branch_dev /一"),
+ Some(3),
+ Some(token),
+ Some(prefix),
+ )
+ .await
+ .unwrap();
+ assert_eq!(page.elements, vec!["z", "a", "z", "一"]);
+ assert_eq!(page.next_page_token.as_deref(), Some(token));
+ let requests = ctx.requests();
+ assert_eq!(requests.len(), 1);
+ let url = reqwest::Url::parse(&format!("http://server{}";,
requests[0].uri)).unwrap();
+ let query: std::collections::HashMap<_, _> =
url.query_pairs().into_owned().collect();
+ assert_eq!(
+ query,
+ std::collections::HashMap::from([
+ ("maxResults".to_string(), "3".to_string()),
+ ("pageToken".to_string(), token.to_string()),
+ ("tagNamePrefix".to_string(), prefix.to_string()),
+ ])
+ );
+ assert!(url.path().ends_with("t%24branch_dev+%2F%E4%B8%80/tags"));
+}
+
+#[tokio::test]
+async fn tag_page_nulls_and_empty_parameters_match_java() {
+ for (body, expected_token) in [
+ (json!({"tags":null,"nextPageToken":"discard"}), None),
+ (json!({"nextPageToken":"discard"}), None),
+ (json!({"tags":[],"nextPageToken":"keep"}), Some("keep")),
+ ] {
+ for max in [None, Some(0), Some(-1)] {
+ let ctx = TestCatalog::new(vec![(StatusCode::OK,
body.clone())]).await;
+ let page = ctx
+ .catalog
+ .list_tags_paged(&Identifier::new("db", "t"), max, Some(""),
Some(""))
+ .await
+ .unwrap();
+ assert!(page.elements.is_empty());
+ assert_eq!(page.next_page_token.as_deref(), expected_token);
+ let requests = ctx.requests();
+ assert!(
+ requests[0].uri.ends_with("/tags?pageToken="),
+ "{requests:?}"
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn tag_page_errors_are_table_errors_and_keep_permissions() {
+ for status in [StatusCode::NOT_FOUND, StatusCode::FORBIDDEN] {
+ let ctx = TestCatalog::new(vec![(
+ status,
+ json!({"resourceType":"TABLE","message":"denied"}),
+ )])
+ .await;
+ let error = ctx
+ .catalog
+ .list_tags_paged(&Identifier::new("db", "t"), None, None, None)
+ .await
+ .unwrap_err();
+ if status == StatusCode::NOT_FOUND {
+ assert!(matches!(error, Error::TableNotExist { .. }));
+ } else {
+ assert!(matches!(
+ error,
+ Error::RestApi {
+ source: RestError::Forbidden { .. }
+ }
+ ));
+ }
+ assert_eq!(ctx.requests().len(), 1);
+ }
+}
+
+#[tokio::test]
+async fn catalog_snapshot_keeps_all_statistics_and_snapshot_metadata() {
+ let ctx = TestCatalog::new(vec![(
+ StatusCode::OK,
+ json!({"snapshot":{
+ "snapshot": version_snapshot(7),
"recordCount":0,"fileSizeInBytes":123,
+ "fileCount":2,"lastFileCreationTime":456,
+ }}),
+ )])
+ .await;
+ let snapshot = ctx
+ .catalog
+ .load_snapshot(&Identifier::new("db", "t$branch_dev"))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(snapshot.snapshot.id(), 7);
+ assert_eq!(snapshot.snapshot.uuid(), Some("snapshot-uuid"));
+ assert_eq!(snapshot.snapshot.base_manifest_list_size(), Some(321));
+ assert_eq!(snapshot.snapshot.delta_manifest_list_size(), Some(0));
+ assert_eq!(snapshot.record_count, Some(0));
+ assert_eq!(snapshot.file_size_in_bytes, Some(123));
+ assert_eq!(snapshot.file_count, Some(2));
+ assert_eq!(snapshot.last_file_creation_time, Some(456));
+ assert_eq!(ctx.requests().len(), 1);
+}
+
+#[tokio::test]
+async fn absent_catalog_snapshot_is_empty_but_missing_table_is_not() {
+ for body in [Value::Null, json!({"snapshot":null}), json!({})] {
+ let ctx = TestCatalog::new(vec![(StatusCode::OK, body)]).await;
+ assert!(ctx
+ .catalog
+ .load_snapshot(&Identifier::new("db", "t"))
+ .await
+ .unwrap()
+ .is_none());
+ assert_eq!(ctx.requests().len(), 1);
+ }
+ let ctx = TestCatalog::new(vec![(
+ StatusCode::NOT_FOUND,
+ json!({"resourceType":"SNAPSHOT","message":"empty"}),
+ )])
+ .await;
+ assert!(ctx
+ .catalog
+ .load_snapshot(&Identifier::new("db", "t"))
+ .await
+ .unwrap()
+ .is_none());
+ assert_eq!(ctx.requests().len(), 1);
+ let ctx = TestCatalog::new(vec![(
+ StatusCode::NOT_FOUND,
+ json!({"resourceType":"TABLE","message":"missing"}),
+ )])
+ .await;
+ assert!(matches!(
+ ctx.catalog.load_snapshot(&Identifier::new("db", "t")).await,
+ Err(Error::TableNotExist { .. })
+ ));
+ assert_eq!(ctx.requests().len(), 1);
+}
+
+#[tokio::test]
+async fn
tag_retention_and_metadata_are_opaque_and_mutations_ignore_success_bodies() {
+ let name = "release /一";
+ let ctx = TestCatalog::new_raw(vec![
+ (StatusCode::OK, "created".to_string()),
+ (
+ StatusCode::OK,
+ json!({"tagName":name,"snapshot":version_snapshot(7),
+ "tagCreateTime":123,"tagTimeRetained":"2 d"})
+ .to_string(),
+ ),
+ (StatusCode::NO_CONTENT, String::new()),
+ ])
+ .await;
+ let id = Identifier::new("db", "t");
+ ctx.catalog
+ .create_tag_with_retention(&id, name, Some(7), Some("2 d"), false)
+ .await
+ .unwrap();
+ let response = ctx.catalog.get_tag(&id, name).await.unwrap();
+ assert_eq!(response.snapshot.id(), 7);
+ let serialized = serde_json::to_value(&response.snapshot).unwrap();
+ assert_eq!(serialized["baseManifestListSize"], 321);
+ assert_eq!(serialized["deltaManifestListSize"], 0);
+ assert_eq!(response.tag_create_time, Some(123));
+ assert_eq!(response.tag_time_retained.as_deref(), Some("2 d"));
+ ctx.catalog.delete_tag(&id, name, false).await.unwrap();
+ let requests = ctx.requests();
+ assert_eq!(
+ requests[0].body,
+ Some(json!({"tagName":name,"snapshotId":7,"timeRetained":"2 d"}))
+ );
+ assert!(requests[1].uri.ends_with("/tags/release+%2F%E4%B8%80"));
+ assert_eq!(requests[2].method, Method::DELETE);
+}
+
+#[tokio::test]
+async fn version_paths_reject_parent_resources_before_requests() {
+ use paimon::table::Instant;
+ let ctx = TestCatalog::new(vec![]).await;
+ for id in [Identifier::new("..", "t"), Identifier::new("db", ".")] {
+ assert!(ctx
+ .catalog
+ .rollback_to(&id, &Instant::snapshot(1), None, None)
+ .await
+ .is_err());
+ assert!(ctx.catalog.load_snapshot(&id).await.is_err());
+ assert!(ctx
+ .catalog
+ .list_tags_paged(&id, None, None, None)
+ .await
+ .is_err());
+ assert!(ctx
+ .catalog
+ .create_tag(&id, "release", None, false)
+ .await
+ .is_err());
+ }
+ for name in ["", " ", ".", ".."] {
+ assert!(ctx
+ .catalog
+ .get_tag(&Identifier::new("db", "t"), name)
+ .await
+ .is_err());
+ assert!(ctx
+ .catalog
+ .delete_tag(&Identifier::new("db", "t"), name, false)
+ .await
+ .is_err());
+ }
+ assert!(ctx.requests().is_empty());
+}
+
#[derive(Debug, PartialEq)]
struct Request {
method: Method,
