Add 6 oracle-verified reader test cases from delta-semantics-tester

src/dat/generated_tables.py

@@ -571,3 +571,95 @@ def create_iceberg_compat_v1(case: TestCaseInfo, spark: SparkSession):
     )
     delta_table.upgradeTableProtocol(3, 7)
     df.repartition(1).write.format("delta").mode("append").save(case.delta_root)
+
+
+# --- Delta Semantics Tester: Oracle-verified test cases ---
+# These test cases were generated by the delta-semantics-tester project,
+# which writes Delta tables via deltalake, reads through multiple engines
+# (delta-rs + DuckDB), and verifies against an independent reference oracle.
+
+
+@reference_table(
+    name="semantics_append_read",
+    description="Basic append/read agreement: write 3 rows, verify all engines return the same data.",
+)
+def create_semantics_append_read(case: TestCaseInfo, spark: SparkSession):
+    columns = ["id", "value", "category"]
+    data = [(1, "alpha", "A"), (2, "beta", "B"), (3, "gamma", "A")]
+    df = spark.createDataFrame(data, schema=columns)
+    df.repartition(1).write.format("delta").save(case.delta_root)
+
+
+@reference_table(
+    name="semantics_multi_append",
+    description="Multiple sequential appends: verify all committed data is visible across engines.",
+)
+def create_semantics_multi_append(case: TestCaseInfo, spark: SparkSession):
+    columns = ["id", "value", "category"]
+    data1 = [(1, "first", "X")]
+    data2 = [(2, "second", "Y")]
+    data3 = [(3, "third", "X")]
+    spark.createDataFrame(data1, schema=columns).repartition(1).write.format("delta").save(case.delta_root)
+    save_expected(case)
+    spark.createDataFrame(data2, schema=columns).repartition(1).write.format("delta").mode("append").save(case.delta_root)
+    save_expected(case)
+    spark.createDataFrame(data3, schema=columns).repartition(1).write.format("delta").mode("append").save(case.delta_root)
+    save_expected(case)
+
+
+@reference_table(
+    name="semantics_partition_filter",
+    description="Partitioned table with predicate filter: verify partition pruning correctness.",
+)
+def create_semantics_partition_filter(case: TestCaseInfo, spark: SparkSession):
+    columns = ["id", "value", "category"]
+    data = [(1, "alpha", "A"), (2, "beta", "B"), (3, "gamma", "A"), (4, "delta", "C")]
+    df = spark.createDataFrame(data, schema=columns)
+    df.repartition(1).write.format("delta").partitionBy("category").save(case.delta_root)
+
+
+@reference_table(
+    name="semantics_delete_read",
+    description="Append then delete: verify deleted rows are not visible to any engine.",
+)
+def create_semantics_delete_read(case: TestCaseInfo, spark: SparkSession):
+    columns = ["id", "value", "category"]
+    data = [(1, "keep", "A"), (2, "remove", "A"), (3, "keep", "B")]
+    df = spark.createDataFrame(data, schema=columns)
+    df.repartition(1).write.format("delta").save(case.delta_root)
+    save_expected(case)
+    dt = DeltaTable.forPath(spark, case.delta_root)
+    dt.delete("id = 2")
+    save_expected(case)
+
+
+@reference_table(
+    name="semantics_optimize_invariance",
+    description="OPTIMIZE must not change logical results: metamorphic invariance check.",
+)
+def create_semantics_optimize_invariance(case: TestCaseInfo, spark: SparkSession):
+    columns = ["id", "value", "category"]
+    data1 = [(1, "alpha", "A"), (2, "beta", "A")]
+    data2 = [(3, "gamma", "B")]
+    spark.createDataFrame(data1, schema=columns).write.format("delta").partitionBy("category").save(case.delta_root)
+    save_expected(case)
+    spark.createDataFrame(data2, schema=columns).repartition(1).write.format("delta").mode("append").partitionBy("category").save(case.delta_root)
+    save_expected(case)
+    dt = DeltaTable.forPath(spark, case.delta_root)
+    dt.optimize().executeCompaction()
+    save_expected(case)
+
+
+@reference_table(
+    name="semantics_time_travel",
+    description="Time travel: read at prior version after delete. Verify correct snapshot reconstruction.",
+)
+def create_semantics_time_travel(case: TestCaseInfo, spark: SparkSession):
+    columns = ["id", "value", "category"]
+    data = [(1, "original", "A"), (2, "will_delete", "A")]
+    df = spark.createDataFrame(data, schema=columns)
+    df.repartition(1).write.format("delta").save(case.delta_root)
+    save_expected(case)
+    dt = DeltaTable.forPath(spark, case.delta_root)
+    dt.delete("id = 2")
+    save_expected(case)