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.

Author: phillipleblanc

Cargo.lock

@@ -458,12 +458,12 @@ tokio-rusqlite = { git = "https://github.com/spiceai/tokio-rusqlite.git", rev =
 # Tracking Issue: https://github.com/allan2/dotenvy/issues/113
 dotenvy = { git = "https://github.com/spiceai/dotenvy.git", rev = "e5cef1871b08003198949dfe2da988633eaad78f" }
 
-iceberg = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "ef85534eca57620fe2a61e3e202ecbbca2471868" } # branch: spiceai-0.9.1-df-54
-iceberg-catalog-glue = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "ef85534eca57620fe2a61e3e202ecbbca2471868" } # branch: spiceai-0.9.1-df-54
-iceberg-catalog-rest = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "ef85534eca57620fe2a61e3e202ecbbca2471868" } # branch: spiceai-0.9.1-df-54
-iceberg-datafusion = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "ef85534eca57620fe2a61e3e202ecbbca2471868" } # branch: spiceai-0.9.1-df-54
-iceberg-storage-opendal = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "ef85534eca57620fe2a61e3e202ecbbca2471868" } # branch: spiceai-0.9.1-df-54
-iceberg_test_utils = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "ef85534eca57620fe2a61e3e202ecbbca2471868" } # branch: spiceai-0.9.1-df-54
+iceberg = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "17183547f127cccf7b47ee25dab99b5dcc8dc340" } # branch: phillip/df-54-snapshot-pinning (snapshot pinning; merge into spiceai-0.9.1-df-54)
+iceberg-catalog-glue = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "17183547f127cccf7b47ee25dab99b5dcc8dc340" } # branch: phillip/df-54-snapshot-pinning (snapshot pinning; merge into spiceai-0.9.1-df-54)
+iceberg-catalog-rest = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "17183547f127cccf7b47ee25dab99b5dcc8dc340" } # branch: phillip/df-54-snapshot-pinning (snapshot pinning; merge into spiceai-0.9.1-df-54)
+iceberg-datafusion = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "17183547f127cccf7b47ee25dab99b5dcc8dc340" } # branch: phillip/df-54-snapshot-pinning (snapshot pinning; merge into spiceai-0.9.1-df-54)
+iceberg-storage-opendal = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "17183547f127cccf7b47ee25dab99b5dcc8dc340" } # branch: phillip/df-54-snapshot-pinning (snapshot pinning; merge into spiceai-0.9.1-df-54)
+iceberg_test_utils = { git = "https://github.com/spiceai/iceberg-rust.git", rev = "17183547f127cccf7b47ee25dab99b5dcc8dc340" } # branch: phillip/df-54-snapshot-pinning (snapshot pinning; merge into spiceai-0.9.1-df-54)
 
 # candle-* crates are already redirected to the spiceai/candle fork via the
 # [patch.crates-io] block higher up in this file. Keeping those here would be

Cargo.toml

@@ -328,6 +328,11 @@ message IcebergTableScanExecNode {
     // Output partitioning, so the deferred node reports the same partition count
     // the scheduler planned before the (lazy) scan actually runs.
     IcebergPartitioning partitioning = 6;
+    // Snapshot pinned at planning time on the scheduler. Every executor task
+    // plans against this snapshot, so all partitions of one query read a single
+    // consistent snapshot even if the table is committed to mid-query. Absent =
+    // read the current snapshot (e.g. table had no snapshot at plan time).
+    optional int64 snapshot_id = 7;
 }
 
 // Serializable form of the scan's output `Partitioning`.

crates/runtime-proto/proto/spice.proto

@@ -23,6 +23,8 @@ use arrow_schema::Schema;
 use ballista_core::serde::BallistaPhysicalExtensionCodec;
 #[cfg(not(windows))]
 use cayenne::provider::CayenneAccelerationExec;
+use data_components::iceberg::delete::IcebergDeletionProvider;
+use datafusion::catalog::TableProvider;
 use datafusion::common::{DataFusionError, Result, TableReference, exec_err};
 use datafusion::execution::{FunctionRegistry, TaskContext};
 use datafusion::logical_expr::Expr;
@@ -37,6 +39,7 @@ use datafusion_proto::physical_plan::AsExecutionPlan;
 use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 #[cfg(not(windows))]
 use datafusion_proto::protobuf::PhysicalPlanNode;
+use iceberg_datafusion::IcebergTableProvider;
 use prost::Message;
 use runtime_datafusion::execution_plan::schema_cast::SchemaCastScanExec;
 use runtime_datafusion::extension::bytes_processed::BytesProcessedExec;
