diff --git a/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java b/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java index bbca857a1cd..f470f07cb07 100644 --- a/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java +++ b/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java @@ -22,7 +22,9 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.function.Function; @@ -30,6 +32,16 @@ import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.ColumnReader; +import org.apache.parquet.column.impl.ColumnReadStoreImpl; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.io.api.Converter; +import org.apache.parquet.io.api.GroupConverter; +import org.apache.parquet.io.api.PrimitiveConverter; +import org.apache.parquet.schema.MessageType; import org.apache.sysds.conf.ConfigurationManager; import org.apache.sysds.hops.OptimizerUtils; import org.apache.sysds.runtime.DMLRuntimeException; @@ -162,6 +174,199 @@ public static int countSelected(int size, boolean[] selected) { return n; } + // ------------------------------------------ + // direct parquet decode of Delta data files + // ------------------------------------------ + + /** Physical-schema metadata key carrying the parquet field id (column mapping mode {@code id}). */ + private static final String PARQUET_FIELD_ID_KEY = "parquet.field.id"; + + /** + * Whether data files can be decoded directly into pre-allocated output columns: the physical read schema must be a + * positional 1:1 image of the logical schema, i.e. no partition columns (not stored in the data files, spliced back + * in by the kernel) and no kernel metadata columns such as {@code row_index} (only requested for deletion-vector + * reads). Deletion vectors themselves are excluded separately via the exact-row-count check. + * + * @param logicalSchema the table's logical schema + * @param physicalSchema the physical read schema from the scan state + * @return true if data files can be decoded directly + */ + public static boolean supportsDirectDecode(StructType logicalSchema, StructType physicalSchema) { + if(physicalSchema.length() != logicalSchema.length()) + return false; + for(int c = 0; c < physicalSchema.length(); c++) + if(physicalSchema.at(c).isMetadataColumn()) + return false; + return true; + } + + /** @param scanFileRow a scan-file row @return the fully-qualified path of its data file */ + public static String dataFilePath(Row scanFileRow) { + return InternalScanFileUtils.getAddFileStatus(scanFileRow).getPath(); + } + + /** + * Decode one Delta data file into pre-allocated typed column arrays at the given absolute row offset, through + * parquet-mr's column API ({@link ColumnReadStoreImpl}/{@link ColumnReader}) with no kernel engine or intermediate + * batch vectors in the path. Columns are resolved by parquet field id first (column mapping mode {@code id}) and + * physical name second; columns absent from the file (schema evolution) keep the array defaults (0 for numerics, + * null for strings), matching the kernel-path null semantics. + * + * @param filePath fully-qualified path of the parquet data file + * @param physicalSchema physical read schema (positionally 1:1 with the output columns) + * @param readCodes per-column type codes (see the {@code T_*} constants) + * @param dest pre-allocated per-column backing arrays + * @param destOff absolute row offset of this file's first row + * @param limit exclusive upper row bound of this file's slice + * @param tablePath table path for error messages + * @return the number of rows decoded + * @throws IOException on read failure + */ + public static int decodeDataFileInto(String filePath, StructType physicalSchema, int[] readCodes, Object[] dest, + int destOff, int limit, String tablePath) throws IOException { + final Configuration conf = ConfigurationManager.getCachedJobConf(); + final int ncol = physicalSchema.length(); + int off = destOff; + try(ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(new Path(filePath), conf))) { + MessageType parquetSchema = reader.getFooter().getFileMetaData().getSchema(); + String createdBy = reader.getFooter().getFileMetaData().getCreatedBy(); + String[] colNames = resolveParquetColumns(physicalSchema, parquetSchema); + GroupConverter root = dummyConverter(parquetSchema.getFieldCount()); + PageReadStore pages; + while((pages = reader.readNextRowGroup()) != null) { + int nrow = (int) pages.getRowCount(); + // guard before decoding: writing past the limit would overflow into the + // next file's slice (or off the array) in the pre-allocated output + if(off + nrow > limit) + throw new DMLRuntimeException("Delta file produced more rows than its " + + "numRecords statistic; refusing direct read of " + tablePath); + ColumnReadStoreImpl store = new ColumnReadStoreImpl(pages, root, parquetSchema, createdBy); + for(int c = 0; c < ncol; c++) { + if(colNames[c] == null) + continue; // column absent from this data file -> keep defaults (nulls) + ColumnDescriptor desc = parquetSchema.getColumnDescription(new String[] {colNames[c]}); + decodeColumnInto(store.getColumnReader(desc), desc.getMaxDefinitionLevel(), nrow, readCodes[c], + dest[c], off); + } + off += nrow; + } + } + return off - destOff; + } + + /** + * Resolve each physical-schema column to the parquet column name of the given file: by parquet field id when the + * schema carries one, by name otherwise, or null when the file does not contain the column at all. + */ + private static String[] resolveParquetColumns(StructType schema, MessageType parquetSchema) { + Map idToName = new HashMap<>(); + Map names = new HashMap<>(); + for(int i = 0; i < parquetSchema.getFieldCount(); i++) { + org.apache.parquet.schema.Type t = parquetSchema.getType(i); + names.put(t.getName(), t.getName()); + if(t.getId() != null) + idToName.put(t.getId().intValue(), t.getName()); + } + String[] resolved = new String[schema.length()]; + for(int c = 0; c < schema.length(); c++) { + Object fid = schema.at(c).getMetadata().get(PARQUET_FIELD_ID_KEY); + String byId = (fid instanceof Number) ? idToName.get(((Number) fid).intValue()) : null; + resolved[c] = (byId != null) ? byId : names.get(schema.at(c).getName()); + } + return resolved; + } + + /** No-op converter tree; the column API requires one, but values are pulled via the typed getters. */ + private static GroupConverter dummyConverter(int nFields) { + final PrimitiveConverter[] leaves = new PrimitiveConverter[nFields]; + for(int i = 0; i < nFields; i++) + leaves[i] = new PrimitiveConverter() { + }; + return new GroupConverter() { + @Override + public Converter getConverter(int fieldIndex) { + return leaves[fieldIndex]; + } + + @Override + public void start() { + } + + @Override + public void end() { + } + }; + } + + /** + * Decode one parquet column of the current row group into a pre-allocated typed array at the given offset. Null + * cells (definition level below max) keep the array default (0 for numerics, null for strings). + */ + private static void decodeColumnInto(ColumnReader creader, int maxDef, int nrow, int readCode, Object dest, + int off) { + switch(readCode) { + case T_DOUBLE: { + double[] a = (double[]) dest; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[off + r] = creader.getDouble(); + creader.consume(); + } + break; + } + case T_FLOAT: { + float[] a = (float[]) dest; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[off + r] = creader.getFloat(); + creader.consume(); + } + break; + } + case T_LONG: { + long[] a = (long[]) dest; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[off + r] = creader.getLong(); + creader.consume(); + } + break; + } + case T_INT: + case T_SHORT: + case T_BYTE: { + // delta short/byte columns are stored as annotated parquet INT32 + int[] a = (int[]) dest; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[off + r] = creader.getInteger(); + creader.consume(); + } + break; + } + case T_BOOLEAN: { + boolean[] a = (boolean[]) dest; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[off + r] = creader.getBoolean(); + creader.consume(); + } + break; + } + case T_STRING: { + String[] a = (String[]) dest; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[off + r] = creader.getBinary().toStringUsingUTF8(); + creader.consume(); + } + break; + } + default: + throw new DMLRuntimeException("Unsupported read code for direct decode: " + readCode); + } + } + /** Floor on the adaptive writer target file size. Below this the per-file metadata/open * overhead (and tiny-file proliferation) outweighs the extra read parallelism. */ public static final long ADAPTIVE_WRITER_MIN_FILE_SIZE = 4L * 1024 * 1024; diff --git a/src/main/java/org/apache/sysds/runtime/io/FrameReaderDelta.java b/src/main/java/org/apache/sysds/runtime/io/FrameReaderDelta.java index 9e8823f7ecf..a87d49448f5 100644 --- a/src/main/java/org/apache/sysds/runtime/io/FrameReaderDelta.java +++ b/src/main/java/org/apache/sysds/runtime/io/FrameReaderDelta.java @@ -36,9 +36,10 @@ /** * Single-threaded native Delta Lake reader for frames, built on the Spark-free Delta Kernel library. It opens the - * latest snapshot of a Delta table, reads its parquet data files through the kernel's default engine (honoring deletion - * vectors), and materializes the columns into a {@link FrameBlock} whose schema and column names are derived from the - * Delta table schema. + * latest snapshot of a Delta table through the kernel (log replay, schema, data-file listing) and decodes the parquet + * data files directly into pre-allocated columns via parquet-mr's column API; tables with deletion vectors, partition + * columns or missing row statistics are read through the kernel's default engine instead (which applies deletion + * vectors and splices partition values). Schema and column names are derived from the Delta table schema. * *

