fixes

Author: xanderbailey

crates/iceberg/src/scan/context.rs

@@ -242,12 +242,19 @@ impl PlanContext {
         // TODO: Ideally we could ditch this intermediate Vec as we return an iterator.
         let mut filtered_mfcs = vec![];
         for manifest_file in manifest_files {
-            // For incremental scans, skip delete manifests entirely —
-            // we only care about newly added data files.
-            if self.snapshot_range.is_some()
-                && manifest_file.content == ManifestContentType::Deletes
-            {
-                continue;
+            // For incremental scans, skip manifests that can't contain relevant entries:
+            // 1. Delete manifests — we only care about newly added data files.
+            // 2. Data manifests whose added_snapshot_id is outside the scan range —
+            //    they can't contain entries added in the snapshots we care about.
+            //    (We still keep the entry-level filter because a manifest can contain
+            //    entries from multiple snapshots via manifest reuse.)
+            if let Some(ref range) = self.snapshot_range {
+                if manifest_file.content == ManifestContentType::Deletes {
+                    continue;
+                }
+                if !range.contains(manifest_file.added_snapshot_id) {
+                    continue;
+                }
             }
 
             let tx = if manifest_file.content == ManifestContentType::Deletes {

crates/iceberg/src/scan/mod.rs

@@ -42,6 +42,7 @@ use crate::runtime::spawn;
 use crate::spec::{DataContentType, Operation, SnapshotRef, TableMetadataRef};
 use crate::table::Table;
 use crate::util::available_parallelism;
+use crate::util::snapshot::ancestors_between;
 use crate::{Error, ErrorKind, Result};
 
 /// A stream of arrow [`RecordBatch`]es.
@@ -75,46 +76,85 @@ impl SnapshotRange {
         to_snapshot_id: i64,
         from_inclusive: bool,
     ) -> Result<Self> {
-        let mut snapshot_ids = HashSet::new();
-        let mut current_id = Some(to_snapshot_id);
-
-        // Walk backwards from to_snapshot to from_snapshot
-        while let Some(id) = current_id {
-            let snapshot = table_metadata.snapshot_by_id(id).ok_or_else(|| {
-                Error::new(ErrorKind::DataInvalid, format!("Snapshot {id} not found"))
+        // Verify from_snapshot exists and determine the exclusive stop point.
+        let from_snapshot = table_metadata
+            .snapshot_by_id(from_snapshot_id)
+            .ok_or_else(|| {
+                Error::new(
+                    ErrorKind::DataInvalid,
+                    format!("Snapshot {from_snapshot_id} not found"),
+                )
             })?;
 
-            // Validate operation is APPEND
+        // ancestors_between returns (oldest_exclusive, latest_inclusive].
+        // For inclusive mode, stop at from's parent so from itself is included.
+        let oldest_exclusive = if from_inclusive {
+            from_snapshot.parent_snapshot_id()
+        } else {
+            Some(from_snapshot_id)
+        };
+
+        let snapshots: Vec<_> =
+            ancestors_between(table_metadata, to_snapshot_id, oldest_exclusive).collect();
+
+        // ancestors_between silently returns the full chain to root if
+        // oldest_exclusive isn't in the ancestry chain. Detect this:
+        // if we got snapshots but from_snapshot_id wasn't encountered as
+        // the stop point, the chain doesn't connect.
+        if from_snapshot_id == to_snapshot_id {
+            // Edge case: from == to. In exclusive mode, range is empty.
+            // In inclusive mode, we should have exactly one snapshot.
+            if !from_inclusive {
+                return Ok(Self {
+                    snapshot_ids: HashSet::new(),
+                });
+            }
+        } else if snapshots.is_empty() {
+            // to_snapshot_id doesn't exist
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!(
+                    "from_snapshot {from_snapshot_id} is not an ancestor of to_snapshot {to_snapshot_id}",
+                ),
+            ));
+        } else {
+            // Verify the oldest snapshot in our walk is actually connected
+            // to from_snapshot_id. The last snapshot's parent (for exclusive)
+            // or the last snapshot itself (for inclusive) should be from_snapshot_id.
+            let oldest_collected = snapshots.last().unwrap();
+            let connects = if from_inclusive {
+                oldest_collected.snapshot_id() == from_snapshot_id
+            } else {
+                oldest_collected.parent_snapshot_id() == Some(from_snapshot_id)
+            };
+            if !connects {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    format!(
+                        "from_snapshot {from_snapshot_id} is not an ancestor of to_snapshot {to_snapshot_id}",
+                    ),
+                ));
+            }
+        }
+
+        // Validate all snapshots have APPEND operations and collect IDs.
+        let mut snapshot_ids = HashSet::with_capacity(snapshots.len());
+        for snapshot in &snapshots {
             if snapshot.summary().operation != Operation::Append {
                 return Err(Error::new(
                     ErrorKind::FeatureUnsupported,
                     format!(
                         "Incremental scan only supports APPEND operations, \
                          snapshot {} has operation: {:?}",
-                        id,
+                        snapshot.snapshot_id(),
                         snapshot.summary().operation
                     ),
                 ));
             }
-
-            if id == from_snapshot_id {
-                if from_inclusive {
-                    snapshot_ids.insert(id);
-                }
-                return Ok(Self { snapshot_ids });
-            }
-
-            snapshot_ids.insert(id);
-            current_id = snapshot.parent_snapshot_id();
+            snapshot_ids.insert(snapshot.snapshot_id());
         }
 
-        // If we get here, from_snapshot was not found in the ancestry chain
-        Err(Error::new(
-            ErrorKind::DataInvalid,
-            format!(
-                "from_snapshot {from_snapshot_id} is not an ancestor of to_snapshot {to_snapshot_id}",
-            ),
-        ))
+        Ok(Self { snapshot_ids })
     }
 
     /// Check if a snapshot_id is within this range
@@ -2494,7 +2534,10 @@ pub mod tests {
 
         assert!(result.is_err());
         let err = result.unwrap_err();
-        assert!(err.to_string().contains("not an ancestor"));
+        assert!(
+            err.to_string().contains("not found"),
+            "Expected 'not found' error, got: {err}"
+        );
     }
 
     #[test]
@@ -2656,4 +2699,80 @@ pub mod tests {
         assert!(range.contains(3055729675574597004), "S2 should be in range");
         assert!(range.contains(3056729675574597004), "S3 should be in range");
     }
+
+    #[tokio::test]
+    async fn test_incremental_scan_returns_only_added_files_in_range() {
+        // Fixture has S1 (append) -> S2 (append, current)
+        // Manifest contains:
+        //   1.parquet: status=Added, snapshot=S2
+        //   2.parquet: status=Deleted, snapshot=S1
+        //   3.parquet: status=Existing, snapshot=S1
+        let mut fixture = TableTestFixture::new();
+        fixture.setup_manifest_files().await;
+
+        let current_snapshot = fixture.table.metadata().current_snapshot().unwrap();
+        let parent_snapshot_id = current_snapshot.parent_snapshot_id().unwrap();
+
+        // Incremental scan from S1 (exclusive) to S2 should return only 1.parquet
+        let table_scan = fixture
+            .table
+            .incremental_append_scan(parent_snapshot_id)
+            .to_snapshot(current_snapshot.snapshot_id())
+            .build()
+            .unwrap();
+
+        let tasks: Vec<_> = table_scan
+            .plan_files()
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+
+        assert_eq!(
+            tasks.len(),
+            1,
+            "Incremental scan should return exactly 1 file"
+        );
+        assert_eq!(
+            tasks[0].data_file_path,
+            format!("{}/1.parquet", &fixture.table_location),
+            "Should only return the file added in S2"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_incremental_scan_exclusive_same_snapshot_returns_empty() {
+        // Fixture has S1 (append) -> S2 (append, current)
+        let mut fixture = TableTestFixture::new();
+        fixture.setup_manifest_files().await;
+
+        let current_snapshot_id = fixture
+            .table
+            .metadata()
+            .current_snapshot()
+            .unwrap()
+            .snapshot_id();
+
+        // Incremental scan from S2 to S2 (exclusive) should return nothing
+        let table_scan = fixture
+            .table
+            .incremental_append_scan(current_snapshot_id)
+            .to_snapshot(current_snapshot_id)
+            .build()
+            .unwrap();
+
+        let tasks: Vec<_> = table_scan
+            .plan_files()
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+
+        assert!(
+            tasks.is_empty(),
+            "Exclusive scan from=to should return no files"
+        );
+    }
 }

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

@@ -1004,30 +1004,27 @@ mod tests {
         let table = get_test_table_from_metadata_file().await;
         let snapshots: Vec<_> = table.metadata().snapshots().collect();
 
-        // Need at least 2 snapshots for incremental scan
         assert!(snapshots.len() >= 2);
         let from_id = snapshots[0].snapshot_id();
         let to_id = snapshots[snapshots.len() - 1].snapshot_id();
 
         let provider =
-            IcebergStaticTableProvider::try_new_incremental(table.clone(), from_id, to_id).await;
-
-        // May fail due to non-APPEND operations in test data, that's OK
-        if let Ok(provider) = provider {
-            let ctx = SessionContext::new();
-            let state = ctx.state();
-
-            let scan_plan = provider.scan(&state, None, &[], None).await.unwrap();
-            let iceberg_scan = scan_plan
-                .as_any()
-                .downcast_ref::<IcebergTableScan>()
-                .expect("Expected IcebergTableScan");
-
-            // Verify incremental scan parameters are set
-            assert_eq!(iceberg_scan.from_snapshot_id(), Some(from_id));
-            assert!(!iceberg_scan.from_snapshot_inclusive());
-            assert_eq!(iceberg_scan.snapshot_id(), Some(to_id));
-        }
+            IcebergStaticTableProvider::try_new_incremental(table.clone(), from_id, to_id)
+                .await
+                .unwrap();
+
+        let ctx = SessionContext::new();
+        let state = ctx.state();
+
+        let scan_plan = provider.scan(&state, None, &[], None).await.unwrap();
+        let iceberg_scan = scan_plan
+            .as_any()
+            .downcast_ref::<IcebergTableScan>()
+            .expect("Expected IcebergTableScan");
+
+        assert_eq!(iceberg_scan.from_snapshot_id(), Some(from_id));
+        assert!(!iceberg_scan.from_snapshot_inclusive());
+        assert_eq!(iceberg_scan.snapshot_id(), Some(to_id));
     }
 
     #[tokio::test]
@@ -1046,22 +1043,20 @@ mod tests {
             from_id,
             to_id,
         )
-        .await;
+        .await
+        .unwrap();
 
-        if let Ok(provider) = provider {
-            let ctx = SessionContext::new();
-            let state = ctx.state();
+        let ctx = SessionContext::new();
+        let state = ctx.state();
 
-            let scan_plan = provider.scan(&state, None, &[], None).await.unwrap();
-            let iceberg_scan = scan_plan
-                .as_any()
-                .downcast_ref::<IcebergTableScan>()
-                .expect("Expected IcebergTableScan");
+        let scan_plan = provider.scan(&state, None, &[], None).await.unwrap();
+        let iceberg_scan = scan_plan
+            .as_any()
+            .downcast_ref::<IcebergTableScan>()
+            .expect("Expected IcebergTableScan");
 
-            // Verify inclusive flag is set
-            assert_eq!(iceberg_scan.from_snapshot_id(), Some(from_id));
-            assert!(iceberg_scan.from_snapshot_inclusive());
-        }
+        assert_eq!(iceberg_scan.from_snapshot_id(), Some(from_id));
+        assert!(iceberg_scan.from_snapshot_inclusive());
     }
 
     #[tokio::test]
@@ -1074,25 +1069,22 @@ mod tests {
         assert!(!snapshots.is_empty());
         let from_id = snapshots[0].snapshot_id();
 
-        let provider =
-            IcebergStaticTableProvider::try_new_appends_after(table.clone(), from_id).await;
-
-        if let Ok(provider) = provider {
-            let ctx = SessionContext::new();
-            let state = ctx.state();
-
-            let scan_plan = provider.scan(&state, None, &[], None).await.unwrap();
-            let iceberg_scan = scan_plan
-                .as_any()
-                .downcast_ref::<IcebergTableScan>()
-                .expect("Expected IcebergTableScan");
-
-            // Verify appends_after configuration
-            assert_eq!(iceberg_scan.from_snapshot_id(), Some(from_id));
-            assert!(!iceberg_scan.from_snapshot_inclusive());
-            // snapshot_id should be None (uses current)
-            assert_eq!(iceberg_scan.snapshot_id(), None);
-        }
+        let provider = IcebergStaticTableProvider::try_new_appends_after(table.clone(), from_id)
+            .await
+            .unwrap();
+
+        let ctx = SessionContext::new();
+        let state = ctx.state();
+
+        let scan_plan = provider.scan(&state, None, &[], None).await.unwrap();
+        let iceberg_scan = scan_plan
+            .as_any()
+            .downcast_ref::<IcebergTableScan>()
+            .expect("Expected IcebergTableScan");
+
+        assert_eq!(iceberg_scan.from_snapshot_id(), Some(from_id));
+        assert!(!iceberg_scan.from_snapshot_inclusive());
+        assert_eq!(iceberg_scan.snapshot_id(), None);
     }
 
     #[tokio::test]