feat(datafusion) Add snapshot_id parameter to pin table reads

crates/integrations/datafusion/public-api.txt

@@ -70,6 +70,9 @@ pub fn iceberg_datafusion::IcebergStaticTableProvider::schema(&self) -> arrow_sc
 pub fn iceberg_datafusion::IcebergStaticTableProvider::supports_filters_pushdown(&self, filters: &[&datafusion_expr::expr::Expr]) -> datafusion_common::error::Result<alloc::vec::Vec<datafusion_expr::table_source::TableProviderFilterPushDown>>
 pub fn iceberg_datafusion::IcebergStaticTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType
 pub struct iceberg_datafusion::table::IcebergTableProvider
+impl iceberg_datafusion::IcebergTableProvider
+pub fn iceberg_datafusion::IcebergTableProvider::snapshot_id(&self) -> core::option::Option<i64>
+pub fn iceberg_datafusion::IcebergTableProvider::with_snapshot_id(self, snapshot_id: core::option::Option<i64>) -> Self
 impl core::clone::Clone for iceberg_datafusion::IcebergTableProvider
 pub fn iceberg_datafusion::IcebergTableProvider::clone(&self) -> iceberg_datafusion::IcebergTableProvider
 impl core::fmt::Debug for iceberg_datafusion::IcebergTableProvider
@@ -114,6 +117,9 @@ pub fn iceberg_datafusion::IcebergStaticTableProvider::schema(&self) -> arrow_sc
 pub fn iceberg_datafusion::IcebergStaticTableProvider::supports_filters_pushdown(&self, filters: &[&datafusion_expr::expr::Expr]) -> datafusion_common::error::Result<alloc::vec::Vec<datafusion_expr::table_source::TableProviderFilterPushDown>>
 pub fn iceberg_datafusion::IcebergStaticTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType
 pub struct iceberg_datafusion::IcebergTableProvider
+impl iceberg_datafusion::IcebergTableProvider
+pub fn iceberg_datafusion::IcebergTableProvider::snapshot_id(&self) -> core::option::Option<i64>
+pub fn iceberg_datafusion::IcebergTableProvider::with_snapshot_id(self, snapshot_id: core::option::Option<i64>) -> Self
 impl core::clone::Clone for iceberg_datafusion::IcebergTableProvider
 pub fn iceberg_datafusion::IcebergTableProvider::clone(&self) -> iceberg_datafusion::IcebergTableProvider
 impl core::fmt::Debug for iceberg_datafusion::IcebergTableProvider

crates/integrations/datafusion/src/table/mod.rs

@@ -72,6 +72,10 @@ pub struct IcebergTableProvider {
     table_ident: TableIdent,
     /// A reference-counted arrow `Schema` (cached at construction)
     schema: ArrowSchemaRef,
+    /// Optional snapshot to read. `None` reads the current snapshot (refreshed
+    /// from the catalog on each scan); `Some` pins reads to that snapshot for
+    /// time-travel. Writes always target the current table state.
+    snapshot_id: Option<i64>,
 }
 
 impl IcebergTableProvider {
@@ -94,9 +98,23 @@ impl IcebergTableProvider {
             catalog,
             table_ident,
             schema,
+            snapshot_id: None,
         })
     }
 
