Fix Delta IO compilation errors due to Delta Kernel API mismatches

durgaprasadml · durgaprasadml · commit 47a7fb4d20a1 · 2026-06-10T19:45:20.000+05:30
diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java
@@ -30,6 +30,7 @@
 import io.delta.kernel.data.MapValue;
 import io.delta.kernel.defaults.engine.DefaultEngine;
 import io.delta.kernel.engine.Engine;
+import io.delta.kernel.engine.FileReadResult;
 import io.delta.kernel.internal.InternalScanFileUtils;
 import io.delta.kernel.internal.data.ScanStateRow;
 import io.delta.kernel.types.ArrayType;
@@ -52,7 +53,6 @@
 import io.delta.kernel.utils.CloseableIterator;
 import io.delta.kernel.utils.FileStatus;
 import java.io.Serializable;
-import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -67,7 +67,6 @@
 import org.apache.beam.sdk.values.PBegin;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.Row;
-import org.apache.hadoop.conf.Configuration;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
@@ -127,25 +126,29 @@ public ReadRows withConfig(Map<String, String> config) {
 
     @Override
     public PCollection<Row> expand(PBegin input) {
-      if (getTablePath() == null) {
+      String tablePath = getTablePath();
+      if (tablePath == null) {
         throw new IllegalArgumentException("Table path must be set.");
       }
 
       org.apache.hadoop.conf.Configuration hadoopConfig = getHadoopConfiguration(getHadoopConfig());
       Engine engine = DefaultEngine.create(hadoopConfig);
-      Table table = Table.forPath(engine, getTablePath());
+      Table table = Table.forPath(engine, tablePath);
 
       try {
         Snapshot snapshot = buildSnapshot(engine, table);
-        Scan scan = snapshot.getScanBuilder(engine).build();
-        StructType readSchema = scan.getSchema(engine);
+        Scan scan = snapshot.getScanBuilder().build();
+        io.delta.kernel.data.Row scanState = scan.getScanState(engine);
+        StructType readSchema = ScanStateRow.getLogicalSchema(scanState);
         org.apache.beam.sdk.schemas.Schema beamSchema = inferBeamSchema(readSchema);
 
-        List<DeltaFileDescriptor> fileDescriptors = buildFileDescriptors(engine, scan);
+        List<DeltaFileDescriptor> fileDescriptors = buildFileDescriptors(engine, scan, tablePath);
 
         return input
-            .apply("CreateFileDescriptors", Create.of(fileDescriptors)
-                .withCoder(SerializableCoder.of(DeltaFileDescriptor.class)))
+            .apply(
+                "CreateFileDescriptors",
+                Create.of(fileDescriptors)
+                    .withCoder(SerializableCoder.of(DeltaFileDescriptor.class)))
             .apply("ReadFile", ParDo.of(new ReadFileFn(beamSchema)))
             .setRowSchema(beamSchema);
 
@@ -155,34 +158,38 @@ public PCollection<Row> expand(PBegin input) {
     }
 
     private Snapshot buildSnapshot(Engine engine, Table table) throws Exception {
-      if (getVersion() != null) {
-        return table.getSnapshotAsOfVersion(engine, getVersion());
-      } else if (getTimestamp() != null) {
-        long epochMillis = org.joda.time.Instant.parse(getTimestamp()).getMillis();
+      Long version = getVersion();
+      String timestamp = getTimestamp();
+      if (version != null) {
+        return table.getSnapshotAsOfVersion(engine, version);
+      } else if (timestamp != null) {
+        long epochMillis = org.joda.time.Instant.parse(timestamp).getMillis();
         return table.getSnapshotAsOfTimestamp(engine, epochMillis);
       } else {
         return table.getLatestSnapshot(engine);
       }
     }
 
-    private List<DeltaFileDescriptor> buildFileDescriptors(Engine engine, Scan scan) throws Exception {
+    private List<DeltaFileDescriptor> buildFileDescriptors(
+        Engine engine, Scan scan, String tablePath) throws Exception {
       List<DeltaFileDescriptor> descriptors = new ArrayList<>();
       try (CloseableIterator<FilteredColumnarBatch> scanFileIter = scan.getScanFiles(engine)) {
         while (scanFileIter.hasNext()) {
           FilteredColumnarBatch scanFilesBatch = scanFileIter.next();
-          try (CloseableIterator<io.delta.kernel.data.Row> scanFileRows = scanFilesBatch.getRows()) {
+          try (CloseableIterator<io.delta.kernel.data.Row> scanFileRows =
+              scanFilesBatch.getRows()) {
             while (scanFileRows.hasNext()) {
               io.delta.kernel.data.Row scanFileRow = scanFileRows.next();
               FileStatus fileStatus = InternalScanFileUtils.getAddFileStatus(scanFileRow);
-              descriptors.add(new DeltaFileDescriptor(
-                  getTablePath(),
-                  fileStatus.getPath(),
-                  fileStatus.getSize(),
-                  fileStatus.getModificationTime(),
-                  getHadoopConfig(),
-                  getVersion(),
-                  getTimestamp()
-              ));
+              descriptors.add(
+                  new DeltaFileDescriptor(
+                      tablePath,
+                      fileStatus.getPath(),
+                      fileStatus.getSize(),
+                      fileStatus.getModificationTime(),
+                      getHadoopConfig(),
+                      getVersion(),
+                      getTimestamp()));
             }
           }
         }
@@ -256,47 +263,50 @@ public ReadFileFn(org.apache.beam.sdk.schemas.Schema beamSchema) {
     @ProcessElement
     public void processElement(ProcessContext c) throws Exception {
       DeltaFileDescriptor desc = c.element();
-      org.apache.hadoop.conf.Configuration hadoopConfig = getHadoopConfiguration(desc.getHadoopConfig());
+      org.apache.hadoop.conf.Configuration hadoopConfig =
+          getHadoopConfiguration(desc.getHadoopConfig());
       Engine engine = DefaultEngine.create(hadoopConfig);
       Table table = Table.forPath(engine, desc.getTablePath());
 
       Snapshot snapshot;
-      if (desc.getVersion() != null) {
-        snapshot = table.getSnapshotAsOfVersion(engine, desc.getVersion());
-      } else if (desc.getTimestamp() != null) {
-        long epochMillis = org.joda.time.Instant.parse(desc.getTimestamp()).getMillis();
+      Long version = desc.getVersion();
+      String timestamp = desc.getTimestamp();
+      if (version != null) {
+        snapshot = table.getSnapshotAsOfVersion(engine, version);
+      } else if (timestamp != null) {
+        long epochMillis = org.joda.time.Instant.parse(timestamp).getMillis();
         snapshot = table.getSnapshotAsOfTimestamp(engine, epochMillis);
       } else {
         snapshot = table.getLatestSnapshot(engine);
       }
 
-      Scan scan = snapshot.getScanBuilder(engine).build();
+      Scan scan = snapshot.getScanBuilder().build();
       io.delta.kernel.data.Row scanState = scan.getScanState(engine);
 
       try (CloseableIterator<FilteredColumnarBatch> scanFileIter = scan.getScanFiles(engine)) {
         while (scanFileIter.hasNext()) {
           FilteredColumnarBatch scanFilesBatch = scanFileIter.next();
-          try (CloseableIterator<io.delta.kernel.data.Row> scanFileRows = scanFilesBatch.getRows()) {
+          try (CloseableIterator<io.delta.kernel.data.Row> scanFileRows =
+              scanFilesBatch.getRows()) {
             while (scanFileRows.hasNext()) {
               io.delta.kernel.data.Row scanFileRow = scanFileRows.next();
               FileStatus fileStatus = InternalScanFileUtils.getAddFileStatus(scanFileRow);
               if (fileStatus.getPath().equals(desc.getFilePath())) {
                 StructType physicalReadSchema = ScanStateRow.getPhysicalDataReadSchema(scanState);
                 CloseableIterator<ColumnarBatch> physicalDataIter =
-                    engine.getParquetHandler().readParquetFiles(
-                        singletonCloseableIterator(fileStatus),
-                        physicalReadSchema,
-                        Optional.empty()).map(FilteredColumnarBatch::getData);
-                try (
-                    CloseableIterator<FilteredColumnarBatch> transformedData =
-                        Scan.transformPhysicalData(
-                            engine,
-                            scanState,
-                            scanFileRow,
-                            physicalDataIter)) {
+                    engine
+                        .getParquetHandler()
+                        .readParquetFiles(
+                            singletonCloseableIterator(fileStatus),
+                            physicalReadSchema,
+                            Optional.empty())
+                        .map(FileReadResult::getData);
+                try (CloseableIterator<FilteredColumnarBatch> transformedData =
+                    Scan.transformPhysicalData(engine, scanState, scanFileRow, physicalDataIter)) {
                   while (transformedData.hasNext()) {
                     FilteredColumnarBatch filteredData = transformedData.next();
-                    try (CloseableIterator<io.delta.kernel.data.Row> rows = filteredData.getRows()) {
+                    try (CloseableIterator<io.delta.kernel.data.Row> rows =
+                        filteredData.getRows()) {
                       while (rows.hasNext()) {
                         io.delta.kernel.data.Row row = rows.next();
                         c.output(convertKernelRowToBeamRow(row, beamSchema));
@@ -323,8 +333,9 @@ private static org.apache.hadoop.conf.Configuration getHadoopConfiguration(
     return config;
   }
 
-  private static org.apache.beam.sdk.schemas.Schema inferBeamSchema(StructType structType) {
-    org.apache.beam.sdk.schemas.Schema.Builder builder = org.apache.beam.sdk.schemas.Schema.builder();
+  static org.apache.beam.sdk.schemas.Schema inferBeamSchema(StructType structType) {
+    org.apache.beam.sdk.schemas.Schema.Builder builder =
+        org.apache.beam.sdk.schemas.Schema.builder();
     for (StructField field : structType.fields()) {
       builder.addField(field.getName(), toBeamFieldType(field.getDataType()));
     }
@@ -355,22 +366,23 @@ private static org.apache.beam.sdk.schemas.Schema.FieldType toBeamFieldType(Data
     } else if (dataType instanceof TimestampType || dataType instanceof DateType) {
       return org.apache.beam.sdk.schemas.Schema.FieldType.DATETIME;
     } else if (dataType instanceof StructType) {
-      return org.apache.beam.sdk.schemas.Schema.FieldType.row(inferBeamSchema((StructType) dataType));
+      return org.apache.beam.sdk.schemas.Schema.FieldType.row(
+          inferBeamSchema((StructType) dataType));
     } else if (dataType instanceof ArrayType) {
-      return org.apache.beam.sdk.schemas.Schema.FieldType.array(toBeamFieldType(((ArrayType) dataType).getElementType()));
+      return org.apache.beam.sdk.schemas.Schema.FieldType.array(
+          toBeamFieldType(((ArrayType) dataType).getElementType()));
     } else if (dataType instanceof MapType) {
       MapType mapType = (MapType) dataType;
       return org.apache.beam.sdk.schemas.Schema.FieldType.map(
-          toBeamFieldType(mapType.getKeyType()),
-          toBeamFieldType(mapType.getValueType()));
+          toBeamFieldType(mapType.getKeyType()), toBeamFieldType(mapType.getValueType()));
     } else {
       throw new IllegalArgumentException("Unsupported Delta type: " + dataType);
     }
   }
 
   private static Row convertKernelRowToBeamRow(
       io.delta.kernel.data.Row deltaRow, org.apache.beam.sdk.schemas.Schema beamSchema) {
-    List<Object> values = new ArrayList<>();
+    List<@Nullable Object> values = new ArrayList<>();
     StructType structType = deltaRow.getSchema();
     for (int i = 0; i < structType.length(); i++) {
       if (deltaRow.isNullAt(i)) {
@@ -383,7 +395,8 @@ private static Row convertKernelRowToBeamRow(
     return Row.withSchema(beamSchema).addValues(values).build();
   }
 
-  private static Object convertValue(io.delta.kernel.data.Row deltaRow, int ordinal, DataType dataType) {
+  private static @Nullable Object convertValue(
+      io.delta.kernel.data.Row deltaRow, int ordinal, DataType dataType) {
     if (deltaRow.isNullAt(ordinal)) {
       return null;
     }
@@ -419,7 +432,7 @@ private static Object convertValue(io.delta.kernel.data.Row deltaRow, int ordina
     } else if (dataType instanceof ArrayType) {
       ArrayValue arrayVal = deltaRow.getArray(ordinal);
       int size = arrayVal.getSize();
-      List<Object> list = new ArrayList<>(size);
+      List<@Nullable Object> list = new ArrayList<>(size);
       DataType elemType = ((ArrayType) dataType).getElementType();
       ColumnVector vec = arrayVal.getElements();
       for (int j = 0; j < size; j++) {
@@ -433,7 +446,7 @@ private static Object convertValue(io.delta.kernel.data.Row deltaRow, int ordina
     } else if (dataType instanceof MapType) {
       MapValue mapVal = deltaRow.getMap(ordinal);
       int size = mapVal.getSize();
-      Map<Object, Object> map = new HashMap<>(size);
+      Map<Object, @Nullable Object> map = new HashMap<>(size);
       DataType keyType = ((MapType) dataType).getKeyType();
       DataType valueType = ((MapType) dataType).getValueType();
       ColumnVector keysVec = mapVal.getKeys();
diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSourceDoFn.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSourceDoFn.java
@@ -127,7 +127,7 @@ public ProcessContinuation processElement(
     SerializableRow scanStateRow = task.getScanStateRow();
     StructType physicalSchema = ScanStateRow.getPhysicalDataReadSchema(scanStateRow);
     StructType logicalSchema = ScanStateRow.getLogicalSchema(scanStateRow);
-    Schema beamSchema = DeltaIO.ReadRows.convertToBeamSchema(logicalSchema);
+    Schema beamSchema = DeltaIO.inferBeamSchema(logicalSchema);
 
     Engine currentEngine = engine;
     if (currentEngine == null) {
@@ -281,7 +281,7 @@ private static Row toBeamRow(io.delta.kernel.data.Row deltaRow, Schema beamSchem
       return map;
     } else if (type instanceof StructType) {
       io.delta.kernel.data.Row nestedRow = row.getStruct(index);
-      Schema nestedBeamSchema = DeltaIO.ReadRows.convertToBeamSchema((StructType) type);
+      Schema nestedBeamSchema = DeltaIO.inferBeamSchema((StructType) type);
       return toBeamRow(nestedRow, nestedBeamSchema);
     }
     throw new UnsupportedOperationException("Unsupported type: " + type.getClass());
@@ -350,7 +350,7 @@ private static Row toBeamRow(io.delta.kernel.data.Row deltaRow, Schema beamSchem
         childVectors[i] = vector.getChild(i);
       }
       io.delta.kernel.data.Row nestedRow = new VectorRow(structType, childVectors, index);
-      Schema nestedBeamSchema = DeltaIO.ReadRows.convertToBeamSchema(structType);
+      Schema nestedBeamSchema = DeltaIO.inferBeamSchema(structType);
       return toBeamRow(nestedRow, nestedBeamSchema);
     }
     throw new UnsupportedOperationException("Unsupported vector type: " + type.getClass());
