Feature: Incremental Append Scan

dev/provision.py

@@ -395,3 +395,48 @@
  )
  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: compact -- rewrites the two letter='b' files into one (operation=replace)
+ # 4: delete number=2
+ # 5: append (5, 'd', 100) -- on evolved schema
+ # 6: 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')")
+ # Compact: letter='b' has two files (from the previous two appends); rewrite them into one.
+ # This commits a non-append (replace) snapshot whose rewritten file the append scan must not pick up.
+ spark.sql(
+ f"""
+ CALL {catalog_name}.system.rewrite_data_files(
+ table => 'default.test_incremental_read',
+ options => map('min-input-files', '2')
+ )
+ """
+ )
+ 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
+ """
+ )