[Iceberg] Merge on read fixes

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadUtils.java

@@ -23,6 +23,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.io.iceberg.IcebergIO.ReadRows.StartingStrategy;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
@@ -49,6 +50,8 @@
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.mapping.NameMappingParser;
 import org.apache.iceberg.parquet.ParquetReader;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.SnapshotUtil;
 import org.apache.parquet.HadoopReadOptions;
 import org.apache.parquet.ParquetReadOptions;
@@ -218,4 +221,37 @@ public static CloseableIterable<Record> maybeApplyFilter(
     }
     return iterable;
   }
+
+  /**
+   * Adds the equality delete fields that {@code projection} lacks, under their parent structs.
+   * Fields already in {@code projection} are kept as they are, even if the table has since renamed
+   * or promoted them.
+   */
+  public static Schema withEqualityFields(
+      Schema tableSchema, Schema projection, Set<Integer> equalityFieldIds) {
+    Set<Integer> missingIds =
+        Sets.difference(equalityFieldIds, TypeUtil.getProjectedIds(projection));
+    if (missingIds.isEmpty()) {
+      return projection;
+    }
+    Types.StructType missing = TypeUtil.select(tableSchema, missingIds).asStruct();
+    return new Schema(withFields(projection.asStruct(), missing).fields());
+  }
+
+  /** Adds the fields of {@code extra} that {@code struct} lacks, recursing into shared structs. */
+  private static Types.StructType withFields(Types.StructType struct, Types.StructType extra) {
+    List<Types.NestedField> fields = Lists.newArrayList(struct.fields());
+    for (Types.NestedField field : extra.fields()) {
+      Types.@Nullable NestedField existing = struct.field(field.fieldId());
+      if (existing == null) {
+        fields.add(field);
+      } else if (existing.type().isStructType()) {
+        Types.StructType merged =
+            withFields(existing.type().asStructType(), field.type().asStructType());
+        fields.set(
+            fields.indexOf(existing), Types.NestedField.from(existing).ofType(merged).build());
+      }
+    }
+    return Types.StructType.of(fields);
+  }
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java

@@ -23,14 +23,18 @@
 
 import java.io.IOException;
 import java.util.ArrayDeque;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Queue;
+import java.util.Set;
 import javax.annotation.Nullable;
 import org.apache.beam.sdk.io.BoundedSource;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.values.Row;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileContent;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
