From f9925e86025864893ae997411e93e4abd86e9f3f Mon Sep 17 00:00:00 2001 From: Jakob-al28 Date: Mon, 6 Jul 2026 19:54:30 +0200 Subject: [PATCH 1/6] [SYSTEMDS-3949] Column API parquet decode for Delta frame reads --- .../sysds/conf/ConfigurationManager.java | 5 + .../java/org/apache/sysds/conf/DMLConfig.java | 2 + .../runtime/io/ColumnApiDeltaEngine.java | 585 ++++++++++++++++++ .../sysds/runtime/io/DeltaKernelUtils.java | 7 +- .../io/DeltaFrameSparkContractTest.java | 273 ++++++++ 5 files changed, 871 insertions(+), 1 deletion(-) create mode 100644 src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java create mode 100644 src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkContractTest.java diff --git a/src/main/java/org/apache/sysds/conf/ConfigurationManager.java b/src/main/java/org/apache/sysds/conf/ConfigurationManager.java index 8b0f5fe06b9..709e806c0ae 100644 --- a/src/main/java/org/apache/sysds/conf/ConfigurationManager.java +++ b/src/main/java/org/apache/sysds/conf/ConfigurationManager.java @@ -263,6 +263,11 @@ public static int getDeltaReaderBatchSize() { return getDMLConfig().getIntValue(DMLConfig.DELTA_READER_BATCH_SIZE); } + /** @return whether the native Delta reader decodes data files via parquet-mr's column API */ + public static boolean isDeltaReaderColumnApi() { + return getDMLConfig().getBooleanValue(DMLConfig.DELTA_READER_COLUMN_API); + } + /** @return matrix rows materialized per columnar batch for the native Delta writer */ public static int getDeltaWriterBatchSize() { return getDMLConfig().getIntValue(DMLConfig.DELTA_WRITER_BATCH_SIZE); diff --git a/src/main/java/org/apache/sysds/conf/DMLConfig.java b/src/main/java/org/apache/sysds/conf/DMLConfig.java index d114ccf69b9..12594a88916 100644 --- a/src/main/java/org/apache/sysds/conf/DMLConfig.java +++ b/src/main/java/org/apache/sysds/conf/DMLConfig.java @@ -72,6 +72,7 @@ public class DMLConfig public static final String CP_PARALLEL_IO = "sysds.cp.parallel.io"; public static final String IO_COMPRESSION_CODEC = "sysds.io.compression.encoding"; public static final String DELTA_READER_BATCH_SIZE = "sysds.io.delta.reader.batchsize"; // int: rows per parquet read batch + public static final String DELTA_READER_COLUMN_API = "sysds.io.delta.reader.columnapi"; // boolean: decode data files via parquet-mr's column API instead of the kernel default row-record reader public static final String DELTA_WRITER_BATCH_SIZE = "sysds.io.delta.writer.batchsize"; // int: matrix rows materialized per columnar batch handed to the engine public static final String DELTA_WRITER_TARGET_FILE_SIZE = "sysds.io.delta.writer.targetfilesize"; // long: upper bound on target data-file size in bytes; adaptive sizing may pick smaller -> more files -> more parallel-read throughput public static final String DELTA_WRITER_ADAPTIVE_FILE_SIZE = "sysds.io.delta.writer.adaptivefilesize"; // boolean: size data files toward one per parallel reader (capped by targetfilesize) @@ -163,6 +164,7 @@ public class DMLConfig _defaultVals.put(CP_PARALLEL_IO, "true" ); _defaultVals.put(IO_COMPRESSION_CODEC, "none"); _defaultVals.put(DELTA_READER_BATCH_SIZE, "4096"); // rows per parquet read batch (Delta Kernel default 1024) + _defaultVals.put(DELTA_READER_COLUMN_API, "true"); // decode data files via parquet-mr's column API _defaultVals.put(DELTA_WRITER_BATCH_SIZE, "4096"); // matrix rows materialized per columnar batch handed to the engine _defaultVals.put(DELTA_WRITER_TARGET_FILE_SIZE, String.valueOf(64L * 1024 * 1024)); // 64MB cap on target data-file size; adaptive sizing may pick smaller -> more files -> more parallel-read throughput _defaultVals.put(DELTA_WRITER_ADAPTIVE_FILE_SIZE, "true"); // size data files toward one per parallel reader diff --git a/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java b/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java new file mode 100644 index 00000000000..2e1d2bdfb22 --- /dev/null +++ b/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java @@ -0,0 +1,585 @@ +/* + * 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.runtime.io; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +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.metadata.ParquetMetadata; +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.runtime.DMLRuntimeException; + +import io.delta.kernel.data.ColumnVector; +import io.delta.kernel.data.ColumnarBatch; +import io.delta.kernel.data.FilteredColumnarBatch; +import io.delta.kernel.engine.Engine; +import io.delta.kernel.engine.ExpressionHandler; +import io.delta.kernel.engine.FileSystemClient; +import io.delta.kernel.engine.JsonHandler; +import io.delta.kernel.engine.ParquetHandler; +import io.delta.kernel.expressions.Column; +import io.delta.kernel.expressions.Predicate; +import io.delta.kernel.types.DataType; +import io.delta.kernel.types.LongType; +import io.delta.kernel.types.StructField; +import io.delta.kernel.types.StructType; +import io.delta.kernel.utils.CloseableIterator; +import io.delta.kernel.utils.DataFileStatus; +import io.delta.kernel.utils.FileStatus; + +/** + * Delta Kernel {@link Engine} whose {@link ParquetHandler} decodes flat data files through + * parquet-mr's low-level column API ({@link ColumnReadStoreImpl}/{@link ColumnReader}) instead of the default + * engine's row-record path ({@code org.apache.parquet.hadoop.ParquetReader}). + * Everything else delegates to the wrapped default engine, and deletion vectors / column mapping are still applied by the kernel. + * + * Beyond plain decoding the fast path honors the {@link ParquetHandler#readParquetFiles} contract cases the + * kernel relies on: the {@code _metadata.row_index} metadata column (requested for every read of a table whose + * protocol carries the {@code deletionVectors} feature) is synthesized from the row-group row offsets, columns are + * resolved by parquet field id first and name second (column mapping mode {@code id}), and columns absent from a + * data file are returned as all-null vectors. Reads with predicates, nested/unsupported types, + * or metadata columns other than {@code row_index} fall back to the wrapped default engine. + */ +public class ColumnApiDeltaEngine implements Engine { + + private final Engine _delegate; + + public ColumnApiDeltaEngine(Engine delegate) { + _delegate = delegate; + } + + @Override + public ExpressionHandler getExpressionHandler() { + return _delegate.getExpressionHandler(); + } + + @Override + public JsonHandler getJsonHandler() { + return _delegate.getJsonHandler(); + } + + @Override + public FileSystemClient getFileSystemClient() { + return _delegate.getFileSystemClient(); + } + + @Override + public ParquetHandler getParquetHandler() { + return new ColumnApiParquetHandler(_delegate.getParquetHandler()); + } + + private static class ColumnApiParquetHandler implements ParquetHandler { + private final ParquetHandler _fallback; + + ColumnApiParquetHandler(ParquetHandler fallback) { + _fallback = fallback; + } + + @Override + public CloseableIterator readParquetFiles(CloseableIterator files, + StructType physicalSchema, Optional predicate) throws IOException { + // fast path only for flat schemas of decodable primitives; predicates, nested + // reads and unknown metadata columns keep the default row-record decode + if(predicate.isPresent() || !supportsFastPath(physicalSchema)) + return _fallback.readParquetFiles(files, physicalSchema, predicate); + return new ColumnApiBatchIterator(files, physicalSchema, ConfigurationManager.getCachedJobConf()); + } + + @Override + public CloseableIterator writeParquetFiles(String directoryPath, + CloseableIterator dataIter, List statsColumns) throws IOException { + return _fallback.writeParquetFiles(directoryPath, dataIter, statsColumns); + } + + @Override + public void writeParquetFileAtomically(String filePath, CloseableIterator data) + throws IOException { + _fallback.writeParquetFileAtomically(filePath, data); + } + + private static boolean supportsFastPath(StructType schema) { + for(int c = 0; c < schema.length(); c++) { + StructField f = schema.at(c); + if(isRowIndexColumn(f)) + continue; // synthesized, need not exist in the data files + if(f.isMetadataColumn() || DeltaKernelUtils.typeCode(f.getDataType()) < 0) + return false; + } + return true; + } + } + + /** The metadata column the kernel asks the parquet handler to fill with the file row index. */ + private static boolean isRowIndexColumn(StructField f) { + return f.isMetadataColumn() && StructField.METADATA_ROW_INDEX_COLUMN_NAME.equals(f.getName()); + } + + /** + * Streams the requested files as one {@link ColumnarBatch} per parquet row group, decoding each column + * directly off {@link ColumnReader} into a pre-sized primitive array. + */ + private static class ColumnApiBatchIterator implements CloseableIterator { + /** 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"; + + private final CloseableIterator _files; + private final StructType _schema; + private final Configuration _conf; + + private ParquetFileReader _reader; + private MessageType _parquetSchema; + private String _createdBy; + private GroupConverter _rootConverter; + private ColumnarBatch _next; + /** Per schema ordinal the resolved parquet column name of the current file, or null if absent there. */ + private String[] _parquetColNames; + /** File row index of the next row group's first row (fallback when the page store carries no offset). */ + private long _fileRowOffset; + + ColumnApiBatchIterator(CloseableIterator files, StructType schema, Configuration conf) { + _files = files; + _schema = schema; + _conf = conf; + } + + @Override + public boolean hasNext() { + if(_next == null) + _next = advance(); + return _next != null; + } + + @Override + public ColumnarBatch next() { + if(!hasNext()) + throw new java.util.NoSuchElementException(); + ColumnarBatch b = _next; + _next = null; + return b; + } + + private ColumnarBatch advance() { + try { + while(true) { + if(_reader != null) { + PageReadStore pages = _reader.readNextRowGroup(); + if(pages != null) + return decodeRowGroup(pages); + _reader.close(); + _reader = null; + } + if(!_files.hasNext()) + return null; + openFile(_files.next()); + } + } + catch(IOException ex) { + throw new DMLRuntimeException("Column-API parquet decode failed", ex); + } + } + + private void openFile(FileStatus file) throws IOException { + _reader = ParquetFileReader.open(HadoopInputFile.fromPath(new Path(file.getPath()), _conf)); + ParquetMetadata meta = _reader.getFooter(); + _parquetSchema = meta.getFileMetaData().getSchema(); + _createdBy = meta.getFileMetaData().getCreatedBy(); + _parquetColNames = resolveParquetColumns(_schema, _parquetSchema); + _fileRowOffset = 0; + final int n = _parquetSchema.getFieldCount(); + final PrimitiveConverter[] leaves = new PrimitiveConverter[n]; + for(int i = 0; i < n; i++) + leaves[i] = new PrimitiveConverter() {}; + _rootConverter = new GroupConverter() { + @Override + public Converter getConverter(int fieldIndex) { + return leaves[fieldIndex]; + } + + @Override + public void start() { + } + + @Override + public void end() { + } + }; + } + + private ColumnarBatch decodeRowGroup(PageReadStore pages) { + final int nrow = (int) pages.getRowCount(); + final int ncol = _schema.length(); + final long rowIndexBase = pages.getRowIndexOffset().orElse(_fileRowOffset); + ColumnReadStoreImpl store = new ColumnReadStoreImpl(pages, _rootConverter, _parquetSchema, _createdBy); + ColumnVector[] vectors = new ColumnVector[ncol]; + for(int c = 0; c < ncol; c++) { + StructField field = _schema.at(c); + if(isRowIndexColumn(field)) { + vectors[c] = rowIndexVector(rowIndexBase, nrow); + continue; + } + String name = _parquetColNames[c]; + if(name == null) { // column absent from this data file (schema evolution) + vectors[c] = new NullVector(field.getDataType(), nrow); + continue; + } + ColumnDescriptor desc = _parquetSchema.getColumnDescription(new String[] {name}); + vectors[c] = decodeColumn(store.getColumnReader(desc), desc.getMaxDefinitionLevel(), nrow, + field.getDataType()); + } + _fileRowOffset += nrow; + return new ArrayBackedBatch(_schema, vectors, nrow); + } + + /** + * Resolve each schema column to the parquet column name of the current file: by parquet field id when the + * physical schema carries one (column mapping mode {@code id}), by name otherwise, 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++) { + StructField f = schema.at(c); + if(isRowIndexColumn(f)) + continue; // synthesized, never read from the file + Object fid = f.getMetadata().get(PARQUET_FIELD_ID_KEY); + String byId = (fid instanceof Number) ? idToName.get(((Number) fid).intValue()) : null; + resolved[c] = (byId != null) ? byId : names.get(f.getName()); + } + return resolved; + } + + private static ColumnVector rowIndexVector(long base, int nrow) { + long[] a = new long[nrow]; + for(int r = 0; r < nrow; r++) + a[r] = base + r; + return TypedVector.longs(LongType.LONG, nrow, null, a); + } + + private static ColumnVector decodeColumn(ColumnReader creader, int maxDef, int nrow, DataType dt) { + boolean[] nulls = maxDef > 0 ? new boolean[nrow] : null; + int code = DeltaKernelUtils.typeCode(dt); + switch(code) { + case DeltaKernelUtils.T_DOUBLE: { + double[] a = new double[nrow]; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[r] = creader.getDouble(); + else + nulls[r] = true; + creader.consume(); + } + return TypedVector.doubles(dt, nrow, nulls, a); + } + case DeltaKernelUtils.T_FLOAT: { + float[] a = new float[nrow]; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[r] = creader.getFloat(); + else + nulls[r] = true; + creader.consume(); + } + return TypedVector.floats(dt, nrow, nulls, a); + } + case DeltaKernelUtils.T_LONG: { + long[] a = new long[nrow]; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[r] = creader.getLong(); + else + nulls[r] = true; + creader.consume(); + } + return TypedVector.longs(dt, nrow, nulls, a); + } + case DeltaKernelUtils.T_INT: + case DeltaKernelUtils.T_SHORT: + case DeltaKernelUtils.T_BYTE: { + // delta short/byte columns are stored as annotated parquet INT32 + int[] a = new int[nrow]; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[r] = creader.getInteger(); + else + nulls[r] = true; + creader.consume(); + } + return TypedVector.ints(dt, nrow, nulls, a); + } + case DeltaKernelUtils.T_BOOLEAN: { + boolean[] a = new boolean[nrow]; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[r] = creader.getBoolean(); + else + nulls[r] = true; + creader.consume(); + } + return TypedVector.booleans(dt, nrow, nulls, a); + } + case DeltaKernelUtils.T_STRING: { + String[] a = new String[nrow]; + for(int r = 0; r < nrow; r++) { + if(creader.getCurrentDefinitionLevel() == maxDef) + a[r] = creader.getBinary().toStringUsingUTF8(); + else + nulls[r] = true; + creader.consume(); + } + return TypedVector.strings(dt, nrow, nulls, a); + } + default: + throw new DMLRuntimeException("Unsupported delta type for column-API decode: " + dt); + } + } + + @Override + public void close() throws IOException { + if(_reader != null) { + _reader.close(); + _reader = null; + } + _files.close(); + } + } + + /** Columnar batch over decoded per-column arrays, with the with* methods the kernel may invoke. */ + private static class ArrayBackedBatch implements ColumnarBatch { + private final StructType _schema; + private final ColumnVector[] _vectors; + private final int _size; + + ArrayBackedBatch(StructType schema, ColumnVector[] vectors, int size) { + _schema = schema; + _vectors = vectors; + _size = size; + } + + @Override + public StructType getSchema() { + return _schema; + } + + @Override + public ColumnVector getColumnVector(int ordinal) { + return _vectors[ordinal]; + } + + @Override + public int getSize() { + return _size; + } + + @Override + public ColumnarBatch withNewColumn(int ordinal, StructField field, ColumnVector vector) { + List fields = new ArrayList<>(_schema.fields()); + fields.add(ordinal, field); + ColumnVector[] vs = new ColumnVector[_vectors.length + 1]; + System.arraycopy(_vectors, 0, vs, 0, ordinal); + vs[ordinal] = vector; + System.arraycopy(_vectors, ordinal, vs, ordinal + 1, _vectors.length - ordinal); + return new ArrayBackedBatch(new StructType(fields), vs, _size); + } + + @Override + public ColumnarBatch withDeletedColumnAt(int ordinal) { + List fields = new ArrayList<>(_schema.fields()); + fields.remove(ordinal); + ColumnVector[] vs = new ColumnVector[_vectors.length - 1]; + System.arraycopy(_vectors, 0, vs, 0, ordinal); + System.arraycopy(_vectors, ordinal + 1, vs, ordinal, _vectors.length - ordinal - 1); + return new ArrayBackedBatch(new StructType(fields), vs, _size); + } + + @Override + public ColumnarBatch withNewSchema(StructType newSchema) { + return new ArrayBackedBatch(newSchema, _vectors, _size); + } + } + + /** All-null vector for a column that is absent from a data file (added to the table after the file was written). */ + private static class NullVector implements ColumnVector { + private final DataType _type; + private final int _size; + + NullVector(DataType type, int size) { + _type = type; + _size = size; + } + + @Override + public DataType getDataType() { + return _type; + } + + @Override + public int getSize() { + return _size; + } + + @Override + public boolean isNullAt(int rowId) { + return true; + } + + @Override + public void close() { + // nothing to release + } + } + + /** Typed vector over one decoded primitive array; exactly one backing array is non-null. */ + private static class TypedVector implements ColumnVector { + private final DataType _type; + private final int _size; + private final boolean[] _nulls; // null => column has no nulls + private double[] _d; + private float[] _f; + private long[] _l; + private int[] _i; + private boolean[] _b; + private String[] _s; + + private TypedVector(DataType type, int size, boolean[] nulls) { + _type = type; + _size = size; + _nulls = nulls; + } + + static TypedVector doubles(DataType t, int n, boolean[] nulls, double[] a) { + TypedVector v = new TypedVector(t, n, nulls); + v._d = a; + return v; + } + + static TypedVector floats(DataType t, int n, boolean[] nulls, float[] a) { + TypedVector v = new TypedVector(t, n, nulls); + v._f = a; + return v; + } + + static TypedVector longs(DataType t, int n, boolean[] nulls, long[] a) { + TypedVector v = new TypedVector(t, n, nulls); + v._l = a; + return v; + } + + static TypedVector ints(DataType t, int n, boolean[] nulls, int[] a) { + TypedVector v = new TypedVector(t, n, nulls); + v._i = a; + return v; + } + + static TypedVector booleans(DataType t, int n, boolean[] nulls, boolean[] a) { + TypedVector v = new TypedVector(t, n, nulls); + v._b = a; + return v; + } + + static TypedVector strings(DataType t, int n, boolean[] nulls, String[] a) { + TypedVector v = new TypedVector(t, n, nulls); + v._s = a; + return v; + } + + @Override + public DataType getDataType() { + return _type; + } + + @Override + public int getSize() { + return _size; + } + + @Override + public boolean isNullAt(int rowId) { + return _nulls != null && _nulls[rowId]; + } + + @Override + public double getDouble(int rowId) { + return _d[rowId]; + } + + @Override + public float getFloat(int rowId) { + return _f[rowId]; + } + + @Override + public long getLong(int rowId) { + return _l[rowId]; + } + + @Override + public int getInt(int rowId) { + return _i[rowId]; + } + + @Override + public short getShort(int rowId) { + return (short) _i[rowId]; + } + + @Override + public byte getByte(int rowId) { + return (byte) _i[rowId]; + } + + @Override + public boolean getBoolean(int rowId) { + return _b[rowId]; + } + + @Override + public String getString(int rowId) { + return _s[rowId]; + } + + @Override + public void close() { + // nothing to release + } + } +} 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..be27dff60a6 100644 --- a/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java +++ b/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java @@ -189,7 +189,12 @@ private static synchronized Configuration deltaConf() { } public static Engine createEngine() { - return DefaultEngine.create(deltaConf()); + Engine engine = DefaultEngine.create(deltaConf()); + //decode data files via parquet-mr's column API rather than the kernel default row-record + //reader, disable via config. + if(ConfigurationManager.isDeltaReaderColumnApi()) + return new ColumnApiDeltaEngine(engine); + return engine; } /** 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..5a0b1136a6d --- /dev/null +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameSparkContractTest.java @@ -0,0 +1,273 @@ +/* + * 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 Kernel {@code ParquetHandler} contract cases that a custom parquet decode path + * (the column-API fast path, {@code sysds.io.delta.reader.columnapi}, on by default) 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)); + } + } +} From a4cfb7092e330a480d4983f5b8b6106de21f675b Mon Sep 17 00:00:00 2001 From: Jakob-al28 Date: Tue, 7 Jul 2026 10:50:42 +0200 Subject: [PATCH 2/6] Apply Eclipse formatter to CI-flagged comment lines --- .../runtime/io/ColumnApiDeltaEngine.java | 31 ++++++++++--------- .../sysds/runtime/io/DeltaKernelUtils.java | 4 +-- .../io/DeltaFrameSparkContractTest.java | 21 ++++++------- 3 files changed, 28 insertions(+), 28 deletions(-) diff --git a/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java b/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java index 2e1d2bdfb22..8e65da633ac 100644 --- a/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java +++ b/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java @@ -60,17 +60,17 @@ import io.delta.kernel.utils.FileStatus; /** - * Delta Kernel {@link Engine} whose {@link ParquetHandler} decodes flat data files through - * parquet-mr's low-level column API ({@link ColumnReadStoreImpl}/{@link ColumnReader}) instead of the default - * engine's row-record path ({@code org.apache.parquet.hadoop.ParquetReader}). - * Everything else delegates to the wrapped default engine, and deletion vectors / column mapping are still applied by the kernel. + * Delta Kernel {@link Engine} whose {@link ParquetHandler} decodes flat data files through parquet-mr's low-level + * column API ({@link ColumnReadStoreImpl}/{@link ColumnReader}) instead of the default engine's row-record path + * ({@code org.apache.parquet.hadoop.ParquetReader}). Everything else delegates to the wrapped default engine, and + * deletion vectors / column mapping are still applied by the kernel. * - * Beyond plain decoding the fast path honors the {@link ParquetHandler#readParquetFiles} contract cases the - * kernel relies on: the {@code _metadata.row_index} metadata column (requested for every read of a table whose - * protocol carries the {@code deletionVectors} feature) is synthesized from the row-group row offsets, columns are - * resolved by parquet field id first and name second (column mapping mode {@code id}), and columns absent from a - * data file are returned as all-null vectors. Reads with predicates, nested/unsupported types, - * or metadata columns other than {@code row_index} fall back to the wrapped default engine. + * Beyond plain decoding the fast path honors the {@link ParquetHandler#readParquetFiles} contract cases the kernel + * relies on: the {@code _metadata.row_index} metadata column (requested for every read of a table whose protocol + * carries the {@code deletionVectors} feature) is synthesized from the row-group row offsets, columns are resolved by + * parquet field id first and name second (column mapping mode {@code id}), and columns absent from a data file are + * returned as all-null vectors. Reads with predicates, nested/unsupported types, or metadata columns other than + * {@code row_index} fall back to the wrapped default engine. */ public class ColumnApiDeltaEngine implements Engine { @@ -147,8 +147,8 @@ private static boolean isRowIndexColumn(StructField f) { } /** - * Streams the requested files as one {@link ColumnarBatch} per parquet row group, decoding each column - * directly off {@link ColumnReader} into a pre-sized primitive array. + * Streams the requested files as one {@link ColumnarBatch} per parquet row group, decoding each column directly off + * {@link ColumnReader} into a pre-sized primitive array. */ private static class ColumnApiBatchIterator implements CloseableIterator { /** Physical-schema metadata key carrying the parquet field id (column mapping mode {@code id}). */ @@ -220,7 +220,8 @@ private void openFile(FileStatus file) throws IOException { final int n = _parquetSchema.getFieldCount(); final PrimitiveConverter[] leaves = new PrimitiveConverter[n]; for(int i = 0; i < n; i++) - leaves[i] = new PrimitiveConverter() {}; + leaves[i] = new PrimitiveConverter() { + }; _rootConverter = new GroupConverter() { @Override public Converter getConverter(int fieldIndex) { @@ -264,8 +265,8 @@ private ColumnarBatch decodeRowGroup(PageReadStore pages) { /** * Resolve each schema column to the parquet column name of the current file: by parquet field id when the - * physical schema carries one (column mapping mode {@code id}), by name otherwise, null when the file does - * not contain the column at all. + * physical schema carries one (column mapping mode {@code id}), by name otherwise, null when the file does not + * contain the column at all. */ private static String[] resolveParquetColumns(StructType schema, MessageType parquetSchema) { Map idToName = new HashMap<>(); 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 be27dff60a6..f4aebb4a053 100644 --- a/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java +++ b/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java @@ -190,8 +190,8 @@ private static synchronized Configuration deltaConf() { public static Engine createEngine() { Engine engine = DefaultEngine.create(deltaConf()); - //decode data files via parquet-mr's column API rather than the kernel default row-record - //reader, disable via config. + // decode data files via parquet-mr's column API rather than the kernel default row-record + // reader, disable via config. if(ConfigurationManager.isDeltaReaderColumnApi()) return new ColumnApiDeltaEngine(engine); return engine; 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 5a0b1136a6d..d8aa4ee40c0 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 @@ -50,20 +50,19 @@ /** * Regression tests pinning the Delta Kernel {@code ParquetHandler} contract cases that a custom parquet decode path * (the column-API fast path, {@code sysds.io.delta.reader.columnapi}, on by default) 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). + * 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. + * 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. + * 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 { From 06c3a4b80bc934bb15f4ec1276111e3e3efa57ed Mon Sep 17 00:00:00 2001 From: Jakob-al28 <04jakob28@gmail.com> Date: Tue, 7 Jul 2026 22:57:56 +0200 Subject: [PATCH 3/6] [SYSTEMDS-3949] Decode Delta data files directly into frame columns --- .../sysds/conf/ConfigurationManager.java | 5 - .../java/org/apache/sysds/conf/DMLConfig.java | 2 - .../runtime/io/ColumnApiDeltaEngine.java | 586 ------------------ .../sysds/runtime/io/DeltaKernelUtils.java | 212 ++++++- .../sysds/runtime/io/FrameReaderDelta.java | 61 +- .../runtime/io/FrameReaderDeltaParallel.java | 30 +- .../io/DeltaFrameSparkContractTest.java | 7 +- 7 files changed, 248 insertions(+), 655 deletions(-) delete mode 100644 src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java diff --git a/src/main/java/org/apache/sysds/conf/ConfigurationManager.java b/src/main/java/org/apache/sysds/conf/ConfigurationManager.java index 709e806c0ae..8b0f5fe06b9 100644 --- a/src/main/java/org/apache/sysds/conf/ConfigurationManager.java +++ b/src/main/java/org/apache/sysds/conf/ConfigurationManager.java @@ -263,11 +263,6 @@ public static int getDeltaReaderBatchSize() { return getDMLConfig().getIntValue(DMLConfig.DELTA_READER_BATCH_SIZE); } - /** @return whether the native Delta reader decodes data files via parquet-mr's column API */ - public static boolean isDeltaReaderColumnApi() { - return getDMLConfig().getBooleanValue(DMLConfig.DELTA_READER_COLUMN_API); - } - /** @return matrix rows materialized per columnar batch for the native Delta writer */ public static int getDeltaWriterBatchSize() { return getDMLConfig().getIntValue(DMLConfig.DELTA_WRITER_BATCH_SIZE); diff --git a/src/main/java/org/apache/sysds/conf/DMLConfig.java b/src/main/java/org/apache/sysds/conf/DMLConfig.java index 12594a88916..d114ccf69b9 100644 --- a/src/main/java/org/apache/sysds/conf/DMLConfig.java +++ b/src/main/java/org/apache/sysds/conf/DMLConfig.java @@ -72,7 +72,6 @@ public class DMLConfig public static final String CP_PARALLEL_IO = "sysds.cp.parallel.io"; public static final String IO_COMPRESSION_CODEC = "sysds.io.compression.encoding"; public static final String DELTA_READER_BATCH_SIZE = "sysds.io.delta.reader.batchsize"; // int: rows per parquet read batch - public static final String DELTA_READER_COLUMN_API = "sysds.io.delta.reader.columnapi"; // boolean: decode data files via parquet-mr's column API instead of the kernel default row-record reader public static final String DELTA_WRITER_BATCH_SIZE = "sysds.io.delta.writer.batchsize"; // int: matrix rows materialized per columnar batch handed to the engine public static final String DELTA_WRITER_TARGET_FILE_SIZE = "sysds.io.delta.writer.targetfilesize"; // long: upper bound on target data-file size in bytes; adaptive sizing may pick smaller -> more files -> more parallel-read throughput public static final String DELTA_WRITER_ADAPTIVE_FILE_SIZE = "sysds.io.delta.writer.adaptivefilesize"; // boolean: size data files toward one per parallel reader (capped by targetfilesize) @@ -164,7 +163,6 @@ public class DMLConfig _defaultVals.put(CP_PARALLEL_IO, "true" ); _defaultVals.put(IO_COMPRESSION_CODEC, "none"); _defaultVals.put(DELTA_READER_BATCH_SIZE, "4096"); // rows per parquet read batch (Delta Kernel default 1024) - _defaultVals.put(DELTA_READER_COLUMN_API, "true"); // decode data files via parquet-mr's column API _defaultVals.put(DELTA_WRITER_BATCH_SIZE, "4096"); // matrix rows materialized per columnar batch handed to the engine _defaultVals.put(DELTA_WRITER_TARGET_FILE_SIZE, String.valueOf(64L * 1024 * 1024)); // 64MB cap on target data-file size; adaptive sizing may pick smaller -> more files -> more parallel-read throughput _defaultVals.put(DELTA_WRITER_ADAPTIVE_FILE_SIZE, "true"); // size data files toward one per parallel reader diff --git a/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java b/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java deleted file mode 100644 index 8e65da633ac..00000000000 --- a/src/main/java/org/apache/sysds/runtime/io/ColumnApiDeltaEngine.java +++ /dev/null @@ -1,586 +0,0 @@ -/* - * 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.runtime.io; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; - -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.metadata.ParquetMetadata; -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.runtime.DMLRuntimeException; - -import io.delta.kernel.data.ColumnVector; -import io.delta.kernel.data.ColumnarBatch; -import io.delta.kernel.data.FilteredColumnarBatch; -import io.delta.kernel.engine.Engine; -import io.delta.kernel.engine.ExpressionHandler; -import io.delta.kernel.engine.FileSystemClient; -import io.delta.kernel.engine.JsonHandler; -import io.delta.kernel.engine.ParquetHandler; -import io.delta.kernel.expressions.Column; -import io.delta.kernel.expressions.Predicate; -import io.delta.kernel.types.DataType; -import io.delta.kernel.types.LongType; -import io.delta.kernel.types.StructField; -import io.delta.kernel.types.StructType; -import io.delta.kernel.utils.CloseableIterator; -import io.delta.kernel.utils.DataFileStatus; -import io.delta.kernel.utils.FileStatus; - -/** - * Delta Kernel {@link Engine} whose {@link ParquetHandler} decodes flat data files through parquet-mr's low-level - * column API ({@link ColumnReadStoreImpl}/{@link ColumnReader}) instead of the default engine's row-record path - * ({@code org.apache.parquet.hadoop.ParquetReader}). Everything else delegates to the wrapped default engine, and - * deletion vectors / column mapping are still applied by the kernel. - * - * Beyond plain decoding the fast path honors the {@link ParquetHandler#readParquetFiles} contract cases the kernel - * relies on: the {@code _metadata.row_index} metadata column (requested for every read of a table whose protocol - * carries the {@code deletionVectors} feature) is synthesized from the row-group row offsets, columns are resolved by - * parquet field id first and name second (column mapping mode {@code id}), and columns absent from a data file are - * returned as all-null vectors. Reads with predicates, nested/unsupported types, or metadata columns other than - * {@code row_index} fall back to the wrapped default engine. - */ -public class ColumnApiDeltaEngine implements Engine { - - private final Engine _delegate; - - public ColumnApiDeltaEngine(Engine delegate) { - _delegate = delegate; - } - - @Override - public ExpressionHandler getExpressionHandler() { - return _delegate.getExpressionHandler(); - } - - @Override - public JsonHandler getJsonHandler() { - return _delegate.getJsonHandler(); - } - - @Override - public FileSystemClient getFileSystemClient() { - return _delegate.getFileSystemClient(); - } - - @Override - public ParquetHandler getParquetHandler() { - return new ColumnApiParquetHandler(_delegate.getParquetHandler()); - } - - private static class ColumnApiParquetHandler implements ParquetHandler { - private final ParquetHandler _fallback; - - ColumnApiParquetHandler(ParquetHandler fallback) { - _fallback = fallback; - } - - @Override - public CloseableIterator readParquetFiles(CloseableIterator files, - StructType physicalSchema, Optional predicate) throws IOException { - // fast path only for flat schemas of decodable primitives; predicates, nested - // reads and unknown metadata columns keep the default row-record decode - if(predicate.isPresent() || !supportsFastPath(physicalSchema)) - return _fallback.readParquetFiles(files, physicalSchema, predicate); - return new ColumnApiBatchIterator(files, physicalSchema, ConfigurationManager.getCachedJobConf()); - } - - @Override - public CloseableIterator writeParquetFiles(String directoryPath, - CloseableIterator dataIter, List statsColumns) throws IOException { - return _fallback.writeParquetFiles(directoryPath, dataIter, statsColumns); - } - - @Override - public void writeParquetFileAtomically(String filePath, CloseableIterator data) - throws IOException { - _fallback.writeParquetFileAtomically(filePath, data); - } - - private static boolean supportsFastPath(StructType schema) { - for(int c = 0; c < schema.length(); c++) { - StructField f = schema.at(c); - if(isRowIndexColumn(f)) - continue; // synthesized, need not exist in the data files - if(f.isMetadataColumn() || DeltaKernelUtils.typeCode(f.getDataType()) < 0) - return false; - } - return true; - } - } - - /** The metadata column the kernel asks the parquet handler to fill with the file row index. */ - private static boolean isRowIndexColumn(StructField f) { - return f.isMetadataColumn() && StructField.METADATA_ROW_INDEX_COLUMN_NAME.equals(f.getName()); - } - - /** - * Streams the requested files as one {@link ColumnarBatch} per parquet row group, decoding each column directly off - * {@link ColumnReader} into a pre-sized primitive array. - */ - private static class ColumnApiBatchIterator implements CloseableIterator { - /** 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"; - - private final CloseableIterator _files; - private final StructType _schema; - private final Configuration _conf; - - private ParquetFileReader _reader; - private MessageType _parquetSchema; - private String _createdBy; - private GroupConverter _rootConverter; - private ColumnarBatch _next; - /** Per schema ordinal the resolved parquet column name of the current file, or null if absent there. */ - private String[] _parquetColNames; - /** File row index of the next row group's first row (fallback when the page store carries no offset). */ - private long _fileRowOffset; - - ColumnApiBatchIterator(CloseableIterator files, StructType schema, Configuration conf) { - _files = files; - _schema = schema; - _conf = conf; - } - - @Override - public boolean hasNext() { - if(_next == null) - _next = advance(); - return _next != null; - } - - @Override - public ColumnarBatch next() { - if(!hasNext()) - throw new java.util.NoSuchElementException(); - ColumnarBatch b = _next; - _next = null; - return b; - } - - private ColumnarBatch advance() { - try { - while(true) { - if(_reader != null) { - PageReadStore pages = _reader.readNextRowGroup(); - if(pages != null) - return decodeRowGroup(pages); - _reader.close(); - _reader = null; - } - if(!_files.hasNext()) - return null; - openFile(_files.next()); - } - } - catch(IOException ex) { - throw new DMLRuntimeException("Column-API parquet decode failed", ex); - } - } - - private void openFile(FileStatus file) throws IOException { - _reader = ParquetFileReader.open(HadoopInputFile.fromPath(new Path(file.getPath()), _conf)); - ParquetMetadata meta = _reader.getFooter(); - _parquetSchema = meta.getFileMetaData().getSchema(); - _createdBy = meta.getFileMetaData().getCreatedBy(); - _parquetColNames = resolveParquetColumns(_schema, _parquetSchema); - _fileRowOffset = 0; - final int n = _parquetSchema.getFieldCount(); - final PrimitiveConverter[] leaves = new PrimitiveConverter[n]; - for(int i = 0; i < n; i++) - leaves[i] = new PrimitiveConverter() { - }; - _rootConverter = new GroupConverter() { - @Override - public Converter getConverter(int fieldIndex) { - return leaves[fieldIndex]; - } - - @Override - public void start() { - } - - @Override - public void end() { - } - }; - } - - private ColumnarBatch decodeRowGroup(PageReadStore pages) { - final int nrow = (int) pages.getRowCount(); - final int ncol = _schema.length(); - final long rowIndexBase = pages.getRowIndexOffset().orElse(_fileRowOffset); - ColumnReadStoreImpl store = new ColumnReadStoreImpl(pages, _rootConverter, _parquetSchema, _createdBy); - ColumnVector[] vectors = new ColumnVector[ncol]; - for(int c = 0; c < ncol; c++) { - StructField field = _schema.at(c); - if(isRowIndexColumn(field)) { - vectors[c] = rowIndexVector(rowIndexBase, nrow); - continue; - } - String name = _parquetColNames[c]; - if(name == null) { // column absent from this data file (schema evolution) - vectors[c] = new NullVector(field.getDataType(), nrow); - continue; - } - ColumnDescriptor desc = _parquetSchema.getColumnDescription(new String[] {name}); - vectors[c] = decodeColumn(store.getColumnReader(desc), desc.getMaxDefinitionLevel(), nrow, - field.getDataType()); - } - _fileRowOffset += nrow; - return new ArrayBackedBatch(_schema, vectors, nrow); - } - - /** - * Resolve each schema column to the parquet column name of the current file: by parquet field id when the - * physical schema carries one (column mapping mode {@code id}), by name otherwise, 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++) { - StructField f = schema.at(c); - if(isRowIndexColumn(f)) - continue; // synthesized, never read from the file - Object fid = f.getMetadata().get(PARQUET_FIELD_ID_KEY); - String byId = (fid instanceof Number) ? idToName.get(((Number) fid).intValue()) : null; - resolved[c] = (byId != null) ? byId : names.get(f.getName()); - } - return resolved; - } - - private static ColumnVector rowIndexVector(long base, int nrow) { - long[] a = new long[nrow]; - for(int r = 0; r < nrow; r++) - a[r] = base + r; - return TypedVector.longs(LongType.LONG, nrow, null, a); - } - - private static ColumnVector decodeColumn(ColumnReader creader, int maxDef, int nrow, DataType dt) { - boolean[] nulls = maxDef > 0 ? new boolean[nrow] : null; - int code = DeltaKernelUtils.typeCode(dt); - switch(code) { - case DeltaKernelUtils.T_DOUBLE: { - double[] a = new double[nrow]; - for(int r = 0; r < nrow; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getDouble(); - else - nulls[r] = true; - creader.consume(); - } - return TypedVector.doubles(dt, nrow, nulls, a); - } - case DeltaKernelUtils.T_FLOAT: { - float[] a = new float[nrow]; - for(int r = 0; r < nrow; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getFloat(); - else - nulls[r] = true; - creader.consume(); - } - return TypedVector.floats(dt, nrow, nulls, a); - } - case DeltaKernelUtils.T_LONG: { - long[] a = new long[nrow]; - for(int r = 0; r < nrow; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getLong(); - else - nulls[r] = true; - creader.consume(); - } - return TypedVector.longs(dt, nrow, nulls, a); - } - case DeltaKernelUtils.T_INT: - case DeltaKernelUtils.T_SHORT: - case DeltaKernelUtils.T_BYTE: { - // delta short/byte columns are stored as annotated parquet INT32 - int[] a = new int[nrow]; - for(int r = 0; r < nrow; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getInteger(); - else - nulls[r] = true; - creader.consume(); - } - return TypedVector.ints(dt, nrow, nulls, a); - } - case DeltaKernelUtils.T_BOOLEAN: { - boolean[] a = new boolean[nrow]; - for(int r = 0; r < nrow; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getBoolean(); - else - nulls[r] = true; - creader.consume(); - } - return TypedVector.booleans(dt, nrow, nulls, a); - } - case DeltaKernelUtils.T_STRING: { - String[] a = new String[nrow]; - for(int r = 0; r < nrow; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getBinary().toStringUsingUTF8(); - else - nulls[r] = true; - creader.consume(); - } - return TypedVector.strings(dt, nrow, nulls, a); - } - default: - throw new DMLRuntimeException("Unsupported delta type for column-API decode: " + dt); - } - } - - @Override - public void close() throws IOException { - if(_reader != null) { - _reader.close(); - _reader = null; - } - _files.close(); - } - } - - /** Columnar batch over decoded per-column arrays, with the with* methods the kernel may invoke. */ - private static class ArrayBackedBatch implements ColumnarBatch { - private final StructType _schema; - private final ColumnVector[] _vectors; - private final int _size; - - ArrayBackedBatch(StructType schema, ColumnVector[] vectors, int size) { - _schema = schema; - _vectors = vectors; - _size = size; - } - - @Override - public StructType getSchema() { - return _schema; - } - - @Override - public ColumnVector getColumnVector(int ordinal) { - return _vectors[ordinal]; - } - - @Override - public int getSize() { - return _size; - } - - @Override - public ColumnarBatch withNewColumn(int ordinal, StructField field, ColumnVector vector) { - List fields = new ArrayList<>(_schema.fields()); - fields.add(ordinal, field); - ColumnVector[] vs = new ColumnVector[_vectors.length + 1]; - System.arraycopy(_vectors, 0, vs, 0, ordinal); - vs[ordinal] = vector; - System.arraycopy(_vectors, ordinal, vs, ordinal + 1, _vectors.length - ordinal); - return new ArrayBackedBatch(new StructType(fields), vs, _size); - } - - @Override - public ColumnarBatch withDeletedColumnAt(int ordinal) { - List fields = new ArrayList<>(_schema.fields()); - fields.remove(ordinal); - ColumnVector[] vs = new ColumnVector[_vectors.length - 1]; - System.arraycopy(_vectors, 0, vs, 0, ordinal); - System.arraycopy(_vectors, ordinal + 1, vs, ordinal, _vectors.length - ordinal - 1); - return new ArrayBackedBatch(new StructType(fields), vs, _size); - } - - @Override - public ColumnarBatch withNewSchema(StructType newSchema) { - return new ArrayBackedBatch(newSchema, _vectors, _size); - } - } - - /** All-null vector for a column that is absent from a data file (added to the table after the file was written). */ - private static class NullVector implements ColumnVector { - private final DataType _type; - private final int _size; - - NullVector(DataType type, int size) { - _type = type; - _size = size; - } - - @Override - public DataType getDataType() { - return _type; - } - - @Override - public int getSize() { - return _size; - } - - @Override - public boolean isNullAt(int rowId) { - return true; - } - - @Override - public void close() { - // nothing to release - } - } - - /** Typed vector over one decoded primitive array; exactly one backing array is non-null. */ - private static class TypedVector implements ColumnVector { - private final DataType _type; - private final int _size; - private final boolean[] _nulls; // null => column has no nulls - private double[] _d; - private float[] _f; - private long[] _l; - private int[] _i; - private boolean[] _b; - private String[] _s; - - private TypedVector(DataType type, int size, boolean[] nulls) { - _type = type; - _size = size; - _nulls = nulls; - } - - static TypedVector doubles(DataType t, int n, boolean[] nulls, double[] a) { - TypedVector v = new TypedVector(t, n, nulls); - v._d = a; - return v; - } - - static TypedVector floats(DataType t, int n, boolean[] nulls, float[] a) { - TypedVector v = new TypedVector(t, n, nulls); - v._f = a; - return v; - } - - static TypedVector longs(DataType t, int n, boolean[] nulls, long[] a) { - TypedVector v = new TypedVector(t, n, nulls); - v._l = a; - return v; - } - - static TypedVector ints(DataType t, int n, boolean[] nulls, int[] a) { - TypedVector v = new TypedVector(t, n, nulls); - v._i = a; - return v; - } - - static TypedVector booleans(DataType t, int n, boolean[] nulls, boolean[] a) { - TypedVector v = new TypedVector(t, n, nulls); - v._b = a; - return v; - } - - static TypedVector strings(DataType t, int n, boolean[] nulls, String[] a) { - TypedVector v = new TypedVector(t, n, nulls); - v._s = a; - return v; - } - - @Override - public DataType getDataType() { - return _type; - } - - @Override - public int getSize() { - return _size; - } - - @Override - public boolean isNullAt(int rowId) { - return _nulls != null && _nulls[rowId]; - } - - @Override - public double getDouble(int rowId) { - return _d[rowId]; - } - - @Override - public float getFloat(int rowId) { - return _f[rowId]; - } - - @Override - public long getLong(int rowId) { - return _l[rowId]; - } - - @Override - public int getInt(int rowId) { - return _i[rowId]; - } - - @Override - public short getShort(int rowId) { - return (short) _i[rowId]; - } - - @Override - public byte getByte(int rowId) { - return (byte) _i[rowId]; - } - - @Override - public boolean getBoolean(int rowId) { - return _b[rowId]; - } - - @Override - public String getString(int rowId) { - return _s[rowId]; - } - - @Override - public void close() { - // nothing to release - } - } -} 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 f4aebb4a053..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; @@ -189,12 +394,7 @@ private static synchronized Configuration deltaConf() { } public static Engine createEngine() { - Engine engine = DefaultEngine.create(deltaConf()); - // decode data files via parquet-mr's column API rather than the kernel default row-record - // reader, disable via config. - if(ConfigurationManager.isDeltaReaderColumnApi()) - return new ColumnApiDeltaEngine(engine); - return engine; + return DefaultEngine.create(deltaConf()); } /** 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 index d8aa4ee40c0..5687461c851 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 @@ -48,10 +48,9 @@ import org.junit.Test; /** - * Regression tests pinning the Delta Kernel {@code ParquetHandler} contract cases that a custom parquet decode path - * (the column-API fast path, {@code sysds.io.delta.reader.columnapi}, on by default) 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). + * 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 From 895217819bfcf73990e20435dc17ffa3cf21dd8c Mon Sep 17 00:00:00 2001 From: Jakob-al28 <04jakob28@gmail.com> Date: Sat, 11 Jul 2026 15:54:31 +0200 Subject: [PATCH 4/6] [SYSTEMDS-3949] Harden direct Delta frame decode and add shape coverage tests Validate each data file's parquet physical layout before decoding and fall back to the kernel engine for files the typed decode cannot handle, e.g. INT32 data files under a schema widened to bigint, which the kernel converts on read. Add DeltaFrameShapeCoverageTest, round-tripping shapes from 1x1 to 1M rows and 1 to 1000 columns on all three read paths (serial direct, parallel direct, forced kernel-buffered). Add tests for type-widened tables and schema evolution leftovers (the widened read fails with an UnsupportedOperationException on the previous code). --- .../sysds/runtime/io/DeltaKernelUtils.java | 112 ++++-- .../sysds/runtime/io/FrameReaderDelta.java | 44 ++- .../runtime/io/FrameReaderDeltaParallel.java | 12 +- .../component/io/DeltaFrameReadWriteTest.java | 218 ++++++++++++ .../io/DeltaFrameShapeCoverageTest.java | 332 ++++++++++++++++++ 5 files changed, 671 insertions(+), 47 deletions(-) create mode 100644 src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java 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 f470f07cb07..c3b9351d3d3 100644 --- a/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java +++ b/src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java @@ -37,11 +37,14 @@ 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.metadata.FileMetaData; 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.parquet.schema.PrimitiveType.PrimitiveTypeName; +import org.apache.parquet.schema.Type.Repetition; import org.apache.sysds.conf.ConfigurationManager; import org.apache.sysds.hops.OptimizerUtils; import org.apache.sysds.runtime.DMLRuntimeException; @@ -117,6 +120,14 @@ public class DeltaKernelUtils { public static final int T_BOOLEAN = 6; public static final int T_STRING = 7; + /** + * Parquet physical type each {@code T_*} column is stored as, indexed by type code (delta int/short/byte columns + * are all stored as annotated parquet INT32). + */ + private static final PrimitiveTypeName[] T_PHYSICAL = {PrimitiveTypeName.DOUBLE, PrimitiveTypeName.FLOAT, + PrimitiveTypeName.INT64, PrimitiveTypeName.INT32, PrimitiveTypeName.INT32, PrimitiveTypeName.INT32, + PrimitiveTypeName.BOOLEAN, PrimitiveTypeName.BINARY}; + //derived configuration cached to avoid copying the (large) base conf on every //engine creation (createEngine is called once per data file in parallel reads); //rebuilt whenever the base conf or the relevant SystemDS settings change. @@ -205,6 +216,20 @@ public static String dataFilePath(Row scanFileRow) { return InternalScanFileUtils.getAddFileStatus(scanFileRow).getPath(); } + /** + * Thrown when a data file's parquet layout cannot be decoded directly into the typed output columns, e.g. a + * physical type narrower than the Delta column type (left behind by type widening). Raised before anything is + * written to the output arrays, so callers can re-read the file through the kernel engine (which performs those + * conversions) instead. + */ + public static final class UnsupportedDirectDecodeException extends DMLRuntimeException { + private static final long serialVersionUID = 1L; + + public UnsupportedDirectDecodeException(String msg) { + super(msg); + } + } + /** * 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 @@ -220,7 +245,10 @@ public static String dataFilePath(Row scanFileRow) { * @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 + * @throws IOException on read failure + * @throws UnsupportedDirectDecodeException if a column's parquet layout does not match the Delta column type + * (thrown before any output is written, so the caller can fall back to the + * kernel for this file) */ public static int decodeDataFileInto(String filePath, StructType physicalSchema, int[] readCodes, Object[] dest, int destOff, int limit, String tablePath) throws IOException { @@ -228,25 +256,27 @@ public static int decodeDataFileInto(String filePath, StructType physicalSchema, 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(); + FileMetaData meta = reader.getFooter().getFileMetaData(); + MessageType parquetSchema = meta.getSchema(); + String createdBy = meta.getCreatedBy(); String[] colNames = resolveParquetColumns(physicalSchema, parquetSchema); + // validate every column before decoding anything: a file whose physical types + // do not match the Delta schema (type widening) must be left to the kernel + final ColumnDescriptor[] descs = new ColumnDescriptor[ncol]; + for(int c = 0; c < ncol; c++) + if(colNames[c] != null) // absent columns keep the array defaults (nulls) + descs[c] = validateDecodable(parquetSchema, colNames[c], readCodes[c], filePath); 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); + checkSliceLimit(off, nrow, limit, 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); + if(descs[c] == null) + continue; + decodeColumnInto(store.getColumnReader(descs[c]), descs[c].getMaxDefinitionLevel(), nrow, + readCodes[c], dest[c], off); } off += nrow; } @@ -254,6 +284,37 @@ public static int decodeDataFileInto(String filePath, StructType physicalSchema, return off - destOff; } + /** + * Guard before writing {@code n} rows at {@code off}: writing past {@code limit} would overflow into the next + * file's slice (or off the array) in the pre-allocated output. Shared by the direct decode and the per-file kernel + * fallback so the check and its message live in one place. + */ + static void checkSliceLimit(int off, int n, int limit, String tablePath) { + if(off + n > limit) + throw new DMLRuntimeException( + "Delta file produced more rows than its numRecords statistic; refusing direct read of " + tablePath); + } + + /** + * Resolve the descriptor of one parquet column and verify it is decodable as the given read code: a non-repeated + * primitive whose physical type is the one the Delta column type is stored as. A mismatch (e.g. an INT32 data file + * under a schema widened to {@code bigint}, or a nested/repeated field where a primitive is expected) is readable + * through the kernel engine but not by the typed direct decode. + */ + private static ColumnDescriptor validateDecodable(MessageType parquetSchema, String colName, int readCode, + String filePath) { + org.apache.parquet.schema.Type t = parquetSchema.getType(colName); + if(!t.isPrimitive() || t.isRepetition(Repetition.REPEATED)) + throw new UnsupportedDirectDecodeException( + "Parquet column '" + colName + "' in " + filePath + " is not a non-repeated primitive"); + PrimitiveTypeName actual = t.asPrimitiveType().getPrimitiveTypeName(); + PrimitiveTypeName expected = T_PHYSICAL[readCode]; + if(actual != expected) + throw new UnsupportedDirectDecodeException("Parquet column '" + colName + "' in " + filePath + " stores " + + actual + " but the Delta schema requires " + expected + " (e.g. a type-widened table)"); + return parquetSchema.getColumnDescription(new String[] {colName}); + } + /** * 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. @@ -304,30 +365,31 @@ public void end() { */ private static void decodeColumnInto(ColumnReader creader, int maxDef, int nrow, int readCode, Object dest, int off) { + final int end = off + nrow; switch(readCode) { case T_DOUBLE: { double[] a = (double[]) dest; - for(int r = 0; r < nrow; r++) { + for(int r = off; r < end; r++) { if(creader.getCurrentDefinitionLevel() == maxDef) - a[off + r] = creader.getDouble(); + a[r] = creader.getDouble(); creader.consume(); } break; } case T_FLOAT: { float[] a = (float[]) dest; - for(int r = 0; r < nrow; r++) { + for(int r = off; r < end; r++) { if(creader.getCurrentDefinitionLevel() == maxDef) - a[off + r] = creader.getFloat(); + a[r] = creader.getFloat(); creader.consume(); } break; } case T_LONG: { long[] a = (long[]) dest; - for(int r = 0; r < nrow; r++) { + for(int r = off; r < end; r++) { if(creader.getCurrentDefinitionLevel() == maxDef) - a[off + r] = creader.getLong(); + a[r] = creader.getLong(); creader.consume(); } break; @@ -337,27 +399,27 @@ private static void decodeColumnInto(ColumnReader creader, int maxDef, int nrow, case T_BYTE: { // delta short/byte columns are stored as annotated parquet INT32 int[] a = (int[]) dest; - for(int r = 0; r < nrow; r++) { + for(int r = off; r < end; r++) { if(creader.getCurrentDefinitionLevel() == maxDef) - a[off + r] = creader.getInteger(); + a[r] = creader.getInteger(); creader.consume(); } break; } case T_BOOLEAN: { boolean[] a = (boolean[]) dest; - for(int r = 0; r < nrow; r++) { + for(int r = off; r < end; r++) { if(creader.getCurrentDefinitionLevel() == maxDef) - a[off + r] = creader.getBoolean(); + a[r] = creader.getBoolean(); creader.consume(); } break; } case T_STRING: { String[] a = (String[]) dest; - for(int r = 0; r < nrow; r++) { + for(int r = off; r < end; r++) { if(creader.getCurrentDefinitionLevel() == maxDef) - a[off + r] = creader.getBinary().toStringUsingUTF8(); + a[r] = creader.getBinary().toStringUsingUTF8(); creader.consume(); } break; 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 a87d49448f5..0127250247d 100644 --- a/src/main/java/org/apache/sysds/runtime/io/FrameReaderDelta.java +++ b/src/main/java/org/apache/sysds/runtime/io/FrameReaderDelta.java @@ -165,18 +165,8 @@ private FrameBlock readDirect(String fname, DeltaKernelUtils.ScanHandle handle, int base = 0; for(int i = 0; i < handle.scanFiles.size(); i++) { - // 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]; - 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); + decodeFileSlice(handle, handle.scanFiles.get(i), readCodes, dest, base, limit, fname); base = limit; } @@ -254,6 +244,38 @@ static Array concatColumn(ValueType vt, int nrow, ArrayList batchCo return ArrayFactory.create(vt, full); } + /** + * Decode one data file into its pre-sized slice {@code [base, limit)} of the destination arrays: through the direct + * parquet column decode by default, or through the kernel engine (which performs the physical-type conversions the + * typed decode declines, e.g. for files left behind by type widening) as a per-file fallback. Fails loud when the + * file produces more or fewer rows than its {@code numRecords} statistic promised, so a lying statistic can neither + * overflow into the next file's slice nor leave a silent gap of array defaults. Thread-safe for distinct files, so + * the parallel reader shares it. + */ + static void decodeFileSlice(DeltaKernelUtils.ScanHandle handle, Row scanFileRow, int[] readCodes, Object[] dest, + int base, int limit, String fname) throws IOException { + int n; + try { + n = DeltaKernelUtils.decodeDataFileInto(DeltaKernelUtils.dataFilePath(scanFileRow), + handle.physicalReadSchema, readCodes, dest, base, limit, fname); + } + catch(DeltaKernelUtils.UnsupportedDirectDecodeException e) { + final int[] off = {base}; + DeltaKernelUtils.readScanFile(DeltaKernelUtils.createEngine(), handle.scanState, handle.physicalReadSchema, + scanFileRow, (cols, size, selected) -> { + int m = DeltaKernelUtils.countSelected(size, selected); + DeltaKernelUtils.checkSliceLimit(off[0], m, limit, fname); + for(int c = 0; c < readCodes.length; c++) + extractColumnInto(cols[c], size, selected, readCodes[c], dest[c], off[0]); + off[0] += m; + }); + n = off[0] - base; + } + if(base + n != limit) + throw new DMLRuntimeException("Delta file produced " + n + " rows, expected " + (limit - base) + + " from its numRecords statistic; refusing direct read of " + fname); + } + static int readCode(DataType dt, String name) { // reuse the shared Delta type -> code mapping; frames additionally reject the // types the matrix reader also cannot map (typeCode returns -1) 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 201a4db9dd1..0e0625824e1 100644 --- a/src/main/java/org/apache/sysds/runtime/io/FrameReaderDeltaParallel.java +++ b/src/main/java/org/apache/sysds/runtime/io/FrameReaderDeltaParallel.java @@ -113,19 +113,9 @@ 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; 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(() -> { - 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(base + n != limit) - throw new DMLRuntimeException("Delta file produced " + n + " rows, expected " + (limit - base) - + " from its numRecords statistic; refusing parallel direct read of " + fname); + decodeFileSlice(handle, scanFileRow, readCodes, dest, base, limit, fname); return null; }); } diff --git a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameReadWriteTest.java b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameReadWriteTest.java index 7012be44426..752ef190ca8 100644 --- a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameReadWriteTest.java +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameReadWriteTest.java @@ -28,11 +28,18 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.NoSuchElementException; import java.util.Optional; import java.util.Random; import org.apache.commons.io.FileUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.sysds.common.Types.FileFormat; import org.apache.sysds.common.Types.ValueType; import org.apache.sysds.conf.CompilerConfig; @@ -50,19 +57,28 @@ import org.apache.sysds.test.TestUtils; import org.junit.Test; +import io.delta.kernel.DataWriteContext; +import io.delta.kernel.Operation; +import io.delta.kernel.Table; +import io.delta.kernel.Transaction; import io.delta.kernel.data.ColumnVector; import io.delta.kernel.data.ColumnarBatch; import io.delta.kernel.data.FilteredColumnarBatch; +import io.delta.kernel.data.Row; import io.delta.kernel.engine.Engine; +import io.delta.kernel.internal.util.Utils; import io.delta.kernel.types.ByteType; import io.delta.kernel.types.DataType; import io.delta.kernel.types.DateType; import io.delta.kernel.types.DoubleType; +import io.delta.kernel.types.IntegerType; import io.delta.kernel.types.LongType; import io.delta.kernel.types.ShortType; import io.delta.kernel.types.StringType; import io.delta.kernel.types.StructType; +import io.delta.kernel.utils.CloseableIterable; import io.delta.kernel.utils.CloseableIterator; +import io.delta.kernel.utils.DataFileStatus; /** * Direct (no DML) round-trip tests for the native Delta Kernel based frame reader/writer. Each test writes a FrameBlock @@ -451,6 +467,87 @@ public void readShortByteColumnsCoercedToInt32() throws Exception { } } + @Test + public void readTypeWidenedIntFilesAsLongColumn() throws Exception { + // what Delta type widening leaves behind: the schema declares bigint while an + // older data file still physically stores INT32. The direct decode must detect + // the narrower physical type and fall back to the kernel engine for that file + // only (the INT64 file stays on the direct path), in both readers. + long[] oldInts = {1, -2, 0, Integer.MAX_VALUE, Integer.MIN_VALUE}; + long[] newLongs = {10L, -20L, 5_000_000_000L, Long.MAX_VALUE, Long.MIN_VALUE}; + Path dir = Files.createTempDirectory("sysds_delta_frame_tw_"); + String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); + try { + writeWidenedLongTable(tablePath, oldInts, newLongs); + // pin the fixture: stats present (so the pre-sized direct path is what runs, + // not the buffered fallback) and exactly one file physically narrower than + // the schema (so the per-file kernel fallback really triggers) + DeltaKernelUtils.ScanHandle handle = DeltaKernelUtils.openScan(DeltaKernelUtils.createEngine(), + DeltaKernelUtils.qualify(tablePath)); + assertTrue("fixture must carry exact row counts", handle.hasExactRowCounts()); + assertEquals("fixture must span two data files", 2, handle.scanFiles.size()); + assertEquals("fixture must contain exactly one INT32-physical data file", 1, countInt32Files(tablePath)); + FrameReader[] readers = {new FrameReaderDelta(), new FrameReaderDeltaParallel()}; + for(FrameReader reader : readers) { + FrameBlock out = reader.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1); + assertEquals("rows", oldInts.length + newLongs.length, out.getNumRows()); + assertEquals("cols", 1, out.getNumColumns()); + assertEquals("widened column surfaces as INT64", ValueType.INT64, out.getSchema()[0]); + for(int r = 0; r < oldInts.length; r++) + assertEquals("int-file cell " + r, oldInts[r], ((Number) out.get(r, 0)).longValue()); + for(int r = 0; r < newLongs.length; r++) + assertEquals("long-file cell " + r, newLongs[r], + ((Number) out.get(oldInts.length + r, 0)).longValue()); + } + } + finally { + FileUtils.deleteQuietly(dir.toFile()); + } + } + + @Test + public void readSchemaEvolvedFilesMissingAndReorderedColumns() throws Exception { + // schema evolution leftovers: one data file written before column 'w' existed + // (its cells must keep the column default, 0 for numerics), and one whose + // parquet column order is reversed relative to the table schema (values must + // land by name, not by position), identically on all three read paths. + StructType tableSchema = new StructType().add("v", LongType.LONG, true).add("w", LongType.LONG, true); + StructType vOnly = new StructType().add("v", LongType.LONG, true); + StructType reversed = new StructType().add("w", LongType.LONG, true).add("v", LongType.LONG, true); + long[] v1 = {1L, 2L, 3L}; + long[] v2 = {10L, 20L}; + long[] w2 = {100L, 200L}; + + Path dir = Files.createTempDirectory("sysds_delta_frame_se_"); + String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); + try { + commitFiles(tablePath, tableSchema, batchOf(vOnly, new LongBackedVector(LongType.LONG, v1)), + batchOf(reversed, new LongBackedVector(LongType.LONG, w2), new LongBackedVector(LongType.LONG, v2))); + FrameReader[] readers = {new FrameReaderDelta(), new FrameReaderDeltaParallel(), new FrameReaderDelta() { + @Override + protected boolean useDirectPath(DeltaKernelUtils.ScanHandle h) { + return false; + } + }}; + for(FrameReader reader : readers) { + FrameBlock out = reader.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1); + assertEquals("rows", v1.length + v2.length, out.getNumRows()); + assertEquals("cols", 2, out.getNumColumns()); + for(int r = 0; r < v1.length; r++) { + assertEquals("old-file v " + r, v1[r], ((Number) out.get(r, 0)).longValue()); + assertEquals("old-file w (missing -> default) " + r, 0L, ((Number) out.get(r, 1)).longValue()); + } + for(int r = 0; r < v2.length; r++) { + assertEquals("reordered-file v " + r, v2[r], ((Number) out.get(v1.length + r, 0)).longValue()); + assertEquals("reordered-file w " + r, w2[r], ((Number) out.get(v1.length + r, 1)).longValue()); + } + } + } + finally { + FileUtils.deleteQuietly(dir.toFile()); + } + } + @Test public void writerRejectsDimensionMismatch() throws Exception { ValueType[] schema = {ValueType.STRING, ValueType.INT64}; @@ -613,6 +710,87 @@ public ColumnVector getColumnVector(int ordinal) { DeltaKernelUtils.commit(engine, DeltaKernelUtils.qualify(tablePath), schema, singleton(fcb)); } + /** + * Creates what Delta type widening leaves behind: a table whose schema declares {@code bigint} while its first data + * file physically stores INT32 (written before the widen) and its second INT64. + */ + private static void writeWidenedLongTable(String tablePath, long[] oldInts, long[] newLongs) throws Exception { + StructType tableSchema = new StructType().add("v", LongType.LONG, true); + StructType intSchema = new StructType().add("v", IntegerType.INTEGER, true); + commitFiles(tablePath, tableSchema, batchOf(intSchema, new LongBackedVector(IntegerType.INTEGER, oldInts)), + batchOf(tableSchema, new LongBackedVector(LongType.LONG, newLongs))); + } + + /** + * Creates a table with the given schema and commits one physically-written data file per batch. The files are + * written through the kernel's parquet handler directly (bypassing the logical-data transform, which would reject + * batch schemas deviating from the table schema) and committed with their statistics, so reads take the pre-sized + * direct path. + */ + private static void commitFiles(String tablePath, StructType tableSchema, ColumnarBatch... batches) + throws Exception { + Engine engine = DeltaKernelUtils.createEngine(); + Table table = Table.forPath(engine, DeltaKernelUtils.qualify(tablePath)); + Transaction txn = table.createTransactionBuilder(engine, "SystemDS-test", Operation.CREATE_TABLE) + .withSchema(engine, tableSchema).build(engine); + Row txnState = txn.getTransactionState(engine); + DataWriteContext ctx = Transaction.getWriteContext(engine, txnState, Collections.emptyMap()); + + List files = new ArrayList<>(); + for(ColumnarBatch batch : batches) + drain(engine.getParquetHandler().writeParquetFiles(ctx.getTargetDirectory(), + singleton(new FilteredColumnarBatch(batch, Optional.empty())), ctx.getStatisticsColumns()), files); + + CloseableIterator actions = Transaction.generateAppendActions(engine, txnState, + Utils.toCloseableIterator(files.iterator()), ctx); + txn.commit(engine, CloseableIterable.inMemoryIterable(actions)); + } + + /** Count the parquet data files whose single column is physically stored as INT32. */ + private static int countInt32Files(String tablePath) throws Exception { + Configuration conf = new Configuration(); + int n = 0; + try(java.util.stream.Stream s = Files.walk(new File(tablePath).toPath())) { + for(Path p : (Iterable) s.filter(f -> f.toString().endsWith(".parquet"))::iterator) { + try(ParquetFileReader r = ParquetFileReader + .open(HadoopInputFile.fromPath(new org.apache.hadoop.fs.Path(p.toString()), conf))) { + PrimitiveTypeName t = r.getFooter().getFileMetaData().getSchema().getType(0).asPrimitiveType() + .getPrimitiveTypeName(); + if(t == PrimitiveTypeName.INT32) + n++; + } + } + } + return n; + } + + private static void drain(CloseableIterator it, List into) throws IOException { + try(CloseableIterator i = it) { + while(i.hasNext()) + into.add(i.next()); + } + } + + /** Batch view over per-column vectors (positionally matching the given schema). */ + private static ColumnarBatch batchOf(StructType schema, ColumnVector... cols) { + return new ColumnarBatch() { + @Override + public StructType getSchema() { + return schema; + } + + @Override + public int getSize() { + return cols[0].getSize(); + } + + @Override + public ColumnVector getColumnVector(int ordinal) { + return cols[ordinal]; + } + }; + } + private static CloseableIterator singleton(FilteredColumnarBatch fcb) { return new CloseableIterator() { private boolean _done = false; @@ -702,6 +880,46 @@ public void close() { } } + /** Column view exposing a long[] as a Delta integer or long column ({@code getInt} narrows). */ + private static class LongBackedVector implements ColumnVector { + private final DataType _dt; + private final long[] _vals; + + LongBackedVector(DataType dt, long[] vals) { + _dt = dt; + _vals = vals; + } + + @Override + public DataType getDataType() { + return _dt; + } + + @Override + public int getSize() { + return _vals.length; + } + + @Override + public boolean isNullAt(int rowId) { + return false; + } + + @Override + public int getInt(int rowId) { + return (int) _vals[rowId]; + } + + @Override + public long getLong(int rowId) { + return _vals[rowId]; + } + + @Override + public void close() { + } + } + /** Column view exposing a byte[] as a Delta byte column. */ private static class ByteVector implements ColumnVector { private final byte[] _vals; diff --git a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java new file mode 100644 index 00000000000..cf417092f51 --- /dev/null +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java @@ -0,0 +1,332 @@ +/* + * 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.assertTrue; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; + +import org.apache.commons.io.FileUtils; +import org.apache.sysds.common.Types.ValueType; +import org.apache.sysds.conf.ConfigurationManager; +import org.apache.sysds.conf.DMLConfig; +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.sysds.runtime.io.FrameWriterDelta; +import org.apache.sysds.test.TestUtils; +import org.junit.Test; + +/** + * Systematic shape/size coverage for the native Delta frame readers: the direct parquet column decode across input + * scales (100 rows to 1 million rows, 1 column to 1000 columns) plus the edge shapes around them: single-cell tables, + * writer batch boundaries (4096 rows), prime/odd dimensions, multi-file layouts, all-null columns, and adversarial cell + * values (NaN/infinities, signed zeros, integer extremes, empty/unicode/very long strings). + * + * Every shape is verified on all three read paths (serial direct decode as the default, parallel direct decode, and the + * forced kernel-engine buffered fallback) against the in-memory input, cell for cell. Column types cycle through all + * six writable frame value types so each typed decode loop is exercised at every shape, and every string column carries + * interspersed nulls, empty strings and multi-byte unicode so the definition-level (null) handling and UTF-8 decode are + * stressed at every size as well. + */ +public class DeltaFrameShapeCoverageTest { + + // nonsense schema/dims handed to the readers to confirm discovery from the table + private static final ValueType[] NO_SCHEMA = new ValueType[] {ValueType.STRING}; + private static final String[] NO_NAMES = new String[] {"x"}; + + // small target file size so large frames roll multiple data files and the + // per-file parallel path really splits (mirrors DeltaFrameReadWriteTest) + private static final long SMALL_TARGET_FILE_SIZE = 512L * 1024; + + // all value types the Delta frame writer can emit, cycled across columns so + // every typed decode loop (string/long/double/boolean/int/float) is hit at + // every covered shape + private static final ValueType[] TYPE_CYCLE = {ValueType.STRING, ValueType.INT64, ValueType.FP64, ValueType.BOOLEAN, + ValueType.INT32, ValueType.FP32}; + + @Test + public void tinyShapesRoundTrip() throws Exception { + // the smallest possible tables, including a 1x1 whose only (string) cell is + // null: a parquet page carrying no data bytes at all + assertRoundTrip(1, 1, 101, false, true); + assertRoundTrip(2, 1, 102, false, true); + assertRoundTrip(3, 2, 103, false, true); + assertRoundTrip(7, 6, 104, false, true); + assertRoundTrip(100, 1, 105, false, true); + assertRoundTrip(100, 3, 106, false, true); + assertRoundTrip(100, 6, 107, false, true); + } + + @Test + public void writerBatchBoundariesRoundTrip() throws Exception { + // the writer chunks frames into 4096-row batches; hit that boundary exactly, + // one below and one above, plus a multiple of it + assertRoundTrip(4095, 2, 201, false, true); + assertRoundTrip(4096, 2, 202, false, true); + assertRoundTrip(4097, 2, 203, false, true); + assertRoundTrip(8192, 5, 204, false, true); + } + + @Test + public void columnScalingRoundTrip() throws Exception { + // 1 -> 1000 columns at fixed row counts + assertRoundTrip(1000, 1, 301, false, true); + assertRoundTrip(100, 10, 302, false, true); + assertRoundTrip(100, 100, 303, false, true); + assertRoundTrip(1000, 64, 304, false, true); + assertRoundTrip(1000, 256, 305, false, true); + assertRoundTrip(100, 1000, 306, false, true); + assertRoundTrip(1000, 1000, 307, false, true); + } + + @Test + public void rowScalingRoundTrip() throws Exception { + // 100 -> 1M rows (the 100-row points are in tinyShapesRoundTrip); prime/odd + // dimensions and multi-file layouts included + assertRoundTrip(997, 7, 401, false, true); + assertRoundTrip(10_000, 3, 402, false, true); + assertRoundTrip(100_000, 6, 403, true, true); + assertRoundTrip(250_000, 2, 404, true, true); + } + + @Test + public void millionRowsRoundTrip() throws Exception { + // the 1M-row end of the row scaling, as a multi-file table so the parallel + // reader really splits across files. The kernel-path parity at this + // scale is asserted by the timing test, so the (slow) forced-buffered + // read is skipped here. + assertRoundTrip(1_000_000, 1, 501, true, false); + assertRoundTrip(1_000_000, 3, 502, true, false); + } + + @Test + public void allNullStringColumnRoundTrip() throws Exception { + // a string column that is entirely null (definition level 0 for every row, + // no data bytes in any page) next to a fully-live numeric column + int nrow = 10_000; + FrameBlock in = TestUtils.generateRandomFrameBlock(nrow, new ValueType[] {ValueType.STRING, ValueType.FP64}, + 61); + for(int r = 0; r < nrow; r++) + in.set(r, 0, null); + roundTripAllReaders("allNullString 10000x2", in, false, true); + } + + @Test + public void extremeValuesRoundTrip() throws Exception { + // adversarial cell values for every type: NaN, infinities, signed zeros, + // subnormals, MIN/MAX of each numeric width, and empty / whitespace / + // multi-byte unicode / control-character / very long strings + ValueType[] schema = {ValueType.FP64, ValueType.FP32, ValueType.INT64, ValueType.INT32, ValueType.BOOLEAN, + ValueType.STRING}; + String[] names = {"d", "f", "l", "i", "b", "s"}; + double[] d = {Double.NaN, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY, Double.MAX_VALUE, + -Double.MAX_VALUE, Double.MIN_VALUE, 0.0, -0.0}; + float[] f = {Float.NaN, Float.POSITIVE_INFINITY, Float.NEGATIVE_INFINITY, Float.MAX_VALUE, -Float.MAX_VALUE, + Float.MIN_VALUE, 0.0f, -0.0f}; + long[] l = {Long.MAX_VALUE, Long.MIN_VALUE, 0L, -1L, 1L, 1L << 40, -(1L << 40), 42L}; + int[] i = {Integer.MAX_VALUE, Integer.MIN_VALUE, 0, -1, 1, 1 << 20, -(1 << 20), 42}; + String longStr = new String(new char[10_000]).replace('\0', 'x'); + String[] s = {null, "", " ", "ü€𐍈", longStr, "line\nbreak", "tab\tsep", "quote\"'"}; + + int nrow = d.length; + FrameBlock in = new FrameBlock(schema, names); + in.ensureAllocatedColumns(nrow); + for(int r = 0; r < nrow; r++) { + in.set(r, 0, d[r]); + in.set(r, 1, f[r]); + in.set(r, 2, l[r]); + in.set(r, 3, i[r]); + in.set(r, 4, r % 2 == 0); + in.set(r, 5, s[r]); + } + roundTripAllReaders("extremes 8x6", in, false, true); + } + + @Test + public void timingDirectVsKernelAcrossShapes() throws Exception { + // direct decode vs kernel-engine path across input scales: a row-scaling + // series and a column-scaling series. + // Each shape asserts result equality of the serial direct, parallel direct + // and kernel-buffered reads, + // then times each path best-of-3 after that warm-up. The not-slower + // assertion only applies where the parquet decode dominates (>= 1M cells); + // tiny tables are dominated by Delta log replay either way. + int[][] shapes = {{100, 3}, {10_000, 3}, {1_000_000, 3}, {1000, 1}, {1000, 100}, {1000, 1000}}; + StringBuilder rep = new StringBuilder(String.format("%n%-12s %6s %11s %13s %11s %8s%n", "shape", "files", + "direct(ms)", "parallel(ms)", "kernel(ms)", "speedup")); + for(int[] shape : shapes) { + final int nrow = shape[0], ncol = shape[1]; + final String label = nrow + "x" + ncol; + final FrameBlock in = genFrame(nrow, ncol, 7000 + nrow + ncol); + final boolean multiFile = (long) nrow * ncol >= 200_000; + withTable(in, multiFile, tablePath -> { + long files = DeltaFrameTestUtils.countParquet(tablePath); + FrameReaderDelta direct = new FrameReaderDelta(); + FrameReaderDeltaParallel parallel = new FrameReaderDeltaParallel(); + FrameReaderDelta kernel = newBufferedReader(); + // correctness on all three paths at this shape (doubles as warm-up) + assertFramesEqual(label + " serial", in, + direct.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); + assertFramesEqual(label + " parallel", in, + parallel.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); + assertFramesEqual(label + " kernel", in, + kernel.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); + double directMs = bestOf3(direct, tablePath); + double parallelMs = bestOf3(parallel, tablePath); + double kernelMs = bestOf3(kernel, tablePath); + rep.append(String.format("%-12s %6d %11.1f %13.1f %11.1f %7.2fx%n", label, files, directMs, parallelMs, + kernelMs, kernelMs / directMs)); + if((long) nrow * ncol >= 1_000_000) + assertTrue(label + ": direct decode should not be slower than the kernel path (direct " + directMs + + "ms vs kernel " + kernelMs + "ms)", directMs <= kernelMs * 1.5); + }); + } + System.out.println(rep); + } + + // ------------------------------------------ + // helpers + // ------------------------------------------ + + private static ValueType[] cycleSchema(int ncol) { + ValueType[] schema = new ValueType[ncol]; + for(int c = 0; c < ncol; c++) + schema[c] = TYPE_CYCLE[c % TYPE_CYCLE.length]; + return schema; + } + + /** + * Deterministic test frame for a covered shape: random data over the cycled schema plus adversarial patterns + * injected into every string column (nulls every 7th row, empty strings every 11th, multi-byte unicode every 13th) + * so the null definition-level handling and UTF-8 decode are covered at every size, including across file and batch + * boundaries of the larger shapes. + */ + private static FrameBlock genFrame(int nrow, int ncol, long seed) { + FrameBlock fb = TestUtils.generateRandomFrameBlock(nrow, cycleSchema(ncol), seed); + ValueType[] schema = fb.getSchema(); + for(int c = 0; c < ncol; c++) { + if(schema[c] != ValueType.STRING) + continue; + for(int r = 0; r < nrow; r++) { + if(r % 7 == 0) + fb.set(r, c, null); + else if(r % 11 == 0) + fb.set(r, c, ""); + else if(r % 13 == 0) + fb.set(r, c, "ü€𐍈" + r); + } + } + return fb; + } + + @FunctionalInterface + private interface TableBody { + void accept(String tablePath) throws Exception; + } + + /** + * Write {@code in} to a fresh temp Delta table and run {@code body} against it. With {@code multiFile} a small + * target file size is configured and the resulting layout is asserted to really span multiple data files. Local + * config and the temp directory are always cleaned up. + */ + private static void withTable(FrameBlock in, boolean multiFile, TableBody body) throws Exception { + if(multiFile) { + DMLConfig conf = new DMLConfig(); + conf.setTextValue(DMLConfig.DELTA_WRITER_TARGET_FILE_SIZE, String.valueOf(SMALL_TARGET_FILE_SIZE)); + ConfigurationManager.setLocalConfig(conf); + } + Path dir = Files.createTempDirectory("sysds_delta_shape_"); + String tablePath = new File(dir.toFile(), "table").getAbsolutePath(); + try { + new FrameWriterDelta().writeFrameToHDFS(in, tablePath, in.getNumRows(), in.getNumColumns()); + if(multiFile) + assertTrue("expected a multi-file Delta table for this shape", + DeltaFrameTestUtils.countParquet(tablePath) > 1); + body.accept(tablePath); + } + finally { + if(multiFile) + ConfigurationManager.clearLocalConfigs(); + FileUtils.deleteQuietly(dir.toFile()); + } + } + + private static void assertRoundTrip(int nrow, int ncol, long seed, boolean multiFile, boolean checkBuffered) + throws Exception { + roundTripAllReaders(nrow + "x" + ncol, genFrame(nrow, ncol, seed), multiFile, checkBuffered); + } + + /** + * Write the frame and assert that the serial direct read (the default path), the parallel direct read, and (when + * {@code checkBuffered}) the forced kernel-engine buffered fallback each reproduce the input cell for cell. + */ + private static void roundTripAllReaders(String label, FrameBlock in, boolean multiFile, boolean checkBuffered) + throws Exception { + withTable(in, multiFile, tablePath -> { + assertFramesEqual(label + " serial", in, + new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); + assertFramesEqual(label + " parallel", in, + new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); + if(checkBuffered) + assertFramesEqual(label + " buffered", in, + newBufferedReader().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); + }); + } + + /** Serial reader that always declines the direct path, forcing the kernel-engine buffered read. */ + private static FrameReaderDelta newBufferedReader() { + return new FrameReaderDelta() { + @Override + protected boolean useDirectPath(DeltaKernelUtils.ScanHandle h) { + return false; + } + }; + } + + /** Best-of-3 wall time of a full table read in milliseconds. */ + private static double bestOf3(FrameReaderDelta reader, String tablePath) throws Exception { + long best = Long.MAX_VALUE; + for(int i = 0; i < 3; i++) { + long t0 = System.nanoTime(); + reader.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1); + best = Math.min(best, System.nanoTime() - t0); + } + return best / 1e6; + } + + private static void assertFramesEqual(String label, FrameBlock expected, FrameBlock actual) { + assertEquals(label + ": rows", expected.getNumRows(), actual.getNumRows()); + assertEquals(label + ": cols", expected.getNumColumns(), actual.getNumColumns()); + int ncol = expected.getNumColumns(); + for(int c = 0; c < ncol; c++) { + assertEquals(label + ": schema col " + c, expected.getSchema()[c], actual.getSchema()[c]); + assertEquals(label + ": name col " + c, expected.getColumnNames()[c], actual.getColumnNames()[c]); + } + int nrow = expected.getNumRows(); + for(int r = 0; r < nrow; r++) + for(int c = 0; c < ncol; c++) + assertEquals(label + ": cell (" + r + "," + c + ")", expected.get(r, c), actual.get(r, c)); + } +} From 4183e54f333e77be39bafa9634e136e267f5866b Mon Sep 17 00:00:00 2001 From: Jakob-al28 <04jakob28@gmail.com> Date: Sun, 12 Jul 2026 16:05:24 +0200 Subject: [PATCH 5/6] [SYSTEMDS-3949] Keep shape coverage test correctness-only --- .../io/DeltaFrameShapeCoverageTest.java | 69 +++---------------- 1 file changed, 8 insertions(+), 61 deletions(-) diff --git a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java index cf417092f51..7b842e8da8f 100644 --- a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java @@ -96,9 +96,10 @@ public void columnScalingRoundTrip() throws Exception { assertRoundTrip(100, 10, 302, false, true); assertRoundTrip(100, 100, 303, false, true); assertRoundTrip(1000, 64, 304, false, true); - assertRoundTrip(1000, 256, 305, false, true); - assertRoundTrip(100, 1000, 306, false, true); - assertRoundTrip(1000, 1000, 307, false, true); + assertRoundTrip(1000, 100, 305, false, true); + assertRoundTrip(1000, 256, 306, false, true); + assertRoundTrip(100, 1000, 307, false, true); + assertRoundTrip(1000, 1000, 308, false, true); } @Test @@ -113,12 +114,11 @@ public void rowScalingRoundTrip() throws Exception { @Test public void millionRowsRoundTrip() throws Exception { - // the 1M-row end of the row scaling, as a multi-file table so the parallel - // reader really splits across files. The kernel-path parity at this - // scale is asserted by the timing test, so the (slow) forced-buffered - // read is skipped here. + // the 1M-row end of the row scaling, as multi-file tables so the parallel + // reader really splits across files; the kernel-path parity at this scale + // is covered via the forced-buffered read of the 1Mx3 table assertRoundTrip(1_000_000, 1, 501, true, false); - assertRoundTrip(1_000_000, 3, 502, true, false); + assertRoundTrip(1_000_000, 3, 502, true, true); } @Test @@ -164,48 +164,6 @@ public void extremeValuesRoundTrip() throws Exception { roundTripAllReaders("extremes 8x6", in, false, true); } - @Test - public void timingDirectVsKernelAcrossShapes() throws Exception { - // direct decode vs kernel-engine path across input scales: a row-scaling - // series and a column-scaling series. - // Each shape asserts result equality of the serial direct, parallel direct - // and kernel-buffered reads, - // then times each path best-of-3 after that warm-up. The not-slower - // assertion only applies where the parquet decode dominates (>= 1M cells); - // tiny tables are dominated by Delta log replay either way. - int[][] shapes = {{100, 3}, {10_000, 3}, {1_000_000, 3}, {1000, 1}, {1000, 100}, {1000, 1000}}; - StringBuilder rep = new StringBuilder(String.format("%n%-12s %6s %11s %13s %11s %8s%n", "shape", "files", - "direct(ms)", "parallel(ms)", "kernel(ms)", "speedup")); - for(int[] shape : shapes) { - final int nrow = shape[0], ncol = shape[1]; - final String label = nrow + "x" + ncol; - final FrameBlock in = genFrame(nrow, ncol, 7000 + nrow + ncol); - final boolean multiFile = (long) nrow * ncol >= 200_000; - withTable(in, multiFile, tablePath -> { - long files = DeltaFrameTestUtils.countParquet(tablePath); - FrameReaderDelta direct = new FrameReaderDelta(); - FrameReaderDeltaParallel parallel = new FrameReaderDeltaParallel(); - FrameReaderDelta kernel = newBufferedReader(); - // correctness on all three paths at this shape (doubles as warm-up) - assertFramesEqual(label + " serial", in, - direct.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); - assertFramesEqual(label + " parallel", in, - parallel.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); - assertFramesEqual(label + " kernel", in, - kernel.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); - double directMs = bestOf3(direct, tablePath); - double parallelMs = bestOf3(parallel, tablePath); - double kernelMs = bestOf3(kernel, tablePath); - rep.append(String.format("%-12s %6d %11.1f %13.1f %11.1f %7.2fx%n", label, files, directMs, parallelMs, - kernelMs, kernelMs / directMs)); - if((long) nrow * ncol >= 1_000_000) - assertTrue(label + ": direct decode should not be slower than the kernel path (direct " + directMs - + "ms vs kernel " + kernelMs + "ms)", directMs <= kernelMs * 1.5); - }); - } - System.out.println(rep); - } - // ------------------------------------------ // helpers // ------------------------------------------ @@ -305,17 +263,6 @@ protected boolean useDirectPath(DeltaKernelUtils.ScanHandle h) { }; } - /** Best-of-3 wall time of a full table read in milliseconds. */ - private static double bestOf3(FrameReaderDelta reader, String tablePath) throws Exception { - long best = Long.MAX_VALUE; - for(int i = 0; i < 3; i++) { - long t0 = System.nanoTime(); - reader.readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1); - best = Math.min(best, System.nanoTime() - t0); - } - return best / 1e6; - } - private static void assertFramesEqual(String label, FrameBlock expected, FrameBlock actual) { assertEquals(label + ": rows", expected.getNumRows(), actual.getNumRows()); assertEquals(label + ": cols", expected.getNumColumns(), actual.getNumColumns()); From c409dc02c7cd2b6826a378857148ffaf78683763 Mon Sep 17 00:00:00 2001 From: Jakob-al28 <04jakob28@gmail.com> Date: Sun, 12 Jul 2026 19:20:45 +0200 Subject: [PATCH 6/6] [SYSTEMDS-3949] Reduce shape coverage test set --- .../io/DeltaFrameShapeCoverageTest.java | 134 ++++-------------- 1 file changed, 30 insertions(+), 104 deletions(-) diff --git a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java index 7b842e8da8f..b902ad22b42 100644 --- a/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java +++ b/src/test/java/org/apache/sysds/test/component/io/DeltaFrameShapeCoverageTest.java @@ -19,7 +19,6 @@ package org.apache.sysds.test.component.io; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import java.io.File; @@ -39,16 +38,14 @@ import org.junit.Test; /** - * Systematic shape/size coverage for the native Delta frame readers: the direct parquet column decode across input - * scales (100 rows to 1 million rows, 1 column to 1000 columns) plus the edge shapes around them: single-cell tables, - * writer batch boundaries (4096 rows), prime/odd dimensions, multi-file layouts, all-null columns, and adversarial cell - * values (NaN/infinities, signed zeros, integer extremes, empty/unicode/very long strings). + * Shape/size coverage for the native Delta frame readers: single-cell tables, the writer batch boundary (4096 rows), 1 + * to 1000 columns, a multi-file layout, all-null columns, and adversarial cell values (NaN/infinities, signed zeros, + * integer extremes, empty/unicode/very long strings). * * Every shape is verified on all three read paths (serial direct decode as the default, parallel direct decode, and the * forced kernel-engine buffered fallback) against the in-memory input, cell for cell. Column types cycle through all - * six writable frame value types so each typed decode loop is exercised at every shape, and every string column carries - * interspersed nulls, empty strings and multi-byte unicode so the definition-level (null) handling and UTF-8 decode are - * stressed at every size as well. + * six writable frame value types so each typed decode loop is exercised at every shape; string nulls, empty strings and + * multi-byte unicode are covered by the all-null column and extreme value tests. */ public class DeltaFrameShapeCoverageTest { @@ -68,69 +65,39 @@ public class DeltaFrameShapeCoverageTest { @Test public void tinyShapesRoundTrip() throws Exception { - // the smallest possible tables, including a 1x1 whose only (string) cell is - // null: a parquet page carrying no data bytes at all - assertRoundTrip(1, 1, 101, false, true); - assertRoundTrip(2, 1, 102, false, true); - assertRoundTrip(3, 2, 103, false, true); - assertRoundTrip(7, 6, 104, false, true); - assertRoundTrip(100, 1, 105, false, true); - assertRoundTrip(100, 3, 106, false, true); - assertRoundTrip(100, 6, 107, false, true); + assertRoundTrip(1, 1, 101, false); + assertRoundTrip(7, 6, 102, false); } @Test - public void writerBatchBoundariesRoundTrip() throws Exception { - // the writer chunks frames into 4096-row batches; hit that boundary exactly, - // one below and one above, plus a multiple of it - assertRoundTrip(4095, 2, 201, false, true); - assertRoundTrip(4096, 2, 202, false, true); - assertRoundTrip(4097, 2, 203, false, true); - assertRoundTrip(8192, 5, 204, false, true); + public void writerBatchBoundaryRoundTrip() throws Exception { + assertRoundTrip(4096, 2, 201, false); } @Test public void columnScalingRoundTrip() throws Exception { - // 1 -> 1000 columns at fixed row counts - assertRoundTrip(1000, 1, 301, false, true); - assertRoundTrip(100, 10, 302, false, true); - assertRoundTrip(100, 100, 303, false, true); - assertRoundTrip(1000, 64, 304, false, true); - assertRoundTrip(1000, 100, 305, false, true); - assertRoundTrip(1000, 256, 306, false, true); - assertRoundTrip(100, 1000, 307, false, true); - assertRoundTrip(1000, 1000, 308, false, true); + // the 1-column and 1000-column endpoints at small row counts + assertRoundTrip(1000, 1, 301, false); + assertRoundTrip(100, 1000, 302, false); } @Test - public void rowScalingRoundTrip() throws Exception { - // 100 -> 1M rows (the 100-row points are in tinyShapesRoundTrip); prime/odd - // dimensions and multi-file layouts included - assertRoundTrip(997, 7, 401, false, true); - assertRoundTrip(10_000, 3, 402, false, true); - assertRoundTrip(100_000, 6, 403, true, true); - assertRoundTrip(250_000, 2, 404, true, true); - } - - @Test - public void millionRowsRoundTrip() throws Exception { - // the 1M-row end of the row scaling, as multi-file tables so the parallel - // reader really splits across files; the kernel-path parity at this scale - // is covered via the forced-buffered read of the 1Mx3 table - assertRoundTrip(1_000_000, 1, 501, true, false); - assertRoundTrip(1_000_000, 3, 502, true, true); + public void multiFileRoundTrip() throws Exception { + // a table large enough to roll multiple data files so the per-file slicing + // of the direct and parallel paths is covered + assertRoundTrip(100_000, 6, 401, true); } @Test public void allNullStringColumnRoundTrip() throws Exception { // a string column that is entirely null (definition level 0 for every row, // no data bytes in any page) next to a fully-live numeric column - int nrow = 10_000; + int nrow = 1000; FrameBlock in = TestUtils.generateRandomFrameBlock(nrow, new ValueType[] {ValueType.STRING, ValueType.FP64}, 61); for(int r = 0; r < nrow; r++) in.set(r, 0, null); - roundTripAllReaders("allNullString 10000x2", in, false, true); + roundTripAllReaders(in, false); } @Test @@ -161,7 +128,7 @@ public void extremeValuesRoundTrip() throws Exception { in.set(r, 4, r % 2 == 0); in.set(r, 5, s[r]); } - roundTripAllReaders("extremes 8x6", in, false, true); + roundTripAllReaders(in, false); } // ------------------------------------------ @@ -175,30 +142,6 @@ private static ValueType[] cycleSchema(int ncol) { return schema; } - /** - * Deterministic test frame for a covered shape: random data over the cycled schema plus adversarial patterns - * injected into every string column (nulls every 7th row, empty strings every 11th, multi-byte unicode every 13th) - * so the null definition-level handling and UTF-8 decode are covered at every size, including across file and batch - * boundaries of the larger shapes. - */ - private static FrameBlock genFrame(int nrow, int ncol, long seed) { - FrameBlock fb = TestUtils.generateRandomFrameBlock(nrow, cycleSchema(ncol), seed); - ValueType[] schema = fb.getSchema(); - for(int c = 0; c < ncol; c++) { - if(schema[c] != ValueType.STRING) - continue; - for(int r = 0; r < nrow; r++) { - if(r % 7 == 0) - fb.set(r, c, null); - else if(r % 11 == 0) - fb.set(r, c, ""); - else if(r % 13 == 0) - fb.set(r, c, "ü€𐍈" + r); - } - } - return fb; - } - @FunctionalInterface private interface TableBody { void accept(String tablePath) throws Exception; @@ -231,25 +174,22 @@ private static void withTable(FrameBlock in, boolean multiFile, TableBody body) } } - private static void assertRoundTrip(int nrow, int ncol, long seed, boolean multiFile, boolean checkBuffered) - throws Exception { - roundTripAllReaders(nrow + "x" + ncol, genFrame(nrow, ncol, seed), multiFile, checkBuffered); + private static void assertRoundTrip(int nrow, int ncol, long seed, boolean multiFile) throws Exception { + roundTripAllReaders(TestUtils.generateRandomFrameBlock(nrow, cycleSchema(ncol), seed), multiFile); } /** - * Write the frame and assert that the serial direct read (the default path), the parallel direct read, and (when - * {@code checkBuffered}) the forced kernel-engine buffered fallback each reproduce the input cell for cell. + * Write the frame and assert that the serial direct read (the default path), the parallel direct read, and the + * forced kernel-engine buffered fallback each reproduce the input cell for cell. */ - private static void roundTripAllReaders(String label, FrameBlock in, boolean multiFile, boolean checkBuffered) - throws Exception { + private static void roundTripAllReaders(FrameBlock in, boolean multiFile) throws Exception { withTable(in, multiFile, tablePath -> { - assertFramesEqual(label + " serial", in, - new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); - assertFramesEqual(label + " parallel", in, - new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); - if(checkBuffered) - assertFramesEqual(label + " buffered", in, - newBufferedReader().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1)); + TestUtils.compareFrames(in, + new FrameReaderDelta().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), true); + TestUtils.compareFrames(in, + new FrameReaderDeltaParallel().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), true); + TestUtils.compareFrames(in, newBufferedReader().readFrameFromHDFS(tablePath, NO_SCHEMA, NO_NAMES, -1, -1), + true); }); } @@ -262,18 +202,4 @@ protected boolean useDirectPath(DeltaKernelUtils.ScanHandle h) { } }; } - - private static void assertFramesEqual(String label, FrameBlock expected, FrameBlock actual) { - assertEquals(label + ": rows", expected.getNumRows(), actual.getNumRows()); - assertEquals(label + ": cols", expected.getNumColumns(), actual.getNumColumns()); - int ncol = expected.getNumColumns(); - for(int c = 0; c < ncol; c++) { - assertEquals(label + ": schema col " + c, expected.getSchema()[c], actual.getSchema()[c]); - assertEquals(label + ": name col " + c, expected.getColumnNames()[c], actual.getColumnNames()[c]); - } - int nrow = expected.getNumRows(); - for(int r = 0; r < nrow; r++) - for(int c = 0; c < ncol; c++) - assertEquals(label + ": cell (" + r + "," + c + ")", expected.get(r, c), actual.get(r, c)); - } }