feat(table): support scan.watermark batch time travel

bindings/c/src/table.rs

@@ -227,8 +227,9 @@ pub unsafe extern "C" fn paimon_table_free(table: *mut paimon_table) {
 }
 
 /// Time-travel selector option names, in the core's resolution priority order.
-const TIME_TRAVEL_SELECTORS: [&str; 4] = [
+const TIME_TRAVEL_SELECTORS: [&str; 5] = [
     "scan.timestamp-millis",
+    "scan.watermark",
     "scan.version",
     "scan.snapshot-id",
     "scan.tag-name",
@@ -300,8 +301,9 @@ pub unsafe extern "C" fn paimon_table_new_read_builder(
 
 /// Create a ReadBuilder from a Table with scan options (e.g. time-travel
 /// selectors `scan.snapshot-id` / `scan.tag-name` / `scan.timestamp-millis` /
-/// `scan.version`). At most one time-travel selector may be set. A selector that
-/// does not resolve to a snapshot is an error (never a silent read-of-latest).
+/// `scan.watermark` / `scan.version`). At most one time-travel selector may be
+/// set. A selector that does not resolve to a snapshot is an error (never a
+/// silent read-of-latest).
 ///
 /// # Safety
 /// `table` must be a valid pointer. `options` must be a valid pointer to
@@ -2298,11 +2300,53 @@ mod tests {
     }
 
     #[test]
-    fn unsupported_scan_option_is_rejected() {
+    fn watermark_conflicting_with_other_selector_is_rejected() {
+        unsafe {
+            let table = boxed_test_table();
+            let k1 = CString::new("scan.watermark").unwrap();
+            let v1 = CString::new("1").unwrap();
+            let k2 = CString::new("scan.snapshot-id").unwrap();
+            let v2 = CString::new("1").unwrap();
+            let opts = [opt(&k1, &v1), opt(&k2, &v2)];
+            let (code, message) = assert_rb_err_code_message(
+                paimon_table_new_read_builder_with_options(table, opts.as_ptr(), 2),
+            );
+            assert_eq!(code, PaimonErrorCode::InvalidInput as i32);
+            assert!(
+                message.contains("scan.watermark") && message.contains("scan.snapshot-id"),
+                "message should name both selectors, got: {message}"
+            );
+            paimon_table_free(table);
+        }
+    }
+
+    #[test]
+    fn unresolved_watermark_does_not_silently_read_latest() {
         unsafe {
+            // The test table commits no watermarks, so any watermark selector
+            // is unresolvable; the binding must error instead of falling back.
             let table = boxed_test_table();
             let k = CString::new("scan.watermark").unwrap();
-            let v = CString::new("0").unwrap();
+            let v = CString::new("1").unwrap();
+            let opts = [opt(&k, &v)];
+            let (code, message) = assert_rb_err_code_message(
+                paimon_table_new_read_builder_with_options(table, opts.as_ptr(), 1),
+            );
+            assert_eq!(code, PaimonErrorCode::InvalidInput as i32);
+            assert!(
+                message.contains("did not resolve"),
+                "message should report the selector did not resolve, got: {message}"
+            );
+            paimon_table_free(table);
+        }
+    }
+
+    #[test]
+    fn unsupported_scan_option_is_rejected() {
+        unsafe {
+            let table = boxed_test_table();
+            let k = CString::new("incremental-between").unwrap();
+            let v = CString::new("1,2").unwrap();
             let opts = [opt(&k, &v)];
             // Core's validate_scan_options rejects this before resolution; the
             // binding surfaces core's Unsupported code.

bindings/python/src/read.rs

@@ -31,8 +31,9 @@ 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] = [
+const TIME_TRAVEL_SELECTORS: [&str; 5] = [
     "scan.timestamp-millis",
+    "scan.watermark",
     "scan.version",
     "scan.snapshot-id",
     "scan.tag-name",

bindings/python/tests/test_read.py

@@ -567,6 +567,16 @@ def test_time_travel_unresolved_snapshot_raises():
             table.new_read_builder({"scan.snapshot-id": "999"})
 
 
+def test_time_travel_unresolved_watermark_raises():
+    with tempfile.TemporaryDirectory() as warehouse:
+        _make_two_snapshot_table(warehouse)
+        table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+        # The Rust commit path never writes watermarks, so no snapshot matches;
+        # the binding must raise instead of silently reading latest.
+        with pytest.raises(ValueError, match="did not resolve"):
+            table.new_read_builder({"scan.watermark": "1"})
+
+
 def test_unsupported_scan_option_raises_not_implemented():
     with tempfile.TemporaryDirectory() as warehouse:
         _make_two_snapshot_table(warehouse)
@@ -659,6 +669,16 @@ def test_time_travel_conflicting_selectors_raises():
         assert "scan.tag-name" in str(exc.value)
 
 
+def test_time_travel_watermark_conflicting_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="Only one time-travel selector") as exc:
+            table.new_read_builder({"scan.watermark": "1", "scan.snapshot-id": "1"})
+        assert "scan.watermark" in str(exc.value)
+        assert "scan.snapshot-id" in str(exc.value)
+
+
 def test_split_serialize_produces_split_v1_binary():
     import struct
 

crates/integrations/datafusion/src/relation_planner.rs

@@ -149,7 +149,7 @@ fn object_name_to_table_reference(
 /// Resolve `VERSION AS OF <expr>` into `scan.version` option.
 ///
 /// The raw value (integer or string) is passed through as-is.
-/// Resolution (tag vs snapshot id) happens at scan time in `TableScan`.
+/// Resolution (tag vs watermark vs snapshot id) happens at scan time in `TableScan`.
 fn resolve_version_as_of(expr: &ast::Expr) -> DFResult<HashMap<String, String>> {
     let version = match expr {
         ast::Expr::Value(v) => match &v.value {
@@ -163,7 +163,7 @@ fn resolve_version_as_of(expr: &ast::Expr) -> DFResult<HashMap<String, String>>
         },
         _ => {
             return Err(datafusion::error::DataFusionError::Plan(format!(
-                "Unsupported VERSION AS OF expression: {expr}. Expected an integer snapshot id or a tag name."
+                "Unsupported VERSION AS OF expression: {expr}. Expected an integer snapshot id, a tag name, or a quoted 'watermark-<value>'."
             )))
         }
     };

crates/integrations/datafusion/tests/time_travel_schema_tests.rs

@@ -101,6 +101,19 @@ fn total_rows(batches: &[datafusion::arrow::record_batch::RecordBatch]) -> usize
     batches.iter().map(|b| b.num_rows()).sum()
 }
 
+fn set_snapshot_watermark(temp_dir: &TempDir, snapshot_id: i64, watermark: i64) {
+    let path = temp_dir
+        .path()
+        .join("default.db")
+        .join("t")
+        .join("snapshot")
+        .join(format!("snapshot-{snapshot_id}"));
+    let mut snapshot: serde_json::Value =
+        serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
+    snapshot["watermark"] = serde_json::json!(watermark);
+    std::fs::write(path, serde_json::to_string(&snapshot).unwrap()).unwrap();
+}
+
 #[tokio::test]
 async fn test_version_as_of_uses_snapshot_schema() {
     let (_tmp, sql_context) = setup_evolved_table().await;
@@ -138,6 +151,58 @@ async fn test_version_as_of_uses_snapshot_schema() {
     assert_eq!(total_rows(&batches), 5);
 }
 
+#[tokio::test]
+async fn test_version_as_of_java_watermark_prefix() {
+    let temp_dir = TempDir::new().expect("Failed to create temp dir");
+    let warehouse = format!("file://{}", temp_dir.path().display());
+    let mut options = Options::new();
+    options.set(CatalogOptions::WAREHOUSE, warehouse);
+    let catalog = Arc::new(FileSystemCatalog::new(options).unwrap());
+    let sql_context = create_sql_context(catalog).await;
+
+    sql_context
+        .sql("CREATE TABLE paimon.default.t (id INT)")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+    for id in 1..=3 {
+        sql_context
+            .sql(&format!("INSERT INTO paimon.default.t VALUES ({id})"))
+            .await
+            .unwrap()
+            .collect()
+            .await
+            .unwrap();
+    }
+    set_snapshot_watermark(&temp_dir, 1, 1);
+    set_snapshot_watermark(&temp_dir, 3, 10);
+
+    for (watermark, expected_rows) in [(1, 1), (9, 3), (10, 3)] {
+        let batches = sql_context
+            .sql(&format!(
+                "SELECT * FROM paimon.default.t VERSION AS OF 'watermark-{watermark}'"
+            ))
+            .await
+            .unwrap()
+            .collect()
+            .await
+            .unwrap();
+        assert_eq!(total_rows(&batches), expected_rows);
+    }
+
+    let df = sql_context
+        .sql("SELECT * FROM paimon.default.t VERSION AS OF 'watermark-11'")
+        .await
+        .unwrap();
+    let err = df.collect().await.expect_err("watermark 11 must not match");
+    assert!(
+        err.to_string().contains("watermark[11]"),
+        "error should name the unmatched watermark: {err}"
+    );
+}
+
 #[tokio::test]
 async fn test_session_scan_version_uses_snapshot_schema() {
     let (_tmp, sql_context) = setup_evolved_table().await;

crates/paimon/src/spec/core_options.rs

@@ -105,7 +105,7 @@ 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";
+pub 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;
@@ -291,9 +291,12 @@ pub struct CoreOptions<'a> {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum TimeTravelSelector<'a> {
     TimestampMillis(i64),
+    /// `scan.watermark`: batch time travel to the earliest snapshot whose
+    /// watermark is greater than or equal to the value (millis).
+    Watermark(i64),
     /// `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.
+    /// scan time as tag name (if a tag exists) → `watermark-<value>` → snapshot
+    /// id (if parseable) → error. `option_name` is kept for error attribution.
     Version {
         value: &'a str,
         option_name: &'static str,
@@ -408,7 +411,6 @@ impl<'a> CoreOptions<'a> {
             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 {
@@ -424,6 +426,7 @@ impl<'a> CoreOptions<'a> {
                     SCAN_SNAPSHOT_ID_OPTION,
                     SCAN_TAG_NAME_OPTION,
                     SCAN_VERSION_OPTION,
+                    SCAN_WATERMARK_OPTION,
                 ]
             } else if mode.eq_ignore_ascii_case("from-timestamp") {
                 &[SCAN_TIMESTAMP_MILLIS_OPTION]
@@ -792,10 +795,13 @@ impl<'a> CoreOptions<'a> {
     }
 
     fn configured_time_travel_selectors(&self) -> Vec<&'static str> {
-        let mut selectors = Vec::with_capacity(4);
+        let mut selectors = Vec::with_capacity(5);
         if self.options.contains_key(SCAN_TIMESTAMP_MILLIS_OPTION) {
             selectors.push(SCAN_TIMESTAMP_MILLIS_OPTION);
         }
+        if self.options.contains_key(SCAN_WATERMARK_OPTION) {
+            selectors.push(SCAN_WATERMARK_OPTION);
+        }
         if self.options.contains_key(SCAN_VERSION_OPTION) {
             selectors.push(SCAN_VERSION_OPTION);
         }
@@ -826,6 +832,8 @@ 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(watermark) = self.parse_i64_option(SCAN_WATERMARK_OPTION)? {
+            Ok(Some(TimeTravelSelector::Watermark(watermark)))
         } else if let Some(value) = self.options.get(SCAN_VERSION_OPTION).map(String::as_str) {
             Ok(Some(TimeTravelSelector::Version {
                 value,
@@ -2064,6 +2072,41 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_watermark_maps_to_watermark_selector() {
+        let options = HashMap::from([(SCAN_WATERMARK_OPTION.to_string(), "1234".to_string())]);
+        assert_eq!(
+            CoreOptions::new(&options)
+                .try_time_travel_selector()
+                .unwrap(),
+            Some(TimeTravelSelector::Watermark(1234))
+        );
+
+        // Strict numeric parsing, like scan.timestamp-millis.
+        let options = HashMap::from([(SCAN_WATERMARK_OPTION.to_string(), "abc".to_string())]);
+        assert!(CoreOptions::new(&options)
+            .try_time_travel_selector()
+            .is_err());
+    }
+
+    #[test]
+    fn test_watermark_conflicts_with_other_selectors() {
+        let options = HashMap::from([
+            (SCAN_WATERMARK_OPTION.to_string(), "1".to_string()),
+            (SCAN_TIMESTAMP_MILLIS_OPTION.to_string(), "2".to_string()),
+        ]);
+        let err = CoreOptions::new(&options)
+            .try_time_travel_selector()
+            .unwrap_err();
+        match err {
+            crate::Error::DataInvalid { message, .. } => {
+                assert!(message.contains(SCAN_WATERMARK_OPTION));
+                assert!(message.contains(SCAN_TIMESTAMP_MILLIS_OPTION));
+            }
+            other => panic!("unexpected: {other:?}"),
+        }
+    }
+
     #[test]
     fn test_snapshot_id_conflicts_with_version_lists_original_keys() {
         let options = HashMap::from([
@@ -2156,7 +2199,6 @@ mod tests {
             "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)
@@ -2194,6 +2236,7 @@ mod tests {
             SCAN_SNAPSHOT_ID_OPTION,
             SCAN_TAG_NAME_OPTION,
             SCAN_VERSION_OPTION,
+            SCAN_WATERMARK_OPTION,
         ] {
             let options = HashMap::from([
                 ("scan.mode".to_string(), "from-snapshot".to_string()),

crates/paimon/src/table/mod.rs

@@ -373,6 +373,7 @@ impl Table {
         let selector_changed = extra.keys().any(|k| {
             k == crate::spec::SCAN_VERSION_OPTION
                 || k == crate::spec::SCAN_TIMESTAMP_MILLIS_OPTION
+                || k == crate::spec::SCAN_WATERMARK_OPTION
                 || k == crate::spec::SCAN_SNAPSHOT_ID_OPTION
                 || k == crate::spec::SCAN_TAG_NAME_OPTION
         });
@@ -400,10 +401,10 @@ impl Table {
     ///
     /// Mirrors Java `AbstractFileStoreTable.copy(dynamicOptions)` →
     /// `tryTimeTravel`: if the merged options contain a time-travel selector
-    /// (`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)`).
+    /// (`scan.version` / `scan.timestamp-millis` / `scan.watermark` /
+    /// `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.