[SYSTEMDS-3949] Column API parquet decode for Delta frame reads

src/main/java/org/apache/sysds/runtime/io/DeltaKernelUtils.java

@@ -22,14 +22,26 @@
 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;
 
 import org.apache.commons.logging.Log;
 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<Integer, String> idToName = new HashMap<>();
+		Map<String, String> 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;

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.
  *
  * <p>
  * 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;
 		}
 

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;
 			});
 		}