From f9925e86025864893ae997411e93e4abd86e9f3f Mon Sep 17 00:00:00 2001 From: Jakob-al28 Date: Mon, 6 Jul 2026 19:54:30 +0200 Subject: [PATCH 1/3] [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/3] 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/3] [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