Skip to content

feat(core): support startup modes in StreamTableScan - #1313

Merged
JingsongLi merged 1 commit into
apache:mainfrom
SteNicholas:PAIMON-807
Nov 3, 2026
Merged

JingsongLi merged 1 commit into
apache:mainfrom
SteNicholas:PAIMON-807

Conversation

@SteNicholas

@SteNicholas SteNicholas commented Nov 3, 2026 •

Copy link
Copy Markdown
Member

Purpose

Linked issue: close #807

StreamTableScan could only start from the latest full state: scan.mode latest and from-snapshot-full were rejected, while from-snapshot and from-timestamp were silently ignored, so a reader asking for the changes since snapshot N got the latest full state instead. This PR resolves the starting position from scan.mode as Java AbstractDataTableScan#createStartingScanner does in streaming mode.

Brief change log

  • Startup modes (stream_table_scan.rs), used only when the scan has neither a checkpoint nor consumer progress:
    • latest: no initial read; changes after the latest snapshot, or from snapshot 1 on an empty table.
    • from-snapshot: changes from max(scan.snapshot-id, earliest). The id after the latest snapshot waits for the next commit; a larger id fails the existing range check, as in Java NextSnapshotFetcher.
    • from-timestamp: changes from the first snapshot committed at or after the timestamp, or from the next commit when every snapshot is older.
    • from-snapshot-full: the full state of max(scan.snapshot-id, earliest), then changes. An empty table waits for its first commit; otherwise a missing snapshot fails with SnapshotNotExist.
    • An unset or default scan.mode resolves as Java CoreOptions#startupMode: scan.snapshot-id selects from-snapshot, a timestamp selects from-timestamp, and no selector selects latest-full.
  • Row kinds: a full state holds the merged live rows of its snapshot as +I, without replaying earlier +U/-D events; later plans keep physical row kinds. With changelog-producer none or input, the full state merges every level, including uncompacted level-0 files of DV tables; lookup and full-compaction keep their level filters.
  • Explicit errors instead of silent fallbacks at startup: tag, watermark and scan.version selectors, incremental reads, and scan.file-creation-time-millis / scan.creation-time-millis, even next to another selector. Every plan still validates the scan options, so an invalid configuration (e.g. from-snapshot-full without scan.snapshot-id) also fails after a restore.
  • Logging: the start chosen from scan.mode or consumer progress is logged at INFO.
  • SnapshotManager::first_id_at_or_after_time_millis: unlike later_or_equal_time_millis, it returns the first of several snapshots sharing a commit time, and unlike Java's earlierThanTimeMills + 1, it skips ids missing from the listing instead of waiting on them.
  • Python binding:
    • table.new_read_builder(options).new_stream_scan() starts from scan.mode, as Java's table.copy(options).newReadBuilder().newStreamScan().
    • new_scan() and new_incremental_scan() check the time-travel selector in table.schema().options() instead of new_read_builder(options) raising. A selector the table resolved passes; one it has not resolved, such as one in a resolved schema, is resolved as the plan would and raises if it matches no snapshot.
    • new_stream_scan() skips that check, so a stream can start from the next commit, an expired snapshot, an early timestamp, or an empty table.

Tests

  • snapshot_manager.rs: test_first_id_at_or_after_time_millis_keeps_ties_and_skips_holes.
  • tests/stream_table_scan_test.rs:
    • startup_modes_wait_on_an_empty_table
    • latest_and_from_snapshot_start_without_an_initial_state (producers none / input, ids below the earliest, the next commit, and beyond the latest)
    • from_timestamp_starts_at_the_first_snapshot_at_or_after_it (shared commit times, timestamps before and after every snapshot, a snapshot-id hole)
    • from_snapshot_full_keeps_row_kinds_across_restore (DV and non-DV tables, uncompacted level 0)
    • initial_lookup_and_full_compaction_select_java_levels_only_once (extended with every startup mode for lookup / full-compaction)
    • consumer_progress_overrides_the_startup_mode
    • startup_options_fail_at_startup_and_invalid_ones_also_on_restore
    • append_table_streams_from_each_startup_mode
  • bindings/python/tests/test_stream_table_scan.py: new_read_builder(options).new_stream_scan() starting at the next commit, an early timestamp, a resolved snapshot, an expired snapshot and an empty table, while new_scan() / new_incremental_scan() raise for the unresolved selectors.
  • bindings/python/tests/test_read.py, test_table.py: unresolved selectors raise from new_scan().
  • bindings/python/tests/test_resolved_table.py: a resolved-schema selector that matches no snapshot raises from new_scan().

