Package org.apache.sysds.runtime.io

# Class DeltaKernelUtils

java.lang.Object
org.apache.sysds.runtime.io.DeltaKernelUtils

public class DeltaKernelUtils extends Object
Shared helpers for the native (Spark-free) Delta Lake read/write paths used by both the matrix and frame readers/writers. Centralizes engine creation, path qualification, the scan loop (snapshot -> data files -> logical columnar batches, honoring deletion vectors), and the write transaction (logical data -> parquet -> commit).

## Nested Class Summary

Nested Classes
Modifier and Type
Class
Description
`static interface`
`DeltaKernelUtils.BatchConsumer`
Consumes a whole columnar batch.
`static final class`
`DeltaKernelUtils.ScanHandle`
Opened latest snapshot of a Delta table: the logical schema plus everything needed to (re)read its data files, including the list of per-data-file scan rows.
`static final class`
`DeltaKernelUtils.UnsupportedDirectDecodeException`
Thrown when a data file's parquet layout cannot be decoded directly into the typed output columns, e.g.

## Field Summary

Fields
Modifier and Type
Field
Description
`static final long`
`ADAPTIVE_WRITER_MIN_FILE_SIZE`
Floor on the adaptive writer target file size.
`static final int`
`T_BOOLEAN`
`static final int`
`T_BYTE`
`static final int`
`T_DOUBLE`
Internal Delta column type codes shared by the matrix and frame readers to dispatch boxing-free primitive column access.
`static final int`
`T_FLOAT`
`static final int`
`T_INT`
`static final int`
`T_LONG`
`static final int`
`T_SHORT`
`static final int`
`T_STRING`

## Method Summary

Modifier and Type
Method
Description
`static long`
`adaptiveWriterTargetFileSize(long estimatedBytes)`
Compute the parquet target data-file size (bytes) for writing a table of the given estimated size.
`static void`
`commit(io.delta.kernel.engine.Engine engine, String tablePath, io.delta.kernel.types.StructType schema, io.delta.kernel.utils.CloseableIterator<io.delta.kernel.data.FilteredColumnarBatch> logicalData)`
Create a Delta table at the target path and commit the given logical data as parquet data files in a single transaction.
`static int`
`countSelected(int size, boolean[] selected)`
`static io.delta.kernel.engine.Engine`
`createEngine()`
`static io.delta.kernel.engine.Engine`
`createWriteEngine(long estimatedBytes)`
Create an engine for writing a table of the given estimated size, configured with an adaptive target data-file size (see `adaptiveWriterTargetFileSize(long)`).
`static String`
`dataFilePath(io.delta.kernel.data.Row scanFileRow)`
`static int`
`decodeDataFileInto(String filePath, io.delta.kernel.types.StructType physicalSchema, int[] readCodes, Object[] dest, int destOff, int limit, String tablePath)`
Decode one Delta data file into pre-allocated typed column arrays at the given absolute row offset, through parquet-mr's column API (`ColumnReadStoreImpl`/`ColumnReader`) with no kernel engine or intermediate batch vectors in the path.
`static DeltaKernelUtils.ScanHandle`
`openScan(io.delta.kernel.engine.Engine engine, String tablePath)`
Open the latest snapshot of a Delta table and enumerate its data files.
`static String`
`qualify(String fname)`
Resolve a (possibly relative) path to a fully-qualified URI so the kernel's default engine can locate the table on the right filesystem.
`static void`
`readScanFile(io.delta.kernel.engine.Engine engine, io.delta.kernel.data.Row scanState, io.delta.kernel.types.StructType physicalReadSchema, io.delta.kernel.data.Row scanFileRow, DeltaKernelUtils.BatchConsumer consumer)`
Read a single Delta data file (identified by its scan-file row), decoding its parquet batches and applying any deletion vector, invoking the consumer once per (logical) batch.
`static io.delta.kernel.types.StructType`
`scan(io.delta.kernel.engine.Engine engine, String tablePath, Function<io.delta.kernel.types.StructType,DeltaKernelUtils.BatchConsumer> consumerFactory)`
Scan the latest snapshot of a Delta table sequentially, invoking the batch consumer for every data batch.
`static boolean`
`supportsDirectDecode(io.delta.kernel.types.StructType logicalSchema, io.delta.kernel.types.StructType physicalSchema)`
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.
`static int`
`typeCode(io.delta.kernel.types.DataType dt)`
Map a Delta Kernel `DataType` to an internal type code (see the `T_*` constants).