+    /// Pins reads to a specific snapshot for time-travel. `None` (the default)
+    /// reads the current snapshot. The snapshot id is threaded into the scan
+    /// node, so it is serialized and honored by a distributed engine as well.
+    pub fn with_snapshot_id(mut self, snapshot_id: Option<i64>) -> Self {
+        self.snapshot_id = snapshot_id;
+        self
+    }
+
+    /// Returns the snapshot this provider reads, if pinned for time-travel.
+    pub fn snapshot_id(&self) -> Option<i64> {
+        self.snapshot_id
+    }
+
     pub(crate) async fn metadata_table(
         &self,
         r#type: MetadataTableType,
@@ -131,10 +149,10 @@ impl TableProvider for IcebergTableProvider {
             .await
             .map_err(to_datafusion_error)?;
 
-        // Create scan with fresh metadata (always use current snapshot)
+        // Create scan with fresh metadata, honoring a pinned snapshot if set.
         Ok(Arc::new(IcebergTableScan::new(
             table,
-            None, // Always use current snapshot for catalog-backed provider
+            self.snapshot_id,
             self.schema.clone(),
             projection,
             filters,
@@ -894,4 +912,132 @@ mod tests {
             "Limit should be None when not specified"
         );
     }
+
+    /// Runs `SELECT * FROM t` against `provider` and returns the result batches.
+    async fn scan_rows(
+        provider: IcebergTableProvider,
+    ) -> Vec<datafusion::arrow::array::RecordBatch> {
+        let ctx = SessionContext::new();
+        ctx.register_table("t", Arc::new(provider)).unwrap();
+        ctx.sql("SELECT * FROM t")
+            .await
+            .unwrap()
+            .collect()
+            .await
+            .unwrap()
+    }
+
+    #[tokio::test]
+    async fn test_pinned_snapshot_reads_historical_data() {
+        use datafusion::assert_batches_sorted_eq;
+
+        let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await;
+
+        // First append -> snapshot with a single row.
+        let writer =
+            IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone())
+                .await
+                .unwrap();
+        let ctx = SessionContext::new();
+        ctx.register_table("t", Arc::new(writer)).unwrap();
+        ctx.sql("INSERT INTO t VALUES (1, 'a')")
+            .await
+            .unwrap()
+            .collect()
+            .await
+            .unwrap();
+
+        // Capture the snapshot produced by the first append.
+        let table_ident = TableIdent::new(namespace.clone(), table_name.clone());
+        let first_snapshot = catalog
+            .load_table(&table_ident)
+            .await
+            .unwrap()
+            .metadata()
+            .current_snapshot()
+            .unwrap()
+            .snapshot_id();
+
+        // Second append -> current snapshot now has two rows.
+        ctx.sql("INSERT INTO t VALUES (2, 'b')")
+            .await
+            .unwrap()
+            .collect()
+            .await
+            .unwrap();
+
+        // Default (unpinned) provider sees the latest state: both rows.
+        let current =
+            IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone())
+                .await
+                .unwrap();
+        assert_batches_sorted_eq!(
+            [
+                "+----+------+",
+                "| id | name |",
+                "+----+------+",
+                "| 1  | a    |",
+                "| 2  | b    |",
+                "+----+------+",
+            ],
+            &scan_rows(current).await
+        );
+
+        // Pinning the first snapshot time-travels: only the first row is visible,
+        // even though a newer snapshot exists.
+        let pinned =
+            IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone())
+                .await
+                .unwrap()
+                .with_snapshot_id(Some(first_snapshot));
+        assert_eq!(pinned.snapshot_id(), Some(first_snapshot));
+
+        // The pin is threaded onto the scan node itself — this is the value a
+        // distributed engine's codec serializes to reproduce the scan remotely.
+        let scan_plan = pinned
+            .scan(&SessionContext::new().state(), None, &[], None)
+            .await
+            .unwrap();
+        let iceberg_scan = scan_plan
+            .downcast_ref::<IcebergTableScan>()
+            .expect("Expected IcebergTableScan");
+        assert_eq!(
+            iceberg_scan.snapshot_id(),
+            Some(first_snapshot),
+            "pinned snapshot should propagate to the scan node"
+        );
+
+        // And it changes what is actually read: only the historical row.
+        assert_batches_sorted_eq!(
+            [
+                "+----+------+",
+                "| id | name |",
+                "+----+------+",
+                "| 1  | a    |",
+                "+----+------+",
+            ],
+            &scan_rows(pinned).await
+        );
+
+        // Clearing the pin (Some -> None) unpins back to the current snapshot,
+        // so the newer row becomes visible again.
+        let unpinned =
+            IcebergTableProvider::try_new(catalog.clone(), namespace.clone(), table_name.clone())
+                .await
+                .unwrap()
+                .with_snapshot_id(Some(first_snapshot))
+                .with_snapshot_id(None);
+        assert_eq!(unpinned.snapshot_id(), None);
+        assert_batches_sorted_eq!(
+            [
+                "+----+------+",
+                "| id | name |",
+                "+----+------+",
+                "| 1  | a    |",
+                "| 2  | b    |",
+                "+----+------+",
+            ],
+            &scan_rows(unpinned).await
+        );
+    }
 }