@@ -124,6 +128,15 @@ public boolean advance() throws IOException {
       InputFile input = decryptor.getInputFile(fileTask);
       Map<Integer, ?> idToConstants =
           PartitionUtils.constantsMap(fileTask.spec(), fileTask.file(), null);
+      GenericDeleteFilter deleteFilter =
+          new GenericDeleteFilter(
+              checkStateNotNull(io),
+              fileTask,
+              fileTask.schema(),
+              ReadUtils.withEqualityFields(
+                  fileTask.schema(), requiredSchema, equalityFieldIds(fileTask)));
+      // also reads the columns that the deletes need: equality fields and the row position
+      org.apache.iceberg.Schema readSchema = deleteFilter.requiredSchema();
 
       CloseableIterable<Record> iterable;
       switch (file.format()) {
@@ -132,10 +145,10 @@ public boolean advance() throws IOException {
           ORC.ReadBuilder orcReader =
               ORC.read(input)
                   .split(fileTask.start(), fileTask.length())
-                  .project(requiredSchema)
+                  .project(readSchema)
                   .createReaderFunc(
                       fileSchema ->
-                          GenericOrcReader.buildReader(requiredSchema, fileSchema, idToConstants))
+                          GenericOrcReader.buildReader(readSchema, fileSchema, idToConstants))
                   .filter(fileTask.residual());
 
           if (nameMapping != null) {
@@ -149,11 +162,10 @@ public boolean advance() throws IOException {
           Parquet.ReadBuilder parquetReader =
               Parquet.read(input)
                   .split(fileTask.start(), fileTask.length())
-                  .project(requiredSchema)
+                  .project(readSchema)
                   .createReaderFunc(
                       fileSchema ->
-                          GenericParquetReaders.buildReader(
-                              requiredSchema, fileSchema, idToConstants))
+                          GenericParquetReaders.buildReader(readSchema, fileSchema, idToConstants))
                   .filter(fileTask.residual());
 
           if (nameMapping != null) {
@@ -167,9 +179,9 @@ public boolean advance() throws IOException {
           Avro.ReadBuilder avroReader =
               Avro.read(input)
                   .split(fileTask.start(), fileTask.length())
-                  .project(requiredSchema)
+                  .project(readSchema)
                   .createReaderFunc(
-                      fileSchema -> DataReader.create(requiredSchema, fileSchema, idToConstants));
+                      fileSchema -> DataReader.create(readSchema, fileSchema, idToConstants));
 
           if (nameMapping != null) {
             avroReader.withNameMapping(NameMappingParser.fromJson(nameMapping));
@@ -180,18 +192,25 @@ public boolean advance() throws IOException {
         default:
           throw new UnsupportedOperationException("Cannot read format: " + file.format());
       }
-      GenericDeleteFilter deleteFilter =
-          new GenericDeleteFilter(
-              checkStateNotNull(io), fileTask, fileTask.schema(), requiredSchema);
       iterable = deleteFilter.filter(iterable);
 
-      iterable = ReadUtils.maybeApplyFilter(iterable, source.getScanConfig());
+      iterable = ReadUtils.maybeApplyFilter(iterable, source.getScanConfig(), readSchema);
       currentIterator = iterable.iterator();
     } while (true);
 
     return false;
   }
 
+  private static Set<Integer> equalityFieldIds(FileScanTask task) {
+    Set<Integer> ids = new HashSet<>();
+    for (DeleteFile delete : task.deletes()) {
+      if (delete.content() == FileContent.EQUALITY_DELETES) {
+        ids.addAll(delete.equalityFieldIds());
+      }
+    }
+    return ids;
+  }
+
   @Override
   public Row getCurrent() throws NoSuchElementException {
     if (current == null) {

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java

@@ -32,7 +32,9 @@
 import org.apache.beam.sdk.io.iceberg.IcebergScanConfig;
 import org.apache.beam.sdk.io.iceberg.ReadUtils;
 import org.apache.beam.sdk.io.iceberg.SerializableDeleteFile;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.apache.iceberg.Accessor;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileContent;
 import org.apache.iceberg.Schema;
@@ -53,6 +55,7 @@
 import org.apache.iceberg.parquet.ParquetMetricsRowGroupFilter;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.NaNUtil;
 import org.apache.iceberg.util.StructLikeSet;
 import org.apache.parquet.hadoop.ParquetFileReader;
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
@@ -214,10 +217,11 @@ public static CloseableIterable<Record> changelogRecordsForTask(
       IcebergScanConfig scanConfig,
       boolean useProjectedSchema) {
     String dataFilePath = task.getDataFile().getPath();
+    Schema dataSchema = useProjectedSchema ? scanConfig.getRequiredSchema() : table.schema();
     Schema outputSchema =
         CdcOutputUtils.readSchemaWithRowMetadata(
             scanConfig.getMetadataColumns(),
-            useProjectedSchema ? scanConfig.getRequiredSchema() : table.schema());
+            ReadUtils.withEqualityFields(table.schema(), dataSchema, equalityFieldIds(task)));
     switch (task.getType()) {
       case ADDED_ROWS:
         DeleteFilter<Record> addedDeletesFilter =
@@ -236,6 +240,18 @@ public static CloseableIterable<Record> changelogRecordsForTask(
     }
   }
 
+  private static Set<Integer> equalityFieldIds(SerializableChangelogTask task) {
+    Set<Integer> ids = new HashSet<>();
+    for (SerializableDeleteFile delete :
+        Iterables.concat(task.getAddedDeletes(), task.getExistingDeletes())) {
+      @Nullable List<Integer> deleteIds = delete.getEqualityFieldIds();
+      if (deleteIds != null) {
+        ids.addAll(deleteIds);
+      }
+    }
+    return ids;
+  }
+
   /**
    * Builds the reader chain for a {@code DELETED_ROWS} task with row-group pushdown when possible.
    * This helps the reader skip entire row groups. For unskipped row groups, the reader should still
@@ -303,7 +319,7 @@ private static CloseableIterable<Record> deletedRowsForTask(
     // byte-range path above already incorporates the eq filter
     Expression eqResidual = Expressions.alwaysTrue();
     if (posFiles.isEmpty() && !eqFiles.isEmpty()) {
-      EqualityPushdownResult eqPushdown = buildEqualityDeletePushdown(table, eqFiles);
+      EqualityPushdownResult eqPushdown = buildEqualityDeletePushdown(table, outputSchema, eqFiles);
       eqResidual = eqPushdown.applicable ? eqPushdown.residual : Expressions.alwaysTrue();
       preloadedDeletes = eqPushdown.preloadedDeletes(null);
     }
@@ -360,14 +376,14 @@ private static CloseableIterable<Record> deletedRowsForTask(
     @Nullable ParquetMetricsRowGroupFilter eqFilter = null;
     EqualityPushdownResult eqPushdown = EqualityPushdownResult.notApplicable();
     if (!eqFiles.isEmpty()) {
-      eqPushdown = buildEqualityDeletePushdown(table, eqFiles);
+      eqPushdown = buildEqualityDeletePushdown(table, outputSchema, eqFiles);
       if (!eqPushdown.applicable) {
         // eq deletes are present but we can't safely identify which row groups they target.
         // A narrowed position-only range could drop eq-deleted rows, so fall back to the
         // default full-range reader. DeleteReader will still apply residual per record.
         return PositionPushdownResult.fallback(eqPushdown.preloadedDeletes(posIndex));
       }
-      eqFilter = new ParquetMetricsRowGroupFilter(table.schema(), eqPushdown.residual);
+      eqFilter = new ParquetMetricsRowGroupFilter(outputSchema, eqPushdown.residual);
     }
     DeleteReader.PreloadedDeletes preloadedDeletes = eqPushdown.preloadedDeletes(posIndex);
 
@@ -479,10 +495,12 @@ private static boolean anyInRange(long[] sortedDeletes, long startInclusive, lon
   /**
    * Returns an {@code IN} expression suitable as a Parquet residual for the given equality-delete
    * files, or {@link Expressions#alwaysTrue()} if pushdown is not applicable. See {@link
-   * #deletedRowsForTask} for the applicability rules.
+   * #deletedRowsForTask} for the applicability rules. The expression names the column as {@code
+   * readSchema} does, since the reader binds it against that schema.
    */
-  private static EqualityPushdownResult buildEqualityDeletePushdown(
-      Table table, List<DeleteFile> eqFiles) {
+  @VisibleForTesting
+  static EqualityPushdownResult buildEqualityDeletePushdown(
+      Table table, Schema readSchema, List<DeleteFile> eqFiles) {
     // All eq delete files in this task must share a single equality field id.
     Set<Integer> sharedIds = null;
     for (DeleteFile df : eqFiles) {
@@ -498,11 +516,14 @@ private static EqualityPushdownResult buildEqualityDeletePushdown(
     }
 
     int fieldId = Iterables.getOnlyElement(sharedIds);
-    Types.NestedField field = table.schema().findField(fieldId);
-    if (field == null) {
+    Types.NestedField field = readSchema.findField(fieldId);
+    Schema deleteSchema = TypeUtil.select(table.schema(), sharedIds);
+    // the delete schema keeps parent structs, so an accessor reaches nested fields
+    @Nullable Accessor<StructLike> accessor = deleteSchema.accessorForField(fieldId);
+    // row-group filters read a column missing from the file as nulls, ignoring its initial default
+    if (field == null || field.initialDefault() != null || accessor == null) {
       return EqualityPushdownResult.notApplicable();
     }
-    Schema deleteSchema = TypeUtil.select(table.schema(), sharedIds);
 
     DeleteLoader loader = new BaseDeleteLoader(df -> table.io().newInputFile(df.location()));
     StructLikeSet set;
@@ -520,20 +541,20 @@ private static EqualityPushdownResult buildEqualityDeletePushdown(
     if (set.size() > IN_PREDICATE_LIMIT) {
       return EqualityPushdownResult.notApplicable(preloadedSets);
     }
-    Class<?> javaClass = field.type().typeId().javaClass();
     List<Object> values = new ArrayList<>(set.size());
     for (StructLike s : set) {
-      @Nullable Object v = s.get(0, javaClass);
-      if (v == null) {
-        // Nulls don't match an IN-expression. pushing down would drop those deletions.
+      @Nullable Object v = accessor.get(s);
+      if (v == null || NaNUtil.isNaN(v)) {
+        // IN-expressions can't match nulls or NaNs. pushing down would drop those deletions.
         return EqualityPushdownResult.notApplicable(preloadedSets);
       }
       values.add(v);
     }
     if (values.isEmpty()) {
       return EqualityPushdownResult.notApplicable(preloadedSets);
     }
-    return EqualityPushdownResult.applicable(Expressions.in(field.name(), values), preloadedSets);
+    return EqualityPushdownResult.applicable(
+        Expressions.in(readSchema.findColumnName(fieldId), values), preloadedSets);
   }
 
   private static final class PositionPushdownResult {
@@ -557,11 +578,11 @@ private PositionPushdownResult(
     }
   }
 
-  private static final class EqualityPushdownResult {
+  static final class EqualityPushdownResult {
     private static final EqualityPushdownResult NOT_APPLICABLE =
         new EqualityPushdownResult(Expressions.alwaysTrue(), Collections.emptyMap(), false);
 
-    private final Expression residual;
+    final Expression residual;
     private final Map<Set<Integer>, StructLikeSet> preloadedSets;
     private final boolean applicable;
 

sdks/java/io/iceberg/src/main/java/org/apache/iceberg/BeamBaseIncrementalChangelogScan.java

@@ -616,7 +616,9 @@ private void processSnapshotForDeletedRowsTasks(
       allDataGroup = allDataGroup.ignoreResiduals();
     }
 
-    String schemaString = SchemaParser.toJson(schema());
+    // as in ManifestGroup, task specs use the table schema: the residual and partition fields may
+    // reference columns outside the projection
+    String schemaString = SchemaParser.toJson(table().schema());
 
     // Cache per specId - same for all files with same specId
     Map<Integer, String> specStringCache = Maps.newHashMap();