feat(scan): support timestamp-based time travel

[dhruvarya-db]

Which issue does this PR close?

None.

What changes are included in this PR?

Adds timestamp-based time travel through snapshot_id_as_of_time and TableScanBuilder::as_of_time:

let scan = table.scan().as_of_time(timestamp_ms).build()?;

The timestamp is in epoch milliseconds with an inclusive cutoff. Selection uses the loaded table's main snapshot history, including rollback events. The first history entry wins equal timestamps, and lookup handles clock-skewed history without assuming timestamp order.

The selected snapshot uses the existing snapshot-ID scan path, including its historical schema, projection, filtering and reader behavior. Unavailable history or a missing snapshot returns an error. Mixing timestamp and snapshot-ID selectors is rejected; repeated calls to the same selector keep the last value. Default scans remain unchanged. Public API signatures are included.

Are these changes tested?

Yes. Tests cover timestamp boundaries and extremes, rollback, ties, clock skew, unavailable history, missing snapshots, selector conflicts, historical schemas, and file-task/Arrow-row equivalence with explicit snapshot-ID scans.

The historical-row regression reads real Parquet data through distinct snapshots: timestamp and explicit-ID scans return the older row (x = 100) with a one-column schema, while the default scan returns 2,048 current rows (x = 1) with eight columns. The test was also verified to fail when timestamp selection was temporarily bypassed; the implementation was then restored.

AI Disclosure

Implemented with assistance from OpenAI Codex. All code has been manually reviewed.

[laskoviymishka]

Nice — this is a clean implementation and the test coverage is better than most first passes I see: unsorted/skewed logs, equal-timestamp ties, rollback records, expired prefixes, schema fallback, and a real end-to-end read against a historical snapshot. I traced the selection logic against Java's SnapshotUtil and it matches exactly — nullableSnapshotIdAsOfTime uses the same strict-> max-timestamp accumulator, so first-wins-on-ties and the ≤60s-skew cases you've encoded are the correct behavior, not a divergence. Good to see those pinned down in tests.

The one thing I'd settle before this merges is the public-API shape. snapshot_id_as_of_time takes &TableMetadata while ancestors_of/ancestors_between in the same module take &TableMetadataRef, and the param is metadata vs their table_metadata. Trivial to align now and semver-breaking once public-api.txt ships, so better before the first publish than after.

A few smaller things, none blocking:

• The tie-break is correct but fragile to a well-meaning >→>= edit; I'd restructure it so first-wins is explicit and locally obvious (left an inline).
• Worth a docstring line that the returned id may not resolve via snapshot_by_id if the snapshot was later expired.
• Cross-engine wrinkle worth a doc note: we match Java (first-wins on ties), but PyIceberg iterates history in reverse and picks last-wins, so duplicate-timestamp queries can select different snapshots across engines. Not something to fix here — just worth acknowledging.
• Test gaps I'd close: the builder-level rollback test steps past the equal-timestamp case (timestamp_ms + 1000), so the one path where ties actually matter isn't covered end-to-end; and there's no case where every qualifying log entry points at an evicted snapshot. The vec![1; 2048] fixture constant could use a name too.
• New public APIs but no CHANGELOG entry.

Once the signature's locked down I'm happy to take another pass.

[laskoviymishka]

I'd switch this to &TableMetadataRef to match ancestors_of/ancestors_between in this module. Right now a caller holding a TableMetadataRef (e.g. StaticTable::metadata()) has to deref for this one fn but not the siblings. &TableMetadata is technically sufficient since we only read an i64 out, but the inconsistency is cheap to fix now and a semver-breaking change once public-api.txt ships. While we're here, the param is metadata where the siblings use table_metadata — worth aligning too. The build() call site can stay self.table.metadata().

[dhruvarya-db]

Updated to table_metadata: &TableMetadataRef

[laskoviymishka]

The tie-break is correct and matches Java's nullableSnapshotIdAsOfTime (strict > on the best timestamp, so the first entry at the max qualifying timestamp wins). My worry is it's easy to break silently: entry is captured in both the guard and the is_none_or closure, and a maintainer "simplifying" > to >= would flip ties to last-wins with no compile error. I'd make first-wins explicit with a reduce so the invariant stays local:

let best = metadata.history()
    .filter(|e| e.timestamp_ms() <= timestamp_ms)
    .reduce(|best, e| if e.timestamp_ms() > best.timestamp_ms() { e } else { best });

and keep a one-line note on why it's > (matches Java; first-appended wins).

[dhruvarya-db]

The suggestion looks much cleaner. Updated!

[laskoviymishka]

Since this is public now, worth spelling out that the returned id isn't guaranteed to still resolve — if the snapshot was expired after that log entry, snapshot_by_id returns None even though this returned Ok. build() handles it, but a standalone caller doing snapshot_by_id(snapshot_id_as_of_time(...)?) won't expect a None after an Ok.

[dhruvarya-db]

Makes sense. Updated the docstring to clarify that the returned snapshot may have expired and snapshot_by_id can still return None.

[laskoviymishka]

2048 here is 2 live files × 1024 rows from the fixture — worth a named const or a short comment, otherwise this assertion fails opaquely if the fixture ever changes.

[dhruvarya-db]

Thanks for the review @laskoviymishka!

Test gaps I'd close: the builder-level rollback test steps past the equal-timestamp case (timestamp_ms + 1000), so the one path where ties actually matter isn't covered end-to-end; and there's no case where every qualifying log entry points at an evicted snapshot

Addressed these test gaps and the other comments.

New public APIs but no CHANGELOG entry.

I didn't realize that we also add unreleased changes to the CHANGELOG (looks like a new process?), I have updated it with this change.