Feature: Incremental Append Scan

Adds IncrementalAppendScan, which accumulates the data appended between
two snapshots. Builds on the BaseScan / ManifestGroupPlanner refactor in
sm/table-scan-refactor; split out of #3364 per reviewer request, reviving
the work from #2235.

- Table.incremental_append_scan(...) builds an IncrementalAppendScan over
  the (from_snapshot_id_exclusive, to_snapshot_id_inclusive] range;
  StagedTable overrides it to raise, mirroring scan().
- Walks the append-only ancestors in the range, dedups the data manifests
  whose added_snapshot_id is in range (Set semantics via ManifestFile
  __eq__/__hash__), and filters manifest entries to ADDED-in-range via a
  new manifest_entry_filter on ManifestGroupPlanner.plan_files.
- Projects onto the table's current schema (not the snapshot schema).
- from_snapshot_id_exclusive is validated with is_parent_ancestor_of, so
  an expired start cursor is accepted as long as the lineage still passes
  through it; equal from/to is rejected. Adds the snapshot helpers
  ancestors_between_ids and is_parent_ancestor_of.
- Unit tests (validation, current-schema projection, type preservation,
  expired-from) and integration tests (append-only, non-append ignored,
  schema evolution within range, partition/metrics pruning, disconnected
  snapshots), plus the test_incremental_read provision fixture.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: smaheshwar-pltr

dev/provision.py

@@ -395,3 +395,37 @@
     )
     spark.sql(f"ALTER TABLE {catalog_name}.default.test_empty_scan_ordered_str WRITE ORDERED BY id")
     spark.sql(f"INSERT INTO {catalog_name}.default.test_empty_scan_ordered_str VALUES 'a', 'c'")
+
+    # Append scan fixture. Snapshots written:
+    #   0: append (1, 'a')
+    #   1: append (2, 'b')
+    #   2: append (3, 'c'), (4, 'b')
+    #   3: delete number=2
+    #   4: append (5, 'd', 100) -- on evolved schema
+    #   5: replace table -- lineage break
+    spark.sql(
+        f"""
+            CREATE OR REPLACE TABLE {catalog_name}.default.test_incremental_read (
+                number integer,
+                letter string
+            )
+            USING iceberg
+            PARTITIONED BY (letter)
+            TBLPROPERTIES ('format-version'='2')
+            """
+    )
+    spark.sql(f"INSERT INTO {catalog_name}.default.test_incremental_read VALUES (1, 'a')")
+    spark.sql(f"INSERT INTO {catalog_name}.default.test_incremental_read VALUES (2, 'b')")
+    spark.sql(f"INSERT INTO {catalog_name}.default.test_incremental_read VALUES (3, 'c'), (4, 'b')")
+    spark.sql(f"DELETE FROM {catalog_name}.default.test_incremental_read WHERE number = 2")
+    spark.sql(f"ALTER TABLE {catalog_name}.default.test_incremental_read ADD COLUMN extra int")
+    spark.sql(f"INSERT INTO {catalog_name}.default.test_incremental_read VALUES (5, 'd', 100)")
+    spark.sql(
+        f"""
+            REPLACE TABLE {catalog_name}.default.test_incremental_read
+            USING iceberg
+            PARTITIONED BY (letter)
+            TBLPROPERTIES ('format-version'='2')
+            AS SELECT number, letter, extra FROM {catalog_name}.default.test_incremental_read
+            """
+    )