From 233edbedede6830bd8e0a13ebb9c5215baaf6ab5 Mon Sep 17 00:00:00 2001 From: Jakob-al28 <04jakob28@gmail.com> Date: Wed, 5 Aug 2026 21:34:09 +0200 Subject: [PATCH] [SYSTEMDS-3949] Extend Delta Spark round trip tests Add three scenarios only an external engine can set up: a log that has rolled into a parquet checkpoint with a delete after it, DATE/TIMESTAMP/DECIMAL columns that have no SystemDS value type and must be rejected on the Delta type and a table SystemDS creates that Spark then appends to. Also stop countParquet from counting checkpoint files as data files. --- .../io/DeltaFrameSparkContractTest.java | 103 +++++++++++++++--- .../io/DeltaFrameSparkInteropTest.java | 44 ++++++-- .../component/io/DeltaFrameTestUtils.java | 11 +- 3 files changed, 135 insertions(+), 23 deletions(-) diff --git a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkContractTest.java b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkContractTest.java index 5687461c851..7ace4b10d0a 100644 --- a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkContractTest.java +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkContractTest.java @@ -22,6 +22,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.io.File; import java.nio.file.Files; @@ -33,7 +34,9 @@ import org.apache.commons.io.FileUtils; import org.apache.sysds.common.Types.ValueType; +import org.apache.sysds.runtime.DMLRuntimeException; import org.apache.sysds.runtime.frame.data.FrameBlock; +import org.apache.sysds.runtime.io.DeltaKernelUtils; import org.apache.sysds.runtime.io.FrameReaderDelta; import org.apache.sysds.runtime.io.FrameReaderDeltaParallel; import org.apache.spark.sql.Dataset; @@ -50,7 +53,7 @@ /** * Regression tests pinning the Delta table layouts that the direct column-API decode path (and its kernel-engine * fallback for deletion vectors and partitioned tables) must honor beyond plain flat reads. Each case is a table layout - * the SystemDS writer never produces itself, so it must be created by the reference engine (Spark/Delta). + * or schema the SystemDS writer never produces itself, so it must be created by the reference engine (Spark/Delta). * * dvFeatureEnabledNoDeleteRead covers a table whose protocol carries the {@code deletionVectors} reader feature but has * no deleted rows - a distinct case from {@link DeltaFrameSparkInteropTest#sparkDeletionVectorsSystemdsRead}, which @@ -62,6 +65,11 @@ * handler must surface it as nulls rather than fail. partitionedTableRead covers partition values, which are not stored * in the data files and must be spliced back in. idColumnMappingRead covers a table using * {@code delta.columnMapping.mode = id}, where columns must be resolvable by parquet field id rather than logical name. + * + * checkpointBackedHistoryRead covers a table whose log has rolled into a parquet checkpoint, with a remove action after + * it, so the snapshot comes from a checkpoint plus trailing commits rather than plain per-commit JSON. + * unsupportedColumnTypeRead covers Delta types with no SystemDS value type. Parquet stores them as types the direct + * decode reads fine, so the rejection has to come from the Delta type before any decoding starts. */ @net.jcip.annotations.NotThreadSafe public class DeltaFrameSparkContractTest { @@ -106,7 +114,7 @@ public void dvFeatureEnabledNoDeleteRead() throws Exception { String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); try { spark.conf().set(DV_DEFAULT, "true"); - indexedDataFrame(rows).write().format("delta").save(tablePath); + indexedDataFrame(0, rows).write().format("delta").save(tablePath); assertFrameMatchesIds(new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), rows, "serial-dvfeat"); @@ -129,7 +137,7 @@ public void schemaEvolutionAddedColumnRead() throws Exception { Path dir = Files.createTempDirectory("sysds_delta_frame_evo_"); String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); try { - indexedDataFrame(oldRows).write().format("delta").save(tablePath); + indexedDataFrame(0, oldRows).write().format("delta").save(tablePath); evolvedDataFrame(oldRows, allRows).write().format("delta").mode("append").option("mergeSchema", "true") .save(tablePath); @@ -168,7 +176,7 @@ public void partitionedTableRead() throws Exception { Path dir = Files.createTempDirectory("sysds_delta_frame_part_"); String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); try { - indexedDataFrame(rows).write().format("delta").partitionBy("c3").save(tablePath); + indexedDataFrame(0, rows).write().format("delta").partitionBy("c3").save(tablePath); assertFrameMatchesIds(new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), rows, "serial-part"); @@ -192,7 +200,7 @@ public void idColumnMappingRead() throws Exception { try { spark.sql("CREATE TABLE delta.`" + tablePath + "` (c0 BIGINT, c1 DOUBLE, c2 STRING, c3 BOOLEAN) " + "USING delta TBLPROPERTIES ('delta.columnMapping.mode'='id')"); - indexedDataFrame(rows).write().format("delta").mode("append").save(tablePath); + indexedDataFrame(0, rows).write().format("delta").mode("append").save(tablePath); assertFrameMatchesIds(new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), rows, "serial-idmap"); @@ -206,6 +214,70 @@ public void idColumnMappingRead() throws Exception { } } + @Test + public void checkpointBackedHistoryRead() throws Exception { + // two appends past delta.checkpointInterval=2 put a parquet checkpoint in + // _delta_log; the delete after it adds a remove action the replay must apply on + // top of the checkpoint + int perCommit = 50, deleteBelow = 30, rows = 2 * perCommit; + Path dir = Files.createTempDirectory("sysds_delta_frame_ckpt_"); + String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); + try { + spark.sql("CREATE TABLE delta.`" + tablePath + "` (c0 BIGINT, c1 DOUBLE, c2 STRING, c3 BOOLEAN) " + + "USING delta TBLPROPERTIES ('delta.checkpointInterval' = '2')"); + indexedDataFrame(0, perCommit).write().format("delta").mode("append").save(tablePath); + indexedDataFrame(perCommit, rows).write().format("delta").mode("append").save(tablePath); + spark.sql("DELETE FROM delta.`" + tablePath + "` WHERE c0 < " + deleteBelow); + assertTrue("expected a checkpoint in _delta_log", DeltaFrameTestUtils.countCheckpoints(tablePath) > 0); + + // add actions replayed out of a checkpoint must still carry the numRecords + // statistic, otherwise the read silently drops to the buffered path + DeltaKernelUtils.ScanHandle handle = DeltaKernelUtils.openScan(DeltaKernelUtils.createEngine(), + DeltaKernelUtils.qualify(tablePath)); + assertTrue("checkpointed log should still expose per-file row counts", handle.hasExactRowCounts()); + + assertFrameMatchesIds(new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), + deleteBelow, rows, "serial-ckpt"); + assertFrameMatchesIds( + new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), deleteBelow, + rows, "parallel-ckpt"); + } + finally { + spark.sql("DROP TABLE IF EXISTS delta.`" + tablePath + "`"); + FileUtils.deleteQuietly(dir.toFile()); + } + } + + @Test + public void unsupportedColumnTypeRead() throws Exception { + // parquet stores these as int32/int64/fixed_len_byte_array, which the direct + // decode would read, so the rejection has to come from the Delta type. + assertUnsupportedColumnRejected("c1 DATE"); + assertUnsupportedColumnRejected("c1 TIMESTAMP"); + assertUnsupportedColumnRejected("c1 DECIMAL(10,2)"); + } + + private void assertUnsupportedColumnRejected(String columnDdl) throws Exception { + Path dir = Files.createTempDirectory("sysds_delta_frame_unsupported_"); + String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); + try { + // the schema alone drives the rejection, so no data has to be written + spark.sql("CREATE TABLE delta.`" + tablePath + "` (c0 BIGINT, " + columnDdl + ") USING delta"); + try { + new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1); + fail("expected read of unsupported column type to fail: " + columnDdl); + } + catch(DMLRuntimeException e) { + assertTrue("message should name the offending column: " + e.getMessage(), + e.getMessage().contains("c1")); + } + } + finally { + spark.sql("DROP TABLE IF EXISTS delta.`" + tablePath + "`"); + FileUtils.deleteQuietly(dir.toFile()); + } + } + // deterministic, exactly-representable cell values keyed by the row id in column 0 private static double dval(int id) { return id * 0.5 - 1.0; @@ -223,15 +295,15 @@ private static String vval(int id) { return "v" + id; } - /** Spark DataFrame with columns c0..c3 (long/double/string/boolean) keyed by the row id in c0. */ - private Dataset indexedDataFrame(int rows) { + /** Spark DataFrame with columns c0..c3 (long/double/string/boolean) keyed by the row id in c0, ids [from,to). */ + private Dataset indexedDataFrame(int from, int to) { StructType schema = DataTypes .createStructType(new StructField[] {DataTypes.createStructField("c0", DataTypes.LongType, false), DataTypes.createStructField("c1", DataTypes.DoubleType, false), DataTypes.createStructField("c2", DataTypes.StringType, false), DataTypes.createStructField("c3", DataTypes.BooleanType, false)}); - List data = new ArrayList<>(rows); - for(int r = 0; r < rows; r++) + List data = new ArrayList<>(to - from); + for(int r = from; r < to; r++) data.add(RowFactory.create((long) r, dval(r), sval(r), bval(r))); return spark.createDataFrame(data, schema); } @@ -252,16 +324,21 @@ private Dataset evolvedDataFrame(int from, int to) { /** Asserts {@code out} holds exactly ids [0,rows) with the exact per-id values in c1..c3. */ private static void assertFrameMatchesIds(FrameBlock out, int rows, String tag) { - assertEquals(tag + " rows", rows, out.getNumRows()); + assertFrameMatchesIds(out, 0, rows, tag); + } + + /** Asserts {@code out} holds exactly ids [fromId,toId) with the exact per-id values in c1..c3. */ + private static void assertFrameMatchesIds(FrameBlock out, int fromId, int toId, String tag) { + assertEquals(tag + " rows", toId - fromId, out.getNumRows()); assertEquals(tag + " cols", 4, out.getNumColumns()); assertEquals(tag + " c0 type", ValueType.INT64, out.getSchema()[0]); assertEquals(tag + " c1 type", ValueType.FP64, out.getSchema()[1]); assertEquals(tag + " c2 type", ValueType.STRING, out.getSchema()[2]); assertEquals(tag + " c3 type", ValueType.BOOLEAN, out.getSchema()[3]); - boolean[] seen = new boolean[rows]; - for(int r = 0; r < rows; r++) { + boolean[] seen = new boolean[toId]; + for(int r = 0; r < out.getNumRows(); r++) { int id = ((Number) out.get(r, 0)).intValue(); - assertTrue(tag + ": unexpected/duplicate id " + id, id >= 0 && id < rows && !seen[id]); + assertTrue(tag + ": unexpected/duplicate id " + id, id >= fromId && id < toId && !seen[id]); seen[id] = true; assertEquals(tag + " id" + id + " c1", dval(id), ((Number) out.get(r, 1)).doubleValue(), 1e-9); assertEquals(tag + " id" + id + " c2", sval(id), out.get(r, 2).toString()); diff --git a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkInteropTest.java b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkInteropTest.java index d17d9ae4005..9bebf0e0bb6 100644 --- a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkInteropTest.java +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkInteropTest.java @@ -59,9 +59,11 @@ * cannot catch a table that SystemDS writes in a way other Delta engines reject (or vice versa). These tests close that * gap by routing a mixed-type (long/double/string/boolean) frame through two independent engines: *
    - *
  • SystemDS writes -> Spark/Delta reads (our output is spec-compliant), and
    • + *
  • SystemDS writes -> Spark/Delta reads (our output is spec-compliant),
    • *
  • Spark/Delta writes -> SystemDS reads, including a multi-file layout and a table with deletion vectors / a - * second commit that the SystemDS writer never produces itself.
    • + * second commit that the SystemDS writer never produces itself, and + *
  • SystemDS creates the table and Spark/Delta appends to it, so a table with a mixed-engine commit history must + * still read back as one table.
    • *
* *

@@ -144,7 +146,7 @@ public void sparkWriteSystemdsReadMultiFile() throws Exception { // the reference Delta engine writes a multi-file mixed-type table; both the // serial and parallel SystemDS frame readers must reconstruct it cell-for-cell. int rows = 600; - Dataset df = indexedDataFrame(rows).repartition(3); // -> multiple data files + Dataset df = indexedDataFrame(0, rows).repartition(3); // -> multiple data files Path dir = Files.createTempDirectory("sysds_delta_frame_p2s_"); String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); try { @@ -176,7 +178,7 @@ public void sparkDeletionVectorsSystemdsRead() throws Exception { // enable deletion vectors for tables created in this block, then delete a // row range so Delta records a DV rather than rewriting the data files. spark.conf().set(DV_DEFAULT, "true"); - indexedDataFrame(rows).write().format("delta").save(tablePath); + indexedDataFrame(0, rows).write().format("delta").save(tablePath); spark.sql("DELETE FROM delta.`" + tablePath + "` WHERE c0 < " + deleteBelow); Set expected = idRange(deleteBelow, rows); @@ -197,6 +199,30 @@ public void sparkDeletionVectorsSystemdsRead() throws Exception { } } + @Test + public void systemdsCreateSparkAppendRead() throws Exception { + // the SystemDS writer always creates from scratch, so Spark has to do the + // appending. both commits have to come back in a single read. + int firstRows = 300, appendRows = 250; + Path dir = Files.createTempDirectory("sysds_delta_frame_s2p_"); + String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); + try { + FrameBlock in = indexedFrame(firstRows); + new FrameWriterDelta().writeFrameToHDFS(in, tablePath, firstRows, in.getNumColumns()); + indexedDataFrame(firstRows, firstRows + appendRows).write().format("delta").mode("append").save(tablePath); + + Set expected = idRange(0, firstRows + appendRows); + assertFrameMatchesIds(new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), + expected, "serial-mixed"); + assertFrameMatchesIds( + new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), expected, + "parallel-mixed"); + } + finally { + FileUtils.deleteQuietly(dir.toFile()); + } + } + private static final String DV_DEFAULT = "spark.databricks.delta.properties.defaults.enableDeletionVectors"; // deterministic, exactly-representable cell values keyed by the row id in column 0 @@ -227,16 +253,18 @@ private static FrameBlock indexedFrame(int rows) { return fb; } - /** Spark DataFrame mirroring {@link #indexedFrame} with columns c0..c3 (long/double/string/boolean). */ - private Dataset indexedDataFrame(int rows) { + /** + * Spark DataFrame mirroring {@link #indexedFrame} with columns c0..c3 (long/double/string/boolean), ids [from,to). + */ + private Dataset indexedDataFrame(int from, int to) { StructType schema = DataTypes .createStructType(new StructField[] {DataTypes.createStructField("c0", DataTypes.LongType, false), DataTypes.createStructField("c1", DataTypes.DoubleType, false), DataTypes.createStructField("c2", DataTypes.StringType, false), DataTypes.createStructField("c3", DataTypes.BooleanType, false)}); - List data = new ArrayList<>(rows); - for(int r = 0; r < rows; r++) + List data = new ArrayList<>(to - from); + for(int r = from; r < to; r++) data.add(RowFactory.create((long) r, dval(r), sval(r), bval(r))); return spark.createDataFrame(data, schema); } diff --git a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameTestUtils.java b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameTestUtils.java index fe025c40eae..122df1e372c 100644 --- a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameTestUtils.java +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameTestUtils.java @@ -31,10 +31,17 @@ private DeltaFrameTestUtils() { // utility class } - /** Count the parquet data files under a Delta table directory. */ + /** Count the parquet data files under a Delta table directory, ignoring checkpoints in the log. */ public static long countParquet(String tablePath) throws Exception { try(Stream s = Files.walk(new File(tablePath).toPath())) { - return s.filter(p -> p.toString().endsWith(".parquet")).count(); + return s.filter(p -> p.toString().endsWith(".parquet") && !p.toString().contains("_delta_log")).count(); + } + } + + /** Count the checkpoint files under a Delta table's {@code _delta_log}. */ + public static long countCheckpoints(String tablePath) throws Exception { + try(Stream s = Files.walk(new File(tablePath, "_delta_log").toPath())) { + return s.filter(p -> p.toString().endsWith(".checkpoint.parquet")).count(); } } }