[SYSTEMDS-3949] Add native Delta Lake frame read/write via Delta Kernel

src/main/java/org/apache/sysds/conf/ConfigurationManager.java

@@ -268,11 +268,16 @@ public static int getDeltaWriterBatchSize() {
 		return getDMLConfig().getIntValue(DMLConfig.DELTA_WRITER_BATCH_SIZE);
 	}
 
-	/** @return target data-file size (bytes) for the native Delta writer */
+	/** @return upper bound (bytes) on the native Delta writer's target data-file size */
 	public static long getDeltaWriterTargetFileSize() {
 		return Long.parseLong(getDMLConfig().getTextValue(DMLConfig.DELTA_WRITER_TARGET_FILE_SIZE));
 	}
 
+	/** @return whether the native Delta writer adaptively sizes data files for parallel reads */
+	public static boolean isDeltaWriterAdaptiveFileSize() {
+		return getDMLConfig().getBooleanValue(DMLConfig.DELTA_WRITER_ADAPTIVE_FILE_SIZE);
+	}
+
 	public static boolean isFederatedSSL(){
 		return getDMLConfig().getBooleanValue(DMLConfig.USE_SSL_FEDERATED_COMMUNICATION);
 	}

src/main/java/org/apache/sysds/conf/DMLConfig.java

@@ -73,7 +73,8 @@ public class DMLConfig
 	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_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: target data-file size in bytes (smaller -> more files -> more parallel-read throughput)
+	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)
 	public static final String PARALLEL_ENCODE      = "sysds.parallel.encode";  // boolean: enable multi-threaded transformencode and apply
 	public static final String PARALLEL_ENCODE_STAGED = "sysds.parallel.encode.staged";
 	public static final String PARALLEL_ENCODE_APPLY_BLOCKS = "sysds.parallel.encode.applyBlocks";
@@ -163,7 +164,8 @@ public class DMLConfig
 		_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_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 target data-file size (Delta Kernel default 128MB) -> more files -> more parallel-read throughput
+		_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
 		_defaultVals.put(PARALLEL_TOKENIZE,      "false");
 		_defaultVals.put(PARALLEL_TOKENIZE_NUM_BLOCKS, "64");
 		_defaultVals.put(FRAME_TO_MATRIX_WARN_CAST, "false");

src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java

@@ -23,6 +23,7 @@
 import org.apache.commons.lang3.ArrayUtils;
 import org.apache.commons.lang3.mutable.MutableBoolean;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.sysds.api.DMLScript;
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.FileFormat;
 import org.apache.sysds.common.Types.ValueType;
@@ -203,13 +204,19 @@ protected FrameBlock readBlobFromHDFS(String fname, long[] dims) throws IOExcept
 			.createFrameReader(iimd.getFileFormat(), getFileFormatProperties())
 			.readFrameFromHDFS(fname, lschema, dc.getRows(), dc.getCols());
 
-		if(iimd.getFileFormat() == FileFormat.CSV)
+		// sanity check correct output (before dereferencing data below)
+		if(data == null)
+			throw new IOException("Unable to load frame from file: " + fname);
+
+		//Delta and CSV discover dimensions (and Delta also schema) at read time, so
+		//refresh the cached metadata to reflect the materialized frame block.
+		if(iimd.getFileFormat() == FileFormat.CSV || iimd.getFileFormat() == FileFormat.DELTA) {
 			_metaData = _metaData instanceof MetaDataFormat ? new MetaDataFormat(data.getDataCharacteristics(),
 				iimd.getFileFormat()) : new MetaData(data.getDataCharacteristics());
+			if(iimd.getFileFormat() == FileFormat.DELTA)
+				_schema = data.getSchema();
+		}
 
-		// sanity check correct output
-		if(data == null)
-			throw new IOException("Unable to load frame from file: " + fname);
 		return data;
 	}
 