### Methods inherited from class java.lang.Object

`equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

## Field Details

### T_DOUBLE

public static final int T_DOUBLE
Internal Delta column type codes shared by the matrix and frame readers to dispatch boxing-free primitive column access.

See Also:
  - Constant Field Values

### T_FLOAT

public static final int T_FLOAT

See Also:
  - Constant Field Values

### T_LONG

public static final int T_LONG

See Also:
  - Constant Field Values

### T_INT

public static final int T_INT

See Also:
  - Constant Field Values

### T_SHORT

public static final int T_SHORT

See Also:
  - Constant Field Values

### T_BYTE

public static final int T_BYTE

See Also:
  - Constant Field Values

### T_BOOLEAN

public static final int T_BOOLEAN

See Also:
  - Constant Field Values

### T_STRING

public static final int T_STRING

See Also:
  - Constant Field Values

### ADAPTIVE_WRITER_MIN_FILE_SIZE

public static final long ADAPTIVE_WRITER_MIN_FILE_SIZE
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.

See Also:
  - Constant Field Values

## Method Details

### typeCode

public static int typeCode(io.delta.kernel.types.DataType dt)
Map a Delta Kernel `DataType` to an internal type code (see the `T_*` constants). Returned once per column so the per-cell read loop can switch on a primitive int instead of repeating `instanceof` checks.

Parameters:
  `dt` - the Delta column data type
Returns:
  the matching `T_*` code, or `-1` if the type is not supported

### countSelected

public static int countSelected(int size, boolean[] selected)

Parameters:
  `size` - number of rows in the batch
  `selected` - per-row selection mask, or `null` if all rows are live
Returns:
  the number of live rows in the batch

### supportsDirectDecode

public static boolean supportsDirectDecode(io.delta.kernel.types.StructType logicalSchema, io.delta.kernel.types.StructType physicalSchema)
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 `row_index` (only requested for deletion-vector reads). Deletion vectors themselves are excluded separately via the exact-row-count check.

Parameters:
  `logicalSchema` - the table's logical schema
  `physicalSchema` - the physical read schema from the scan state
Returns:
  true if data files can be decoded directly

### dataFilePath

public static String dataFilePath(io.delta.kernel.data.Row scanFileRow)

Parameters:
  `scanFileRow` - a scan-file row @return the fully-qualified path of its data file

### decodeDataFileInto

public static int decodeDataFileInto(String filePath, io.delta.kernel.types.StructType physicalSchema, int[] readCodes, Object[] dest, int destOff, int limit, String tablePath) throws IOException
Decode one Delta data file into pre-allocated typed column arrays at the given absolute row offset, through parquet-mr's column API (`ColumnReadStoreImpl`/`ColumnReader`) with no kernel engine or intermediate batch vectors in the path. Columns are resolved by parquet field id first (column mapping mode `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.

Parameters:
  `filePath` - fully-qualified path of the parquet data file
  `physicalSchema` - physical read schema (positionally 1:1 with the output columns)
  `readCodes` - per-column type codes (see the `T_*` constants)
  `dest` - pre-allocated per-column backing arrays
  `destOff` - absolute row offset of this file's first row
  `limit` - exclusive upper row bound of this file's slice
  `tablePath` - table path for error messages
Returns:
  the number of rows decoded
Throws:
  `IOException` - on read failure
  `DeltaKernelUtils.UnsupportedDirectDecodeException` - if a column's parquet layout does not match the Delta column type (thrown before any output is written, so the caller can fall back to the kernel for this file)

