[SteNicholas]

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]

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.

[SteNicholas]

@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]

+1