feat(core): support startup modes in StreamTableScan

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -468,7 +468,14 @@ class Table:
     def location(self) -> str: ...
     def schema(self) -> TableSchema: ...
     def new_system_read(self, name: str) -> SystemRead: ...
-    def new_read_builder(self, options: Optional[Dict[str, str]] = None) -> ReadBuilder: ...
+    def new_read_builder(self, options: Optional[Dict[str, str]] = None) -> ReadBuilder:
+        """Create a read builder on a copy of this table with ``options``.
+
+        Batch scans raise when a time-travel selector resolves to no snapshot.
+        ``new_stream_scan`` resolves ``scan.mode`` at its first plan instead, so
+        its start may be the next commit, an expired snapshot, or an empty table.
+        """
+        ...
     def new_stream_scan(self) -> StreamTableScan: ...
     def new_full_text_search_builder(self) -> FullTextSearchBuilder: ...
     def new_hybrid_search_builder(self) -> HybridSearchBuilder: ...

bindings/python/src/read.rs

@@ -155,13 +155,13 @@ impl PyReadBuilder {
 
     /// 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.
+    /// schema. A selector that resolves to no snapshot raises from batch scans
+    /// only; stream scans resolve their start from `scan.mode`.
     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.
+        // conflict error via its Java-parity silent fallback, so the batch scan
+        // check would otherwise misattribute the failure to a single selector.
+        // Surface the real conflict, listing the keys the user set.
         // scan.version must first be adapted by the core: Java allows it to
         // overwrite a selector of the same kind after resolving tag precedence.
         let present: Vec<&str> = TIME_TRAVEL_SELECTORS
@@ -175,31 +175,52 @@ impl PyReadBuilder {
                 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::new(Arc::new(traveled)))
+    }
+
+    /// Raise if the table's time-travel selector resolves to no snapshot. A
+    /// selector the table has not resolved, e.g. one in a resolved schema, is
+    /// resolved here as its plan would.
+    fn ensure_time_travel_resolved(&self, py: Python<'_>) -> PyResult<()> {
+        let Some((name, value)) = find_time_travel_selector(self.table.schema().options()) else {
+            return Ok(());
+        };
+        if self.table.has_resolved_travel_snapshot() {
+            return Ok(());
+        }
+        py.detach(|| runtime().block_on(self.table.resolve_read_snapshot()))
+            .map(|_| ())
+            .map_err(|e| {
+                PyValueError::new_err(format!(
+                    "time-travel selector {name}={value} did not resolve to any snapshot: {e}"
+                ))
+            })
+    }
+
+    /// Build a scan without checking the time-travel selector, which stream
+    /// scans resolve themselves.
+    fn table_scan(&self) -> PyTableScan {
+        PyTableScan {
+            table: Arc::clone(&self.table),
+            projection: self.projection.clone(),
+            read_type: self.read_type.clone(),
+            limit: self.limit,
+            filter: self.filter.clone(),
+            partition_filter: self.partition_filter.clone(),
+            row_ranges: self.row_ranges.clone(),
+            case_sensitive: self.case_sensitive,
+            incremental_scan: None,
+            global_index_result: None,
+            row_position_slice: None,
+            row_position_shard: None,
+            chunk_shuffle: None,
+            shard: None,
+            file_filters: None,
         }
-        Ok(Self {
-            table: Arc::new(traveled),
-            projection: None,
-            read_type: None,
-            limit: None,
-            filter: None,
-            partition_filter: None,
-            row_ranges: None,
-            case_sensitive: true,
-            blob_parallelism: None,
-            include_row_kind: false,
-        })
     }
 }
 
@@ -336,31 +357,18 @@ impl PyReadBuilder {
         Ok(slf)
     }
 