### createEngine

public static io.delta.kernel.engine.Engine createEngine()

### adaptiveWriterTargetFileSize

public static long adaptiveWriterTargetFileSize(long estimatedBytes)
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 `ADAPTIVE_WRITER_MIN_FILE_SIZE` unless the configured target is itself smaller than that floor (in which case the configured target wins).

Parameters:
  `estimatedBytes` - estimate of the table's size (the block in-memory size is a fine proxy)
Returns:
  the target max parquet data-file size in bytes

### createWriteEngine

public static io.delta.kernel.engine.Engine createWriteEngine(long estimatedBytes)
Create an engine for writing a table of the given estimated size, configured with an adaptive target data-file size (see `adaptiveWriterTargetFileSize(long)`). A fresh (uncached) configuration is built since writes happen once per table, not per data file.

Parameters:
  `estimatedBytes` - estimate of the table's size (the block in-memory size is a fine proxy)
Returns:
  a Delta Kernel engine for the write

### qualify

public static String qualify(String fname)
Resolve a (possibly relative) path to a fully-qualified URI so the kernel's default engine can locate the table on the right filesystem.

Parameters:
  `fname` - input path
Returns:
  fully-qualified table path

### openScan

public static DeltaKernelUtils.ScanHandle openScan(io.delta.kernel.engine.Engine engine, String tablePath) throws IOException
Open the latest snapshot of a Delta table and enumerate its data files.

Parameters:
  `engine` - delta kernel engine
  `tablePath` - fully-qualified table path
Returns:
  a handle carrying the schema, scan state, physical read schema and one scan-file row per data file
Throws:
  `IOException` - on metadata read failure

### readScanFile

public static void readScanFile(io.delta.kernel.engine.Engine engine, io.delta.kernel.data.Row scanState, io.delta.kernel.types.StructType physicalReadSchema, io.delta.kernel.data.Row scanFileRow, DeltaKernelUtils.BatchConsumer consumer) throws IOException
Read a single Delta data file (identified by its scan-file row), decoding its parquet batches and applying any deletion vector, invoking the consumer once per (logical) batch. Safe to call concurrently for distinct files as long as each call uses its own `engine`.

Parameters:
  `engine` - delta kernel engine
  `scanState` - scan state from `openScan(io.delta.kernel.engine.Engine, java.lang.String)`
  `physicalReadSchema` - physical read schema from `openScan(io.delta.kernel.engine.Engine, java.lang.String)`
  `scanFileRow` - the data file's scan-file row
  `consumer` - batch consumer
Throws:
  `IOException` - on read failure

### scan

public static io.delta.kernel.types.StructType scan(io.delta.kernel.engine.Engine engine, String tablePath, Function<io.delta.kernel.types.StructType,DeltaKernelUtils.BatchConsumer> consumerFactory) throws IOException
Scan the latest snapshot of a Delta table sequentially, invoking the batch consumer for every data batch. The consumer is created lazily from the table schema (so callers can size buffers / derive per-column types up front).

Parameters:
  `engine` - delta kernel engine
  `tablePath` - fully-qualified table path
  `consumerFactory` - builds the batch consumer from the table schema
Returns:
  the logical table schema
Throws:
  `IOException` - on read failure

### commit

public static void commit(io.delta.kernel.engine.Engine engine, String tablePath, io.delta.kernel.types.StructType schema, io.delta.kernel.utils.CloseableIterator<io.delta.kernel.data.FilteredColumnarBatch> logicalData) throws IOException
Create a Delta table at the target path and commit the given logical data as parquet data files in a single transaction. Any existing file/directory at the target path is deleted first, so a write fully replaces what was there (matching the overwrite semantics of the other SystemDS writers). Append / incremental table updates are not supported.

Parameters:
  `engine` - delta kernel engine
  `tablePath` - fully-qualified table path
  `schema` - table schema to create
  `logicalData` - logical (unpartitioned) data batches to write
Throws:
  `IOException` - on write failure