@@ -293,6 +300,9 @@ protected void writeBlobToHDFS(String fname, String ofmt, int rep, FileFormatPro
 
 		FrameWriter writer = FrameWriterFactory.createFrameWriter(fmt, fprop);
 		writer.writeFrameToHDFS(_data, fname, getNumRows(), getNumColumns());
+
+		if(DMLScript.STATISTICS)
+			CacheStatistics.incrementHDFSWrites();
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/frame/data/columns/ArrayFactory.java

@@ -123,6 +123,87 @@ public static <T> RaggedArray<T> create(T[] col, int m) {
 		return new RaggedArray<>(col, m);
 	}
 
+	/**
+	 * Wrap a fully populated raw typed column array into an {@link Array} of the given value type. The runtime type of
+	 * {@code col} must match the primitive backing type of {@code vt} (e.g. {@code double[]} for {@link ValueType#FP64},
+	 * {@code String[]} for {@link ValueType#STRING}).
+	 *
+	 * <p>For {@link ValueType#BOOLEAN} this mirrors {@link #allocateBoolean(int)}: a {@code boolean[]} longer than
+	 * {@link #bitSetSwitchPoint} is bit-packed into a compact {@link BitSetArray} (so a bulk decoder that fills a
+	 * plain {@code boolean[]} still ends up with the same representation as every other frame allocation path),
+	 * while shorter columns stay a plain {@link BooleanArray}.</p>
+	 *
+	 * @param vt  the value type of the column
+	 * @param col the backing array to wrap
+	 * @return an {@link Array} view over {@code col} (boolean columns may be bit-packed rather than wrapped in place)
+	 */
+	public static Array<?> create(ValueType vt, Object col) {
+		switch(vt) {
+			case FP64:
+				return create((double[]) col);
+			case FP32:
+				return create((float[]) col);
+			case INT64:
+				return create((long[]) col);
+			case UINT4:
+			case UINT8:
+			case INT32:
+				return create((int[]) col);
+			case BOOLEAN: {
+				boolean[] b = (boolean[]) col;
+				return b.length > bitSetSwitchPoint ? new BitSetArray(b) : create(b);
+			}
+			case CHARACTER:
+				return create((char[]) col);
+			case HASH64:
+				return createHash64((long[]) col);
+			case HASH32:
+				return createHash32((int[]) col);
+			case UNKNOWN:
+			case STRING:
+			default:
+				return create((String[]) col);
+		}
+	}
+
+	/**
+	 * Allocate the raw backing array for a column of the given value type: the inverse of
+	 * {@link #create(ValueType, Object)}. Returns {@code double[]} for {@link ValueType#FP64},
+	 * {@code int[]} for INT32/UINT/HASH32, {@code long[]} for INT64/HASH64, {@code String[]} for STRING, etc. The
+	 * runtime array type matches what {@link #create(ValueType, Object)} expects, so a bulk decoder can fill this
+	 * primitive array directly and then wrap it via {@code create(vt, backing)}.
+	 *
+	 * @param vt   the value type of the column
+	 * @param nRow the number of rows to allocate
+	 * @return a freshly allocated raw backing array of the matching primitive/object type
+	 */
+	public static Object allocateBacking(ValueType vt, int nRow) {
+		switch(vt) {
+			case FP64:
+				return new double[nRow];
+			case FP32:
+				return new float[nRow];
+			case INT64:
+			case HASH64:
+				return new long[nRow];
+			case UINT4:
+			case UINT8:
+				LOG.warn("Not supported allocation of UInt 4 or 8 array: defaulting to Int32");
+				// fall through: UINT4/UINT8 are backed by int[] (wrapped as Int32)
+			case INT32:
+			case HASH32:
+				return new int[nRow];
+			case BOOLEAN:
+				return new boolean[nRow];
+			case CHARACTER:
+				return new char[nRow];
+			case UNKNOWN:
+			case STRING:
+			default:
+				return new String[nRow];
+		}
+	}
+
 	public static long getInMemorySize(ValueType type, int _numRows, boolean containsNull) {
 		if(containsNull) {
 			switch(type) {
@@ -221,27 +302,8 @@ public static Array<?> allocate(ValueType v, int nRow) {
 		switch(v) {
 			case BOOLEAN:
 				return allocateBoolean(nRow);
-			case UINT4:
-			case UINT8:
-				LOG.warn("Not supported allocation of UInt 4 or 8 array: defaulting to Int32");
-			case INT32:
-				return new IntegerArray(new int[nRow]);
-			case INT64:
-				return new LongArray(new long[nRow]);
-			case FP32:
-				return new FloatArray(new float[nRow]);
-			case FP64:
-				return new DoubleArray(new double[nRow]);
-			case CHARACTER:
-				return new CharArray(new char[nRow]);
-			case HASH64:
-				return new HashLongArray(new long[nRow]);
-			case HASH32:
-				return new HashIntegerArray(new int[nRow]);
-			case UNKNOWN:
-			case STRING:
 			default:
-				return new StringArray(new String[nRow]);
+				return create(v, allocateBacking(v, nRow));
 		}
 	}
 

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

@@ -26,9 +26,12 @@
 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.sysds.conf.ConfigurationManager;
+import org.apache.sysds.hops.OptimizerUtils;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.util.HDFSTool;
 
@@ -77,6 +80,8 @@
  */
 public class DeltaKernelUtils {
 
+	private static final Log LOG = LogFactory.getLog(DeltaKernelUtils.class.getName());
+
 	private static final String ENGINE_INFO = "Apache SystemDS";
 
 	/** Reused thread-safe JSON reader for the per-file Delta stats (numRecords). */
@@ -157,17 +162,25 @@ public static int countSelected(int size, boolean[] selected) {
 		return n;
 	}
 
+	/** 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;
+
+	private static Configuration buildConf(Configuration base, int batchSize, long targetFileSize) {
+		Configuration c = new Configuration(base);
+		c.setInt(CONF_READER_BATCH_SIZE, batchSize);
+		c.setLong(CONF_WRITER_TARGET_FILE_SIZE, targetFileSize);
+		return c;
+	}
+
 	private static synchronized Configuration deltaConf() {
 		Configuration base = ConfigurationManager.getCachedJobConf();
 		int batchSize = ConfigurationManager.getDeltaReaderBatchSize();
 		long targetFileSize = ConfigurationManager.getDeltaWriterTargetFileSize();
 		if(cachedConf == null || cachedConfBase != base
 			|| cachedBatchSize != batchSize || cachedTargetFileSize != targetFileSize)
 		{
-			Configuration c = new Configuration(base);
-			c.setInt(CONF_READER_BATCH_SIZE, batchSize);
-			c.setLong(CONF_WRITER_TARGET_FILE_SIZE, targetFileSize);
-			cachedConf = c;
+			cachedConf = buildConf(base, batchSize, targetFileSize);
 			cachedConfBase = base;
 			cachedBatchSize = batchSize;
 			cachedTargetFileSize = targetFileSize;
@@ -179,6 +192,47 @@ public static Engine createEngine() {
 		return DefaultEngine.create(deltaConf());
 	}
 
+	/**
+	 * Compute the parquet target data-file size (bytes) for writing a table of the given
+	 * estimated size. With adaptive sizing enabled the writer aims for roughly one data
+	 * file per expected parallel reader (so the native per-file parallel read can use all
+	 * threads): never above the configured target, and never below
+	 * {@code ADAPTIVE_WRITER_MIN_FILE_SIZE} unless the configured target is itself smaller
+	 * than that floor (in which case the configured target wins).
+	 *
+	 * @param estimatedBytes estimate of the table's size (the block in-memory size is a fine proxy)
+	 * @return the target max parquet data-file size in bytes
+	 */
+	public static long adaptiveWriterTargetFileSize(long estimatedBytes) {
+		long configured = ConfigurationManager.getDeltaWriterTargetFileSize();
+		if(!ConfigurationManager.isDeltaWriterAdaptiveFileSize() || estimatedBytes <= 0)
+			return configured;
+		int par = Math.max(1, OptimizerUtils.getParallelBinaryReadParallelism());
+		long perReader = Math.max(1, estimatedBytes / par);
+		//never above the configured cap, never below the floor (unless the cap itself is lower)
+		long target = Math.min(configured, Math.max(ADAPTIVE_WRITER_MIN_FILE_SIZE, perReader));
+		if(LOG.isDebugEnabled())
+			LOG.debug("Delta adaptive file size: est=" + estimatedBytes + "B par=" + par + " -> target=" + target
+				+ "B (cap=" + configured + "B, floor=" + ADAPTIVE_WRITER_MIN_FILE_SIZE + "B)");
+		return target;
+	}
+
+	/**
+	 * Create an engine for writing a table of the given estimated size, configured with an
+	 * adaptive target data-file size (see {@link #adaptiveWriterTargetFileSize(long)}). A fresh
+	 * (uncached) configuration is built since writes happen once per table, not per data file.
+	 *
+	 * @param estimatedBytes estimate of the table's size (the block in-memory size is a fine proxy)
+	 * @return a Delta Kernel engine for the write
+	 */
+	public static Engine createWriteEngine(long estimatedBytes) {
+		//the reader batch size is irrelevant on the write path but is set to keep the
+		//conf shape identical to deltaConf(); only the target file size matters here.
+		Configuration c = buildConf(ConfigurationManager.getCachedJobConf(),
+			ConfigurationManager.getDeltaReaderBatchSize(), adaptiveWriterTargetFileSize(estimatedBytes));
+		return DefaultEngine.create(c);
+	}
+
 	/**
 	 * Resolve a (possibly relative) path to a fully-qualified URI so the
 	 * kernel's default engine can locate the table on the right filesystem.