Feature: Incremental Append Scan

Author: smaheshwar-pltr

dev/provision.py

@@ -395,3 +395,42 @@
     )
     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'")
+
+    # Fixture for IncrementalAppendScan. Partitioning by `letter` makes the file layout
+    # deterministic (one file per partition per insert). The schema is evolved mid-fixture
+    # to exercise scans that span the change, and ends with a REPLACE TABLE to break
+    # snapshot lineage for the disconnected-snapshots test.
+    #
+    # 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 the 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
+            """
+    )