API and Format

  • StreamTableScan now honors scan.mode latest, from-snapshot, from-timestamp and from-snapshot-full. Tables that set scan.snapshot-id or a timestamp without a checkpoint now start there instead of at the latest full state, and the unsupported startup options above now fail instead of being ignored.
  • Python: Table.new_read_builder(options) no longer raises for a time-travel selector that resolves to no snapshot; new_scan() and new_incremental_scan() raise instead.
  • No storage format change.

Documentation

  • New docs/src/streaming-reading.md: read loop, checkpoints, startup modes, row kinds and limitations; added to the mkdocs.yml nav.
  • docs/src/python-binding.md: continuous reads with new_read_builder(options).new_stream_scan(), and the batch scans that raise for unresolved selectors.

🤖 Generated with Claude Code

@JingsongLi

Copy link
Copy Markdown
Contributor

I suggest removing unresolved_selector and validating directly at the batch scan entry point using the selector in table.schema().options() and table.has_resolved_travel_snapshot().

The table already retains the information needed for this check, so there is no need to cache a separate diagnostic string on PyReadBuilder. new_scan() can derive the error from that state, with new_incremental_scan() reusing the same check. new_stream_scan() should continue to let the stream scan resolve its own starting position.

StreamTableScan could only start from the latest full state: scan.mode
latest and from-snapshot-full were rejected, and from-snapshot and
from-timestamp were silently ignored. Resolve the starting position from
scan.mode, as Java AbstractDataTableScan#createStartingScanner does for
streaming, whenever the scan has neither a checkpoint nor consumer
progress:

- latest: the changes committed after the latest snapshot.
- from-snapshot: the changes from snapshot max(scan.snapshot-id,
  earliest) onward.
- from-timestamp: the changes from the first snapshot committed at or
  after the timestamp onward.
- from-snapshot-full: the full state of snapshot max(scan.snapshot-id,
  earliest), followed by the changes after it.

As in Java CoreOptions#startupMode, an unset or default scan.mode
selects from-snapshot for scan.snapshot-id, from-timestamp for a
timestamp, and latest-full without a selector.

The full state emits the merged live rows as +I, while changes keep
their physical row kinds. Startup options a stream cannot honor, such as
tag, watermark and scan.version selectors or incremental reads, now fail
instead of being silently ignored.

In the Python binding, new_read_builder(options).new_stream_scan()
starts from scan.mode, matching Java's
table.copy(options).newReadBuilder().newStreamScan(). Batch scans
created by new_scan() and new_incremental_scan() raise when a
time-travel selector resolves to no snapshot.

Document streaming reads in a new Streaming Reading page.
@SteNicholas

Copy link
Copy Markdown
Member Author

@JingsongLi, done in 38e6458. PyReadBuilder no longer caches unresolved_selector: new_scan(), and new_incremental_scan() through it, reads the selector from table.schema().options() and passes when table.has_resolved_travel_snapshot() is true, while new_stream_scan() still lets the stream scan resolve its own start.

One adjustment: has_resolved_travel_snapshot() is also false for a table that never went through copy_with_time_travel, such as a Table.from_resolved_schema / copy_with_resolved_schema table or a catalog table with a persisted scan.snapshot-id or scan.tag-name, whose selector the core resolves at plan time (test_resolved_table.py reads such tables through new_read_builder().new_scan()). Failing on that state alone would reject them, so when a selector is set but not yet resolved, new_scan() calls table.resolve_read_snapshot(), the same resolution the plan runs, and raises only if it fails. Only that case does IO; a table without a selector, or with a resolved one, is checked from state alone. test_resolved_table.py now also covers a resolved-schema selector that matches no snapshot.

@JingsongLi JingsongLi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

@JingsongLi
JingsongLi merged commit 1752ab8 into apache:main Nov 3, 2026
15 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Support latest, from-snapshot, from-timestamp and from-snapshot-full startup modes in StreamTableScan

2 participants