feat: expand native planning across resolved tables and scan modes

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/):

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__(

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(|| {

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))
     }