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 e357a660 [python] Construct native tables from resolved REST responses
(#922)
e357a660 is described below
commit e357a6602c36fca5f7d53f1f56644c90f18fdf85
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Sep 24 19:18:15 2026 +0800
[python] Construct native tables from resolved REST responses (#922)
---
bindings/python/README.md | 5 +
.../python/python/pypaimon_rust/datafusion.pyi | 7 ++
bindings/python/src/table.rs | 29 +++++
bindings/python/tests/test_resolved_table.py | 118 +++++++++++++++++++++
crates/paimon-rest-server/src/lib.rs | 1 +
crates/paimon/src/api/api_response.rs | 4 +
crates/paimon/src/table/rest_env.rs | 102 +++++++++++++++---
crates/paimon/tests/mock_server.rs | 4 +
8 files changed, 256 insertions(+), 14 deletions(-)
diff --git a/bindings/python/README.md b/bindings/python/README.md
index b6505d1d..496aecc9 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -197,6 +197,11 @@ 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.
+`Table.from_rest_response(response_json, database=..., table=...,
rest_options=...)`
+reuses the matching REST table response and merged catalog options, skipping
+config/get-table requests while preserving REST snapshots and token refresh.
+Use `copy_with_resolved_schema` to apply branch or dynamic options.
+
## 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 4b0532ef..9500b25a 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -166,6 +166,13 @@ class Table:
REST authorization or credential refresh is required.
"""
...
+ @staticmethod
+ def from_rest_response(
+ response_json: str, *, database: str, table: str, rest_options:
Dict[str, str],
+ ) -> "Table":
+ """Reuse matching REST metadata and merged catalog options."""
+ ...
+
def copy_with_resolved_schema(self, schema_json: str, *, branch:
Optional[str] = None) -> "Table":
"""Replace all fields/options, preserving FileIO, REST context and
branch."""
...
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 47701b5c..799af4a5 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -21,6 +21,7 @@ use std::sync::Arc;
use paimon::catalog::Identifier;
use paimon::io::FileIO;
use paimon::spec::TableSchema;
+use paimon::Options;
use paimon_datafusion::runtime::runtime;
use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
@@ -83,6 +84,34 @@ impl PyTable {
Ok(Self::new(Arc::new(table)))
}
+ /// Reuse the matching REST table response and merged catalog options.
+ /// Skips config/get-table requests, preserving REST snapshots and token
refresh.
+ #[staticmethod]
+ #[pyo3(signature = (response_json, *, database, table, rest_options))]
+ fn from_rest_response(
+ py: Python<'_>,
+ response_json: &str,
+ database: &str,
+ table: &str,
+ rest_options: HashMap<String, String>,
+ ) -> PyResult<Self> {
+ let response: paimon::api::GetTableResponse =
+ serde_json::from_str(response_json).map_err(|err| {
+ PyValueError::new_err(format!("Invalid REST table response
JSON: {err}"))
+ })?;
+ let identifier = Identifier::new(database, table);
+ let table = py
+ .detach(|| {
+ runtime().block_on(paimon::table::Table::from_rest_response(
+ identifier,
+ response,
+ Options::from_map(rest_options),
+ ))
+ })
+ .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))]
diff --git a/bindings/python/tests/test_resolved_table.py
b/bindings/python/tests/test_resolved_table.py
index 92937112..566885c7 100644
--- a/bindings/python/tests/test_resolved_table.py
+++ b/bindings/python/tests/test_resolved_table.py
@@ -194,3 +194,121 @@ def
test_catalog_schema_copy_validates_branch_and_structure(resolved_source):
schema["fields"][1]["id"] = schema["fields"][0]["id"]
with pytest.raises(ValueError):
table.copy_with_resolved_schema(json.dumps(schema))
+
+
[email protected]("external", [False, True])
[email protected]("object_name", ["t", "t$branch_main", "t$branch_dev"])
+def test_resolved_rest_response_keeps_snapshot_and_token_refresh(
+ resolved_source, external, object_name
+):
+ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+ from threading import Thread
+
+ root, schema = resolved_source
+ snapshot = json.loads((root / "snapshot" / "snapshot-1").read_text())
+ requests = []
+ token_requests = []
+
+ class Handler(BaseHTTPRequestHandler):
+ def do_GET(self):
+ requests.append(self.path)
+ if self.path.endswith('/token'):
+ token_requests.append(self.path)
+ # Expire the first token to verify that subsequent FileIO
refreshes it.
+ response = {"token": {}, "expiresAtMillis": (
+ 0 if len(token_requests) == 1 else 4102444800000)}
+ elif self.path.endswith('/snapshot'):
+ # Disk has snapshot 2; REST snapshot 1 must remain
authoritative.
+ response = {"snapshot": {"snapshot": snapshot}}
+ else:
+ self.send_error(500, "Unexpected metadata request")
+ return
+ body = json.dumps(response).encode()
+ self.send_response(200)
+ self.send_header('Content-Type', 'application/json')
+ self.send_header('Content-Length', str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ def log_message(self, *args):
+ pass
+
+ server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+ thread = Thread(target=server.serve_forever, daemon=True)
+ thread.start()
+ try:
+ # PyPaimon and older REST servers do not include `database`.
+ response = {"id": "table-uuid", "name": object_name, "path": str(root),
+ "isExternal": external, "schemaId": schema['id'],
"schema": schema}
+ table = Table.from_rest_response(
+ json.dumps(response),
+ database='db',
+ table=object_name,
+ rest_options={
+ 'uri': 'http://127.0.0.1:%d' % server.server_port,
+ 'warehouse': 'test', 'token.provider': 'bear', 'token':
'test-token',
+ 'data-token.enabled': 'true',
+ },
+ )
+ expected_branch = (
+ object_name.split('$branch_', 1)[-1]
+ if '$branch_' in object_name else 'main'
+ )
+ assert table.branch() == expected_branch
+ if '$branch_' in object_name:
+ with pytest.raises(NotImplementedError, match='Writing to Paimon
branch'):
+ table.new_batch_write_builder().new_write()
+ assert len(token_requests) == (0 if external else 1)
+ assert all(path.endswith('/token') for path in requests)
+ assert _read(table) == (1, [{'id': 1, 'name': 'a'}])
+ assert all(path.endswith(('/token', '/snapshot')) for path in requests)
+ encoded_name = object_name.replace('$', '%24')
+ assert any(path.endswith(
+ f'/databases/db/tables/{encoded_name}/snapshot') for path in
requests)
+ assert len(token_requests) == (0 if external else 2)
+ finally:
+ server.shutdown()
+ server.server_close()
+ thread.join()
+
+
[email protected](("database", "table"), [("db", "wrong"), ("wrong",
"t")])
+def test_resolved_rest_response_rejects_identity_mismatch(resolved_source,
database, table):
+ root, schema = resolved_source
+ response = {"id": "table-uuid", "database": "db", "name": "t", "path":
str(root),
+ "isExternal": True, "schemaId": schema["id"], "schema": schema}
+ # Identity validation must run before REST auth/cache initialization.
+ with pytest.raises(ValueError, match="does not match requested
identifier"):
+ Table.from_rest_response(
+ json.dumps(response), database=database, table=table,
rest_options={})
+
+
[email protected](("change", "message"), [
+ ("duplicate_id", "duplicate field id"),
+ ("missing_primary_key", "primary key"),
+ ("missing_partition_key", "partition fields"),
+])
+def test_resolved_rest_response_validates_schema_structure(resolved_source,
change, message):
+ root, schema = resolved_source
+ if change == "duplicate_id":
+ schema["fields"][1]["id"] = schema["fields"][0]["id"]
+ elif change == "missing_primary_key":
+ schema["primaryKeys"] = ["missing"]
+ else:
+ schema["partitionKeys"] = ["missing"]
+ response = {"id": "table-uuid", "database": "db", "name": "t", "path":
str(root),
+ "isExternal": True, "schemaId": schema["id"], "schema": schema}
+ with pytest.raises(ValueError, match=message):
+ Table.from_rest_response(json.dumps(response), database="db",
table="t", rest_options={
+ "uri": "http://127.0.0.1:1";, "warehouse": "test",
+ "token.provider": "bear", "token": "test-token",
+ })
+
+
[email protected]('response', ['{', '{}'])
+def test_resolved_rest_response_rejects_missing_metadata(response):
+ with pytest.raises(ValueError):
+ Table.from_rest_response(response, database='db', table='t',
rest_options={
+ 'uri': 'http://127.0.0.1:1', 'warehouse': 'test',
+ 'token.provider': 'bear', 'token': 'test-token',
+ })
diff --git a/crates/paimon-rest-server/src/lib.rs
b/crates/paimon-rest-server/src/lib.rs
index 7d2a0959..8cf89f04 100644
--- a/crates/paimon-rest-server/src/lib.rs
+++ b/crates/paimon-rest-server/src/lib.rs
@@ -461,6 +461,7 @@ async fn get_table(path: RestPath, Extension(state):
Extension<Arc<AppState>>) -
// FileSystemCatalog has no UUID concept; the full name is a stable id
// that satisfies the client's RESTEnv requirement.
Some(identifier.full_name()),
+ Some(identifier.database().to_string()),
Some(table),
Some(location),
Some(false),
diff --git a/crates/paimon/src/api/api_response.rs
b/crates/paimon/src/api/api_response.rs
index 5df155c2..943f861c 100644
--- a/crates/paimon/src/api/api_response.rs
+++ b/crates/paimon/src/api/api_response.rs
@@ -140,6 +140,8 @@ pub struct GetTableResponse {
pub audit: AuditRESTResponse,
/// The unique identifier of the table.
pub id: Option<String>,
+ /// The database containing the table.
+ pub database: Option<String>,
/// The name of the table.
pub name: Option<String>,
/// The path to the table.
@@ -242,6 +244,7 @@ impl GetTableResponse {
#[allow(clippy::too_many_arguments)]
pub fn new(
id: Option<String>,
+ database: Option<String>,
name: Option<String>,
path: Option<String>,
is_external: Option<bool>,
@@ -252,6 +255,7 @@ impl GetTableResponse {
Self {
audit,
id,
+ database,
name,
path,
is_external,
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index 6440536f..af810f08 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -20,9 +20,9 @@
use crate::api::rest_api::RESTApi;
use crate::api::rest_error::RestError;
use crate::catalog::{Identifier, RESTTokenFileIO};
-use crate::common::Options;
+use crate::common::{CatalogOptions, Options};
use crate::error::Error;
-use crate::io::cache::LocalCache;
+use crate::io::cache::{create_local_cache_with_namespace, LocalCache};
use crate::io::FileIO;
use crate::spec::{CoreOptions, TableSchema, PATH_OPTION};
use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit};
@@ -30,6 +30,41 @@ use crate::table::{ObjectTable, Table};
use crate::Result;
use std::sync::Arc;
+impl Table {
+ /// Reuse the matching REST response and merged catalog options without
config/get-table requests.
+ /// Preserves REST snapshots, credential refresh and local caching.
+ pub async fn from_rest_response(
+ identifier: Identifier,
+ response: crate::api::GetTableResponse,
+ rest_options: Options,
+ ) -> Result<Self> {
+ identifier.validate()?;
+ // Reject a stale or misrouted response before initializing auth or
local-cache resources.
+ response_identifier(&identifier, &response)?;
+ rest_options
+ .get(CatalogOptions::WAREHOUSE)
+ .ok_or_else(|| RestError::BadRequest {
+ message: format!("Missing required option: {}",
CatalogOptions::WAREHOUSE),
+ })?;
+ let api = Arc::new(RESTApi::new(rest_options.clone(), false).await?);
+ let data_token_enabled = api
+ .options()
+ .get(CatalogOptions::DATA_TOKEN_ENABLED)
+ .map(|v| v.eq_ignore_ascii_case("true"))
+ .unwrap_or(false);
+ let local_cache = create_local_cache_with_namespace(&rest_options,
api.options())?;
+ RESTEnv::build_table(
+ &identifier,
+ response,
+ api,
+ rest_options,
+ data_token_enabled,
+ local_cache,
+ )
+ .await
+ }
+}
+
/// REST environment that holds the REST API client, identifier, and uuid
/// needed to create a `RESTSnapshotCommit`.
#[derive(Clone)]
@@ -143,6 +178,7 @@ impl RESTEnv {
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
) -> Result<Table> {
+ let identifier = response_identifier(identifier, &response)?;
let schema = response.schema.ok_or_else(|| Error::DataInvalid {
message: format!("Table {} response missing schema",
identifier.full_name()),
source: None,
@@ -180,6 +216,7 @@ impl RESTEnv {
table_path.clone(),
)]));
}
+ table_schema.validate_resolved_structure()?;
let is_external = response.is_external.ok_or_else(||
Error::DataInvalid {
message: format!(
@@ -188,7 +225,7 @@ impl RESTEnv {
),
source: None,
})?;
- validate_catalog_managed_format_table(identifier, &table_schema,
is_external)?;
+ validate_catalog_managed_format_table(&identifier, &table_schema,
is_external)?;
let uuid = response.id.ok_or_else(|| Error::DataInvalid {
message: format!(
@@ -199,7 +236,7 @@ impl RESTEnv {
})?;
let file_io = Self::build_file_io(
- identifier,
+ &identifier,
&table_path,
api.clone(),
&options,
@@ -217,14 +254,20 @@ impl RESTEnv {
data_token_enabled,
local_cache,
);
-
- Ok(Table::new(
+ let parsed_identifier = identifier.parsed_object_name()?;
+ let branch = parsed_identifier.branch_or_default().to_string();
+ let branch_reference = parsed_identifier.branch().is_some();
+ let table = Table::new(
file_io,
- identifier.clone(),
+ identifier,
table_path,
table_schema,
Some(rest_env),
- ))
+ );
+
+ let mut table =
table.copy_with_resolved_schema(table.schema().clone(), &branch)?;
+ table.branch_reference = branch_reference;
+ Ok(table)
}
pub(crate) async fn build_object_table(
@@ -235,6 +278,7 @@ impl RESTEnv {
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
) -> Result<ObjectTable> {
+ let identifier = response_identifier(identifier, &response)?;
let schema = response.schema.ok_or_else(|| Error::DataInvalid {
message: format!("Table {} response missing schema",
identifier.full_name()),
source: None,
@@ -260,6 +304,7 @@ impl RESTEnv {
let mut schema_options = schema.options().clone();
schema_options.insert(PATH_OPTION.to_string(), object_path.clone());
let table_schema = TableSchema::new(schema_id,
&schema).copy_with_options(schema_options);
+ table_schema.validate_resolved_structure()?;
let is_external = response.is_external.ok_or_else(||
Error::DataInvalid {
message: format!(
"Table {} response missing is_external",
@@ -269,7 +314,7 @@ impl RESTEnv {
})?;
let file_io = Self::build_file_io(
- identifier,
+ &identifier,
&object_path,
api,
&options,
@@ -279,7 +324,7 @@ impl RESTEnv {
)
.await?;
- ObjectTable::try_new(file_io, identifier.clone(), &table_schema)
+ ObjectTable::try_new(file_io, identifier, &table_schema)
}
async fn build_file_io(
@@ -315,11 +360,13 @@ impl RESTEnv {
&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
+ let parsed_identifier = self.identifier.parsed_object_name()?;
+ let object = if parsed_identifier.branch() == Some(branch) {
+ format!("{}$branch_{branch}", parsed_identifier.table())
+ } else if branch == crate::catalog::DEFAULT_MAIN_BRANCH {
+ parsed_identifier.table().to_string()
} else {
- format!("{object}$branch_{branch}")
+ format!("{}$branch_{branch}", parsed_identifier.table())
};
let identifier = Identifier::new(self.identifier.database(), object);
match self.api.load_snapshot(&identifier).await {
@@ -341,6 +388,33 @@ impl RESTEnv {
}
}
+fn response_identifier(
+ requested: &Identifier,
+ response: &crate::api::GetTableResponse,
+) -> Result<Identifier> {
+ let database = response
+ .database
+ .as_deref()
+ .unwrap_or_else(|| requested.database());
+ let name = response.name.as_deref().ok_or_else(|| Error::DataInvalid {
+ message: format!("Table response for database '{database}' missing
name"),
+ source: None,
+ })?;
+ let identifier = Identifier::new(database, name);
+ identifier.validate()?;
+ if &identifier != requested {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Table response identifier '{}' does not match requested
identifier '{}'",
+ identifier.full_name(),
+ requested.full_name()
+ ),
+ source: None,
+ });
+ }
+ Ok(identifier)
+}
+
/// Refuse a Format Table that asks for catalog-managed partitions it cannot
have: an engine
/// implementation reads the table directory itself, and only an internal
table's partitions
/// belong to the catalog.
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index ec593d13..3afa0ac0 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -758,6 +758,7 @@ impl RESTServer {
// Create table response
let response = GetTableResponse::new(
Some(table_name.clone()),
+ Some(db.clone()),
Some(table_name),
None,
Some(true),
@@ -1620,6 +1621,7 @@ impl RESTServer {
// Update the table name in response and insert at new location
let new_table_response = GetTableResponse::new(
Some(request.destination.object().to_string()),
+ Some(request.destination.database().to_string()),
Some(request.destination.object().to_string()),
table_response.path,
table_response.is_external,
@@ -1682,6 +1684,7 @@ impl RESTServer {
s.tables.entry(key).or_insert_with(|| {
GetTableResponse::new(
Some(table.to_string()),
+ Some(database.to_string()),
Some(table.to_string()),
None,
Some(true),
@@ -1818,6 +1821,7 @@ impl RESTServer {
key,
GetTableResponse::new(
Some(table.to_string()),
+ Some(database.to_string()),
Some(table.to_string()),
Some(path.to_string()),
Some(true),
