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 49a6121 [python] Wire per-read scan options (time travel) into the
Rust read kernel (#441)
49a6121 is described below
commit 49a61216e87b203e52c390006118508f97419256
Author: Junrui Lee <[email protected]>
AuthorDate: Sat Jul 4 23:26:15 2026 +0800
[python] Wire per-read scan options (time travel) into the Rust read kernel
(#441)
Recognize scan.snapshot-id and scan.tag-name as distinct time-travel
selectors with strict semantics: scan.snapshot-id parses an id only (never
falls back to a tag lookup), scan.tag-name resolves a tag only (never falls
back to a snapshot id), while scan.version keeps the ambiguous tag->id
compatibility of SQL VERSION AS OF. Selector resolution errors are
attributed to the original option name, conflicts among all four selectors
are rejected, and all four invalidate the cached travel snapshot in
copy_with_options so re-selecting on an already-travelled copy never
silently reuses the stale snapshot.
Add CoreOptions::validate_scan_options() to reject scan options the Rust
reader does not implement yet (incremental-between*, scan.watermark,
scan.mode != default) with Error::Unsupported, and invoke it on the merged
options inside copy_with_time_travel before travel resolution. Add
Table::has_resolved_travel_snapshot() so callers can distinguish an
unresolved selector from a real time-travelled read.
---
.../python/python/pypaimon_rust/datafusion.pyi | 2 +-
bindings/python/src/error.rs | 9 +-
bindings/python/src/read.rs | 74 ++++++
bindings/python/src/table.rs | 16 +-
bindings/python/tests/test_read.py | 134 +++++++++++
crates/integrations/datafusion/src/sql_context.rs | 12 +-
.../datafusion/tests/time_travel_schema_tests.rs | 41 ++++
crates/paimon/src/spec/core_options.rs | 261 ++++++++++++++++++++-
crates/paimon/src/table/mod.rs | 25 +-
crates/paimon/src/table/table_scan.rs | 10 +-
crates/paimon/src/table/time_travel.rs | 172 +++++++++++++-
crates/paimon/src/table/write_builder.rs | 7 +-
12 files changed, 727 insertions(+), 36 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 184c79e..8373789 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -61,7 +61,7 @@ class Table:
def identifier(self) -> str: ...
def location(self) -> str: ...
def schema(self) -> TableSchema: ...
- def new_read_builder(self) -> ReadBuilder: ...
+ def new_read_builder(self, options: Optional[Dict[str, str]] = None) ->
ReadBuilder: ...
def new_write_builder(self) -> "WriteBuilder": ...
class CommitMessage: ...
diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs
index 992a9f1..7255bf3 100644
--- a/bindings/python/src/error.rs
+++ b/bindings/python/src/error.rs
@@ -15,11 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use pyo3::exceptions::PyValueError;
+use pyo3::exceptions::{PyNotImplementedError, PyValueError};
use pyo3::PyErr;
pub fn to_py_err(err: paimon::Error) -> PyErr {
- PyValueError::new_err(err.to_string())
+ match err {
+ // Unimplemented scan semantics: distinct from malformed input so upper
+ // layers can catch NotImplementedError and decide on a fallback.
+ paimon::Error::Unsupported { .. } =>
PyNotImplementedError::new_err(err.to_string()),
+ _ => PyValueError::new_err(err.to_string()),
+ }
}
pub fn df_to_py_err(err: datafusion::error::DataFusionError) -> PyErr {
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index b708457..3cd662c 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::collections::HashMap;
use std::sync::Arc;
use arrow::pyarrow::ToPyArrow;
@@ -29,6 +30,38 @@ use pyo3::types::{PyBytes, PyDict};
use crate::error::to_py_err;
use crate::predicate::dict_to_predicate;
+/// Time-travel selector option names, in the core's resolution priority order.
+const TIME_TRAVEL_SELECTORS: [&str; 4] = [
+ "scan.timestamp-millis",
+ "scan.version",
+ "scan.snapshot-id",
+ "scan.tag-name",
+];
+
+/// Extract a Python dict of scan options into a String->String map, requiring
+/// string keys and values (non-string → TypeError) so option semantics stay
+/// unambiguous.
+pub(crate) fn extract_options(options: &Bound<'_, PyDict>) ->
PyResult<HashMap<String, String>> {
+ let mut out = HashMap::with_capacity(options.len());
+ for (k, v) in options.iter() {
+ let key: String = k
+ .extract()
+ .map_err(|_| PyTypeError::new_err("scan option keys must be
strings"))?;
+ let val: String = v.extract().map_err(|_| {
+ PyTypeError::new_err(format!("scan option '{key}' value must be a
string"))
+ })?;
+ out.insert(key, val);
+ }
+ Ok(out)
+}
+
+/// Return the first configured time-travel selector as (name, value), if any.
+fn find_time_travel_selector(opts: &HashMap<String, String>) -> Option<(&str,
&str)> {
+ TIME_TRAVEL_SELECTORS
+ .iter()
+ .find_map(|&name| opts.get(name).map(|v| (name, v.as_str())))
+}
+
/// Apply projection/limit/filter from a config snapshot onto a core
ReadBuilder.
/// Shared by PyTableScan::plan and PyTableRead::read so scan and read stay
consistent.
fn apply_read_config(
@@ -85,6 +118,47 @@ impl PyReadBuilder {
filter: None,
}
}
+
+ /// Build on a table copy resolved from scan options. Resolves time travel
+ /// (may do IO) so `with_filter` later validates against the travelled
+ /// schema. Raises if a selector is set but resolves to no snapshot, so a
+ /// mistyped snapshot-id can never silently read latest.
+ pub fn from_options(table: Arc<Table>, opts: HashMap<String, String>) ->
PyResult<Self> {
+ // Reject conflicting time-travel selectors here. The core swallows the
+ // conflict error via its Java-parity silent fallback, so the strict
+ // gate below would otherwise misattribute the failure to a single
+ // selector. Surface the real conflict, listing the keys the user set.
+ let present: Vec<&str> = TIME_TRAVEL_SELECTORS
+ .iter()
+ .copied()
+ .filter(|name| opts.contains_key(*name))
+ .collect();
+ if present.len() > 1 {
+ return Err(PyValueError::new_err(format!(
+ "Only one time-travel selector may be set, found: {}",
+ present.join(", ")
+ )));
+ }
+ let selector =
+ find_time_travel_selector(&opts).map(|(n, v)| (n.to_string(),
v.to_string()));
+ let rt = runtime();
+ let traveled = rt
+ .block_on(async { table.copy_with_time_travel(opts).await })
+ .map_err(to_py_err)?;
+ if let Some((name, value)) = selector {
+ if !traveled.has_resolved_travel_snapshot() {
+ return Err(PyValueError::new_err(format!(
+ "time-travel selector {name}={value} did not resolve to
any snapshot"
+ )));
+ }
+ }
+ Ok(Self {
+ table: Arc::new(traveled),
+ projection: None,
+ limit: None,
+ filter: None,
+ })
+ }
}
#[pymethods]
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 2864b02..a59ca8f 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -18,6 +18,7 @@
use std::sync::Arc;
use pyo3::prelude::*;
+use pyo3::types::PyDict;
use crate::read::PyReadBuilder;
use crate::schema::PyTableSchema;
@@ -49,9 +50,18 @@ impl PyTable {
PyTableSchema::new(self.inner.schema().clone())
}
- /// Create a [`PyReadBuilder`] for DataFrame-style scan planning.
- fn new_read_builder(&self) -> PyReadBuilder {
- PyReadBuilder::new(Arc::clone(&self.inner))
+ /// Create a [`PyReadBuilder`]. With `options`, resolves scan options
(incl.
+ /// 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> {
+ match options {
+ Some(dict) if !dict.is_empty() => {
+ let opts = crate::read::extract_options(dict)?;
+ PyReadBuilder::from_options(Arc::clone(&self.inner), opts)
+ }
+ _ => Ok(PyReadBuilder::new(Arc::clone(&self.inner))),
+ }
}
/// Create a [`PyWriteBuilder`] for the batch write loop.
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index f81d583..c37f564 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -523,3 +523,137 @@ def test_read_with_filter_smoke():
splits = b.new_scan().plan().splits()
batches = b.new_read().read(splits)
assert isinstance(batches, list)
+
+
+def _make_two_snapshot_table(warehouse):
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.tdb")
+ ctx.sql("CREATE TABLE paimon.tdb.t (id INT, name STRING)")
+ ctx.sql("INSERT INTO paimon.tdb.t VALUES (1, 'a')") # snapshot 1
+ ctx.sql("INSERT INTO paimon.tdb.t VALUES (2, 'b'), (3, 'c')") # snapshot 2
+ return ctx
+
+
+def _rows(batches):
+ return sum(b.num_rows for b in batches)
+
+
+def test_time_travel_by_snapshot_id():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ builder = table.new_read_builder({"scan.snapshot-id": "1"})
+ splits = builder.new_scan().plan().splits()
+ batches = builder.new_read().read(splits)
+ assert _rows(batches) == 1 # only snapshot 1's row
+
+
+def test_time_travel_by_tag_name():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = _make_two_snapshot_table(warehouse)
+ ctx.sql("CALL sys.create_tag(table => 'tdb.t', tag => 'v1',
snapshot_id => 1)")
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ builder = table.new_read_builder({"scan.tag-name": "v1"})
+ splits = builder.new_scan().plan().splits()
+ assert _rows(builder.new_read().read(splits)) == 1
+
+
+def test_time_travel_unresolved_snapshot_raises():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ with pytest.raises(ValueError, match="did not resolve"):
+ table.new_read_builder({"scan.snapshot-id": "999"})
+
+
+def test_unsupported_scan_option_raises_not_implemented():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ with pytest.raises(NotImplementedError):
+ table.new_read_builder({"incremental-between": "1,2"})
+
+
+def test_explicit_scan_mode_with_matching_selector_reads_snapshot():
+ # Java's CoreOptions.setDefaultValues() writes scan.mode=from-snapshot next
+ # to scan.snapshot-id, so configs from the Java toolchain carry both keys.
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ builder = table.new_read_builder(
+ {"scan.mode": "from-snapshot", "scan.snapshot-id": "1"})
+ splits = builder.new_scan().plan().splits()
+ assert _rows(builder.new_read().read(splits)) == 1
+
+
+def test_explicit_scan_mode_without_selector_raises():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ with pytest.raises(ValueError, match="from-snapshot"):
+ table.new_read_builder({"scan.mode": "from-snapshot"})
+
+
+def test_unimplemented_scan_mode_raises_not_implemented():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ with pytest.raises(NotImplementedError):
+ table.new_read_builder({"scan.mode": "incremental"})
+
+
+def test_scan_option_non_string_value_raises_type_error():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ with pytest.raises(TypeError):
+ table.new_read_builder({"scan.snapshot-id": 1})
+
+
+def test_new_read_builder_none_options_reads_latest():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ builder = table.new_read_builder() # no options → latest
+ splits = builder.new_scan().plan().splits()
+ assert _rows(builder.new_read().read(splits)) == 3
+
+
+def test_time_travel_filter_uses_travelled_schema():
+ # Snapshot 1 predates the 'age' column. Travelling to it and filtering on
+ # 'age' must fail (not in the travelled schema), while filtering on a
+ # column present in that schema ('name') must succeed. This proves
+ # with_filter validates against the travelled schema, not the latest one.
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.edb")
+ ctx.sql("CREATE TABLE paimon.edb.t (id INT, name STRING)")
+ ctx.sql("INSERT INTO paimon.edb.t VALUES (1, 'a')") #
snapshot 1, schema 0
+ ctx.sql("ALTER TABLE paimon.edb.t ADD COLUMN age INT")
+ ctx.sql("INSERT INTO paimon.edb.t VALUES (2, 'b', 20)") #
snapshot 2, schema 1
+ catalog = PaimonCatalog({"warehouse": warehouse})
+
+ # Filtering on the post-travel-absent column must fail.
+ travelled =
catalog.get_table("edb.t").new_read_builder({"scan.snapshot-id": "1"})
+ with pytest.raises(ValueError):
+ travelled.with_filter({"method": "equal", "field": "age",
"literals": [20]})
+
+ # Filtering on a column present in the travelled schema must succeed.
+ travelled_ok =
catalog.get_table("edb.t").new_read_builder({"scan.snapshot-id": "1"})
+ plan = travelled_ok.with_filter(
+ {"method": "equal", "field": "name", "literals": ["a"]}
+ ).new_scan().plan()
+ assert plan is not None
+
+
+def test_time_travel_conflicting_selectors_raises():
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ with pytest.raises(ValueError, match="Only one time-travel selector")
as exc:
+ table.new_read_builder({"scan.snapshot-id": "1", "scan.tag-name":
"t"})
+ # both offending keys are named
+ assert "scan.snapshot-id" in str(exc.value)
+ assert "scan.tag-name" in str(exc.value)
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index c6a652c..515cead 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -960,10 +960,18 @@ impl SQLContext {
/// with a clear message is safer than writing against a different schema
/// than concurrent reads observe.
fn ensure_no_time_travel_for_write(&self, operation: &str) -> DFResult<()>
{
- use paimon::spec::{SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_VERSION_OPTION};
+ use paimon::spec::{
+ SCAN_SNAPSHOT_ID_OPTION, SCAN_TAG_NAME_OPTION,
SCAN_TIMESTAMP_MILLIS_OPTION,
+ SCAN_VERSION_OPTION,
+ };
let options = self.dynamic_options.read().unwrap();
- for key in [SCAN_VERSION_OPTION, SCAN_TIMESTAMP_MILLIS_OPTION] {
+ for key in [
+ SCAN_VERSION_OPTION,
+ SCAN_TIMESTAMP_MILLIS_OPTION,
+ SCAN_SNAPSHOT_ID_OPTION,
+ SCAN_TAG_NAME_OPTION,
+ ] {
if options.contains_key(key) {
return Err(DataFusionError::Plan(format!(
"Cannot execute {operation} while time-travel option
'{key}' is set; \
diff --git a/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
b/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
index 71e50c9..e798665 100644
--- a/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
+++ b/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
@@ -237,6 +237,47 @@ async fn test_session_scan_version_uses_snapshot_schema() {
assert_eq!(column_names(&batches), vec!["id", "name", "age"]);
}
+#[tokio::test]
+async fn test_write_guard_rejects_snapshot_id_and_tag_name_selectors() {
+ // scan.snapshot-id and scan.tag-name are first-class time-travel
selectors,
+ // so the SQL write guard must reject writes while either is set.
Otherwise a
+ // write proceeds against the latest catalog table while reads in the same
+ // session resolve a historical snapshot (e.g. the data-evolution UPDATE
path
+ // reads _ROW_ID from a pinned snapshot and then writes to the latest
state,
+ // updating against stale row ids).
+ for (key, value) in [
+ ("paimon.scan.snapshot-id", "1"),
+ ("paimon.scan.tag-name", "some-tag"),
+ ] {
+ let (_tmp, sql_context) = setup_evolved_table().await;
+ sql_context
+ .sql(&format!("SET '{key}' = '{value}'"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ for sql in [
+ "UPDATE paimon.default.t SET name = 'x' WHERE id = 1",
+ "DELETE FROM paimon.default.t WHERE id = 1",
+ "TRUNCATE TABLE paimon.default.t",
+ ] {
+ let err = match sql_context.sql(sql).await {
+ Err(e) => e.to_string(),
+ Ok(df) => match df.collect().await {
+ Err(e) => e.to_string(),
+ Ok(_) => panic!("{sql} should fail while {key} is set"),
+ },
+ };
+ assert!(
+ err.contains("time-travel option"),
+ "{sql} should be rejected while {key} is set: {err}"
+ );
+ }
+ }
+}
+
#[tokio::test]
async fn test_relation_planner_version_as_of_uses_snapshot_schema() {
let (_tmp, sql_context) = setup_evolved_table().await;
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index f5f71c5..c5f1e50 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -67,6 +67,13 @@ const DEFAULT_COMMIT_MIN_RETRY_WAIT_MS: u64 = 1_000;
const DEFAULT_COMMIT_MAX_RETRY_WAIT_MS: u64 = 10_000;
pub const SCAN_TIMESTAMP_MILLIS_OPTION: &str = "scan.timestamp-millis";
pub const SCAN_VERSION_OPTION: &str = "scan.version";
+pub const SCAN_SNAPSHOT_ID_OPTION: &str = "scan.snapshot-id";
+pub const SCAN_TAG_NAME_OPTION: &str = "scan.tag-name";
+const INCREMENTAL_BETWEEN_OPTION: &str = "incremental-between";
+const INCREMENTAL_BETWEEN_TIMESTAMP_OPTION: &str =
"incremental-between-timestamp";
+const INCREMENTAL_BETWEEN_SCAN_MODE_OPTION: &str =
"incremental-between-scan-mode";
+const SCAN_WATERMARK_OPTION: &str = "scan.watermark";
+const SCAN_MODE_OPTION: &str = "scan.mode";
const DEFAULT_SOURCE_SPLIT_TARGET_SIZE: i64 = 128 * 1024 * 1024;
const DEFAULT_SOURCE_SPLIT_OPEN_FILE_COST: i64 = 4 * 1024 * 1024;
const DEFAULT_MANIFEST_COMPRESSION: &str = "zstd";
@@ -191,9 +198,25 @@ pub struct CoreOptions<'a> {
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum TimeTravelSelector<'a> {
TimestampMillis(i64),
- /// Raw version string from `VERSION AS OF`. Resolved at scan time:
- /// tag name (if tag exists) → snapshot id (if parseable as i64) → error.
- Version(&'a str),
+ /// `scan.version` (SQL `VERSION AS OF`): ambiguous by design. Resolved at
+ /// scan time as tag name (if a tag exists) → snapshot id (if parseable) →
+ /// error. `option_name` is kept for error attribution.
+ Version {
+ value: &'a str,
+ option_name: &'static str,
+ },
+ /// `scan.snapshot-id`: an explicit snapshot id. Resolved strictly by
+ /// parsing `value` as an id — never falls back to a tag lookup.
+ SnapshotId {
+ value: &'a str,
+ option_name: &'static str,
+ },
+ /// `scan.tag-name`: an explicit tag name. Resolved strictly by tag lookup
—
+ /// never falls back to a snapshot id.
+ TagName {
+ value: &'a str,
+ option_name: &'static str,
+ },
}
impl<'a> CoreOptions<'a> {
@@ -201,6 +224,64 @@ impl<'a> CoreOptions<'a> {
Self { options }
}
+ /// Reject scan options whose semantics the Rust core does not yet
implement.
+ ///
+ /// These are not malformed input — they are unimplemented scan modes — so
+ /// they surface as `Error::Unsupported` (mapped to `NotImplementedError`
at
+ /// the Python boundary). Explicit `scan.mode=from-snapshot` /
+ /// `from-timestamp` are the modes Java's `CoreOptions.setDefaultValues()`
+ /// writes next to the corresponding selector, so they are accepted when
+ /// that selector is present (the batch-read semantics are identical to
+ /// leaving the mode at `default`); an explicit mode without its selector
+ /// is malformed input (`Error::DataInvalid`), mirroring Java's
+ /// `SchemaValidation`. All other non-default modes are unimplemented.
+ pub fn validate_scan_options(&self) -> crate::Result<()> {
+ for key in [
+ INCREMENTAL_BETWEEN_OPTION,
+ INCREMENTAL_BETWEEN_TIMESTAMP_OPTION,
+ INCREMENTAL_BETWEEN_SCAN_MODE_OPTION,
+ SCAN_WATERMARK_OPTION,
+ ] {
+ if self.options.contains_key(key) {
+ return Err(crate::Error::Unsupported {
+ message: format!("Scan option '{key}' is not supported by
the Rust reader yet"),
+ });
+ }
+ }
+ if let Some(mode) = self.options.get(SCAN_MODE_OPTION) {
+ let selector_keys: &[&str] = if
mode.eq_ignore_ascii_case("default") {
+ return Ok(());
+ } else if mode.eq_ignore_ascii_case("from-snapshot") {
+ &[
+ SCAN_SNAPSHOT_ID_OPTION,
+ SCAN_TAG_NAME_OPTION,
+ SCAN_VERSION_OPTION,
+ ]
+ } else if mode.eq_ignore_ascii_case("from-timestamp") {
+ &[SCAN_TIMESTAMP_MILLIS_OPTION]
+ } else {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Scan option 'scan.mode={mode}' is not supported by
the Rust reader yet"
+ ),
+ });
+ };
+ if !selector_keys
+ .iter()
+ .any(|key| self.options.contains_key(*key))
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Scan option 'scan.mode={mode}' requires one of {} to
be set",
+ selector_keys.join(", ")
+ ),
+ source: None,
+ });
+ }
+ }
+ Ok(())
+ }
+
pub fn deletion_vectors_enabled(&self) -> bool {
self.options
.get(DELETION_VECTORS_ENABLED_OPTION)
@@ -414,13 +495,19 @@ impl<'a> CoreOptions<'a> {
}
fn configured_time_travel_selectors(&self) -> Vec<&'static str> {
- let mut selectors = Vec::with_capacity(2);
+ let mut selectors = Vec::with_capacity(4);
if self.options.contains_key(SCAN_TIMESTAMP_MILLIS_OPTION) {
selectors.push(SCAN_TIMESTAMP_MILLIS_OPTION);
}
if self.options.contains_key(SCAN_VERSION_OPTION) {
selectors.push(SCAN_VERSION_OPTION);
}
+ if self.options.contains_key(SCAN_SNAPSHOT_ID_OPTION) {
+ selectors.push(SCAN_SNAPSHOT_ID_OPTION);
+ }
+ if self.options.contains_key(SCAN_TAG_NAME_OPTION) {
+ selectors.push(SCAN_TAG_NAME_OPTION);
+ }
selectors
}
@@ -442,8 +529,25 @@ impl<'a> CoreOptions<'a> {
if let Some(ts) = self.parse_i64_option(SCAN_TIMESTAMP_MILLIS_OPTION)?
{
Ok(Some(TimeTravelSelector::TimestampMillis(ts)))
- } else if let Some(version) =
self.options.get(SCAN_VERSION_OPTION).map(String::as_str) {
- Ok(Some(TimeTravelSelector::Version(version)))
+ } else if let Some(value) =
self.options.get(SCAN_VERSION_OPTION).map(String::as_str) {
+ Ok(Some(TimeTravelSelector::Version {
+ value,
+ option_name: SCAN_VERSION_OPTION,
+ }))
+ } else if let Some(value) = self
+ .options
+ .get(SCAN_SNAPSHOT_ID_OPTION)
+ .map(String::as_str)
+ {
+ Ok(Some(TimeTravelSelector::SnapshotId {
+ value,
+ option_name: SCAN_SNAPSHOT_ID_OPTION,
+ }))
+ } else if let Some(value) =
self.options.get(SCAN_TAG_NAME_OPTION).map(String::as_str) {
+ Ok(Some(TimeTravelSelector::TagName {
+ value,
+ option_name: SCAN_TAG_NAME_OPTION,
+ }))
} else {
Ok(None)
}
@@ -1050,7 +1154,10 @@ mod tests {
version_core
.try_time_travel_selector()
.expect("version selector"),
- Some(TimeTravelSelector::Version("my-tag"))
+ Some(TimeTravelSelector::Version {
+ value: "my-tag",
+ option_name: SCAN_VERSION_OPTION
+ })
);
let version_num_options =
@@ -1060,10 +1167,51 @@ mod tests {
version_num_core
.try_time_travel_selector()
.expect("version numeric selector"),
- Some(TimeTravelSelector::Version("3"))
+ Some(TimeTravelSelector::Version {
+ value: "3",
+ option_name: SCAN_VERSION_OPTION
+ })
);
}
+ #[test]
+ fn test_snapshot_id_and_tag_name_map_to_distinct_selectors() {
+ let snap = HashMap::from([(SCAN_SNAPSHOT_ID_OPTION.to_string(),
"2".to_string())]);
+ assert_eq!(
+ CoreOptions::new(&snap).try_time_travel_selector().unwrap(),
+ Some(TimeTravelSelector::SnapshotId {
+ value: "2",
+ option_name: SCAN_SNAPSHOT_ID_OPTION
+ })
+ );
+ let tag = HashMap::from([(SCAN_TAG_NAME_OPTION.to_string(),
"t1".to_string())]);
+ assert_eq!(
+ CoreOptions::new(&tag).try_time_travel_selector().unwrap(),
+ Some(TimeTravelSelector::TagName {
+ value: "t1",
+ option_name: SCAN_TAG_NAME_OPTION
+ })
+ );
+ }
+
+ #[test]
+ fn test_snapshot_id_conflicts_with_version_lists_original_keys() {
+ let options = HashMap::from([
+ (SCAN_SNAPSHOT_ID_OPTION.to_string(), "1".to_string()),
+ (SCAN_TAG_NAME_OPTION.to_string(), "t".to_string()),
+ ]);
+ let err = CoreOptions::new(&options)
+ .try_time_travel_selector()
+ .unwrap_err();
+ match err {
+ crate::Error::DataInvalid { message, .. } => {
+ assert!(message.contains(SCAN_SNAPSHOT_ID_OPTION));
+ assert!(message.contains(SCAN_TAG_NAME_OPTION));
+ }
+ other => panic!("unexpected: {other:?}"),
+ }
+ }
+
#[test]
fn test_write_options_defaults() {
let options = HashMap::new();
@@ -1080,4 +1228,101 @@ mod tests {
let core = CoreOptions::new(&options);
assert_eq!(core.write_parquet_buffer_size(), 32 * 1024 * 1024);
}
+
+ #[test]
+ fn test_validate_scan_options_rejects_unsupported() {
+ for key in [
+ "incremental-between",
+ "incremental-between-timestamp",
+ "incremental-between-scan-mode",
+ "scan.watermark",
+ ] {
+ let options = HashMap::from([(key.to_string(), "x".to_string())]);
+ let err = CoreOptions::new(&options)
+ .validate_scan_options()
+ .unwrap_err();
+ assert!(matches!(err, crate::Error::Unsupported { message } if
message.contains(key)));
+ }
+ }
+
+ #[test]
+ fn test_validate_scan_options_scan_mode_whitelist() {
+ // absent OK
+ assert!(CoreOptions::new(&HashMap::new())
+ .validate_scan_options()
+ .is_ok());
+ // default OK
+ let ok = HashMap::from([("scan.mode".to_string(),
"default".to_string())]);
+ assert!(CoreOptions::new(&ok).validate_scan_options().is_ok());
+ // unimplemented modes Unsupported
+ for mode in ["compacted-full", "incremental", "latest", "latest-full"]
{
+ let bad = HashMap::from([("scan.mode".to_string(),
mode.to_string())]);
+ let err =
CoreOptions::new(&bad).validate_scan_options().unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { message } if
message.contains("scan.mode")),
+ "scan.mode={mode} should be Unsupported"
+ );
+ }
+ }
+
+ #[test]
+ fn test_validate_scan_options_explicit_mode_with_matching_selector() {
+ // Java's CoreOptions.setDefaultValues() writes scan.mode=from-snapshot
+ // next to scan.snapshot-id, so these combinations are standard input.
+ for selector in [
+ SCAN_SNAPSHOT_ID_OPTION,
+ SCAN_TAG_NAME_OPTION,
+ SCAN_VERSION_OPTION,
+ ] {
+ let options = HashMap::from([
+ ("scan.mode".to_string(), "from-snapshot".to_string()),
+ (selector.to_string(), "1".to_string()),
+ ]);
+ assert!(
+ CoreOptions::new(&options).validate_scan_options().is_ok(),
+ "scan.mode=from-snapshot with {selector} should be accepted"
+ );
+ }
+ let options = HashMap::from([
+ ("scan.mode".to_string(), "from-timestamp".to_string()),
+ (SCAN_TIMESTAMP_MILLIS_OPTION.to_string(), "1".to_string()),
+ ]);
+ assert!(CoreOptions::new(&options).validate_scan_options().is_ok());
+ }
+
+ #[test]
+ fn test_validate_scan_options_explicit_mode_without_selector() {
+ // An explicit mode missing its selector must fail loudly instead of
+ // silently reading latest (mirrors Java SchemaValidation).
+ let options = HashMap::from([("scan.mode".to_string(),
"from-snapshot".to_string())]);
+ let err = CoreOptions::new(&options)
+ .validate_scan_options()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::DataInvalid { ref message, .. } if
message.contains("from-snapshot")),
+ "got {err:?}"
+ );
+
+ let options = HashMap::from([("scan.mode".to_string(),
"from-timestamp".to_string())]);
+ let err = CoreOptions::new(&options)
+ .validate_scan_options()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::DataInvalid { ref message, .. } if
message.contains("from-timestamp")),
+ "got {err:?}"
+ );
+
+ // A mismatched selector doesn't satisfy the mode either.
+ let options = HashMap::from([
+ ("scan.mode".to_string(), "from-timestamp".to_string()),
+ (SCAN_SNAPSHOT_ID_OPTION.to_string(), "1".to_string()),
+ ]);
+ assert!(CoreOptions::new(&options).validate_scan_options().is_err());
+ }
+
+ #[test]
+ fn test_validate_scan_options_allows_supported_selectors() {
+ let options = HashMap::from([(SCAN_SNAPSHOT_ID_OPTION.to_string(),
"1".to_string())]);
+ assert!(CoreOptions::new(&options).validate_scan_options().is_ok());
+ }
}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 4a59948..10b4648 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -95,7 +95,7 @@ pub use write_builder::WriteBuilder;
use crate::catalog::Identifier;
use crate::io::FileIO;
-use crate::spec::{DataField, Snapshot, TableSchema};
+use crate::spec::{CoreOptions, DataField, Snapshot, TableSchema};
use std::collections::HashMap;
/// Table represents a table in the catalog.
@@ -214,7 +214,10 @@ impl Table {
// scans of such a copy fail until `copy_with_time_travel` re-resolves
// it). Unrelated options keep the snapshot/schema pair intact.
let selector_changed = extra.keys().any(|k| {
- k == crate::spec::SCAN_VERSION_OPTION || k ==
crate::spec::SCAN_TIMESTAMP_MILLIS_OPTION
+ k == crate::spec::SCAN_VERSION_OPTION
+ || k == crate::spec::SCAN_TIMESTAMP_MILLIS_OPTION
+ || k == crate::spec::SCAN_SNAPSHOT_ID_OPTION
+ || k == crate::spec::SCAN_TAG_NAME_OPTION
});
Self {
file_io: self.file_io.clone(),
@@ -238,14 +241,18 @@ impl Table {
///
/// Mirrors Java `AbstractFileStoreTable.copy(dynamicOptions)` →
/// `tryTimeTravel`: if the merged options contain a time-travel selector
- /// (`scan.version` / `scan.timestamp-millis`) that resolves to a snapshot,
- /// the table's fields and keys come from that snapshot's schema while the
- /// options stay the merged ones (Java `TableSchema.copy(newOptions)`).
+ /// (`scan.version` / `scan.timestamp-millis` / `scan.snapshot-id` /
+ /// `scan.tag-name`) that resolves to a snapshot, the table's fields and
+ /// keys come from that snapshot's schema while the options stay the merged
+ /// ones (Java `TableSchema.copy(newOptions)`).
/// Like Java, resolution failures fall back silently to the current
/// schema (the `if let Ok` below swallows them); an invalid selector
/// still fails later at scan planning.
pub async fn copy_with_time_travel(&self, extra: HashMap<String, String>)
-> Result<Self> {
let mut table = self.copy_with_options(extra);
+ // Reject unimplemented scan options on the merged view before any IO,
so
+ // both table-level and per-read options are covered.
+ CoreOptions::new(table.schema().options()).validate_scan_options()?;
// travel_to_snapshot returns Ok(None) without IO when the merged
// options contain no selector.
if let Ok(Some(snapshot)) =
@@ -269,6 +276,14 @@ impl Table {
self.time_traveled
}
+ /// Whether a time-travel selector in this copy's options resolved to a
+ /// snapshot. Lets external callers (e.g. the Python binding) distinguish
+ /// "selector set but unresolved" (silent fallback to latest) from a real
+ /// travelled read, so they can reject the former instead of reading
latest.
+ pub fn has_resolved_travel_snapshot(&self) -> bool {
+ self.travel_snapshot.is_some()
+ }
+
/// The snapshot resolved by [`Table::copy_with_time_travel`] from this
/// copy's options, if any. Lets scans skip re-resolving the selector.
pub(crate) fn travel_snapshot(&self) -> Option<&Snapshot> {
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 93a1c35..e8ee0a9 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -627,12 +627,18 @@ impl<'a> TableScan<'a> {
/// Plan the full scan: resolve snapshot (via options or latest), then
read manifests and build DataSplits.
///
/// Time travel is resolved from table options:
- /// - only one of `scan.version`, `scan.timestamp-millis` may be set
- /// - `scan.version` → tag name (if exists) → snapshot id (if parseable) →
error
+ /// - only one of `scan.version`, `scan.timestamp-millis`,
+ /// `scan.snapshot-id`, `scan.tag-name` may be set
+ /// - `scan.version` → tag name (if exists) → snapshot id (if parseable) →
+ /// error (ambiguous by design, like SQL `VERSION AS OF`)
+ /// - `scan.snapshot-id` → snapshot id only (never a tag lookup)
+ /// - `scan.tag-name` → tag name only (never parsed as a snapshot id)
/// - `scan.timestamp-millis` → find the latest snapshot <= that timestamp
/// - otherwise → read the latest snapshot
///
/// Reference:
[TimeTravelUtil.tryTravelToSnapshot](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java)
+ /// for `scan.version`; the strict selectors mirror Java's typed
+ /// `scan.snapshot-id` / `scan.tag-name` handling.
pub async fn plan(&self) -> crate::Result<Plan> {
self.ensure_query_auth_allowed()?;
let data_evolution_read_field_ids = self.projected_read_field_ids()?;
diff --git a/crates/paimon/src/table/time_travel.rs
b/crates/paimon/src/table/time_travel.rs
index 7a5d07d..ebb67ae 100644
--- a/crates/paimon/src/table/time_travel.rs
+++ b/crates/paimon/src/table/time_travel.rs
@@ -48,22 +48,47 @@ pub(crate) async fn travel_to_snapshot(
}),
}
}
- Some(TimeTravelSelector::Version(v)) => {
- // Tag first, then snapshot id, else error.
+ Some(TimeTravelSelector::Version {
+ value: v,
+ option_name,
+ }) => {
+ // `scan.version` is ambiguous by design: tag first, then snapshot
id.
let tag_manager = TagManager::new(file_io.clone(),
table_path.to_string());
if tag_manager.tag_exists(v).await? {
- match tag_manager.get(v).await? {
- Some(s) => Ok(Some(s)),
- None => Err(Error::DataInvalid {
- message: format!("Tag '{v}' doesn't exist."),
- source: None,
- }),
- }
+ resolve_tag(&tag_manager, v).await.map(Some)
} else if let Ok(id) = v.parse::<i64>() {
snapshot_manager.get_snapshot(id).await.map(Some)
} else {
Err(Error::DataInvalid {
- message: format!("Version '{v}' is not a valid tag name or
snapshot id."),
+ message: format!("{option_name} '{v}' is not a valid tag
name or snapshot id."),
+ source: None,
+ })
+ }
+ }
+ Some(TimeTravelSelector::SnapshotId {
+ value: v,
+ option_name,
+ }) => {
+ // An explicit snapshot id: parse strictly, never resolve a tag.
+ match v.parse::<i64>() {
+ Ok(id) => snapshot_manager.get_snapshot(id).await.map(Some),
+ Err(_) => Err(Error::DataInvalid {
+ message: format!("{option_name} '{v}' is not a valid
snapshot id."),
+ source: None,
+ }),
+ }
+ }
+ Some(TimeTravelSelector::TagName {
+ value: v,
+ option_name: _,
+ }) => {
+ // An explicit tag name: resolve strictly by tag, never as a
snapshot id.
+ let tag_manager = TagManager::new(file_io.clone(),
table_path.to_string());
+ if tag_manager.tag_exists(v).await? {
+ resolve_tag(&tag_manager, v).await.map(Some)
+ } else {
+ Err(Error::DataInvalid {
+ message: format!("Tag '{v}' doesn't exist."),
source: None,
})
}
@@ -72,6 +97,17 @@ pub(crate) async fn travel_to_snapshot(
}
}
+/// Fetch a tag known to exist, mapping an unexpectedly-missing tag to an
error.
+async fn resolve_tag(tag_manager: &TagManager, name: &str) ->
crate::Result<Snapshot> {
+ match tag_manager.get(name).await? {
+ Some(s) => Ok(s),
+ None => Err(Error::DataInvalid {
+ message: format!("Tag '{name}' doesn't exist."),
+ source: None,
+ }),
+ }
+}
+
#[cfg(test)]
mod tests {
use crate::catalog::Identifier;
@@ -375,6 +411,22 @@ mod tests {
assert_eq!(retraveled.travel_snapshot().map(|s| s.id()), Some(2));
}
+ #[tokio::test]
+ async fn test_invalid_snapshot_id_error_names_original_option() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let opts = options(&[("scan.snapshot-id", "abc")]);
+ let err = super::travel_to_snapshot(&file_io, &table_path, &opts)
+ .await
+ .expect_err("non-numeric snapshot-id must fail");
+ match err {
+ crate::Error::DataInvalid { message, .. } => {
+ assert!(message.contains("scan.snapshot-id"), "got:
{message}");
+ assert!(!message.contains("scan.version"), "got: {message}");
+ }
+ other => panic!("unexpected: {other:?}"),
+ }
+ }
+
#[tokio::test]
async fn test_time_travel_read_uses_snapshot_schema() {
use futures::TryStreamExt;
@@ -421,4 +473,104 @@ mod tests {
let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(rows, 5);
}
+
+ #[tokio::test]
+ async fn test_copy_with_time_travel_rejects_unsupported_scan_option() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+ let err = table
+ .copy_with_time_travel(options(&[("scan.watermark", "5")]))
+ .await
+ .expect_err("unsupported scan option must fail");
+ assert!(
+ matches!(err, crate::Error::Unsupported { message } if
message.contains("scan.watermark"))
+ );
+ }
+
+ #[tokio::test]
+ async fn test_has_resolved_travel_snapshot_reflects_resolution() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ let resolved = table
+ .copy_with_time_travel(options(&[("scan.snapshot-id", "1")]))
+ .await
+ .unwrap();
+ assert!(resolved.has_resolved_travel_snapshot());
+
+ // Nonexistent snapshot id: core silently falls back, helper reports
false.
+ let unresolved = table
+ .copy_with_time_travel(options(&[("scan.snapshot-id", "999")]))
+ .await
+ .unwrap();
+ assert!(!unresolved.has_resolved_travel_snapshot());
+
+ // No selector at all: false.
+ let none = table.copy_with_time_travel(HashMap::new()).await.unwrap();
+ assert!(!none.has_resolved_travel_snapshot());
+ }
+
+ #[tokio::test]
+ async fn
test_changing_snapshot_id_or_tag_selector_after_travel_invalidates_cache() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ let traveled = table
+ .copy_with_time_travel(options(&[("scan.version", "1")]))
+ .await
+ .unwrap();
+ assert_eq!(traveled.travel_snapshot().map(|s| s.id()), Some(1));
+
+ // scan.snapshot-id and scan.tag-name are first-class time-travel
+ // selectors, so re-selecting through either on an already-travelled
+ // copy must invalidate the cached snapshot — otherwise the stale
+ // snapshot would be reused and the new selector silently ignored.
+ for selector in ["scan.snapshot-id", "scan.tag-name"] {
+ let stale = traveled.copy_with_options(options(&[(selector,
"2")]));
+ assert!(
+ stale.travel_snapshot().is_none(),
+ "{selector} must invalidate the cached travel snapshot"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_snapshot_id_selector_rejects_tag_name() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ // A tag named "v1-tag" exists, but it is not a valid snapshot id.
+ let sm = SnapshotManager::new(file_io.clone(), table_path.clone());
+ let snapshot1 = sm.get_snapshot(1).await.unwrap();
+ let tm = TagManager::new(file_io.clone(), table_path.clone());
+ tm.create("v1-tag", &snapshot1).await.unwrap();
+
+ // scan.snapshot-id must only accept a numeric snapshot id; it must not
+ // fall back to resolving a tag of the same name.
+ let err = super::travel_to_snapshot(
+ &file_io,
+ &table_path,
+ &options(&[("scan.snapshot-id", "v1-tag")]),
+ )
+ .await
+ .expect_err("scan.snapshot-id must not resolve a tag name");
+ assert!(
+ matches!(err, crate::Error::DataInvalid { ref message, .. }
+ if message.contains("scan.snapshot-id")),
+ "expected snapshot-id parse error, got {err:?}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_tag_name_selector_rejects_snapshot_id() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ // No tag named "1" exists, but snapshot 1 does.
+ let err =
+ super::travel_to_snapshot(&file_io, &table_path,
&options(&[("scan.tag-name", "1")]))
+ .await
+ .expect_err("scan.tag-name must not resolve a snapshot id");
+ assert!(
+ matches!(err, crate::Error::DataInvalid { ref message, .. }
+ if message.contains("Tag '1'")),
+ "expected missing-tag error, got {err:?}"
+ );
+ }
}
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index 703fd2c..9671e3b 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -92,9 +92,10 @@ impl<'a> WriteBuilder<'a> {
crate::spec::CoreOptions::new(self.table.schema().options()).try_time_travel_selector();
if !matches!(selector, Ok(None)) {
return Err(crate::Error::Unsupported {
- message: "Cannot write to a table with a time-travel option
set \
- (scan.version / scan.timestamp-millis)"
- .to_string(),
+ message:
+ "Cannot write to a table with a time-travel option set \
+ (scan.version / scan.timestamp-millis /
scan.snapshot-id / scan.tag-name)"
+ .to_string(),
});
}
let write = TableWrite::new(self.table, self.commit_user.clone())?;
