You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/09/16 04:16:35 UTC
(paimon-rust) branch main updated: feat: expand native planning across resolved tables and scan modes (#847)
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 c92d8736 feat: expand native planning across resolved tables and scan modes (#847)
c92d8736 is described below
commit c92d87363d69a5beb24427be69d4f78210788b68
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Wed Sep 16 12:16:29 2026 +0800
feat: expand native planning across resolved tables and scan modes (#847)
---
bindings/python/README.md | 30 ++++
.../python/python/pypaimon_rust/datafusion.pyi | 21 ++-
bindings/python/src/context.rs | 28 ++-
bindings/python/src/table.rs | 72 +++++++-
bindings/python/tests/test_resolved_table.py | 196 +++++++++++++++++++++
bindings/python/tests/test_table.py | 38 ++++
crates/paimon-rest-server/README.md | 1 +
crates/paimon-rest-server/src/lib.rs | 47 ++++-
crates/paimon-rest-server/tests/e2e.rs | 64 +++++++
crates/paimon/src/api/api_response.rs | 17 ++
crates/paimon/src/api/mod.rs | 7 +-
crates/paimon/src/api/resource_paths.rs | 5 +
crates/paimon/src/api/rest_api.rs | 17 ++
crates/paimon/src/table/mod.rs | 31 +++-
crates/paimon/src/table/rest_env.rs | 21 +++
crates/paimon/src/table/snapshot_manager.rs | 192 +++++++++++++++++++-
crates/paimon/src/table/table_scan.rs | 20 +++
crates/paimon/src/table/time_travel.rs | 44 +++++
crates/paimon/tests/first_row_scan_test.rs | 109 ++++++++++++
crates/paimon/tests/mock_server.rs | 41 +++++
20 files changed, 973 insertions(+), 28 deletions(-)
diff --git a/bindings/python/README.md b/bindings/python/README.md
index 6f173a7d..53f3a8ee 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -107,6 +107,36 @@ print(f"\nRead: {batches_tt[0].num_rows} rows")
print(batches_tt[0])
```
+### Tables resolved outside the Rust catalog
+
+`Table.from_resolved_schema(location, schema_json, *, database="default",
+table="table", branch="main", options=None)` accepts a Java-format TableSchema
+JSON document. It preserves the supplied fields, field IDs and complete table
+options, including removed options, without reloading a catalog schema.
+`options` configures FileIO; `branch` selects the snapshot/schema/tag namespace.
+
+Use `new_read_builder()` without extra options to keep that resolved schema.
+Snapshot selectors in the schema's options still select the requested snapshot.
+Passing options to `new_read_builder(options)` instead uses the normal schema
+and snapshot time-travel resolution.
+
+For names containing dots, use `catalog.get_table(("namespace.database", "table.with.dots"))`
+to preserve the database and table components. Both string and tuple identifiers
+support `$branch_<name>` on the table component and reject system-table suffixes.
+
+For REST tables, first use `PaimonCatalog.get_table()`, then
+`table.copy_with_resolved_schema(schema_json, branch=None)`. This replaces the
+complete fields/options while retaining the table location, identity, FileIO
+provider and REST environment. The optional branch selects its metadata namespace
+without reading a branch schema file. Omit it to retain the original branch.
+Cached time-travel resolution is discarded so the supplied options select the
+snapshot, with the externally resolved fields preserved.
+
+REST tables load the latest snapshot through the catalog, including empty
+results and branch-scoped requests. Permission and service failures (including
+HTTP 501) are propagated as in Java, and the
+FileIO provider continues to refresh catalog credentials after schema replacement.
+
## Setup
Install [uv](https://docs.astral.sh/uv/getting-started/installation/):
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 3eecaf72..c8abd01b 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -16,7 +16,7 @@
# under the License.
from os import PathLike
-from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, TypeAlias, Union
+from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, Tuple, TypeAlias, Union
import pyarrow
@@ -109,6 +109,23 @@ class PartitionStat:
def total_size_bytes(self) -> int: ...
class Table:
+ @staticmethod
+ def from_resolved_schema(
+ location: str, schema_json: str, *, database: str = "default",
+ table: str = "table", branch: str = "main",
+ options: Optional[Dict[str, str]] = None,
+ ) -> "Table":
+ """Preserve a resolved Java-format TableSchema; options configures FileIO.
+
+ No catalog lookup or schema time-travel resolution is performed. Snapshot
+ selection still uses the supplied schema's options. Use a catalog when
+ REST authorization or credential refresh is required.
+ """
+ ...
+ def copy_with_resolved_schema(self, schema_json: str, *, branch: Optional[str] = None) -> "Table":
+ """Replace all fields/options, preserving FileIO, REST context and branch."""
+ ...
+
def identifier(self) -> str: ...
def branch(self) -> str: ...
def location(self) -> str: ...
@@ -170,7 +187,7 @@ class PaimonCatalog:
def __datafusion_catalog_provider__(self, session: Any) -> object: ...
def list_databases(self) -> List[str]: ...
def list_tables(self, database_name: str) -> List[str]: ...
- def get_table(self, identifier: str) -> Table: ...
+ def get_table(self, identifier: Union[str, Tuple[str, str]]) -> Table: ...
class PythonScalarUDF:
def __init__(
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 72899145..9fea4b9f 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -168,22 +168,32 @@ impl PaimonCatalog {
})
}
- /// Get a table handle by `"db.table"` or `"db.table$branch_name"` identifier.
- fn get_table(&self, py: Python<'_>, identifier: &str) -> PyResult<PyTable> {
- let parts: Vec<&str> = identifier.splitn(2, '.').collect();
- if parts.len() != 2 || parts[0].is_empty() || parts[1].is_empty() {
- return Err(PyValueError::new_err(format!(
- "expected identifier in 'db.table' format, got '{identifier}'"
- )));
+ /// Get a table by "db.table" or a (database, table) tuple.
+ /// Tuple components preserve dots; the table can include a branch suffix.
+ fn get_table(&self, py: Python<'_>, identifier: &Bound<'_, PyAny>) -> PyResult<PyTable> {
+ let (database, object_name) = if let Ok(name) = identifier.extract::<String>() {
+ let (database, table) = name.split_once('.').ok_or_else(|| {
+ PyValueError::new_err(format!(
+ "expected identifier in 'db.table' format, got '{name}'"
+ ))
+ })?;
+ (database.to_string(), table.to_string())
+ } else {
+ identifier.extract::<(String, String)>()?
+ };
+ if database.is_empty() || object_name.is_empty() {
+ return Err(PyValueError::new_err(
+ "database and table names must not be empty",
+ ));
}
- let id = Identifier::new(parts[0], parts[1]);
+ let id = Identifier::new(&database, &object_name);
let parsed = id.parsed_object_name().map_err(to_py_err)?;
if parsed.system_table().is_some() {
return Err(PyValueError::new_err(
"get_table() does not support system-table identifiers",
));
}
- let base_id = Identifier::new(parts[0], parsed.table());
+ let base_id = Identifier::new(&database, parsed.table());
let branch = parsed.branch().map(str::to_string);
let catalog = Arc::clone(&self.catalog);
let table = py.detach(|| {
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 4e26f048..50ca8072 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -18,7 +18,11 @@
use std::collections::HashMap;
use std::sync::Arc;
+use paimon::catalog::Identifier;
+use paimon::io::FileIO;
+use paimon::spec::TableSchema;
use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyDict;
@@ -43,6 +47,55 @@ impl PyTable {
#[pymethods]
impl PyTable {
+ /// Construct a filesystem table from a Java-format TableSchema JSON document.
+ ///
+ /// Preserves the caller's resolved schema and complete table options without
+ /// loading catalog metadata or resolving a different schema for time travel.
+ /// `options` configures FileIO, not table reads. `branch` selects the metadata
+ /// namespace. REST authorization and credential refresh require a catalog.
+ #[staticmethod]
+ #[pyo3(signature = (location, schema_json, *, database="default", table="table", branch="main", options=None))]
+ fn from_resolved_schema(
+ location: String,
+ schema_json: &str,
+ database: &str,
+ table: &str,
+ branch: &str,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<Self> {
+ let schema: TableSchema = serde_json::from_str(schema_json)
+ .map_err(|err| PyValueError::new_err(format!("Invalid table schema JSON: {err}")))?;
+ let properties = options
+ .map(crate::read::extract_options)
+ .transpose()?
+ .unwrap_or_default();
+ let file_io = FileIO::from_path(&location)
+ .and_then(|builder| builder.with_props(properties).build())
+ .map_err(to_py_err)?;
+ let table = paimon::table::Table::from_resolved_schema(
+ file_io,
+ Identifier::new(database, table),
+ location,
+ schema,
+ branch,
+ )
+ .map_err(to_py_err)?;
+ Ok(Self::new(Arc::new(table)))
+ }
+
+ /// Replace the complete schema while retaining FileIO, REST credentials and branch.
+ /// The caller has already resolved fields and options; no schema is reloaded.
+ #[pyo3(signature = (schema_json, *, branch=None))]
+ fn copy_with_resolved_schema(&self, schema_json: &str, branch: Option<&str>) -> PyResult<Self> {
+ let schema: TableSchema = serde_json::from_str(schema_json)
+ .map_err(|err| PyValueError::new_err(format!("Invalid table schema JSON: {err}")))?;
+ let table = self
+ .inner
+ .copy_with_resolved_schema(schema, branch.unwrap_or(self.inner.branch()))
+ .map_err(to_py_err)?;
+ Ok(Self::new(Arc::new(table)))
+ }
+
fn identifier(&self) -> String {
let id = self.inner.identifier();
format!("{}.{}", id.database(), id.object())
@@ -65,11 +118,16 @@ impl PyTable {
/// time travel) before building, so filters validate against the resolved
/// schema. Empty/absent options are a zero-cost latest read.
#[pyo3(signature = (options=None))]
- fn new_read_builder(&self, options: Option<&Bound<'_, PyDict>>) -> PyResult<PyReadBuilder> {
+ fn new_read_builder(
+ &self,
+ py: Python<'_>,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<PyReadBuilder> {
match options {
Some(dict) if !dict.is_empty() => {
let opts = crate::read::extract_options(dict)?;
- PyReadBuilder::from_options(Arc::clone(&self.inner), opts)
+ let table = Arc::clone(&self.inner);
+ py.detach(|| PyReadBuilder::from_options(table, opts))
}
_ => Ok(PyReadBuilder::new(Arc::clone(&self.inner))),
}
@@ -81,11 +139,13 @@ impl PyTable {
}
// ---------------- #285: observability ----------------
- fn latest_snapshot(&self) -> PyResult<Option<PySnapshot>> {
+ fn latest_snapshot(&self, py: Python<'_>) -> PyResult<Option<PySnapshot>> {
let sm = self.inner.snapshot_manager();
- let snap = runtime()
- .block_on(sm.get_latest_snapshot())
- .map_err(to_py_err)?;
+ let snap = py.detach(|| {
+ runtime()
+ .block_on(sm.get_latest_snapshot())
+ .map_err(to_py_err)
+ })?;
Ok(snap.map(PySnapshot::new))
}
diff --git a/bindings/python/tests/test_resolved_table.py b/bindings/python/tests/test_resolved_table.py
new file mode 100644
index 00000000..92937112
--- /dev/null
+++ b/bindings/python/tests/test_resolved_table.py
@@ -0,0 +1,196 @@
+# 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.
+
+import copy
+import json
+import shutil
+from pathlib import Path
+
+import pyarrow as pa
+import pytest
+
+from pypaimon_rust.datafusion import PaimonCatalog, SQLContext, Table
+
+
+@pytest.fixture
+def resolved_source(tmp_path):
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": str(tmp_path)})
+ ctx.sql("CREATE SCHEMA paimon.db")
+ ctx.sql("CREATE TABLE paimon.db.t (id INT, name STRING)")
+ ctx.sql("INSERT INTO paimon.db.t VALUES (1, 'a')")
+ ctx.sql("INSERT INTO paimon.db.t VALUES (2, 'b')")
+ table = PaimonCatalog({"warehouse": str(tmp_path)}).get_table("db.t")
+ root = Path(table.location())
+ schema = json.loads((root / "schema" / "schema-0").read_text())
+ return root, schema
+
+
+def _read(table, predicate=None):
+ builder = table.new_read_builder()
+ if predicate is not None:
+ builder = builder.with_filter(predicate)
+ plan = builder.new_scan().plan()
+ batches = builder.new_read().read(plan.splits())
+ return plan.snapshot_id(), pa.Table.from_batches(batches).to_pylist() if batches else []
+
+
+def test_resolved_schema_preserves_field_ids_and_names(resolved_source):
+ root, schema = resolved_source
+ # An external catalog has resolved a rename; on-disk data still uses `name`.
+ # Filters, projection and physical reads must all use the supplied schema.
+ schema["id"] = 1
+ schema["fields"][1]["name"] = "renamed"
+ table = Table.from_resolved_schema(str(root), json.dumps(schema), database="db", table="t")
+ assert table.identifier() == "db.t"
+ assert table.location() == str(root)
+ assert [field.name() for field in table.schema().fields()] == ["id", "renamed"]
+ assert _read(table, {"method": "equal", "field": "renamed", "literals": ["b"]}) == (
+ 2, [{"id": 2, "renamed": "b"}])
+ builder = table.new_read_builder().with_projection(["renamed"])
+ batches = builder.new_read().read(builder.new_scan().plan().splits())
+ assert sorted(pa.Table.from_batches(batches).column("renamed").to_pylist()) == ["a", "b"]
+
+
+def test_resolved_options_replace_persisted_options(resolved_source):
+ root, schema = resolved_source
+ persisted = copy.deepcopy(schema)
+ persisted["options"]["scan.snapshot-id"] = "1"
+ (root / "schema" / "schema-0").write_text(json.dumps(persisted))
+ latest = Table.from_resolved_schema(str(root), json.dumps(schema))
+ assert _read(latest)[0] == 2
+ assert sorted(row["id"] for row in _read(latest)[1]) == [1, 2]
+ schema["options"]["scan.snapshot-id"] = "1"
+ historical = Table.from_resolved_schema(str(root), json.dumps(schema))
+ assert _read(historical) == (1, [{"id": 1, "name": "a"}])
+ assert "scan.snapshot-id" not in latest.schema().options()
+
+
+def test_resolved_branch_snapshot_tag_and_empty_plan(resolved_source):
+ root, schema = resolved_source
+ branch_root = root / "branch" / "branch-dev"
+ (branch_root / "snapshot").mkdir(parents=True)
+ shutil.copy(root / "snapshot" / "snapshot-1", branch_root / "snapshot" / "snapshot-1")
+ (branch_root / "tag").mkdir()
+ shutil.copy(root / "snapshot" / "snapshot-1", branch_root / "tag" / "tag-release")
+ # Construction/planning need no catalog or latest schema file in the branch.
+ table = Table.from_resolved_schema(str(root), json.dumps(schema), branch="dev")
+ assert table.branch() == "dev"
+ assert table.latest_snapshot().id() == 1
+ assert table.new_read_builder().new_scan().plan().snapshot_id() == 1
+ schema["options"]["scan.tag-name"] = "release"
+ tagged = Table.from_resolved_schema(str(root), json.dumps(schema), branch="dev")
+ plan = tagged.new_read_builder().with_filter(
+ {"method": "equal", "field": "id", "literals": [99]}).new_scan().plan()
+ assert plan.snapshot_id() == 1
+ assert plan.splits() == []
+ schema["options"].pop("scan.tag-name")
+ empty = Table.from_resolved_schema(str(root), json.dumps(schema), branch="empty")
+ assert empty.new_read_builder().new_scan().plan().snapshot_id() is None
+
+
+def test_resolved_schema_does_not_load_catalog(tmp_path):
+ schema = {"version": 3, "id": 0, "fields": [{"id": 0, "name": "id", "type": "INT"}],
+ "highestFieldId": 0, "partitionKeys": [], "primaryKeys": [],
+ "options": {}, "timeMillis": 0}
+ table = Table.from_resolved_schema(tmp_path.as_uri(), json.dumps(schema), options={})
+ assert table.new_read_builder().new_scan().plan().snapshot_id() is None
+
+
+@pytest.mark.parametrize("invalid", ["{", "{}", '{"fields": null}'])
+def test_resolved_schema_rejects_invalid_json(tmp_path, invalid):
+ with pytest.raises(ValueError, match="Invalid table schema JSON"):
+ Table.from_resolved_schema(str(tmp_path), invalid)
+
+
+@pytest.mark.parametrize("change", ["duplicate_id", "duplicate_name", "missing_pk"])
+def test_resolved_schema_validates_structure(resolved_source, change):
+ root, schema = resolved_source
+ if change == "duplicate_id":
+ schema["fields"][1]["id"] = schema["fields"][0]["id"]
+ elif change == "duplicate_name":
+ schema["fields"][1]["name"] = schema["fields"][0]["name"]
+ else:
+ schema["primaryKeys"] = ["missing"]
+ with pytest.raises(ValueError):
+ Table.from_resolved_schema(str(root), json.dumps(schema))
+
+
+@pytest.mark.parametrize("kwargs", [{"branch": "../escape"}, {"database": ""}, {"table": "../t"}])
+def test_resolved_schema_validates_metadata_identity(resolved_source, kwargs):
+ root, schema = resolved_source
+ with pytest.raises(ValueError):
+ Table.from_resolved_schema(str(root), json.dumps(schema), **kwargs)
+
+
+def test_resolved_file_io_options_require_strings(resolved_source):
+ root, schema = resolved_source
+ with pytest.raises(TypeError):
+ Table.from_resolved_schema(str(root), json.dumps(schema), options={"key": True})
+
+
+def test_resolved_schema_keeps_query_authorization_guard(resolved_source):
+ root, schema = resolved_source
+ schema["options"]["query-auth.enabled"] = "true"
+ table = Table.from_resolved_schema(str(root), json.dumps(schema))
+ builder = table.new_read_builder()
+ with pytest.raises(NotImplementedError, match="query-auth"):
+ builder.new_scan().plan()
+ with pytest.raises(NotImplementedError, match="query-auth"):
+ builder.new_read().read([])
+
+
+def test_catalog_schema_copy_replaces_options_and_keeps_branch(resolved_source):
+ root, schema = resolved_source
+ original = Table.from_resolved_schema(str(root), json.dumps(schema))
+ schema["options"]["scan.snapshot-id"] = "1"
+ historical = original.copy_with_resolved_schema(json.dumps(schema))
+ assert _read(historical) == (1, [{"id": 1, "name": "a"}])
+ schema["options"].pop("scan.snapshot-id")
+ schema["fields"][1]["name"] = "renamed"
+ schema["id"] = 1
+ resolved = historical.copy_with_resolved_schema(json.dumps(schema))
+ assert _read(resolved, {"method": "equal", "field": "renamed", "literals": ["b"]}) == (
+ 2, [{"id": 2, "renamed": "b"}])
+ assert _read(historical)[0] == 1
+ branch_root = root / "branch" / "branch-dev"
+ (branch_root / "snapshot").mkdir(parents=True)
+ shutil.copy(root / "snapshot" / "snapshot-1", branch_root / "snapshot" / "snapshot-1")
+ # No branch schema file: the catalog has already provided the complete schema.
+ branch = resolved.copy_with_resolved_schema(json.dumps(schema), branch="dev")
+ assert branch.branch() == "dev"
+ assert branch.new_read_builder().new_scan().plan().snapshot_id() == 1
+ assert branch.copy_with_resolved_schema(json.dumps(schema)).branch() == "dev"
+ assert branch.copy_with_resolved_schema(json.dumps(schema), branch="main").latest_snapshot().id() == 2
+
+
+@pytest.mark.parametrize("schema_json", ["{", "{}"])
+def test_catalog_schema_copy_rejects_invalid_json(resolved_source, schema_json):
+ root, schema = resolved_source
+ table = Table.from_resolved_schema(str(root), json.dumps(schema))
+ with pytest.raises(ValueError, match="Invalid table schema JSON"):
+ table.copy_with_resolved_schema(schema_json)
+
+
+def test_catalog_schema_copy_validates_branch_and_structure(resolved_source):
+ root, schema = resolved_source
+ table = Table.from_resolved_schema(str(root), json.dumps(schema))
+ with pytest.raises(ValueError):
+ table.copy_with_resolved_schema(json.dumps(schema), branch="../escape")
+ schema["fields"][1]["id"] = schema["fields"][0]["id"]
+ with pytest.raises(ValueError):
+ table.copy_with_resolved_schema(json.dumps(schema))
diff --git a/bindings/python/tests/test_table.py b/bindings/python/tests/test_table.py
index aa68dd8b..cd7d11af 100644
--- a/bindings/python/tests/test_table.py
+++ b/bindings/python/tests/test_table.py
@@ -118,3 +118,41 @@ def test_branch_incremental_scan_uses_branch_snapshot_bounds(branch_tables):
"id": [1], "dt": ["blue"]}
with pytest.raises(ValueError, match="out of available range"):
builder.new_incremental_scan(0, 2).plan()
+
+
+@pytest.mark.parametrize("branch", [None, "blue", "empty"])
+def test_tuple_identifier_preserves_dots_and_branch(branch_tables, branch):
+ main, _, _ = branch_tables
+ root = Path(main.location())
+ warehouse = root.parent.parent
+ database = warehouse / "namespace.database.db"
+ root.parent.rename(database)
+ (database / "t").rename(database / "table.with.dots")
+ catalog = PaimonCatalog({"warehouse": str(warehouse)})
+ name = "table.with.dots" + ("$branch_" + branch if branch else "")
+ table = catalog.get_table(("namespace.database", name))
+ assert table.branch() == (branch or "main")
+ assert table.location() == str(database / "table.with.dots")
+ builder = table.new_read_builder()
+ plan = builder.new_scan().plan()
+ assert plan.snapshot_id() == {None: 2, "blue": 1, "empty": None}[branch]
+ if branch != "empty":
+ rows = pa.Table.from_batches(builder.new_read().read(plan.splits()))
+ assert sorted(rows.column("id").to_pylist()) == ([1] if branch else [1, 2])
+
+
+@pytest.mark.parametrize("identifier", [
+ ("", "t"), ("db", ""), ("db", "t$snapshots"), ("db", "t$branch_../escape"),
+ ("db", "t$branch_"), ("db",), ("db", "t", "extra"),
+])
+def test_tuple_identifier_validation(tmp_path, identifier):
+ catalog = PaimonCatalog({"warehouse": str(tmp_path)})
+ with pytest.raises(ValueError):
+ catalog.get_table(identifier)
+
+
+@pytest.mark.parametrize("identifier", [("db", 1), None, 1])
+def test_invalid_identifier_types(tmp_path, identifier):
+ catalog = PaimonCatalog({"warehouse": str(tmp_path)})
+ with pytest.raises(TypeError):
+ catalog.get_table(identifier)
diff --git a/crates/paimon-rest-server/README.md b/crates/paimon-rest-server/README.md
index 2b8a1ea7..d2757a15 100644
--- a/crates/paimon-rest-server/README.md
+++ b/crates/paimon-rest-server/README.md
@@ -68,6 +68,7 @@ Served under the configured prefix (`/v1/...` by default):
| --- | --- | --- |
| GET | `/v1/config` | server config |
| GET / POST | `/databases` | list / create database |
+| GET | `/databases/:db/tables/:table/snapshot` | latest snapshot (including `$branch_<name>`), or null for an empty table |
| GET / POST / DELETE | `/databases/{db}` | get / alter (no-op) / drop database |
| GET / POST | `/databases/{db}/tables` | list / create table |
| GET / POST / DELETE | `/databases/{db}/tables/{table}` | get / alter / drop table |
diff --git a/crates/paimon-rest-server/src/lib.rs b/crates/paimon-rest-server/src/lib.rs
index 403b4220..60aca42d 100644
--- a/crates/paimon-rest-server/src/lib.rs
+++ b/crates/paimon-rest-server/src/lib.rs
@@ -52,8 +52,9 @@ use serde_json::json;
use paimon::api::{
AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse, ConfigResponse, CreateTableRequest,
- ErrorResponse, GetDatabaseResponse, GetTableResponse, ListDatabasesResponse,
- ListPartitionsResponse, ListTablesResponse, RESTUtil, RenameTableRequest, ResourcePaths,
+ ErrorResponse, GetDatabaseResponse, GetTableResponse, GetTableSnapshotResponse,
+ ListDatabasesResponse, ListPartitionsResponse, ListTablesResponse, RESTUtil,
+ RenameTableRequest, ResourcePaths, TableSnapshot,
};
use paimon::catalog::{list_partitions_from_file_system, Catalog, Identifier};
use paimon::common::{CatalogOptions, Options};
@@ -165,6 +166,10 @@ fn build_router(prefix: &str, state: Arc<AppState>) -> Router {
&format!("{base}/databases/:db/tables/:table"),
get(get_table).post(alter_table).delete(drop_table),
)
+ .route(
+ &format!("{base}/databases/:db/tables/:table/snapshot"),
+ get(load_snapshot),
+ )
.route(&format!("{base}/tables/rename"), post(rename_table))
.route(
&format!("{base}/databases/:db/tables/:table/commit"),
@@ -522,6 +527,44 @@ async fn commit(
}
}
+/// Load the latest snapshot from the filesystem catalog, respecting branch suffixes.
+async fn load_snapshot(path: RestPath, Extension(state): Extension<Arc<AppState>>) -> Response {
+ let identifier = Identifier::new(path.get("db"), path.get("table"));
+ let parsed = match identifier.parsed_object_name() {
+ Ok(parsed) if parsed.system_table().is_none() => parsed,
+ Ok(_) => {
+ return error_response(Error::Unsupported {
+ message: "System tables do not expose a table snapshot".to_string(),
+ })
+ }
+ Err(error) => return error_response(error),
+ };
+ let base = Identifier::new(identifier.database(), parsed.table());
+ let table = match state.catalog.get_table(&base).await {
+ Ok(table) => table,
+ Err(error) => return error_response(error),
+ };
+ let manager = table
+ .snapshot_manager()
+ .with_branch(parsed.branch_or_default());
+ match manager.get_latest_snapshot().await {
+ Ok(snapshot) => {
+ let response = GetTableSnapshotResponse {
+ snapshot: snapshot.map(|snapshot| TableSnapshot {
+ record_count: snapshot.total_record_count(),
+ snapshot,
+ // The snapshot alone does not contain these statistics.
+ file_size_in_bytes: None,
+ file_count: None,
+ last_file_creation_time: None,
+ }),
+ };
+ (StatusCode::OK, Json(response)).into_response()
+ }
+ Err(error) => error_response(error),
+ }
+}
+
/// List a table's partitions, computed from the latest snapshot on disk.
///
/// Mirrors `Catalog::list_partitions`: resolve the table, then derive partition
diff --git a/crates/paimon-rest-server/tests/e2e.rs b/crates/paimon-rest-server/tests/e2e.rs
index db27bb3e..1c48a140 100644
--- a/crates/paimon-rest-server/tests/e2e.rs
+++ b/crates/paimon-rest-server/tests/e2e.rs
@@ -525,3 +525,67 @@ async fn altering_the_declared_type_is_rejected() {
.await
.expect("still readable");
}
+
+#[tokio::test]
+async fn test_load_snapshot_empty_latest_and_branch() {
+ use paimon::spec::{CommitKind, Snapshot};
+ let ctx = setup().await;
+ ctx.catalog
+ .create_database("db", false, HashMap::new())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("db", "snapshots");
+ ctx.catalog
+ .create_table(&identifier, append_only_schema(), false)
+ .await
+ .unwrap();
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ let api = table.rest_env().unwrap().api();
+ assert!(api.load_snapshot(&identifier).await.unwrap().is_none());
+ for id in [1, 2] {
+ let snapshot = Snapshot::builder()
+ .version(3)
+ .id(id)
+ .schema_id(0)
+ .base_manifest_list("base".into())
+ .delta_manifest_list("delta".into())
+ .commit_user("test".into())
+ .commit_identifier(id)
+ .commit_kind(CommitKind::APPEND)
+ .time_millis(1000)
+ .total_record_count(Some(id * 3))
+ .build();
+ table
+ .snapshot_manager()
+ .commit_snapshot(&snapshot)
+ .await
+ .unwrap();
+ if id == 1 {
+ table
+ .snapshot_manager()
+ .with_branch("dev")
+ .commit_snapshot(&snapshot)
+ .await
+ .unwrap();
+ }
+ }
+ let snapshot = api.load_snapshot(&identifier).await.unwrap().unwrap();
+ assert_eq!(snapshot.snapshot.id(), 2);
+ assert_eq!(snapshot.record_count, Some(6));
+ assert_eq!(snapshot.file_count, None);
+ let branch = api
+ .load_snapshot(&Identifier::new("db", "snapshots$branch_dev"))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(branch.snapshot.id(), 1);
+ assert!(api
+ .load_snapshot(&Identifier::new("db", "snapshots$branch_empty"))
+ .await
+ .unwrap()
+ .is_none());
+ assert!(api
+ .load_snapshot(&Identifier::new("db", "missing"))
+ .await
+ .is_err());
+}
diff --git a/crates/paimon/src/api/api_response.rs b/crates/paimon/src/api/api_response.rs
index be95d989..5df155c2 100644
--- a/crates/paimon/src/api/api_response.rs
+++ b/crates/paimon/src/api/api_response.rs
@@ -114,6 +114,23 @@ impl AuditRESTResponse {
}
}
+/// Latest snapshot and table statistics returned by a REST catalog.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TableSnapshot {
+ pub snapshot: Snapshot,
+ pub record_count: Option<i64>,
+ pub file_size_in_bytes: Option<i64>,
+ pub file_count: Option<i64>,
+ pub last_file_creation_time: Option<i64>,
+}
+
+/// Response for loading the latest catalog snapshot.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct GetTableSnapshotResponse {
+ pub snapshot: Option<TableSnapshot>,
+}
+
/// Response for getting a table.
#[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 f7c0e078..55a1dd85 100644
--- a/crates/paimon/src/api/mod.rs
+++ b/crates/paimon/src/api/mod.rs
@@ -41,9 +41,10 @@ pub use api_request::{
// Re-export response types
pub use api_response::{
AuditRESTResponse, AuthTableQueryResponse, ConfigResponse, ErrorResponse, GetDatabaseResponse,
- GetFunctionResponse, GetTableResponse, GetTableTokenResponse, GetTagResponse, GetViewResponse,
- ListDatabasesResponse, ListFunctionsResponse, ListPartitionsResponse, ListPermissionsResponse,
- ListPoliciesResponse, ListTablesResponse, ListViewsResponse, PagedList,
+ GetFunctionResponse, GetTableResponse, GetTableSnapshotResponse, GetTableTokenResponse,
+ GetTagResponse, GetViewResponse, ListDatabasesResponse, ListFunctionsResponse,
+ ListPartitionsResponse, ListPermissionsResponse, ListPoliciesResponse, ListTablesResponse,
+ ListViewsResponse, PagedList, TableSnapshot,
};
// Re-export management types
diff --git a/crates/paimon/src/api/resource_paths.rs b/crates/paimon/src/api/resource_paths.rs
index 470b2f9b..62f1bcf4 100644
--- a/crates/paimon/src/api/resource_paths.rs
+++ b/crates/paimon/src/api/resource_paths.rs
@@ -116,6 +116,11 @@ impl ResourcePaths {
)
}
+ /// Get the latest table snapshot endpoint path, including an optional branch suffix.
+ pub fn table_snapshot(&self, database_name: &str, table_name: &str) -> String {
+ format!("{}/snapshot", self.table(database_name, table_name))
+ }
+
/// Get the views endpoint path for a database.
pub fn views(&self, database_name: &str) -> String {
format!(
diff --git a/crates/paimon/src/api/rest_api.rs b/crates/paimon/src/api/rest_api.rs
index 6fe0c8fe..34d76f46 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -393,6 +393,23 @@ impl RESTApi {
self.client.get(&path, None::<&[(&str, &str)]>).await
}
+ /// Load the latest snapshot and statistics from the catalog.
+ pub async fn load_snapshot(
+ &self,
+ identifier: &Identifier,
+ ) -> Result<Option<super::TableSnapshot>> {
+ validate_non_empty_multi(&[
+ (identifier.database(), "database name"),
+ (identifier.object(), "table name"),
+ ])?;
+ let path = self
+ .resource_paths
+ .table_snapshot(identifier.database(), identifier.object());
+ let response: super::GetTableSnapshotResponse =
+ self.client.get(&path, None::<&[(&str, &str)]>).await?;
+ Ok(response.snapshot)
+ }
+
/// Rename a table.
pub async fn rename_table(&self, source: &Identifier, destination: &Identifier) -> Result<()> {
validate_non_empty_multi(&[
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 7d682ced..f46c835f 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -327,7 +327,8 @@ impl Table {
}
pub fn snapshot_manager(&self) -> SnapshotManager {
- let manager = SnapshotManager::new(self.file_io.clone(), self.location.clone());
+ let manager = SnapshotManager::new(self.file_io.clone(), self.location.clone())
+ .with_rest_env(self.rest_env.clone());
if self.is_main_branch() {
manager
} else {
@@ -471,6 +472,34 @@ impl Table {
}
}
+ /// Replace the complete schema with one already resolved by an external caller.
+ ///
+ /// Like `from_resolved_schema`, this preserves field IDs and options exactly
+ /// and does not perform time travel. Unlike that constructor, it retains the
+ /// FileIO provider, REST environment and identity of this table. The branch
+ /// selects its metadata namespace without loading a different schema.
+ /// Any cached time-travel resolution is discarded: subsequent scans resolve
+ /// the supplied options without replacing the supplied fields.
+ pub fn copy_with_resolved_schema(&self, schema: TableSchema, branch: &str) -> Result<Self> {
+ schema.validate_resolved_structure()?;
+ validate_branch_name(branch)?;
+ let schema_manager = SchemaManager::new(self.file_io.clone(), self.location.clone());
+ let schema_manager = if branch == DEFAULT_MAIN_BRANCH {
+ schema_manager
+ } else {
+ schema_manager.with_branch(branch)
+ };
+ Ok(Self {
+ schema,
+ schema_manager,
+ branch: branch.to_string(),
+ branch_reference: self.branch_reference || branch != DEFAULT_MAIN_BRANCH,
+ time_traveled: false,
+ travel_snapshot: None,
+ ..self.clone()
+ })
+ }
+
/// Create a read-only copy pinned to an already resolved snapshot.
///
/// Replaces any selector that originally resolved the snapshot with an
diff --git a/crates/paimon/src/table/rest_env.rs b/crates/paimon/src/table/rest_env.rs
index 4e8a1f7c..e814f966 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -305,6 +305,27 @@ impl RESTEnv {
builder.build()
}
+ /// Load the authoritative latest snapshot, scoped to this table's branch.
+ pub(crate) async fn load_snapshot(
+ &self,
+ branch: &str,
+ ) -> Result<Option<crate::spec::Snapshot>> {
+ let object = self.identifier.parsed_object_name()?.table().to_string();
+ let object = if branch == crate::catalog::DEFAULT_MAIN_BRANCH {
+ object
+ } else {
+ format!("{object}$branch_{branch}")
+ };
+ let identifier = Identifier::new(self.identifier.database(), object);
+ match self.api.load_snapshot(&identifier).await {
+ Ok(snapshot) => Ok(snapshot.map(|snapshot| snapshot.snapshot)),
+ Err(Error::RestApi {
+ source: RestError::NoSuchResource { resource_type, .. },
+ }) if resource_type.as_deref() == Some("SNAPSHOT") => Ok(None),
+ Err(error) => Err(map_rest_error_for_table(error, &identifier)),
+ }
+ }
+
/// Create a `RESTSnapshotCommit` from this environment.
pub fn snapshot_commit(&self) -> Arc<dyn SnapshotCommit> {
Arc::new(RESTSnapshotCommit::new(
diff --git a/crates/paimon/src/table/snapshot_manager.rs b/crates/paimon/src/table/snapshot_manager.rs
index 1de8baa9..dbc76397 100644
--- a/crates/paimon/src/table/snapshot_manager.rs
+++ b/crates/paimon/src/table/snapshot_manager.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! Snapshot manager for reading snapshot metadata using FileIO.
+//! Snapshot manager for reading file and catalog snapshot metadata.
//!
//! Reference:[org.apache.paimon.utils.SnapshotManager](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java).
use crate::catalog::DEFAULT_MAIN_BRANCH;
@@ -29,7 +29,7 @@ const SNAPSHOT_PREFIX: &str = "snapshot-";
const LATEST_HINT: &str = "LATEST";
const EARLIEST_HINT: &str = "EARLIEST";
-/// Manager for snapshot files using unified FileIO.
+/// Manager for snapshot files and REST catalog snapshot resolution.
///
/// Reference: [org.apache.paimon.utils.SnapshotManager](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java).
#[derive(Debug, Clone)]
@@ -37,6 +37,7 @@ pub struct SnapshotManager {
file_io: FileIO,
table_path: String,
branch: String,
+ rest_env: Option<super::RESTEnv>,
}
impl SnapshotManager {
@@ -46,9 +47,15 @@ impl SnapshotManager {
file_io,
table_path,
branch: DEFAULT_MAIN_BRANCH.to_string(),
+ rest_env: None,
}
}
+ pub(crate) fn with_rest_env(mut self, rest_env: Option<super::RESTEnv>) -> Self {
+ self.rest_env = rest_env;
+ self
+ }
+
pub fn file_io(&self) -> &FileIO {
&self.file_io
}
@@ -74,6 +81,7 @@ impl SnapshotManager {
file_io: self.file_io.clone(),
table_path: self.table_path.clone(),
branch: branch.to_string(),
+ rest_env: self.rest_env.clone(),
}
}
@@ -137,11 +145,22 @@ impl SnapshotManager {
/// Get the latest snapshot id.
///
- /// First tries the LATEST hint file. If the hint is valid and no next snapshot
- /// exists, returns it. Otherwise falls back to listing snapshot files.
+ /// REST tables use the catalog snapshot. Otherwise, first tries the LATEST
+ /// hint file. If the hint is valid and no next snapshot exists, returns it.
+ /// Otherwise falls back to listing snapshot files.
///
/// Reference: [HintFileUtils.findLatest](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/utils/HintFileUtils.java)
pub async fn get_latest_snapshot_id(&self) -> crate::Result<Option<i64>> {
+ if self.rest_env.is_some() {
+ return Ok(self
+ .get_latest_snapshot()
+ .await?
+ .map(|snapshot| snapshot.id()));
+ }
+ self.latest_snapshot_id_from_filesystem().await
+ }
+
+ async fn latest_snapshot_id_from_filesystem(&self) -> crate::Result<Option<i64>> {
let hint_path = self.latest_hint_path();
if let Some(hint_id) = self.read_hint(&hint_path).await {
if hint_id > 0 {
@@ -231,7 +250,12 @@ impl SnapshotManager {
/// Get the latest snapshot, or None if no snapshots exist.
pub async fn get_latest_snapshot(&self) -> crate::Result<Option<Snapshot>> {
- let snapshot_id = match self.get_latest_snapshot_id().await? {
+ if let Some(env) = &self.rest_env {
+ // Java REST NotImplementedException (HTTP 501) is a service error,
+ // not SnapshotLoader's UnsupportedOperationException fallback.
+ return env.load_snapshot(&self.branch).await;
+ }
+ let snapshot_id = match self.latest_snapshot_id_from_filesystem().await? {
Some(id) => id,
None => return Ok(None),
};
@@ -514,6 +538,164 @@ mod tests {
.build()
}
+ struct RestFixture {
+ table: crate::table::Table,
+ response: std::sync::Arc<std::sync::Mutex<(u16, serde_json::Value)>>,
+ requests: std::sync::Arc<std::sync::Mutex<Vec<String>>>,
+ server: tokio::task::JoinHandle<()>,
+ }
+
+ impl Drop for RestFixture {
+ fn drop(&mut self) {
+ self.server.abort();
+ }
+ }
+
+ async fn rest_fixture() -> RestFixture {
+ use crate::api::rest_api::RESTApi;
+ use crate::catalog::Identifier;
+ use crate::common::Options;
+ use crate::spec::{DataType, IntType, Schema, TableSchema};
+ use axum::{
+ http::{HeaderMap, StatusCode, Uri},
+ Json, Router,
+ };
+ use std::sync::{Arc, Mutex};
+
+ let response = Arc::new(Mutex::new((
+ 200_u16,
+ serde_json::json!({
+ "snapshot": {"snapshot": test_snapshot(7), "recordCount": 10}
+ }),
+ )));
+ let requests = Arc::new(Mutex::new(Vec::new()));
+ let handler_response = response.clone();
+ let handler_requests = requests.clone();
+ let app = Router::new().fallback(move |uri: Uri, headers: HeaderMap| {
+ let (status, value) = handler_response.lock().unwrap().clone();
+ handler_requests.lock().unwrap().push(uri.to_string());
+ assert_eq!(headers["authorization"], "Bearer test-token");
+ async move { (StatusCode::from_u16(status).unwrap(), Json(value)) }
+ });
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let mut options = Options::new();
+ options.set("uri", format!("http://{}", listener.local_addr().unwrap()));
+ options.set("prefix", "test");
+ options.set("token.provider", "bear");
+ options.set("token", "test-token");
+ let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+ let api = Arc::new(RESTApi::new(options.clone(), false).await.unwrap());
+ let id = Identifier::new("database", "table");
+ let env = crate::table::RESTEnv::new(id.clone(), "uuid".into(), api, options, false, None);
+ let (io, manager) = setup("/rest-table").await;
+ manager.commit_snapshot(&test_snapshot(2)).await.unwrap();
+ let schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .build()
+ .unwrap(),
+ );
+ let table = crate::table::Table::new(io, id, "/rest-table".into(), schema, Some(env));
+ RestFixture {
+ table,
+ response,
+ requests,
+ server,
+ }
+ }
+
+ #[tokio::test]
+ async fn rest_latest_snapshot_and_id_use_catalog_without_snapshot_file() {
+ let fixture = rest_fixture().await;
+ let sm = fixture.table.snapshot_manager();
+ assert_eq!(sm.get_latest_snapshot().await.unwrap().unwrap().id(), 7);
+ assert_eq!(sm.get_latest_snapshot_id().await.unwrap(), Some(7));
+ assert!(sm.get_snapshot(7).await.is_err());
+ assert_eq!(
+ *fixture.requests.lock().unwrap(),
+ vec!["/v1/test/databases/database/tables/table/snapshot"; 2]
+ );
+ }
+
+ #[tokio::test]
+ async fn rest_empty_snapshot_is_authoritative() {
+ let fixture = rest_fixture().await;
+ let sm = fixture.table.snapshot_manager();
+ for body in [serde_json::json!({"snapshot": null}), serde_json::json!({})] {
+ *fixture.response.lock().unwrap() = (200, body);
+ assert!(sm.get_latest_snapshot().await.unwrap().is_none());
+ assert_eq!(sm.get_latest_snapshot_id().await.unwrap(), None);
+ }
+ *fixture.response.lock().unwrap() = (
+ 404,
+ serde_json::json!({
+ "code": 404, "resourceType": "SNAPSHOT", "message": "No snapshot"
+ }),
+ );
+ assert!(sm.get_latest_snapshot().await.unwrap().is_none());
+ }
+
+ #[tokio::test]
+ async fn rest_snapshot_errors_never_fall_back_to_filesystem() {
+ let fixture = rest_fixture().await;
+ let sm = fixture.table.snapshot_manager();
+ for code in [401, 403, 404, 500, 501, 503] {
+ *fixture.response.lock().unwrap() = (
+ code,
+ serde_json::json!({
+ "code": code, "resourceType": "TABLE", "message": "unavailable"
+ }),
+ );
+ assert!(sm.get_latest_snapshot().await.is_err(), "status {code}");
+ assert!(sm.get_latest_snapshot_id().await.is_err(), "status {code}");
+ }
+ *fixture.response.lock().unwrap() = (200, serde_json::json!({"snapshot": {}}));
+ assert!(sm.get_latest_snapshot().await.is_err());
+ }
+
+ #[tokio::test]
+ async fn resolved_schema_copy_preserves_catalog_and_branch() {
+ let fixture = rest_fixture().await;
+ // Changing the resolved schema must not discard the catalog provider.
+ let schema = fixture
+ .table
+ .schema()
+ .copy_with_options(std::collections::HashMap::from([(
+ "source.split.target-size".into(),
+ "1mb".into(),
+ )]));
+ let table = fixture
+ .table
+ .copy_with_resolved_schema(schema, "main")
+ .unwrap();
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(7)
+ );
+ let sm = table.snapshot_manager().with_branch("dev");
+ assert_eq!(sm.get_latest_snapshot_id().await.unwrap(), Some(7));
+ assert_eq!(
+ fixture.requests.lock().unwrap().last().unwrap(),
+ "/v1/test/databases/database/tables/table%24branch_dev/snapshot"
+ );
+ assert_eq!(
+ sm.with_branch("main")
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(7)
+ );
+ assert_eq!(
+ fixture.requests.lock().unwrap().last().unwrap(),
+ "/v1/test/databases/database/tables/table/snapshot"
+ );
+ }
+
fn test_snapshot_with_watermark(id: i64, watermark: Option<i64>) -> Snapshot {
Snapshot::builder()
.version(3)
diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs
index dda84796..a83880c8 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -2260,6 +2260,26 @@ impl<'a> PaimonTableScan<'a> {
result
}
+ } else if pk_comparator.is_some()
+ && (deletion_vectors_enabled
+ || matches!(
+ core_options.merge_engine(),
+ Ok(crate::spec::MergeEngine::FirstRow)
+ ))
+ && data_files.iter().all(|file| {
+ file.level != 0 && file.delete_row_count.is_none_or(|count| count == 0)
+ })
+ {
+ // Java MergeTreeSplitGenerator packs materialized DV/first-row
+ // files by size even across levels. Clustered files need not be
+ // sorted by PK, so marking them as merge-required is unsafe.
+ split_for_batch(data_files, target_split_size, open_file_cost)
+ .into_iter()
+ .map(|files| SplitGroup {
+ files,
+ raw_convertible: true,
+ })
+ .collect()
} else if let Some(ref comparator) = pk_comparator {
// Merge-tree path: keep key-overlapping files in one split and
// mark which splits the sort-merge reader can skip (mirrors
diff --git a/crates/paimon/src/table/time_travel.rs b/crates/paimon/src/table/time_travel.rs
index 8187dce4..f5676bce 100644
--- a/crates/paimon/src/table/time_travel.rs
+++ b/crates/paimon/src/table/time_travel.rs
@@ -286,6 +286,50 @@ mod tests {
(file_io, table_path.to_string())
}
+ #[tokio::test]
+ async fn externally_resolved_schema_replaces_cached_time_travel() {
+ let (io, path) = setup_evolved_table().await;
+ let latest = latest_table(&io, &path);
+ let historical = latest
+ .copy_with_time_travel(options(&[("scan.snapshot-id", "1")]))
+ .await
+ .unwrap();
+ assert_eq!(historical.travel_snapshot().unwrap().id(), 1);
+ assert_eq!(historical.schema().id(), 0);
+ // The external caller deliberately keeps the old fields while removing
+ // the snapshot selector, as Java copyWithoutTimeTravel can do.
+ let resolved = historical
+ .copy_with_resolved_schema(schema_v0(), "main")
+ .unwrap();
+ assert_eq!(
+ super::resolve_snapshot(&resolved)
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 2
+ );
+ assert_eq!(resolved.schema().id(), 0);
+ assert_eq!(historical.travel_snapshot().unwrap().id(), 1);
+ let selected = resolved
+ .copy_with_resolved_schema(
+ schema_v0().copy_with_options(options(&[("scan.snapshot-id", "1")])),
+ "main",
+ )
+ .unwrap();
+ assert_eq!(
+ super::resolve_snapshot(&selected)
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 1
+ );
+ assert!(resolved
+ .copy_with_resolved_schema(schema_v0(), "../invalid")
+ .is_err());
+ }
+
fn latest_table(file_io: &FileIO, table_path: &str) -> Table {
make_table(file_io, table_path, schema_v1())
}
diff --git a/crates/paimon/tests/first_row_scan_test.rs b/crates/paimon/tests/first_row_scan_test.rs
index c9f4a158..7a36ba6f 100644
--- a/crates/paimon/tests/first_row_scan_test.rs
+++ b/crates/paimon/tests/first_row_scan_test.rs
@@ -160,3 +160,112 @@ async fn first_row_incremental_preserves_events_instead_of_merging() {
vec![(1, 10), (1, 99), (2, 20), (3, 30)]
);
}
+
+#[tokio::test]
+async fn materialized_dv_files_use_raw_size_packing_across_levels() {
+ for engine in ["deduplicate", "first-row"] {
+ for merge_on_read in ["false", "true"] {
+ for (target_size, expected_splits) in [("1b", 2), ("1mb", 1)] {
+ let path =
+ format!("memory:/materialized_dv/{engine}/{merge_on_read}/{target_size}");
+ // Model Java-created clustering metadata. Rust's create-time
+ // validation still rejects first-row DVs; resolved table reads
+ // must nevertheless support this Java-valid combination.
+ let schema = pk_schema(&[("merge-engine", engine)]);
+ let options = [
+ ("merge-engine", engine),
+ ("deletion-vectors.enabled", "true"),
+ ("deletion-vectors.merge-on-read", merge_on_read),
+ ("pk-clustering-override", "true"),
+ ("clustering.columns", "value"),
+ ("source.split.target-size", target_size),
+ ("source.split.open-file-cost", "1b"),
+ ]
+ .into_iter()
+ .map(|(key, value)| (key.to_string(), value.to_string()));
+ let schema = schema.copy_with_options(options.collect());
+ let (io, table) = memory_table(&path, schema);
+ setup_dirs(&io, &path).await;
+ persist_table_schema(&io, &path, table.schema()).await;
+ // Materialized files can have overlapping min/max keys across
+ // levels without sharing a live key (e.g. clustered output).
+ for (level, ids, values) in
+ [(1, vec![1, 3], vec![10, 30]), (2, vec![2, 4], vec![20, 40])]
+ {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(ids, values))
+ .await
+ .unwrap();
+ let mut messages = writer.prepare_commit().await.unwrap();
+ for message in &mut messages {
+ for file in &mut message.new_files {
+ file.level = level;
+ }
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert_eq!(plan.splits().len(), expected_splits, "{path}");
+ assert!(
+ plan.splits().iter().all(|split| split.raw_convertible()),
+ "{path}"
+ );
+ assert_eq!(
+ rows(&builder, &plan).await,
+ vec![(1, 10), (2, 20), (3, 30), (4, 40)]
+ );
+
+ // Adding overlapping L0 must still keep every key version in
+ // one merge split, even when the size target is one byte.
+ write_batch(&table, &make_batch(vec![1, 5], vec![99, 50])).await;
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let mut expected = vec![(1, 10), (2, 20), (3, 30), (4, 40)];
+ if merge_on_read == "true" {
+ assert_eq!(plan.splits().len(), 1, "{path}");
+ assert!(!plan.splits()[0].raw_convertible(), "{path}");
+ if engine == "deduplicate" {
+ expected[0].1 = 99;
+ }
+ expected.push((5, 50));
+ }
+ assert_eq!(rows(&builder, &plan).await, expected, "{path}");
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn first_row_dv_merge_on_read_merges_l0_before_value_filtering() {
+ let table = table_with_versions("memory:/first_row/dv_l0", false).await;
+ let table = table.copy_with_options(
+ [
+ ("deletion-vectors.enabled", "true"),
+ ("deletion-vectors.merge-on-read", "true"),
+ ("pk-clustering-override", "true"),
+ ("clustering.columns", "value"),
+ ]
+ .into_iter()
+ .map(|(k, v)| (k.to_string(), v.to_string()))
+ .collect(),
+ );
+ for (value, expected) in [(10, vec![(1, 10)]), (99, vec![])] {
+ let mut builder = table.new_read_builder();
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("value", Datum::Int(value))
+ .unwrap(),
+ );
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert_eq!(plan.snapshot_id(), Some(2));
+ assert_eq!(plan.splits().len(), 1);
+ let split = &plan.splits()[0];
+ assert!(!split.raw_convertible());
+ assert_eq!(split.data_files().len(), 2);
+ assert!(split.data_files().iter().all(|file| file.level == 0));
+ assert_eq!(rows(&builder, &plan).await, expected);
+ }
+}
diff --git a/crates/paimon/tests/mock_server.rs b/crates/paimon/tests/mock_server.rs
index c8a64f9c..ec593d13 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -769,6 +769,43 @@ impl RESTServer {
(StatusCode::OK, Json(serde_json::json!(""))).into_response()
}
+ /// Load the snapshot from the same filesystem fixtures used by table reads.
+ pub async fn load_snapshot(
+ Path((db, table)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let identifier = Identifier::new(&db, &table);
+ let parsed = identifier.parsed_object_name().unwrap();
+ let key = format!("{db}.{}", parsed.table());
+ let response = state.inner.lock().unwrap().tables.get(&key).cloned();
+ let Some(response) = response else {
+ return resource_error(StatusCode::NOT_FOUND, "TABLE", &table);
+ };
+ let location = response.path.unwrap();
+ let file_io = paimon::io::FileIO::from_path(&location)
+ .unwrap()
+ .build()
+ .unwrap();
+ let manager = paimon::table::SnapshotManager::new(file_io, location)
+ .with_branch(parsed.branch_or_default());
+ match manager.get_latest_snapshot().await {
+ Ok(snapshot) => (
+ StatusCode::OK,
+ Json(json!({
+ "snapshot": snapshot.map(|snapshot| json!({"snapshot": snapshot}))
+ })),
+ )
+ .into_response(),
+ Err(error) => (
+ StatusCode::INTERNAL_SERVER_ERROR,
+ Json(json!({
+ "code": 500, "message": error.to_string()
+ })),
+ )
+ .into_response(),
+ }
+ }
+
/// Handle GET /databases/:db/tables/:table - get a specific table.
pub async fn get_table(
Path((db, table)): Path<(String, String)>,
@@ -2097,6 +2134,10 @@ pub async fn start_mock_server(
let app = Router::new()
// Config endpoint (for RESTApi initialization)
.route("/v1/config", get(RESTServer::get_config))
+ .route(
+ &format!("{prefix}/databases/:db/tables/:table/snapshot"),
+ get(RESTServer::load_snapshot),
+ )
// Database routes
.route(
&format!("{prefix}/databases"),