feat(iceberg): pinned snapshot reads in IcebergTableProvider - #45
Merged
phillipleblanc merged 1 commit intoJun 21, 2026
Merged
Conversation
Add an optional snapshot_id to the catalog-backed IcebergTableProvider (with_snapshot_id builder, defaulting to None = current snapshot). scan() threads it into both file planning (TableScanBuilder::snapshot_id) and the produced IcebergTableScan, so a pinned snapshot is honored consistently -- including when the scan is serialized and replayed by a distributed engine (e.g. Ballista), letting every task of one query read the same snapshot. Backward-compatible: existing callers pass no snapshot and keep reading the current snapshot.
There was a problem hiding this comment.
Pull request overview
This PR adds an optional “pinned snapshot” capability to the catalog-backed IcebergTableProvider in the DataFusion integration so that distributed execution (e.g., Ballista) can ensure all tasks for a single query plan against—and read from—the same Iceberg snapshot.
Changes:
- Introduces
snapshot_id: Option<i64>onIcebergTableProvider(defaultNone) with awith_snapshot_id(Option<i64>)builder andsnapshot_id()getter. - Threads the pinned snapshot into both file planning (
TableScanBuilder::snapshot_id) and the producedIcebergTableScanso execution honors the same snapshot.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
phillipleblanc
added a commit
to spiceai/spiceai
that referenced
this pull request
Jun 21, 2026
Cross-task snapshot consistency: without pinning, each Ballista task re-planned the scan against the table's current snapshot, so a concurrent commit mid-query could leave partitions reading different snapshots. Now the scheduler captures the plan-time snapshot (the scan's own, else the table's current), serializes it in the recipe, and each executor task pins its IcebergTableProvider clone to it via with_snapshot_id — so all partitions of one query read one snapshot. Requires the iceberg fork change adding IcebergTableProvider::with_snapshot_id (spiceai/iceberg-rust#45); re-pins the iceberg deps to that rev.
phillipleblanc
changed the title
sgrebnov
approved these changes
Jun 21, 2026
phillipleblanc
added a commit
to spiceai/spiceai
that referenced
this pull request
Jun 21, 2026
Fork PR spiceai/iceberg-rust#45 (IcebergTableProvider::with_snapshot_id) merged into spiceai-0.9.1-df-54; move the iceberg-* pins from the feature-branch rev to the merged branch rev 5c5557c3. Lock diff is iceberg-rev-only.
pull Bot
pushed a commit
to TheRakeshPurohit/spiceai
that referenced
this pull request
Jun 21, 2026
…spiceai#11378) * feat(cluster): support distributed (ballista) scans of Iceberg tables * refactor(cluster): address review on iceberg distributed scan codec - Fail with a structured error (instead of saturating to usize/u32/u64 MAX) when a projection index or limit doesn't fit the wire/decode type. - Use the per-job TaskContext's target_partitions when replanning on the executor, so the rebuilt scan buckets into the same partition count the scheduler planned. - Add a unit test covering IcebergScanExec -> IcebergTableScanExecNode encode. * fix(cluster): keep iceberg tables excluded from health monitoring The IcebergClusterTableProvider wrapper (and the IcebergDeletionProvider it wraps for read-write datasets) hid the underlying IcebergTableProvider from find_concrete_table_provider, which the datasets health monitor uses to skip Iceberg availability checks (spiceai#6994). Peel both wrappers so the skip still fires. Adds inner() accessors + a regression test. * refactor(cluster): defer iceberg scan to execute(), drop blocking decode Removes the block_in_place/block_on sync-over-async bridge from the iceberg codec decode path. try_decode now resolves the registered provider synchronously (get_table_sync) and builds a deferred IcebergScanExec; the async TableProvider::scan is replayed lazily inside execute()'s RecordBatch stream (proper async context), so no catalog I/O runs during plan deserialization. Output partitioning is serialized so the deferred node reports the right partition count before the lazy scan runs. Also future-proofs IcebergClusterTableProvider: #[deny(clippy::missing_trait_methods)] forces every TableProvider method (incl. scan_with_args/update/truncate) to be explicitly forwarded, so a new defaulted trait method can't silently bypass the distributed-scan wrapping. * refactor(cluster): address review on deferred iceberg scan - Memoize the re-derived scan per IcebergScanExec instance (tokio OnceCell) so it is planned once, not once per partition: avoids redundant plan_files() and keeps all partitions of an instance on one snapshot/fileset. - encode/decode_partitioning now fail on out-of-range counts/indices instead of saturating (consistent with projection/limit handling). - Fix deferred DisplayAs double-bracketing the projection. * feat(cluster): pin the iceberg snapshot across distributed scan tasks Cross-task snapshot consistency: without pinning, each Ballista task re-planned the scan against the table's current snapshot, so a concurrent commit mid-query could leave partitions reading different snapshots. Now the scheduler captures the plan-time snapshot (the scan's own, else the table's current), serializes it in the recipe, and each executor task pins its IcebergTableProvider clone to it via with_snapshot_id — so all partitions of one query read one snapshot. Requires the iceberg fork change adding IcebergTableProvider::with_snapshot_id (spiceai/iceberg-rust#45); re-pins the iceberg deps to that rev. * fix(cluster): locate iceberg cluster provider through all wrappers on decode Use find_concrete_table_provider::<IcebergClusterTableProvider> instead of peeling only the federation/metadata adaptors, so a distributed scan still resolves the provider when the Iceberg dataset is additionally wrapped (e.g. EmbeddingTable for embeddings, IndexedTableProvider for a search index). * fix(cluster): replay iceberg scan with the task's RuntimeEnv Build the deferred re-plan SessionContext with new_with_config_rt using the executing task's RuntimeEnv (object stores, memory pool, disk manager) instead of a default one, so the scan replay matches the executor's environment rather than diverging from it. * chore(deps): re-pin iceberg to the merged spiceai-0.9.1-df-54 rev Fork PR spiceai/iceberg-rust#45 (IcebergTableProvider::with_snapshot_id) merged into spiceai-0.9.1-df-54; move the iceberg-* pins from the feature-branch rev to the merged branch rev 5c5557c3. Lock diff is iceberg-rev-only.
github-actions Bot
pushed a commit
to spiceai/spiceai
that referenced
this pull request
Jun 23, 2026
…#11378) * feat(cluster): support distributed (ballista) scans of Iceberg tables * refactor(cluster): address review on iceberg distributed scan codec - Fail with a structured error (instead of saturating to usize/u32/u64 MAX) when a projection index or limit doesn't fit the wire/decode type. - Use the per-job TaskContext's target_partitions when replanning on the executor, so the rebuilt scan buckets into the same partition count the scheduler planned. - Add a unit test covering IcebergScanExec -> IcebergTableScanExecNode encode. * fix(cluster): keep iceberg tables excluded from health monitoring The IcebergClusterTableProvider wrapper (and the IcebergDeletionProvider it wraps for read-write datasets) hid the underlying IcebergTableProvider from find_concrete_table_provider, which the datasets health monitor uses to skip Iceberg availability checks (#6994). Peel both wrappers so the skip still fires. Adds inner() accessors + a regression test. * refactor(cluster): defer iceberg scan to execute(), drop blocking decode Removes the block_in_place/block_on sync-over-async bridge from the iceberg codec decode path. try_decode now resolves the registered provider synchronously (get_table_sync) and builds a deferred IcebergScanExec; the async TableProvider::scan is replayed lazily inside execute()'s RecordBatch stream (proper async context), so no catalog I/O runs during plan deserialization. Output partitioning is serialized so the deferred node reports the right partition count before the lazy scan runs. Also future-proofs IcebergClusterTableProvider: #[deny(clippy::missing_trait_methods)] forces every TableProvider method (incl. scan_with_args/update/truncate) to be explicitly forwarded, so a new defaulted trait method can't silently bypass the distributed-scan wrapping. * refactor(cluster): address review on deferred iceberg scan - Memoize the re-derived scan per IcebergScanExec instance (tokio OnceCell) so it is planned once, not once per partition: avoids redundant plan_files() and keeps all partitions of an instance on one snapshot/fileset. - encode/decode_partitioning now fail on out-of-range counts/indices instead of saturating (consistent with projection/limit handling). - Fix deferred DisplayAs double-bracketing the projection. * feat(cluster): pin the iceberg snapshot across distributed scan tasks Cross-task snapshot consistency: without pinning, each Ballista task re-planned the scan against the table's current snapshot, so a concurrent commit mid-query could leave partitions reading different snapshots. Now the scheduler captures the plan-time snapshot (the scan's own, else the table's current), serializes it in the recipe, and each executor task pins its IcebergTableProvider clone to it via with_snapshot_id — so all partitions of one query read one snapshot. Requires the iceberg fork change adding IcebergTableProvider::with_snapshot_id (spiceai/iceberg-rust#45); re-pins the iceberg deps to that rev. * fix(cluster): locate iceberg cluster provider through all wrappers on decode Use find_concrete_table_provider::<IcebergClusterTableProvider> instead of peeling only the federation/metadata adaptors, so a distributed scan still resolves the provider when the Iceberg dataset is additionally wrapped (e.g. EmbeddingTable for embeddings, IndexedTableProvider for a search index). * fix(cluster): replay iceberg scan with the task's RuntimeEnv Build the deferred re-plan SessionContext with new_with_config_rt using the executing task's RuntimeEnv (object stores, memory pool, disk manager) instead of a default one, so the scan replay matches the executor's environment rather than diverging from it. * chore(deps): re-pin iceberg to the merged spiceai-0.9.1-df-54 rev Fork PR spiceai/iceberg-rust#45 (IcebergTableProvider::with_snapshot_id) merged into spiceai-0.9.1-df-54; move the iceberg-* pins from the feature-branch rev to the merged branch rev 5c5557c3. Lock diff is iceberg-rev-only.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Adds an optional pinned snapshot to the catalog-backed
IcebergTableProvider:snapshot_id: Option<i64>field (defaultNone),with_snapshot_id(Option<i64>)builder +snapshot_id()getter,scan()threads it into both file planning (TableScanBuilder::snapshot_id)and the produced
IcebergTableScan.Why
Spice's distributed (Ballista) Iceberg reads serialize a scan recipe and replay
TableProvider::scanon each executor task. Without a pinned snapshot, each taskplans against the table's current snapshot, so a concurrent commit mid-query
could leave different partitions reading different snapshots. Pinning the
plan-time snapshot lets every task of one query read a single consistent
snapshot.
Compatibility
Fully backward-compatible:
snapshot_iddefaults toNone, which preserves theexisting "read the current snapshot" behavior for all current callers. The scan
node already honored a snapshot in
execute()(build_table_scan); this onlylets the catalog-backed provider set it.
Tested:
cargo test -p iceberg-datafusion --lib(86 passed).