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:

    • T_FLOAT

      public static final int T_FLOAT
      See Also:

    • T_LONG

      public static final int T_LONG
      See Also:

    • T_INT

      public static final int T_INT
      See Also:

    • T_SHORT

      public static final int T_SHORT
      See Also:

    • T_BYTE

      public static final int T_BYTE
      See Also:

    • T_BOOLEAN

      public static final int T_BOOLEAN
      See Also:

    • T_STRING

      public static final int T_STRING
      See Also:

    • 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:

  • 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