* Data is extracted column-at-a-time into primitive arrays (no per-cell boxing or {@code FrameBlock.set} dispatch) and @@ -92,7 +93,7 @@ protected FrameBlock readWithHandle(String fname, Engine engine, DeltaKernelUtil if(total == 0) return new FrameBlock(plan.vt, plan.cnames, 0); if(total <= Integer.MAX_VALUE) - return readDirect(fname, engine, handle, plan, (int) total); + return readDirect(fname, handle, plan, (int) total); } // fallback: row counts unknown or deletion vectors present -> decode into @@ -135,24 +136,27 @@ protected static ReadPlan planColumns(DeltaKernelUtils.ScanHandle handle) { } /** - * Whether the metadata-driven direct read fast path can be used for this table (exact per-file row counts and no - * deletion vectors, so the output can be pre-sized and each file decoded straight into its row offset). Visible for - * testing: the buffered fallback is otherwise only reachable for tables lacking row statistics or carrying deletion - * vectors, which the SystemDS Delta writer never produces. + * Whether the metadata-driven direct read fast path can be used for this table: exact per-file row counts (no + * deletion vectors), so the output can be pre-sized, and a physical read schema that maps 1:1 onto the output + * columns (no partition columns or kernel metadata columns to splice back in), so each data file can be decoded + * straight into its row offset without the kernel engine. The buffered kernel-path fallback covers everything else + * (deletion vectors, missing statistics, partitioned tables). * * @param handle the opened scan handle * @return true if the direct path is applicable */ protected boolean useDirectPath(DeltaKernelUtils.ScanHandle handle) { - return handle.hasExactRowCounts(); + return handle.hasExactRowCounts() && + DeltaKernelUtils.supportsDirectDecode(handle.schema, handle.physicalReadSchema); } /** - * Fast path: decode each data file straight into pre-sized typed column arrays at a metadata-derived row offset. - * One allocation per column, single pass, no intermediate per-batch buffers or serial concatenation. + * Fast path: decode each data file straight into pre-sized typed column arrays at a metadata-derived row offset, + * through parquet-mr's column API with no kernel engine or intermediate batch vectors in the path. One allocation + * per column, single pass, no per-batch buffers or serial concatenation. */ - private FrameBlock readDirect(String fname, Engine engine, DeltaKernelUtils.ScanHandle handle, ReadPlan plan, - int nrow) throws IOException { + private FrameBlock readDirect(String fname, DeltaKernelUtils.ScanHandle handle, ReadPlan plan, int nrow) + throws IOException { final int ncol = plan.ncol; final int[] readCodes = plan.readCodes; final Object[] dest = new Object[ncol]; @@ -161,27 +165,18 @@ private FrameBlock readDirect(String fname, Engine engine, DeltaKernelUtils.Scan int base = 0; for(int i = 0; i < handle.scanFiles.size(); i++) { - // exclusive upper row bound for this file's slice; a file decoding more - // rows than its numRecords statistic would otherwise overflow into the - // next file's region or off the array + // exclusive upper row bound for this file's slice (enforced inside the + // decode before each row group is written) final int limit = base + (int) handle.numRecords[i]; - final int[] cur = new int[] {base}; - DeltaKernelUtils.readScanFile(engine, handle.scanState, handle.physicalReadSchema, handle.scanFiles.get(i), - (cols, size, selected) -> { - int n = DeltaKernelUtils.countSelected(size, selected); - if(cur[0] + n > limit) - throw new DMLRuntimeException("Delta file produced more rows than its " - + "numRecords statistic; refusing direct read of " + fname); - for(int c = 0; c < ncol; c++) - extractColumnInto(cols[c], size, selected, readCodes[c], dest[c], cur[0]); - cur[0] += n; - }); - // also fail loud on underflow: a file decoding fewer rows than its - // numRecords statistic would leave the tail of the slice at the array - // default (0/null) while nrow still reports the (inflated) statistic. - if(cur[0] != limit) - throw new DMLRuntimeException("Delta file produced " + (cur[0] - base) + " rows, expected " - + (limit - base) + " from its numRecords statistic; refusing direct read of " + fname); + String path = DeltaKernelUtils.dataFilePath(handle.scanFiles.get(i)); + int n = DeltaKernelUtils.decodeDataFileInto(path, handle.physicalReadSchema, readCodes, dest, base, limit, + fname); + // fail loud on underflow: a file decoding fewer rows than its numRecords + // statistic would leave the tail of the slice at the array default (0/null) + // while nrow still reports the (inflated) statistic. + if(base + n != limit) + throw new DMLRuntimeException("Delta file produced " + n + " rows, expected " + (limit - base) + + " from its numRecords statistic; refusing direct read of " + fname); base = limit; } diff --git a/src/main/java/org/apache/sysds/runtime/io/FrameReaderDeltaParallel.java b/src/main/java/org/apache/sysds/runtime/io/FrameReaderDeltaParallel.java index 106264afe6c..201a4db9dd1 100644 --- a/src/main/java/org/apache/sysds/runtime/io/FrameReaderDeltaParallel.java +++ b/src/main/java/org/apache/sysds/runtime/io/FrameReaderDeltaParallel.java @@ -89,7 +89,8 @@ public FrameBlock readFrameFromHDFS(String fname, ValueType[] schema, String[] n /** * Fast path: each thread decodes one data file straight into the final typed column arrays at a metadata-derived - * row offset. Single allocation per column, fully parallel. + * row offset, through parquet-mr's column API with no kernel engine in the path (and hence no per-file engine + * creation). Single allocation per column, fully parallel. */ private FrameBlock readDirect(String fname, DeltaKernelUtils.ScanHandle handle, ReadPlan plan, int nrow) throws IOException { @@ -112,28 +113,19 @@ private FrameBlock readDirect(String fname, DeltaKernelUtils.ScanHandle handle, for(int i = 0; i < nfiles; i++) { final Row scanFileRow = handle.scanFiles.get(i); final int base = rowOffset[i]; - // exclusive upper row bound for this file's slice; a file decoding more - // rows than its numRecords statistic would otherwise overflow into the - // next file's region (concurrent overlapping writes) or off the array + // exclusive upper row bound for this file's slice; enforced inside the + // decode before each row group is written, so a file with more rows than + // its numRecords statistic cannot overflow into the next file's region final int limit = base + (int) handle.numRecords[i]; tasks.add(() -> { - int[] cur = new int[] {base}; - Engine eng = DeltaKernelUtils.createEngine(); - DeltaKernelUtils.readScanFile(eng, handle.scanState, handle.physicalReadSchema, scanFileRow, - (cols, size, selected) -> { - int n = DeltaKernelUtils.countSelected(size, selected); - if(cur[0] + n > limit) - throw new DMLRuntimeException("Delta file produced more rows than its " - + "numRecords statistic; refusing parallel direct read of " + fname); - for(int c = 0; c < ncol; c++) - extractColumnInto(cols[c], size, selected, readCodes[c], dest[c], cur[0]); - cur[0] += n; - }); + String path = DeltaKernelUtils.dataFilePath(scanFileRow); + int n = DeltaKernelUtils.decodeDataFileInto(path, handle.physicalReadSchema, readCodes, dest, base, + limit, fname); // fail loud on underflow too: fewer decoded rows than the statistic // would leave this slice's tail at the array default (0/null). - if(cur[0] != limit) - throw new DMLRuntimeException("Delta file produced " + (cur[0] - base) + " rows, expected " - + (limit - base) + " from its numRecords statistic; refusing parallel direct read of " + fname); + if(base + n != limit) + throw new DMLRuntimeException("Delta file produced " + n + " rows, expected " + (limit - base) + + " from its numRecords statistic; refusing parallel direct read of " + fname); return null; }); } 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 new file mode 100644 index 00000000000..5687461c851 --- /dev/null +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkContractTest.java @@ -0,0 +1,271 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.sysds.test.component.io; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import org.apache.commons.io.FileUtils; +import org.apache.sysds.common.Types.ValueType; +import org.apache.sysds.runtime.frame.data.FrameBlock; +import org.apache.sysds.runtime.io.FrameReaderDelta; +import org.apache.sysds.runtime.io.FrameReaderDeltaParallel; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * 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). + * + * 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 + * covers row filtering once rows are actually deleted. Here the kernel still appends the {@code _metadata.row_index} + * metadata column to every read once the feature is enabled, which does not exist in the data files and must not be + * requested from them. + * + * schemaEvolutionAddedColumnRead covers a column added after the first commit, so older data files lack it and the + * 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. + */ +@net.jcip.annotations.NotThreadSafe +public class DeltaFrameSparkContractTest { + + // nonsense schema/dims handed to the reader to confirm it discovers everything from the table + private static final ValueType[] NO_SCHEMA = new ValueType[] {ValueType.STRING}; + private static final String[] NO_NAMES = new String[] {"x"}; + + private static final String DV_DEFAULT = "spark.databricks.delta.properties.defaults.enableDeletionVectors"; + + private static SparkSession spark; + + @BeforeClass + public static void startSpark() { + // each test class runs in its own fork (surefire reuseForks=false), so this + // is the only SparkSession in the JVM and gets the Delta extensions injected. + SparkSession.clearActiveSession(); + SparkSession.clearDefaultSession(); + spark = SparkSession.builder().appName("sysds-delta-frame-contract").master("local[2]") + .config("spark.ui.enabled", "false").config("spark.sql.shuffle.partitions", "2") + .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") + .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog").getOrCreate(); + } + + @AfterClass + public static void stopSpark() { + if(spark != null) + spark.stop(); + SparkSession.clearActiveSession(); + SparkSession.clearDefaultSession(); + spark = null; + } + + @Test + public void dvFeatureEnabledNoDeleteRead() throws Exception { + // enabling deletion vectors adds the feature to the table protocol, which + // makes the kernel append the _metadata.row_index metadata column to the physical + // read schema of every read, no row has to be deleted. The parquet handler must + // populate that column (it is not stored in the data files) instead of failing. + int rows = 500; + Path dir = Files.createTempDirectory("sysds_delta_frame_dvfeat_"); + String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); + try { + spark.conf().set(DV_DEFAULT, "true"); + indexedDataFrame(rows).write().format("delta").save(tablePath); + + assertFrameMatchesIds(new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), + rows, "serial-dvfeat"); + assertFrameMatchesIds( + new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), rows, + "parallel-dvfeat"); + } + finally { + spark.conf().unset(DV_DEFAULT); + FileUtils.deleteQuietly(dir.toFile()); + } + } + + @Test + public void schemaEvolutionAddedColumnRead() throws Exception { + // column c4 is added by the second commit, so the data files of the first commit + // do not contain it; the parquet handler must return nulls for it there (the + // kernel hands every file the same table-level physical read schema). + int oldRows = 200, allRows = 300; + Path dir = Files.createTempDirectory("sysds_delta_frame_evo_"); + String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); + try { + indexedDataFrame(oldRows).write().format("delta").save(tablePath); + evolvedDataFrame(oldRows, allRows).write().format("delta").mode("append").option("mergeSchema", "true") + .save(tablePath); + + for(FrameBlock out : new FrameBlock[] { + new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), + new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)}) { + assertEquals("rows", allRows, out.getNumRows()); + assertEquals("cols", 5, out.getNumColumns()); + assertEquals("c4 type", ValueType.STRING, out.getSchema()[4]); + Set seen = new HashSet<>(); + for(int r = 0; r < out.getNumRows(); r++) { + int id = ((Number) out.get(r, 0)).intValue(); + assertTrue("unexpected/duplicate id " + id, id >= 0 && id < allRows && seen.add(id)); + assertEquals("id" + id + " c1", dval(id), ((Number) out.get(r, 1)).doubleValue(), 1e-9); + assertEquals("id" + id + " c2", sval(id), out.get(r, 2).toString()); + Object c4 = out.get(r, 4); + if(id < oldRows) + assertNull("id" + id + " c4 must be null (file predates the column)", c4); + else + assertEquals("id" + id + " c4", vval(id), c4.toString()); + } + assertEquals(allRows, seen.size()); + } + } + finally { + FileUtils.deleteQuietly(dir.toFile()); + } + } + + @Test + public void partitionedTableRead() throws Exception { + // partition values are not stored in the data files; the kernel splices them back + // into every batch (ColumnarBatch.withNewColumn), so this pins the batch-reshaping + // side of the contract that plain unpartitioned round-trips never touch. + int rows = 300; + 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); + + assertFrameMatchesIds(new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), + rows, "serial-part"); + assertFrameMatchesIds( + new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), rows, + "parallel-part"); + } + finally { + FileUtils.deleteQuietly(dir.toFile()); + } + } + + @Test + public void idColumnMappingRead() throws Exception { + // with delta.columnMapping.mode=id the parquet columns carry field ids and + // physical names; the handler must resolve columns through the + // mapped physical schema, preferring field ids per the ParquetHandler contract. + int rows = 400; + Path dir = Files.createTempDirectory("sysds_delta_frame_idmap_"); + 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.columnMapping.mode'='id')"); + indexedDataFrame(rows).write().format("delta").mode("append").save(tablePath); + + assertFrameMatchesIds(new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), + rows, "serial-idmap"); + assertFrameMatchesIds( + new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), rows, + "parallel-idmap"); + } + 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; + } + + private static String sval(int id) { + return "s" + id; + } + + private static boolean bval(int id) { + return id % 2 == 0; + } + + 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) { + 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++) + data.add(RowFactory.create((long) r, dval(r), sval(r), bval(r))); + return spark.createDataFrame(data, schema); + } + + /** Like {@link #indexedDataFrame} for ids [from,to) but with an additional string column c4. */ + private Dataset evolvedDataFrame(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), + DataTypes.createStructField("c4", DataTypes.StringType, true)}); + 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), vval(r))); + return spark.createDataFrame(data, schema); + } + + /** 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()); + 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++) { + int id = ((Number) out.get(r, 0)).intValue(); + assertTrue(tag + ": unexpected/duplicate id " + id, id >= 0 && id < rows && !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()); + assertEquals(tag + " id" + id + " c3", Boolean.valueOf(bval(id)), out.get(r, 3)); + } + } +}