From 9febbe7d4c90b3686fdd5329f1cad6a93b59d838 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 12:08:34 +0800 Subject: [PATCH 1/2] [core] Support REST catalog version management --- .../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 + .../python/tests/test_catalog_versions.py | 172 +++++++++ .../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 ++ .../paimon/src/catalog/rest/rest_catalog.rs | 131 ++++++- crates/paimon/src/spec/snapshot.rs | 16 + crates/paimon/src/table/instant.rs | 48 +++ 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, 1105 insertions(+), 83 deletions(-) create mode 100644 bindings/python/src/catalog_version.rs create mode 100644 bindings/python/tests/test_catalog_versions.py create mode 100644 crates/paimon/src/table/instant.rs diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 44cee5906..8c1bc75ba 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 000000000..ba92f1fe6 --- /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 { + self.inner.tag_create_time + } + fn tag_time_retained(&self) -> Option<&str> { + self.inner.tag_time_retained.as_deref() + } + fn to_json(&self) -> PyResult { + 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 { + self.inner.record_count + } + fn file_size_in_bytes(&self) -> Option { + self.inner.file_size_in_bytes + } + fn file_count(&self) -> Option { + self.inner.file_count + } + fn last_file_creation_time(&self) -> Option { + self.inner.last_file_creation_time + } + fn to_json(&self) -> PyResult { + 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 d72233505..170b3eabd 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, } -fn catalog_table_identifier(identifier: &Bound<'_, PyAny>) -> PyResult { +fn catalog_identifier(identifier: &Bound<'_, PyAny>) -> PyResult { let (database, object_name) = if let Ok(name) = identifier.extract::() { 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) -> PyResult { + 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, + 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 { + 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, + page_token: Option<&str>, + tag_name_prefix: Option<&str>, + ) -> PyResult<(Vec, Option)> { + 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> { + 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, + 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::())?; this.add_class::()?; + this.add_class::()?; + this.add_class::()?; + this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs index b94999cb9..4e9b277a5 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 0b4508c82..72d17d428 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 000000000..477600844 --- /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) + + +@pytest.fixture +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 +/%一"]} + + +@pytest.mark.parametrize("instant,payload", [ + (Instant.snapshot(7), {"type": "snapshot", "snapshotId": 7}), + (Instant.tag("release /一"), {"type": "tag", "tagName": "release /一"}), +]) +@pytest.mark.parametrize("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 + + +@pytest.mark.parametrize("operation", ["create_tag", "delete_tag", "rollback_to"]) +@pytest.mark.parametrize("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 + + +@pytest.mark.parametrize("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 0161536c0..3124a8630 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 b46300c31..132b11bde 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, + #[serde(skip_serializing_if = "Option::is_none")] + pub table_id: Option, +} + /// 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 aa489f622..8863c5976 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, } +/// 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>, + pub next_page_token: Option, +} + /// 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 55527ae01..f21a1e6ab 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 45ec5a861..c7329a3bf 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> { - 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 = 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 { 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, + page_token: Option<&str>, + tag_name_prefix: Option<&str>, + ) -> Result> { + 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, + 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 12e73a849..3c5923b15 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> { + 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, + _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, + _page_token: Option<&str>, + _tag_name_prefix: Option<&str>, + ) -> Result> { + 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 3f4f623d2..e39032c98 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> { + 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, + 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, + page_token: Option<&str>, + tag_name_prefix: Option<&str>, + ) -> Result> { + 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, } } diff --git a/crates/paimon/src/spec/snapshot.rs b/crates/paimon/src/spec/snapshot.rs index 1af888b02..ffb674624 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, /// 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, /// 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 { + 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 { + 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/crates/paimon/src/table/instant.rs b/crates/paimon/src/table/instant.rs new file mode 100644 index 000000000..c6e5f0ac9 --- /dev/null +++ b/crates/paimon/src/table/instant.rs @@ -0,0 +1,48 @@ +// 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. + +//! Table rollback targets, matching Java's snapshot and tag instants. + +use serde::{Deserialize, Serialize}; + +/// 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) -> 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 08a847430..33382b119 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 eeb2ad473..e70828c81 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 1b5a4f005..7dbfde172 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, From c568813d96cdff3ddf8badea212ecd0e6fd776cd Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 13:21:05 +0800 Subject: [PATCH 2/2] [core] Align tag error mapping tests with Java semantics --- .../paimon/src/catalog/rest/rest_catalog.rs | 50 +++++++++++++++++-- 1 file changed, 46 insertions(+), 4 deletions(-) diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs b/crates/paimon/src/catalog/rest/rest_catalog.rs index e39032c98..93875d288 100644 --- a/crates/paimon/src/catalog/rest/rest_catalog.rs +++ b/crates/paimon/src/catalog/rest/rest_catalog.rs @@ -1235,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" )); }