diff --git a/src/dat/generated_tables.py b/src/dat/generated_tables.py index 72ce72d0b..52258e76b 100644 --- a/src/dat/generated_tables.py +++ b/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)