-    fn new_scan(&self) -> PyTableScan {
-        PyTableScan {
-            table: Arc::clone(&self.table),
-            projection: self.projection.clone(),
-            read_type: self.read_type.clone(),
-            limit: self.limit,
-            filter: self.filter.clone(),
-            partition_filter: self.partition_filter.clone(),
-            row_ranges: self.row_ranges.clone(),
-            case_sensitive: self.case_sensitive,
-            incremental_scan: None,
-            global_index_result: None,
-            row_position_slice: None,
-            row_position_shard: None,
-            chunk_shuffle: None,
-            shard: None,
-            file_filters: None,
-        }
+    /// Raises if a time-travel selector resolves to no snapshot.
+    fn new_scan(&self, py: Python<'_>) -> PyResult<PyTableScan> {
+        self.ensure_time_travel_resolved(py)?;
+        Ok(self.table_scan())
     }
 
-    /// Create a stateful Java-compatible stream scan.
+    /// Create a stateful Java-compatible stream scan. Its first plan resolves
+    /// the start from `scan.mode`, so selectors need not resolve here.
     pub(crate) fn new_stream_scan(&self) -> PyResult<PyStreamTableScan> {
         Ok(PyStreamTableScan {
             scan: self
-                .new_scan()
+                .table_scan()
                 .read_builder()?
                 .new_stream_scan()
                 .map_err(to_py_err)?,
@@ -374,11 +382,12 @@ impl PyReadBuilder {
     #[pyo3(signature = (start_snapshot_id, end_snapshot_id, mode = "delta"))]
     fn new_incremental_scan(
         &self,
+        py: Python<'_>,
         start_snapshot_id: i64,
         end_snapshot_id: i64,
         mode: &str,
     ) -> PyResult<PyTableScan> {
-        let mut scan = self.new_scan();
+        let mut scan = self.new_scan(py)?;
         scan.incremental_scan = Some(PyIncrementalScan {
             start_snapshot_id,
             end_snapshot_id,
@@ -1559,7 +1568,7 @@ assert routing.plan().snapshot_id() == 2
             });
             let mut builder = PyReadBuilder::new(Arc::clone(&table));
             builder.limit = Some(0);
-            let scan = Py::new(py, builder.new_scan()).unwrap();
+            let scan = Py::new(py, builder.new_scan(py).unwrap()).unwrap();
             for name in ["skip_level0", "only_read_real_buckets"] {
                 let kwargs = PyDict::new(py);
                 kwargs.set_item(name, true).unwrap();
@@ -1638,7 +1647,7 @@ assert routing.plan().snapshot_id() == 2
             assert_eq!(restored.new_files, vec![expected.files()[0].file().clone()]);
 
             builder.limit = None;
-            let scan = Py::new(py, builder.new_scan()).unwrap();
+            let scan = Py::new(py, builder.new_scan(py).unwrap()).unwrap();
             assert_eq!(
                 scan.bind(py).call_method0("plan").unwrap().len().unwrap(),
                 0
@@ -1682,7 +1691,7 @@ assert routing.plan().snapshot_id() == 2
                     "plan_all includes L0 even when explicit file filters exclude it"
                 );
             }
-            let fresh = Py::new(py, builder.new_scan()).unwrap();
+            let fresh = Py::new(py, builder.new_scan(py).unwrap()).unwrap();
             assert_eq!(
                 fresh.bind(py).call_method0("plan").unwrap().len().unwrap(),
                 0,

bindings/python/tests/test_read.py

@@ -1387,7 +1387,7 @@ def test_time_travel_by_timestamp_string():
             assert result.column("id").to_pylist() == [1]
         for value in ["invalid", "1970-01-01 00:00:00"]:
             with pytest.raises(ValueError, match="did not resolve"):
-                table.new_read_builder({"scan.timestamp": value})
+                table.new_read_builder({"scan.timestamp": value}).new_scan()
         with pytest.raises(ValueError, match="Only one"):
             table.new_read_builder({"scan.timestamp": "1970-01-03",
                                     "scan.timestamp-millis": "172800000"})
@@ -1418,15 +1418,15 @@ def test_scan_version_adapts_before_binding_selector_validation():
                 {"scan.version": "1", "scan.tag-name": "v1"},
                 {"scan.version": "v1", "scan.snapshot-id": "1"}]:
             with pytest.raises(ValueError, match="did not resolve"):
-                table.new_read_builder(opts)
+                table.new_read_builder(opts).new_scan()
 
 
 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"})
+            table.new_read_builder({"scan.snapshot-id": "999"}).new_scan()
 
 
 def test_time_travel_unresolved_watermark_raises():
@@ -1436,7 +1436,7 @@ def test_time_travel_unresolved_watermark_raises():
         # 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"})
+            table.new_read_builder({"scan.watermark": "1"}).new_scan()
 
 
 def test_unsupported_scan_option_raises_not_implemented():

bindings/python/tests/test_resolved_table.py

@@ -77,6 +77,10 @@ def test_resolved_options_replace_persisted_options(resolved_source):
     schema["options"]["scan.snapshot-id"] = "1"
     historical = Table.from_resolved_schema(str(root), json.dumps(schema))
     assert _read(historical) == (1, [{"id": 1, "name": "a"}])
+    schema["options"]["scan.snapshot-id"] = "3"
+    missing = Table.from_resolved_schema(str(root), json.dumps(schema))
+    with pytest.raises(ValueError, match="scan.snapshot-id=3 did not resolve"):
+        missing.new_read_builder().new_scan()
     assert "scan.snapshot-id" not in latest.schema().options()
 
 

bindings/python/tests/test_stream_table_scan.py

@@ -27,24 +27,32 @@
 from pypaimon_rust.datafusion import PaimonCatalog, StreamTableScan, SQLContext
 
 
-@pytest.fixture
-def table(tmp_path):
+def _create_table(tmp_path, name):
     options = {"warehouse": str(tmp_path)}
     ctx = SQLContext()
     ctx.register_catalog("paimon", options)
     ctx.sql("CREATE SCHEMA paimon.snapdb")
     # These cursor tests require one APPEND snapshot per prepared batch.
-    ctx.sql("CREATE TABLE paimon.snapdb.t (id INT, value INT, PRIMARY KEY (id)) "
+    ctx.sql(f"CREATE TABLE paimon.snapdb.{name} (id INT, value INT, PRIMARY KEY (id)) "
             "WITH ('bucket'='4', 'changelog-producer'='input', 'write-only'='true', "
             "'source.split.target-size'='1 b', 'source.split.open-file-cost'='1 b')")
-    table = PaimonCatalog(options).get_table("snapdb.t")
+    return PaimonCatalog(options).get_table(f"snapdb.{name}")
+
+
+def _write(table, value):
+    builder = table.new_batch_write_builder()
+    writer = builder.new_write()
+    writer.write_arrow(pa.record_batch({'id': pa.array(range(32), type=pa.int32()),
+                                 'value': pa.array(range(value, value + 32), type=pa.int32())}))
+    builder.new_commit().commit(writer.prepare_commit())
+    writer.close()
+
+
+@pytest.fixture
+def table(tmp_path):
+    table = _create_table(tmp_path, "t")
     for value in (100, 200):
-        builder = table.new_batch_write_builder()
-        writer = builder.new_write()
-        writer.write_arrow(pa.record_batch({'id': pa.array(range(32), type=pa.int32()),
-                                     'value': pa.array(range(value, value + 32), type=pa.int32())}))
-        builder.new_commit().commit(writer.prepare_commit())
-        writer.close()
+        _write(table, value)
     return table
 
 
@@ -225,6 +233,61 @@ def test_invalid_configuration_keeps_the_scan_usable(table):
     assert scan.plan().snapshot_id() == 2
 
 
+def test_read_builder_options_start_stream_scans_at_the_first_plan(tmp_path, table):
+    following = {'scan.mode': 'from-snapshot', 'scan.snapshot-id': '3'}
+    earlier = {'scan.mode': 'from-timestamp', 'scan.timestamp-millis': '1'}
+    for options in (following, earlier):
+        with pytest.raises(ValueError, match='did not resolve'):
+            table.new_read_builder(options).new_scan()
+
+    plan = table.new_read_builder(earlier).new_stream_scan().plan()
+    assert plan.snapshot_id() == 1
+    assert all(split.is_streaming() for split in plan.splits())
+
+    builder = table.new_read_builder({'scan.mode': 'from-snapshot-full', 'scan.snapshot-id': '1'})
+    scan = builder.new_stream_scan()
+    plan = scan.plan()
+    assert plan.snapshot_id() == 1
+    assert not any(split.is_streaming() for split in plan.splits())
+    assert _rows(builder, plan) == [(i, 100 + i) for i in range(32)]
+    plan = scan.plan()
+    assert plan.snapshot_id() == 2
+    assert _rows(builder, plan) == [(i, 200 + i) for i in range(32)]
+
+    builder = table.new_read_builder(following)
+    scan = builder.new_stream_scan()
+    assert scan.plan() is None
+    assert scan.checkpoint() == 3
+    _write(table, 300)
+    plan = scan.plan()
+    assert plan.snapshot_id() == 3
+    assert all(split.is_streaming() for split in plan.splits())
+    assert _rows(builder, plan) == [(i, 300 + i) for i in range(32)]
+
+    snapshot_dir = tmp_path / 'snapdb.db' / 't' / 'snapshot'
+    (snapshot_dir / 'snapshot-1').unlink()
+    (snapshot_dir / 'EARLIEST').write_text('2')
+    expired = table.new_read_builder({'scan.mode': 'from-snapshot', 'scan.snapshot-id': '1'})
+    with pytest.raises(ValueError, match='did not resolve'):
+        expired.new_incremental_scan(1, 2)
+    assert expired.new_stream_scan().plan().snapshot_id() == 2
+
+
+def test_read_builder_options_start_stream_scans_on_an_empty_table(tmp_path):
+    empty = _create_table(tmp_path, "empty")
+    builder = empty.new_read_builder({'scan.mode': 'from-snapshot-full', 'scan.snapshot-id': '1'})
+    with pytest.raises(ValueError, match='did not resolve'):
+        builder.new_scan()
+    scan = builder.new_stream_scan()
+    assert scan.plan() is None
+    _write(empty, 100)
+    plan = scan.plan()
+    assert plan.snapshot_id() == 1
+    assert not any(split.is_streaming() for split in plan.splits())
+    assert _rows(builder, plan) == [(i, 100 + i) for i in range(32)]
+    assert scan.checkpoint() == 2
+
+
 def test_snapshot_reader_is_an_internal_component():
     from pypaimon_rust import datafusion
     assert not hasattr(datafusion, 'SnapshotReader')

bindings/python/tests/test_table.py

@@ -116,7 +116,7 @@ def test_branch_tags_are_isolated(branch_tables):
     plan = blue.new_read_builder({"scan.tag-name": "blue"}).new_scan().plan()
     assert plan.snapshot_id() == 1
     with pytest.raises(ValueError, match="main"):
-        blue.new_read_builder({"scan.tag-name": "main"})
+        blue.new_read_builder({"scan.tag-name": "main"}).new_scan()
 
 
 @pytest.mark.parametrize("suffix", [