diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 0ede5a1fb..09ed065fb 100644 --- a/bindings/python/python/pypaimon_rust/datafusion.pyi +++ b/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: ... diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs index 2a03d03b4..4f2839927 100644 --- a/bindings/python/src/read.rs +++ b/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, opts: HashMap) -> PyResult { // 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 { + 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 { 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 { - 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, diff --git a/bindings/python/tests/test_read.py b/bindings/python/tests/test_read.py index 08dea16ef..7f98cc55c 100644 --- a/bindings/python/tests/test_read.py +++ b/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,7 +1418,7 @@ 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(): @@ -1426,7 +1426,7 @@ def test_time_travel_unresolved_snapshot_raises(): _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(): diff --git a/bindings/python/tests/test_resolved_table.py b/bindings/python/tests/test_resolved_table.py index 36cfe8b9f..e67a2c462 100644 --- a/bindings/python/tests/test_resolved_table.py +++ b/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() diff --git a/bindings/python/tests/test_stream_table_scan.py b/bindings/python/tests/test_stream_table_scan.py index ad226bb44..2d6fbf1a3 100644 --- a/bindings/python/tests/test_stream_table_scan.py +++ b/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') diff --git a/bindings/python/tests/test_table.py b/bindings/python/tests/test_table.py index 2a5b10f6f..f40fd7b82 100644 --- a/bindings/python/tests/test_table.py +++ b/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", [ diff --git a/crates/paimon/src/table/snapshot_manager.rs b/crates/paimon/src/table/snapshot_manager.rs index 0c5368744..1c35cb705 100644 --- a/crates/paimon/src/table/snapshot_manager.rs +++ b/crates/paimon/src/table/snapshot_manager.rs @@ -358,6 +358,36 @@ impl SnapshotManager { Ok(result) } + /// Returns the id of the first listed snapshot committed at or after + /// `timestamp_millis`, or `latest + 1` when every snapshot is older. + /// Returns `None` when the table has no snapshot. + /// + /// Unlike `later_or_equal_time_millis`, the result is the first of several + /// snapshots sharing a commit time. Unlike Java, which adds one to + /// `earlierThanTimeMills`, it skips ids missing from the listing. + /// + /// Reference: [TimeTravelUtil.earlierThanTimeMills](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java) + pub(crate) async fn first_id_at_or_after_time_millis( + &self, + timestamp_millis: i64, + ) -> crate::Result> { + let ids = self.list_all_ids().await?; + let Some(&latest) = ids.last() else { + return Ok(None); + }; + let (mut lo, mut hi) = (0usize, ids.len()); + while lo < hi { + let mid = lo + (hi - lo) / 2; + let commit_time = self.get_snapshot(ids[mid]).await?.time_millis() as i64; + if commit_time < timestamp_millis { + lo = mid + 1; + } else { + hi = mid; + } + } + Ok(Some(ids.get(lo).copied().unwrap_or(latest + 1))) + } + /// Returns the first snapshot whose watermark is later than or equal to the given /// `watermark`. Snapshots without a watermark — `None`, or `Some(i64::MIN)`, /// Flink's no-watermark sentinel — are skipped. If no such snapshot exists, @@ -681,6 +711,43 @@ mod tests { ); } + #[tokio::test] + async fn test_first_id_at_or_after_time_millis_keeps_ties_and_skips_holes() { + let (_, sm) = setup("memory:/test_first_id_at_or_after_time").await; + assert_eq!(sm.first_id_at_or_after_time_millis(0).await.unwrap(), None); + for (id, time) in [(2, 1000), (3, 2000), (4, 2000), (6, 3000)] { + let snapshot = Snapshot::builder() + .version(3) + .id(id) + .schema_id(0) + .base_manifest_list("base-list".to_string()) + .delta_manifest_list("delta-list".to_string()) + .commit_user("test-user".to_string()) + .commit_identifier(0) + .commit_kind(CommitKind::APPEND) + .time_millis(time) + .build(); + sm.commit_snapshot(&snapshot).await.unwrap(); + } + for (timestamp, expected) in [ + (500, 2), + (1000, 2), + (1001, 3), + (2000, 3), + (2001, 6), + (3000, 6), + (3001, 7), + ] { + assert_eq!( + sm.first_id_at_or_after_time_millis(timestamp) + .await + .unwrap(), + Some(expected), + "timestamp {timestamp}" + ); + } + } + fn test_snapshot_with_watermark(id: i64, watermark: Option) -> Snapshot { Snapshot::builder() .version(3) diff --git a/crates/paimon/src/table/snapshot_scan_tests.rs b/crates/paimon/src/table/snapshot_scan_tests.rs index 74fe00a43..5c91ff02f 100644 --- a/crates/paimon/src/table/snapshot_scan_tests.rs +++ b/crates/paimon/src/table/snapshot_scan_tests.rs @@ -253,13 +253,6 @@ async fn full_startup_modes_empty_tables_and_stream_boundaries() { assert_eq!(plan.snapshot_id(), None); assert!(plan.splits().is_empty()); } - for mode in ["latest", "from-snapshot-full"] { - let selected = table.copy_with_options(HashMap::from([("scan.mode".into(), mode.into())])); - assert!(matches!( - selected.new_read_builder().new_stream_scan(), - Err(paimon::Error::Unsupported { .. }) - )); - } let selected = table.copy_with_options(HashMap::from([ ("scan.mode".into(), "from-snapshot-full".into()), ("scan.snapshot-id".into(), "1".into()), diff --git a/crates/paimon/src/table/stream_table_scan.rs b/crates/paimon/src/table/stream_table_scan.rs index 88b52f3a4..4ad6c15e1 100644 --- a/crates/paimon/src/table/stream_table_scan.rs +++ b/crates/paimon/src/table/stream_table_scan.rs @@ -18,16 +18,20 @@ //! Stateful snapshot selection and progress for continuous table reads. use super::table_scan::SnapshotScanMode; -use super::{Plan, TableScan}; -use crate::spec::{ChangelogProducer, CommitKind, CoreOptions, Snapshot}; +use super::{Plan, SnapshotManager, TableScan}; +use crate::spec::{ChangelogProducer, CommitKind, CoreOptions, Snapshot, TimeTravelSelector}; use crate::{Error, Result}; +use std::collections::HashMap; /// Continuous table scan. `checkpoint` is the next snapshot to consume. /// -/// The first plan reads the latest full state. Follow-up plans read APPEND -/// deltas when no changelog producer is configured, or physical changelogs -/// otherwise. Empty follow-up plans and irrelevant commits are skipped, as in -/// Java. `None` means there is no available next snapshot; clients can poll. +/// Without a checkpoint or consumer progress, the first plan resolves its +/// start from `scan.mode`; every plan still validates the scan options. A +/// full state holds the merged live rows of its snapshot as inserts. Follow-up +/// plans keep the physical events of APPEND deltas when no changelog producer +/// is configured, or of changelogs otherwise. Empty follow-up plans and +/// irrelevant commits are skipped, as in Java. `None` means there is no +/// available next snapshot; clients can poll. /// Progress is persisted only by an explicit `notify_checkpoint_complete`. /// Clones retain the current progress and read configuration; subsequent /// planning, restoration and configuration changes are independent. @@ -42,29 +46,110 @@ pub struct StreamTableScan { missing_snapshot_polls: usize, } +/// Where a stream starts without a checkpoint or consumer progress, as Java +/// `AbstractDataTableScan#createStartingScanner` in streaming mode. +#[derive(Debug, Clone, Copy)] +enum StartupMode { + /// The latest full state, then follow-ups. + LatestFull, + /// Follow-ups after the latest snapshot. + Latest, + /// Follow-ups from this snapshot, without an initial state. + FromSnapshot(i64), + /// This snapshot's full state, then follow-ups. + FromSnapshotFull(i64), + /// Follow-ups from the first snapshot committed at or after this time, or + /// from the next commit, whatever its time, when every snapshot is older. + FromTimestamp(i64), +} + +/// Where the first plan starts: a full state, or follow-ups from a snapshot. +enum StartingPosition { + Full(Box), + Next(i64), +} + +impl StartupMode { + /// Resolve `scan.mode` and its selector as Java `CoreOptions#startupMode`. + /// Options a stream cannot honor fail instead of reading the latest state. + fn resolve(options: &HashMap) -> Result { + let core = CoreOptions::new(options); + core.validate_scan_options()?; + // A stream supports neither option, even next to a selector. + for key in [ + "scan.file-creation-time-millis", + "scan.creation-time-millis", + ] { + if options.contains_key(key) { + return Err(Error::Unsupported { + message: format!("Stream scans do not yet support {key}"), + }); + } + } + let selector = core.try_time_travel_selector()?; + let mode = options + .get("scan.mode") + .map(|mode| mode.to_ascii_lowercase()); + // Validation guarantees each remaining mode's selector; in particular, + // from-snapshot-full has only scan.snapshot-id. + let snapshot_id_mode: fn(i64) -> Self = match mode.as_deref().unwrap_or("default") { + "latest-full" => return Ok(Self::LatestFull), + "latest" => return Ok(Self::Latest), + "incremental" => return Err(incremental_in_streaming()), + "from-snapshot-full" => Self::FromSnapshotFull, + "default" | "from-snapshot" | "from-timestamp" => Self::FromSnapshot, + other => { + return Err(Error::Unsupported { + message: format!("Stream scans do not yet support scan.mode={other}"), + }) + } + }; + match selector { + Some(TimeTravelSelector::TimestampMillis(timestamp)) => { + Ok(Self::FromTimestamp(timestamp)) + } + Some(TimeTravelSelector::SnapshotId { value, option_name }) => { + parse_snapshot_id(value, option_name).map(snapshot_id_mode) + } + Some(TimeTravelSelector::TagName { .. }) => Err(Error::DataInvalid { + message: "Cannot scan from tag in streaming mode.".into(), + source: None, + }), + Some(TimeTravelSelector::Watermark(_)) => Err(Error::DataInvalid { + message: "Cannot scan from watermark in streaming mode.".into(), + source: None, + }), + // Java resolves scan.version against tags before choosing a mode. + Some(TimeTravelSelector::Version { .. }) => Err(Error::Unsupported { + message: "Stream scans do not yet support scan.version; use scan.snapshot-id" + .into(), + }), + None if core.incremental_timestamp_window()?.is_some() => { + Err(incremental_in_streaming()) + } + None => Ok(Self::LatestFull), + } + } +} + +fn parse_snapshot_id(value: &str, option_name: &str) -> Result { + value.parse().map_err(|_| Error::DataInvalid { + message: format!("{option_name} '{value}' is not a valid snapshot id."), + source: None, + }) +} + +fn incremental_in_streaming() -> Error { + Error::DataInvalid { + message: "Cannot read incremental in streaming mode.".into(), + source: None, + } +} + impl StreamTableScan { pub(super) fn new(scan: TableScan<'_>) -> Result { let mut scan = scan.into_owned_paimon_scan()?; let options = CoreOptions::new(scan.snapshot_table().schema().options()); - // These aliases are supported by the batch starting scanner. A - // continuous reader needs a different starting position: latest must - // skip the initial state, and from-snapshot-full must start at that - // snapshot. Keep rejecting them until those stream policies exist. - if scan - .snapshot_table() - .schema() - .options() - .get("scan.mode") - .is_some_and(|mode| { - mode.eq_ignore_ascii_case("latest") - || mode.eq_ignore_ascii_case("from-snapshot-full") - }) - { - return Err(Error::Unsupported { - message: "Stream scans do not yet support scan.mode=latest or from-snapshot-full" - .into(), - }); - } let producer = options.try_changelog_producer()?; let follow_up_mode = if producer == ChangelogProducer::None { SnapshotScanMode::Delta @@ -152,6 +237,10 @@ impl StreamTableScan { .get(consumer_id) .await?; if let Some(next) = next { + log::info!( + "Stream scan of {} resumes consumer '{consumer_id}' at snapshot {next}", + self.scan.snapshot_table().identifier().full_name() + ); self.restore(Some(next))?; } } @@ -159,15 +248,20 @@ impl StreamTableScan { } let manager = self.scan.snapshot_table().snapshot_manager(); if self.next_snapshot_id.is_none() { - let Some(snapshot) = manager.get_latest_snapshot().await? else { - return Ok(None); - }; - let next = snapshot.id() + 1; - let watermark = snapshot.watermark(); - let plan = self.plan_initial_snapshot(snapshot, scan).await?; - self.next_snapshot_id = Some(next); - self.watermark = watermark; - return Ok(Some(plan)); + match self.starting_position(&manager).await? { + None => return Ok(None), + Some(StartingPosition::Full(snapshot)) => { + let next = snapshot.id() + 1; + let watermark = snapshot.watermark(); + let plan = self.plan_initial_snapshot(*snapshot, scan).await?; + self.next_snapshot_id = Some(next); + self.watermark = watermark; + return Ok(Some(plan)); + } + // Record the position so a checkpoint taken before data + // arrives restores to it, then read follow-ups. + Some(StartingPosition::Next(next)) => self.next_snapshot_id = Some(next), + } } let mode = self.follow_up_mode; loop { @@ -216,6 +310,59 @@ impl StreamTableScan { } } + /// Resolve the starting position from `scan.mode`, or `None` to wait for a + /// snapshot. A start below the earliest snapshot moves up to it, as in Java. + async fn starting_position( + &self, + manager: &SnapshotManager, + ) -> Result> { + let table = self.scan.snapshot_table(); + let startup = StartupMode::resolve(table.schema().options())?; + let position = match startup { + StartupMode::LatestFull => manager + .get_latest_snapshot() + .await? + .map(|snapshot| StartingPosition::Full(Box::new(snapshot))), + StartupMode::FromSnapshotFull(id) => match manager.earliest_snapshot_id().await? { + Some(earliest) => Some(StartingPosition::Full(Box::new( + manager.get_snapshot(id.max(earliest)).await?, + ))), + None => None, + }, + // A table without snapshots streams everything from the first one. + StartupMode::Latest => Some(StartingPosition::Next( + manager + .get_latest_snapshot_id() + .await? + .map_or(1, |id| id + 1), + )), + StartupMode::FromSnapshot(id) => manager + .earliest_snapshot_id() + .await? + .map(|earliest| StartingPosition::Next(id.max(earliest))), + StartupMode::FromTimestamp(timestamp) => manager + .first_id_at_or_after_time_millis(timestamp) + .await? + .map(StartingPosition::Next), + }; + match &position { + Some(StartingPosition::Full(snapshot)) => log::info!( + "Stream scan of {} starts from {startup:?} with the full state of snapshot {}", + table.identifier().full_name(), + snapshot.id() + ), + Some(StartingPosition::Next(next)) => log::info!( + "Stream scan of {} starts from {startup:?} at snapshot {next}", + table.identifier().full_name() + ), + None => {} + } + Ok(position) + } + + /// Plan the full state of `snapshot`. Lookup tables skip level 0 and + /// full-compaction tables read only the last level, as in Java; rows in + /// the other levels arrive later as changelogs from compaction. async fn plan_initial_snapshot( &self, snapshot: Snapshot, diff --git a/crates/paimon/tests/stream_table_scan_test.rs b/crates/paimon/tests/stream_table_scan_test.rs index ea853eee5..75bed4445 100644 --- a/crates/paimon/tests/stream_table_scan_test.rs +++ b/crates/paimon/tests/stream_table_scan_test.rs @@ -18,7 +18,7 @@ mod common; use common::incremental_helpers::{ - make_batch, memory_table, persist_table_schema, setup_dirs, write_batch, + make_batch, make_batch_with_kinds, memory_table, persist_table_schema, setup_dirs, write_batch, }; use futures::TryStreamExt; use paimon::spec::{DataType, IntType, Schema, TableSchema}; @@ -44,8 +44,8 @@ async fn table(name: &str, producer: &str) -> Table { table } -async fn values(table: &Table, plan: Plan) -> Vec<(i32, i32)> { - use arrow_array::Int32Array; +async fn events(table: &Table, plan: Plan) -> Vec<(String, i32, i32)> { + use arrow_array::{Int32Array, StringArray}; let batches: Vec<_> = table .new_read_builder() .new_read() @@ -55,8 +55,13 @@ async fn values(table: &Table, plan: Plan) -> Vec<(i32, i32)> { .try_collect() .await .unwrap(); - let mut values = Vec::new(); + let mut events = Vec::new(); for batch in batches { + let kinds = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); let ids = batch .column(1) .as_any() @@ -67,8 +72,24 @@ async fn values(table: &Table, plan: Plan) -> Vec<(i32, i32)> { .as_any() .downcast_ref::() .unwrap(); - values.extend((0..batch.num_rows()).map(|row| (ids.value(row), vals.value(row)))); + events.extend((0..batch.num_rows()).map(|row| { + ( + kinds.value(row).to_string(), + ids.value(row), + vals.value(row), + ) + })); } + events.sort_unstable(); + events +} + +async fn values(table: &Table, plan: Plan) -> Vec<(i32, i32)> { + let mut values: Vec<_> = events(table, plan) + .await + .into_iter() + .map(|(_, id, value)| (id, value)) + .collect(); values.sort_unstable(); values } @@ -93,6 +114,15 @@ async fn patch_snapshot(table: &Table, id: i64, key: &str, value: serde_json::Va .unwrap(); } +fn with_options(table: &Table, options: &[(&str, &str)]) -> Table { + table.copy_with_options( + options + .iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect(), + ) +} + #[tokio::test] async fn empty_poll_then_full_initial_and_follow_up() { let table = table("memory:/stream-phases", "none").await; @@ -530,4 +560,431 @@ async fn initial_lookup_and_full_compaction_select_java_levels_only_once() { vec![(0, 0)] ); } + for (producer, levels, expected) in [ + ("lookup", None, vec![(1, 10)]), + ("full-compaction", None, vec![]), + ("full-compaction", Some("2"), vec![(1, 10)]), + ] { + let mut options = HashMap::from([ + ("changelog-producer".into(), producer.into()), + ("scan.mode".into(), "from-snapshot-full".into()), + ("scan.snapshot-id".into(), "2".into()), + ]); + if let Some(levels) = levels { + options.insert("num-levels".into(), levels.into()); + } + let copy = table.copy_with_options(options); + let mut scan = copy.new_stream_scan().unwrap(); + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(2)); + assert_eq!(values(©, plan).await, expected, "{producer}/{levels:?}"); + assert_eq!(scan.checkpoint(), Some(3)); + } + for producer in ["lookup", "full-compaction"] { + for (options, expected) in [ + (&[("scan.mode", "latest")][..], vec![]), + (&[("scan.snapshot-id", "2")][..], vec![(1, 10), (2, 20)]), + ( + &[("scan.timestamp-millis", "1")][..], + vec![(0, 0), (1, 10), (2, 20)], + ), + ] { + let copy = with_options( + &table, + &[options, &[("changelog-producer", producer)]].concat(), + ); + let mut scan = copy.new_stream_scan().unwrap(); + let mut rows = Vec::new(); + while let Some(plan) = scan.plan().await.unwrap() { + assert!(plan.splits().iter().all(|split| split.is_streaming())); + rows.extend(values(©, plan).await); + } + assert_eq!(rows, expected, "{producer}/{options:?}"); + assert_eq!(scan.checkpoint(), Some(4), "{producer}/{options:?}"); + } + } +} + +#[tokio::test] +async fn startup_modes_wait_on_an_empty_table() { + for (index, (options, checkpoint, full)) in [ + (&[("scan.mode", "latest")][..], Some(1), false), + (&[("scan.snapshot-id", "1")][..], None, false), + (&[("scan.timestamp-millis", "1")][..], None, false), + ( + &[ + ("scan.mode", "from-snapshot-full"), + ("scan.snapshot-id", "1"), + ][..], + None, + true, + ), + ] + .into_iter() + .enumerate() + { + let table = table(&format!("memory:/stream-empty-start-{index}"), "none").await; + let mut scan = with_options(&table, options).new_stream_scan().unwrap(); + assert!(scan.plan().await.unwrap().is_none(), "{options:?}"); + assert_eq!(scan.checkpoint(), checkpoint, "{options:?}"); + write_batch(&table, &make_batch(vec![1], vec![10])).await; + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(1), "{options:?}"); + assert!( + plan.splits() + .iter() + .all(|split| split.is_streaming() != full), + "{options:?}" + ); + assert_eq!(values(&table, plan).await, vec![(1, 10)], "{options:?}"); + assert_eq!(scan.checkpoint(), Some(2), "{options:?}"); + } +} + +#[tokio::test] +async fn latest_and_from_snapshot_start_without_an_initial_state() { + for producer in ["none", "input"] { + let table = table( + &format!("memory:/stream-from-snapshot-{producer}"), + producer, + ) + .await; + for id in 1..=3 { + write_batch(&table, &make_batch(vec![id], vec![id * 10])).await; + } + let mut latest = with_options(&table, &[("scan.mode", "latest")]) + .new_stream_scan() + .unwrap(); + assert!(latest.plan().await.unwrap().is_none(), "{producer}"); + assert_eq!(latest.checkpoint(), Some(4), "{producer}"); + for options in [ + &[("scan.mode", "from-snapshot"), ("scan.snapshot-id", "2")][..], + &[("scan.snapshot-id", "2")][..], + ] { + let selected = with_options(&table, options); + let mut scan = selected.new_stream_scan().unwrap(); + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(2), "{producer}/{options:?}"); + assert!(plan.splits().iter().all(|split| split.is_streaming())); + assert_eq!(values(&selected, plan).await, vec![(2, 20)], "{producer}"); + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(values(&selected, plan).await, vec![(3, 30)], "{producer}"); + assert_eq!(scan.checkpoint(), Some(4)); + } + + for (id, fails) in [(4, false), (5, true)] { + let mut scan = with_options(&table, &[("scan.snapshot-id", &id.to_string())]) + .new_stream_scan() + .unwrap(); + for _ in 0..15 { + assert!(scan.plan().await.unwrap().is_none(), "{producer}/{id}"); + } + let result = scan.plan().await; + if fails { + let error = result.unwrap_err().to_string(); + assert!(error.contains("outside the available snapshots"), "{error}"); + } else { + assert!(result.unwrap().is_none(), "{producer}/{id}"); + } + assert_eq!(scan.checkpoint(), Some(id), "{producer}/{id}"); + } + let result = with_options( + &table, + &[ + ("scan.mode", "from-snapshot-full"), + ("scan.snapshot-id", "4"), + ], + ) + .new_stream_scan() + .unwrap() + .plan() + .await; + assert!( + matches!( + result, + Err(paimon::Error::SnapshotNotExist { snapshot_id: 4 }) + ), + "{producer}: {result:?}" + ); + + let manager = table.snapshot_manager(); + manager.delete_snapshot(1).await.unwrap(); + manager.write_earliest_hint(2).await.unwrap(); + for mode in ["from-snapshot", "from-snapshot-full"] { + let selected = with_options(&table, &[("scan.mode", mode), ("scan.snapshot-id", "1")]); + let mut scan = selected.new_stream_scan().unwrap(); + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(2), "{producer}/{mode}"); + assert_eq!(scan.checkpoint(), Some(3), "{producer}/{mode}"); + } + + write_batch(&table, &make_batch(vec![4], vec![40])).await; + let plan = latest.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(4), "{producer}"); + assert!(plan.splits().iter().all(|split| split.is_streaming())); + assert_eq!(values(&table, plan).await, vec![(4, 40)], "{producer}"); + assert_eq!(latest.checkpoint(), Some(5), "{producer}"); + } +} + +#[tokio::test] +async fn from_timestamp_starts_at_the_first_snapshot_at_or_after_it() { + let table = table("memory:/stream-from-timestamp", "none").await; + for id in 1..=4 { + write_batch(&table, &make_batch(vec![id], vec![id * 10])).await; + } + for (id, time) in [(1, 1000), (2, 2000), (3, 2000), (4, 3000)] { + patch_snapshot(&table, id, "timeMillis", time.into()).await; + } + for (timestamp, first, checkpoint) in [ + ("500", Some(1), Some(2)), + ("2000", Some(2), Some(3)), + ("2001", Some(4), Some(5)), + ("3001", None, Some(5)), + ] { + for options in [ + &[ + ("scan.mode", "from-timestamp"), + ("scan.timestamp-millis", timestamp), + ][..], + &[("scan.timestamp-millis", timestamp)][..], + ] { + let mut scan = with_options(&table, options).new_stream_scan().unwrap(); + let plan = scan.plan().await.unwrap(); + assert_eq!( + plan.as_ref().and_then(Plan::snapshot_id), + first, + "{options:?}" + ); + assert!(plan + .iter() + .flat_map(|plan| plan.splits()) + .all(|split| split.is_streaming())); + assert_eq!(scan.checkpoint(), checkpoint, "{options:?}"); + } + } + + table.snapshot_manager().delete_snapshot(3).await.unwrap(); + let mut scan = with_options(&table, &[("scan.timestamp-millis", "2500")]) + .new_stream_scan() + .unwrap(); + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(4)); + assert_eq!(values(&table, plan).await, vec![(4, 40)]); + assert_eq!(scan.checkpoint(), Some(5)); +} + +#[tokio::test] +async fn from_snapshot_full_keeps_row_kinds_across_restore() { + let event = |kind: &str, id: i32, value: i32| (kind.to_string(), id, value); + for producer in ["none", "input"] { + for dv in ["false", "true"] { + let name = format!("memory:/stream-row-kinds-{producer}-{dv}"); + let table = with_options( + &table(&name, producer).await, + &[("deletion-vectors.enabled", dv), ("write-only", "true")], + ); + write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await; + write_batch(&table, &make_batch_with_kinds(vec![1], vec![11], vec![2])).await; + write_batch(&table, &make_batch_with_kinds(vec![2], vec![20], vec![3])).await; + let selected = with_options( + &table, + &[ + ("scan.mode", "from-snapshot-full"), + ("scan.snapshot-id", "2"), + ], + ); + let mut scan = selected.new_stream_scan().unwrap(); + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(2)); + assert!(plan.splits().iter().all(|split| !split.is_streaming())); + assert_eq!( + events(&selected, plan).await, + vec![event("+I", 1, 11), event("+I", 2, 20)], + "{producer}/{dv}" + ); + let plan = scan.plan().await.unwrap().unwrap(); + assert!(plan.splits().iter().all(|split| split.is_streaming())); + assert_eq!( + events(&selected, plan).await, + vec![event("-D", 2, 20)], + "{producer}/{dv}" + ); + + write_batch(&table, &make_batch_with_kinds(vec![1], vec![12], vec![2])).await; + let mut restored = selected.new_stream_scan().unwrap(); + restored.restore(scan.checkpoint()).unwrap(); + let plan = restored.plan().await.unwrap().unwrap(); + assert_eq!( + events(&selected, plan).await, + vec![event("+U", 1, 12)], + "{producer}/{dv}" + ); + } + } +} + +#[tokio::test] +async fn consumer_progress_overrides_the_startup_mode() { + let table = table("memory:/stream-consumer-startup", "none").await; + for id in 1..=3 { + write_batch(&table, &make_batch(vec![id], vec![id * 10])).await; + } + table.consumer_manager().reset("job", 2).await.unwrap(); + for options in [ + &[("scan.mode", "latest")][..], + &[ + ("scan.mode", "from-snapshot-full"), + ("scan.snapshot-id", "1"), + ][..], + &[("scan.timestamp-millis", "1")][..], + &[("scan.tag-name", "t")][..], + ] { + let selected = with_options(&table, &[options, &[("consumer-id", "job")]].concat()); + let mut scan = selected.new_stream_scan().unwrap(); + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(2), "{options:?}"); + assert!(plan.splits().iter().all(|split| split.is_streaming())); + } +} + +#[tokio::test] +async fn startup_options_fail_at_startup_and_invalid_ones_also_on_restore() { + let table = table("memory:/stream-startup-errors", "none").await; + for (options, unsupported, restorable) in [ + (&[("scan.mode", "from-snapshot-full")][..], false, false), + ( + &[("scan.mode", "from-snapshot"), ("scan.tag-name", "t")][..], + false, + true, + ), + (&[("scan.tag-name", "t")][..], false, true), + (&[("scan.watermark", "5")][..], false, true), + (&[("scan.snapshot-id", "x")][..], false, true), + ( + &[ + ("scan.mode", "incremental"), + ("incremental-between-timestamp", "1,2"), + ][..], + false, + true, + ), + (&[("incremental-between-timestamp", "1,2")][..], false, true), + (&[("scan.version", "1")][..], true, true), + (&[("scan.file-creation-time-millis", "1")][..], true, true), + (&[("scan.creation-time-millis", "1")][..], true, true), + ( + &[ + ("scan.snapshot-id", "1"), + ("scan.file-creation-time-millis", "1"), + ][..], + true, + true, + ), + ( + &[ + ("scan.timestamp-millis", "1"), + ("scan.creation-time-millis", "1"), + ][..], + true, + true, + ), + (&[("scan.mode", "compacted-full")][..], true, false), + ] { + let selected = with_options(&table, options); + let result = selected.new_stream_scan().unwrap().plan().await; + if unsupported { + assert!( + matches!(result, Err(paimon::Error::Unsupported { .. })), + "{options:?}: {result:?}" + ); + } else { + assert!( + matches!(result, Err(paimon::Error::DataInvalid { .. })), + "{options:?}: {result:?}" + ); + } + let mut restored = selected.new_stream_scan().unwrap(); + restored.restore(Some(1)).unwrap(); + assert_eq!(restored.plan().await.is_ok(), restorable, "{options:?}"); + } +} + +#[tokio::test] +async fn append_table_streams_from_each_startup_mode() { + let name = "memory:/stream-append-startup"; + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .build() + .unwrap(); + let (file_io, table) = memory_table(name, TableSchema::new(0, &schema)); + setup_dirs(&file_io, name).await; + persist_table_schema(&file_io, name, table.schema()).await; + for id in 1..=3 { + write_batch(&table, &make_batch(vec![id], vec![id * 10])).await; + } + let mut waiting = Vec::new(); + for (options, first, full, expected, checkpoint) in [ + (&[][..], Some(3), true, vec![(1, 10), (2, 20), (3, 30)], 4), + (&[("scan.mode", "latest")][..], None, false, vec![], 4), + ( + &[("scan.snapshot-id", "2")][..], + Some(2), + false, + vec![(2, 20)], + 3, + ), + ( + &[ + ("scan.mode", "from-snapshot-full"), + ("scan.snapshot-id", "2"), + ][..], + Some(2), + true, + vec![(1, 10), (2, 20)], + 3, + ), + ( + &[("scan.timestamp", "2000-01-01 00:00:00")][..], + Some(1), + false, + vec![(1, 10)], + 2, + ), + ( + &[("scan.timestamp", "2999-01-01")][..], + None, + false, + vec![], + 4, + ), + ] { + let selected = with_options(&table, options); + let mut scan = selected.new_stream_scan().unwrap(); + let plan = scan.plan().await.unwrap(); + assert_eq!( + plan.as_ref().and_then(Plan::snapshot_id), + first, + "{options:?}" + ); + assert_eq!(scan.checkpoint(), Some(checkpoint), "{options:?}"); + if let Some(plan) = plan { + assert!( + plan.splits() + .iter() + .all(|split| split.is_streaming() != full), + "{options:?}" + ); + assert_eq!(values(&selected, plan).await, expected, "{options:?}"); + } else { + waiting.push((options, scan)); + } + } + write_batch(&table, &make_batch(vec![4], vec![40])).await; + for (options, mut scan) in waiting { + let plan = scan.plan().await.unwrap().unwrap(); + assert_eq!(plan.snapshot_id(), Some(4), "{options:?}"); + assert_eq!(values(&table, plan).await, vec![(4, 40)], "{options:?}"); + } } diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 6f2ec7ad3..1cd5bf1da 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -49,6 +49,7 @@ nav: - Home: index.md - Getting Started: getting-started.md - Batch Incremental Reading: incremental-reading.md + - Streaming Reading: streaming-reading.md - Table Compaction: compaction.md - SQL Integration: sql.md - Performance: benchmark.md diff --git a/docs/src/python-binding.md b/docs/src/python-binding.md index 0f7eb061c..8ac271136 100644 --- a/docs/src/python-binding.md +++ b/docs/src/python-binding.md @@ -180,6 +180,25 @@ Builder filters, projections, and limits still apply. both batch and streaming splits to Java binary encoding, preserving the streaming flag. +For continuous reads, `new_stream_scan()` returns a stateful scan that starts +from `scan.mode`. See [Streaming Reading](streaming-reading.md) for startup +modes, checkpoints, and row kinds: + +```python +rb = table.new_read_builder({"scan.mode": "from-snapshot", "scan.snapshot-id": "5"}) +scan = rb.new_stream_scan() +read = rb.with_include_row_kind(True).new_read() +plan = scan.plan() +if plan is not None: + batches = read.read(plan.splits()) + # Process batches, then acknowledge the progress. + scan.notify_checkpoint_complete(scan.checkpoint()) +``` + +`new_scan()` and `new_incremental_scan()` raise when a time-travel selector +resolves to no snapshot, while `new_stream_scan()` resolves its start at the +first plan, so snapshot 5 may also be the next commit or an expired one. + For append and Data Evolution tables, select half-open row positions or one balanced shard on a scan: @@ -456,6 +475,7 @@ rb = table.new_read_builder({"scan.tag-name": "release-1.0"}) !!! warning Only one time-travel selector may be set. Providing multiple selectors will raise a `ValueError`. + A selector that resolves to no snapshot raises a `ValueError` from `new_scan()` and `new_incremental_scan()`. `scan.timestamp` selects the latest snapshot committed at or before the given local time. It accepts `YYYY-MM-DD`, `YYYY-MM-DD HH:MM:SS[.fraction]`, or diff --git a/docs/src/streaming-reading.md b/docs/src/streaming-reading.md new file mode 100644 index 000000000..86bc08fa3 --- /dev/null +++ b/docs/src/streaming-reading.md @@ -0,0 +1,124 @@ + + +# Streaming Reading + +`StreamTableScan` reads a table continuously: an optional full state first, then +the changes of each later snapshot. Filters and projection configured on the +read builder apply to every plan. + +## Read Loop + +`plan()` returns the next plan, or `None` when no new snapshot is available +yet. `checkpoint()` is the next snapshot to read. Acknowledge it only after the +plan's rows have been processed: + +```rust +use futures::TryStreamExt; +use std::time::Duration; + +let read_builder = table.new_read_builder(); +let mut scan = read_builder.new_stream_scan()?; +let read = read_builder.new_read()?; +loop { + let Some(plan) = scan.plan().await? else { + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + }; + let batches = read + .to_arrow_with_row_kind(plan.splits())? + .try_collect::>() + .await?; + // Process `batches`, then acknowledge the progress. + scan.notify_checkpoint_complete(scan.checkpoint()).await?; +} +``` + +`notify_checkpoint_complete` persists the progress when the scan has a consumer, +set by table option `consumer-id` or `StreamTableScan::with_consumer_id`. A new +scan with the same consumer resumes from it. `restore(Some(id))` resumes from a +checkpoint kept by the caller, and `restore(None)` starts over from consumer +progress or `scan.mode`. + +## Startup Modes + +An explicit checkpoint takes precedence over consumer progress, and both take +precedence over `scan.mode`. Without either, the first plan starts from +`scan.mode`: + +| `scan.mode` | Required option | First plan | +|-------------|-----------------|------------| +| `latest-full` (default) | | Full state of the latest snapshot | +| `latest` | | Changes after the snapshot that is latest at the first plan; an empty table streams from its first snapshot | +| `from-snapshot` | `scan.snapshot-id` | Changes from that snapshot, without a full state | +| `from-snapshot-full` | `scan.snapshot-id` | Full state of that snapshot | +| `from-timestamp` | `scan.timestamp-millis` or `scan.timestamp` | Changes from the first snapshot committed at or after that time | + +Later plans read the changes of each following snapshot. Without `scan.mode`, +`scan.snapshot-id` selects `from-snapshot`, and `scan.timestamp-millis` or +`scan.timestamp` selects `from-timestamp`, as in Java. + +On an empty table, every mode waits for the first commit. A snapshot ID below +the earliest snapshot starts from the earliest snapshot. With `from-snapshot`, +the ID after the latest snapshot waits for the next commit, and a larger ID +fails after some empty polls. With `from-snapshot-full`, once the table has a +snapshot, each plan fails with `SnapshotNotExist` until the selected one +exists. A timestamp only selects the first snapshot: when it is later than +every snapshot, the stream starts at the next commit, whatever its commit time, +rather than waiting until then. + +```rust +use std::collections::HashMap; + +let table = table.copy_with_options(HashMap::from([ + ("scan.mode".to_string(), "from-snapshot".to_string()), + ("scan.snapshot-id".to_string(), "5".to_string()), +])); +let mut scan = table.new_stream_scan()?; +``` + +The starting position is resolved only when the scan starts from `scan.mode`. +There, the first plan fails for options a stream cannot honor: `scan.tag-name`, +`scan.watermark`, `scan.version`, `incremental-between-timestamp`, +`scan.file-creation-time-millis`, and `scan.creation-time-millis`. Every plan +still validates the scan options, so an invalid or unsupported configuration, +such as `from-snapshot-full` without `scan.snapshot-id` or +`scan.mode=compacted-full`, also fails after `restore`. + +## Row Kinds + +`TableRead::to_arrow_with_row_kind` adds a leading `rowkind` column: + +- A full state contains the merged live rows of its snapshot as `+I`. It does + not replay earlier updates or deletes. +- Later plans keep the physical events of their files: `+I`, `-U`, `+U`, and + `-D`. Without a changelog producer, they come from the data files of `APPEND` + snapshots; otherwise, from changelog files. + +With `changelog-producer=none` or `input`, the full state merges every level, +including uncompacted level-0 files of deletion-vector tables. With `lookup`, +the full state skips level 0; with `full-compaction`, it reads only the last +level. Rows in the other levels arrive later as changelog from compaction. + +## Limitations + +- Long-lived changelogs are not read. Restoring a checkpoint older than the + earliest snapshot fails. +- `scan.bounded.watermark`, `streaming.read.snapshot.delay`, and + `consumer.ignore-progress` are not supported.