Skip to content

[SYSTEMDS-3949] Extend Delta Spark round trip tests #2580

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Baunsgaard merged 1 commit into from
Aug 6, 2026
Merged

[SYSTEMDS-3949] Extend Delta Spark round trip tests #2580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 90 additions & 13 deletions src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkContractTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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
Expand All @@ -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 {
Expand Down Expand Up @@ -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");
Expand All @@ -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);

Expand Down Expand Up @@ -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");
Expand All @@ -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");
Expand All @@ -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;
Expand All @@ -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<Row> 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<Row> 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<Row> data = new ArrayList<>(rows);
for(int r = 0; r < rows; r++)
List<Row> 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);
}
Expand All @@ -252,16 +324,21 @@ private Dataset<Row> 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());
Expand Down
44 changes: 36 additions & 8 deletions src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkInteropTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
* <ul>
* <li>SystemDS writes -&gt; Spark/Delta reads (our output is spec-compliant), and</li>
* <li>SystemDS writes -&gt; Spark/Delta reads (our output is spec-compliant),</li>
* <li>Spark/Delta writes -&gt; SystemDS reads, including a multi-file layout and a table with deletion vectors / a
* second commit that the SystemDS writer never produces itself.</li>
* second commit that the SystemDS writer never produces itself, and</li>
* <li>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.</li>
* </ul>
*
* <p>
Expand Down Expand Up @@ -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<Row> df = indexedDataFrame(rows).repartition(3); // -> multiple data files
Dataset<Row> 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 {
Expand Down Expand Up @@ -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<Integer> expected = idRange(deleteBelow, rows);
Expand All @@ -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<Integer> 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
Expand Down Expand Up @@ -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<Row> indexedDataFrame(int rows) {
/**
* Spark DataFrame mirroring {@link #indexedFrame} with columns c0..c3 (long/double/string/boolean), ids [from,to).
*/
private Dataset<Row> 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<Row> data = new ArrayList<>(rows);
for(int r = 0; r < rows; r++)
List<Row> 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);
}
Expand Down
11 changes: 9 additions & 2 deletions src/test/java/org/apache/sysds/test/component/io/DeltaFrameTestUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Path> 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<Path> s = Files.walk(new File(tablePath, "_delta_log").toPath())) {
return s.filter(p -> p.toString().endsWith(".checkpoint.parquet")).count();
}
}
}
Loading