@@ -208,21 +211,31 @@ impl PhysicalExtensionCodec for SpicePhysicalCodec {
                          distributed Iceberg scans require the Iceberg data connector"
                     );
                 };
-                // The concrete Iceberg provider (not the cluster wrapper), so
-                // replaying its scan() yields the bare scan without re-wrapping.
-                let inner_provider = Arc::clone(cluster.inner());
+                // Resolve the concrete IcebergTableProvider (the cluster wrapper's
+                // inner, peeling the deletion wrapper on the read-write path) and
+                // pin it to the snapshot the scheduler chose, so every executor task
+                // of this query reads the same snapshot. Scanning this provider
+                // directly also yields the bare scan without re-wrapping.
+                let Some(iceberg_provider) = concrete_iceberg_provider(cluster.inner()) else {
+                    return exec_err!(
+                        "IcebergClusterTableProvider for {table_ref} does not wrap an \
+                         IcebergTableProvider; cannot reconstruct the distributed scan"
+                    );
+                };
+                let pinned: Arc<dyn TableProvider> =
+                    Arc::new(iceberg_provider.clone().with_snapshot_id(node.snapshot_id));
 
                 // Output schema = table schema projected by the recipe, taken
                 // synchronously from the registered provider.
-                let table_schema = inner_provider.schema();
+                let table_schema = pinned.schema();
                 let output_schema = match &projection {
                     Some(p) => Arc::new(table_schema.project(p)?),
                     None => table_schema,
                 };
 
                 Ok(Arc::new(IcebergScanExec::new_deferred(
                     table_ref,
-                    inner_provider,
+                    pinned,
                     projection,
                     filters,
                     limit,
@@ -323,6 +336,8 @@ impl PhysicalExtensionCodec for SpicePhysicalCodec {
                         filters,
                         limit,
                         partitioning: Some(partitioning),
+                        // Pin the plan-time snapshot so every executor task reads it.
+                        snapshot_id: scan_exec.snapshot_id(),
                     },
                 )),
             }
@@ -355,6 +370,19 @@ impl PhysicalExtensionCodec for SpicePhysicalCodec {
     }
 }
 
+/// Returns the concrete [`IcebergTableProvider`] behind a cluster wrapper's inner
+/// provider — directly (read path) or through the [`IcebergDeletionProvider`] the
+/// read-write path inserts. The returned provider is used (cloned + snapshot
+/// pinned) to replay the scan on the executor.
+fn concrete_iceberg_provider(inner: &Arc<dyn TableProvider>) -> Option<&IcebergTableProvider> {
+    if let Some(p) = inner.downcast_ref::<IcebergTableProvider>() {
+        return Some(p);
+    }
+    inner
+        .downcast_ref::<IcebergDeletionProvider>()
+        .and_then(|d| d.inner().downcast_ref::<IcebergTableProvider>())
+}
+
 /// Serializes a scan's output [`Partitioning`] into its wire form, so the
 /// deferred node on the executor reports the same partition count the scheduler
 /// planned (before the lazy scan runs).

crates/runtime/src/cluster/datafusion/codec/spice_physical_codec.rs

@@ -61,6 +61,7 @@ use datafusion::physical_plan::{
 };
 use datafusion::prelude::{SessionConfig, SessionContext};
 use futures::TryStreamExt;
+use iceberg_datafusion::physical_plan::IcebergTableScan;
 use runtime_datafusion::config::cluster_config::SpiceClusterConfig;
 use tokio::sync::OnceCell;
 
@@ -202,6 +203,24 @@ impl IcebergScanExec {
     pub fn limit(&self) -> Option<usize> {
         self.limit
     }
+
+    /// The snapshot to pin for this scan, derived from the wrapped scan at plan
+    /// time: the scan's explicit snapshot if any, else the table's current
+    /// snapshot. The codec serializes this so every executor task plans against
+    /// the same snapshot — giving one consistent snapshot across all partitions
+    /// of a distributed query even under concurrent commits. Returns `None` for a
+    /// deferred node (its provider already carries the pin) or if the wrapped plan
+    /// isn't an `IcebergTableScan`.
+    #[must_use]
+    pub fn snapshot_id(&self) -> Option<i64> {
+        match &self.source {
+            ScanSource::Planned(inner) => inner.downcast_ref::<IcebergTableScan>().and_then(|s| {
+                s.snapshot_id()
+                    .or_else(|| s.table().metadata().current_snapshot_id())
+            }),
+            ScanSource::Deferred { .. } => None,
+        }
+    }
 }
 
 impl DisplayAs for IcebergScanExec {