diff --git a/bindings/python/README.md b/bindings/python/README.md
index 6f173a7d4..53f3a8ee0 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 3eecaf729..c8abd01bc 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 72899145d..9fea4b9fa 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 4e26f0485..50ca80729 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 000000000..929371126
--- /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 aa68dd8b3..cd7d11af0 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 2b8a1ea7d..d2757a156 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 403b42203..60aca42dd 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 db27bb3eb..1c48a140f 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 be95d9894..5df155c2e 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 f7c0e0786..55a1dd850 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 470b2f9b0..62f1bcf40 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 6fe0c8fe3..34d76f464 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 7d682cedd..f46c835f2 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 4e8a1f7c5..e814f966a 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 1de8baa9e..dbc763977 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 dda84796c..a83880c80 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 8187dce4c..f5676bce7 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 c9f4a158a..7a36ba6fe 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 c8a64f9cd..ec593d134 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"),