This is an automated email from the ASF dual-hosted git repository. ahmedabu98 pushed a commit to branch iceberg-mor-projection-fixes in repository https://gitbox.apache.org/repos/asf/beam.git
commit 11a635cf4cb07190cec2c4cc26da2825c3129810 Author: Ahmed Abualsaud <[email protected]> AuthorDate: Mon Oct 19 20:52:01 2026 -0700 m-o-r fixes --- .../org/apache/beam/sdk/io/iceberg/ReadUtils.java | 36 +++++ .../apache/beam/sdk/io/iceberg/ScanTaskReader.java | 41 ++++-- .../beam/sdk/io/iceberg/cdc/CdcReadUtils.java | 55 +++++--- .../iceberg/BeamBaseIncrementalChangelogScan.java | 4 +- .../beam/sdk/io/iceberg/IcebergIOReadTest.java | 87 ++++++++++++ .../beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java | 156 ++++++++++++++++++++- .../cdc/IncrementalChangelogSourceTest.java | 110 +++++++++++++++ 7 files changed, 456 insertions(+), 33 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadUtils.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadUtils.java index 3e70ae12687..fa8ee0c0add 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadUtils.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadUtils.java @@ -23,6 +23,7 @@ import java.util.Collection; 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.io.InputFile; 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 class ReadUtils { } 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); + } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java index c6ddd0a7e25..85851c32b4a 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java @@ -23,14 +23,18 @@ import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; 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 @@ class ScanTaskReader extends BoundedSource.BoundedReader<Row> { 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 @@ class ScanTaskReader extends BoundedSource.BoundedReader<Row> { 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 @@ class ScanTaskReader extends BoundedSource.BoundedReader<Row> { 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 @@ class ScanTaskReader extends BoundedSource.BoundedReader<Row> { 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 @@ class ScanTaskReader extends BoundedSource.BoundedReader<Row> { 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) { diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java index cc43cb40e8f..e2e3904c949 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java @@ -32,7 +32,9 @@ import java.util.stream.Collectors; 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.io.SeekableInputStream; 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 final class CdcReadUtils { 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 final class CdcReadUtils { } } + 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 @@ public final class CdcReadUtils { // 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 @@ public final class CdcReadUtils { @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 @@ public final class CdcReadUtils { /** * 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 @@ public final class CdcReadUtils { } 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,12 +541,11 @@ public final class CdcReadUtils { 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); @@ -533,7 +553,8 @@ public final class CdcReadUtils { 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 @@ public final class CdcReadUtils { } } - 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; diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/iceberg/BeamBaseIncrementalChangelogScan.java b/sdks/java/io/iceberg/src/main/java/org/apache/iceberg/BeamBaseIncrementalChangelogScan.java index a7fe19218c1..74009e739b5 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/iceberg/BeamBaseIncrementalChangelogScan.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/iceberg/BeamBaseIncrementalChangelogScan.java @@ -616,7 +616,9 @@ public class BeamBaseIncrementalChangelogScan 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(); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java index 9a8076bbb7d..e692221d58b 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOReadTest.java @@ -28,6 +28,7 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsInAnyOrder; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeFalse; import static org.junit.Assume.assumeTrue; import java.io.File; @@ -75,7 +76,12 @@ import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericAppenderFactory; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.EqualityDeleteWriter; +import org.apache.iceberg.deletes.PositionDelete; +import org.apache.iceberg.deletes.PositionDeleteWriter; +import org.apache.iceberg.encryption.EncryptedFiles; import org.apache.iceberg.hadoop.HadoopInputFile; import org.apache.iceberg.mapping.MappedField; import org.apache.iceberg.mapping.MappedFields; @@ -458,6 +464,87 @@ public class IcebergIOReadTest { testPipeline.run(); } + /** A projected read applies deletes on the row position, 'id' and nested 'metadata.source'. */ + @Test + public void testScanWithMergeOnReadDeletes() throws Exception { + assumeFalse(useIncrementalScan); + TableIdentifier tableId = + TableIdentifier.of("default", "table" + Long.toString(UUID.randomUUID().hashCode(), 16)); + Table table = + warehouse.createTable( + tableId, + TestFixtures.NESTED_SCHEMA, + null, + ImmutableMap.of(TableProperties.FORMAT_VERSION, "2")); + DataFile file = + warehouse.writeRecords( + testName.getMethodName() + ".parquet", + table.schema(), + ImmutableList.of( + createRecord(table.schema(), ImmutableMap.of("id", 1L, "data", "a")), + createRecord(table.schema(), ImmutableMap.of("id", 2L, "data", "b")), + createRecord(table.schema(), ImmutableMap.of("id", 3L, "data", "c")), + createRecord(table.schema(), ImmutableMap.of("id", 4L, "data", "d")), + icebergGenericRecord( + table.schema().asStruct(), + ImmutableMap.of( + "id", 5L, "data", "e", "metadata", ImmutableMap.of("source", "x"))))); + table.newFastAppend().appendFile(file).commit(); + org.apache.iceberg.Schema idSchema = table.schema().select("id"); + GenericAppenderFactory deletes = + new GenericAppenderFactory(table.schema(), table.spec(), new int[] {1}, idSchema, null); + PositionDeleteWriter<Record> positionDeletes = + deletes.newPosDeleteWriter( + EncryptedFiles.plainAsEncryptedOutput( + table.io().newOutputFile(file.location() + ".pos.parquet")), + FileFormat.PARQUET, + null); + try (positionDeletes) { + positionDeletes.write(PositionDelete.<Record>create().set(file.location(), 1L)); + } + EqualityDeleteWriter<Record> equalityDeletes = + deletes.newEqDeleteWriter( + EncryptedFiles.plainAsEncryptedOutput( + table.io().newOutputFile(file.location() + ".eq.parquet")), + FileFormat.PARQUET, + null); + try (equalityDeletes) { + equalityDeletes.write(createRecord(idSchema, ImmutableMap.of("id", 3L))); + } + org.apache.iceberg.Schema sourceSchema = table.schema().select("metadata.source"); + EqualityDeleteWriter<Record> nestedDeletes = + new GenericAppenderFactory(table.schema(), table.spec(), new int[] {4}, sourceSchema, null) + .newEqDeleteWriter( + EncryptedFiles.plainAsEncryptedOutput( + table.io().newOutputFile(file.location() + ".nested.parquet")), + FileFormat.PARQUET, + null); + try (nestedDeletes) { + nestedDeletes.write( + icebergGenericRecord( + sourceSchema.asStruct(), + ImmutableMap.of("metadata", ImmutableMap.of("source", "x")))); + } + table + .newRowDelta() + .addDeletes(positionDeletes.toDeleteFile()) + .addDeletes(equalityDeletes.toDeleteFile()) + .addDeletes(nestedDeletes.toDeleteFile()) + .commit(); + + IcebergIO.ReadRows read = + IcebergIO.readRows(catalogConfig()) + .from(tableId) + .withFilter("\"data\" <> 'd'") + .keeping(singletonList("data")); + Schema outputSchema = icebergSchemaToBeamSchema(table.schema().select("data")); + + PAssert.that(testPipeline.apply(read)) + .containsInAnyOrder(Row.withSchema(outputSchema).addValue("a").build()); + + testPipeline.run(); + } + @Test public void testReadSchemaWithRandomlyOrderedIds() throws IOException { TableIdentifier tableId = TableIdentifier.of("default", testName.getMethodName()); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java index 8c23a844be6..3a7b884b036 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java @@ -63,7 +63,13 @@ public class CdcReadUtilsTest { new org.apache.iceberg.Schema( ImmutableList.of( Types.NestedField.required(1, "id", Types.LongType.get()), - Types.NestedField.optional(2, "data", Types.StringType.get())), + Types.NestedField.optional(2, "data", Types.StringType.get()), + Types.NestedField.optional( + 3, + "metadata", + Types.StructType.of( + Types.NestedField.optional(4, "source", Types.StringType.get()), + Types.NestedField.optional(5, "region", Types.StringType.get())))), ImmutableSet.of(1)); @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); @@ -206,19 +212,27 @@ public class CdcReadUtilsTest { record(0L, "already-hidden-by-position"), record(1L, "still-live-one"), record(2L, "already-hidden-by-equality"), - record(3L, "still-live-two"))); + record(3L, "still-live-two"), + record(4L, "already-hidden-by-nested-equality", "hidden"))); DeleteFile existingPositionDelete = writePositionDelete(table, dataFile, "cdc-deleted-file-pos-delete.parquet", 0L); DeleteFile existingEqualityDelete = writeEqualityDelete( table, dataFile, "cdc-deleted-file-eq-delete.parquet", "already-hidden-by-equality"); + DeleteFile existingNestedEqualityDelete = + writeSourceEqualityDelete( + table, + dataFile, + "cdc-deleted-file-nested-eq-delete.parquet", + record(-1L, null, "hidden")); SerializableChangelogTask task = task( SerializableChangelogTask.Type.DELETED_FILE, dataFile, ImmutableList.of(), - ImmutableList.of(existingPositionDelete, existingEqualityDelete), + ImmutableList.of( + existingPositionDelete, existingEqualityDelete, existingNestedEqualityDelete), table); CloseableIterable<Record> records = @@ -276,6 +290,107 @@ public class CdcReadUtilsTest { assertEquals(ImmutableList.of(1L), idsOf(records)); } + @Test + public void deletedRowsPushesDownNestedEqualityDeletes() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", testName.getMethodName()); + Table table = warehouse.createTable(tableId, CDC_SCHEMA); + DataFile dataFile = + warehouse.writeRecords( + "cdc-nested-eq.parquet", + table.schema(), + ImmutableList.of( + record(0L, "a", "new-equality-delete"), + record(1L, "b", "already-hidden-by-equality"), + record(2L, "c", "new-position-delete"))); + DeleteFile existingEqualityDelete = + writeSourceEqualityDelete( + table, + dataFile, + "cdc-nested-existing-eq-delete.parquet", + record(-1L, null, "already-hidden-by-equality")); + DeleteFile addedEqualityDelete = + writeSourceEqualityDelete( + table, + dataFile, + "cdc-nested-added-eq-delete.parquet", + record(-1L, null, "new-equality-delete")); + DeleteFile addedPositionDelete = + writePositionDelete(table, dataFile, "cdc-nested-pos-delete.parquet", 2L); + + // the pushed down IN expression references the full column path + assertEquals( + Expressions.in("metadata.source", "new-equality-delete").toString(), + CdcReadUtils.buildEqualityDeletePushdown( + table, table.schema(), ImmutableList.of(addedEqualityDelete)) + .residual + .toString()); + + SerializableChangelogTask equalityOnly = + task( + SerializableChangelogTask.Type.DELETED_ROWS, + dataFile, + ImmutableList.of(addedEqualityDelete), + ImmutableList.of(existingEqualityDelete), + table); + SerializableChangelogTask positionAndEquality = + task( + SerializableChangelogTask.Type.DELETED_ROWS, + dataFile, + ImmutableList.of(addedPositionDelete, addedEqualityDelete), + ImmutableList.of(existingEqualityDelete), + table); + // the projection leaves out metadata.source, or keeps only its sibling + for (List<String> keep : + ImmutableList.of(ImmutableList.of("id"), ImmutableList.of("id", "metadata.region"))) { + IcebergScanConfig scanConfig = scanConfig(table, tableId, keep); + assertEquals( + ImmutableList.of(0L), + idsOf(CdcReadUtils.changelogRecordsForTask(equalityOnly, table, scanConfig, true))); + assertEquals( + ImmutableList.of(0L, 2L), + idsOf( + CdcReadUtils.changelogRecordsForTask(positionAndEquality, table, scanConfig, true))); + } + } + + @Test + public void deletedRowsHandlesNullNestedEqualityDeletesWithoutPushdown() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", testName.getMethodName()); + Table table = warehouse.createTable(tableId, CDC_SCHEMA); + DataFile dataFile = + warehouse.writeRecords( + "cdc-nested-null-eq.parquet", + table.schema(), + ImmutableList.of( + record(0L, "live", "x"), + record(1L, "null-source", null), + record(2L, "null-metadata"))); + DeleteFile nullEqualityDeletes = + writeSourceEqualityDelete( + table, + dataFile, + "cdc-nested-null-eq-delete.parquet", + record(-1L, null, null), + record(-2L, null)); + + assertEquals( + Expressions.alwaysTrue().toString(), + CdcReadUtils.buildEqualityDeletePushdown( + table, table.schema(), ImmutableList.of(nullEqualityDeletes)) + .residual + .toString()); + SerializableChangelogTask task = + task( + SerializableChangelogTask.Type.DELETED_ROWS, + dataFile, + ImmutableList.of(nullEqualityDeletes), + ImmutableList.of(), + table); + assertEquals( + ImmutableList.of(1L, 2L), + idsOf(CdcReadUtils.changelogRecordsForTask(task, table, scanConfig(table, tableId), true))); + } + private static Record record(long id, String data) { GenericRecord record = GenericRecord.create(CDC_SCHEMA); record.setField("id", id); @@ -283,6 +398,14 @@ public class CdcReadUtilsTest { return record; } + private static Record record(long id, String data, @Nullable String source) { + GenericRecord metadata = GenericRecord.create(CDC_SCHEMA.findType("metadata").asStructType()); + metadata.setField("source", source); + Record record = record(id, data); + record.setField("metadata", metadata); + return record; + } + private static DeleteFile writePositionDelete( Table table, DataFile dataFile, String filename, long... positions) throws IOException { GenericAppenderFactory appenderFactory = @@ -320,7 +443,32 @@ public class CdcReadUtilsTest { return writer.toDeleteFile(); } + /** Writes whole rows as equality deletes on {@code metadata.source}. */ + private static DeleteFile writeSourceEqualityDelete( + Table table, DataFile dataFile, String filename, Record... deletes) throws IOException { + int sourceId = table.schema().findField("metadata.source").fieldId(); + GenericAppenderFactory appenderFactory = + new GenericAppenderFactory( + table.schema(), table.spec(), new int[] {sourceId}, table.schema(), null); + EqualityDeleteWriter<Record> writer = + appenderFactory.newEqDeleteWriter( + EncryptedFiles.plainAsEncryptedOutput( + table.io().newOutputFile(dataFile.location() + "." + filename)), + FileFormat.PARQUET, + null); + try (writer) { + for (Record delete : deletes) { + writer.write(delete); + } + } + return writer.toDeleteFile(); + } + private IcebergScanConfig scanConfig(Table table, TableIdentifier tableId) { + return scanConfig(table, tableId, ImmutableList.of("id")); + } + + private IcebergScanConfig scanConfig(Table table, TableIdentifier tableId, List<String> keep) { return IcebergScanConfig.builder() .setCatalogConfig( IcebergCatalogConfig.builder() @@ -329,7 +477,7 @@ public class CdcReadUtilsTest { .build()) .setTableIdentifier(tableId) .setSchema(IcebergUtils.icebergSchemaToBeamSchema(table.schema())) - .setKeepFields(ImmutableList.of("id")) + .setKeepFields(keep) .build(); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/IncrementalChangelogSourceTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/IncrementalChangelogSourceTest.java index c055e609f76..83dc4986e52 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/IncrementalChangelogSourceTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/IncrementalChangelogSourceTest.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.iceberg.cdc; +import static org.apache.beam.sdk.io.iceberg.IcebergIOReadTest.icebergGenericRecord; import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; import static org.hamcrest.MatcherAssert.assertThat; @@ -52,12 +53,19 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Immuta import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.iceberg.ChangelogOperation; import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericAppenderFactory; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.EqualityDeleteWriter; +import org.apache.iceberg.deletes.PositionDelete; +import org.apache.iceberg.deletes.PositionDeleteWriter; +import org.apache.iceberg.encryption.EncryptedFiles; import org.apache.iceberg.expressions.Literal; import org.apache.iceberg.types.Types; import org.joda.time.Instant; @@ -418,6 +426,93 @@ public class IncrementalChangelogSourceTest { pipeline.run().waitUntilFinish(); } + /** A merge-on-read delete is filtered on a partition column that 'keep' leaves out. */ + @Test + public void mergeOnReadDeleteIsFilteredOnColumnLeftOutOfOutput() throws Exception { + TableIdentifier tableId = tableId(); + PartitionSpec spec = PartitionSpec.builderFor(EVENT_SCHEMA).day("event_time").build(); + Table table = warehouse.createTable(tableId, EVENT_SCHEMA, spec, tableProperties()); + LocalDateTime time = LocalDateTime.parse("2026-01-01T10:00:00"); + DataFile file = + warehouse.writeRecords( + testName.getMethodName() + "-events.parquet", + table.schema(), + spec, + DataFiles.data(spec, "event_time_day=2026-01-01"), + ImmutableList.of(eventRecord(1L, "a", time), eventRecord(2L, "b", time.plusHours(4)))); + table.newFastAppend().appendFile(file).commit(); + commitPositionDelete(table, file, 1L); + + IcebergScanConfig scanConfig = + baseConfigBuilder(table, tableId) + .setKeepFields(ImmutableList.of("id", "data")) + .setFilterString("event_time > TIMESTAMP '2026-01-01 12:00:00'") + .setToSnapshot(table.currentSnapshot().snapshotId()) + .build(); + + PAssert.that( + pipeline + .apply(new IncrementalChangelogSource(scanConfig)) + .apply(ParDo.of(new FormatValueKindAndRow()))) + .containsInAnyOrder("INSERT:2:b", "DELETE:2:b"); + + pipeline.run().waitUntilFinish(); + } + + /** + * An equality delete on a nested field that 'keep' leaves out, while keeping a sibling field, + * matches only its own rows. + */ + @Test + public void nestedEqualityDeleteIsFilteredOnColumnLeftOutOfOutput() throws Exception { + TableIdentifier tableId = tableId(); + Table table = + warehouse.createTable(tableId, TestFixtures.NESTED_SCHEMA, null, tableProperties()); + table.updateSchema().addColumn("metadata", "region", Types.StringType.get()).commit(); + DataFile file = + commitAppend( + table, + "nested.parquet", + ImmutableList.of( + TestFixtures.createRecord(table.schema(), ImmutableMap.of("id", 1L, "data", "a")), + icebergGenericRecord( + table.schema().asStruct(), + ImmutableMap.of( + "id", 2L, "data", "b", "metadata", ImmutableMap.of("source", "x"))))); + org.apache.iceberg.Schema sourceSchema = table.schema().select("metadata.source"); + EqualityDeleteWriter<Record> writer = + new GenericAppenderFactory(table.schema(), table.spec(), new int[] {4}, sourceSchema, null) + .newEqDeleteWriter( + EncryptedFiles.plainAsEncryptedOutput( + table.io().newOutputFile(file.location() + ".deletes.parquet")), + FileFormat.PARQUET, + null); + try (writer) { + writer.write( + icebergGenericRecord( + sourceSchema.asStruct(), + ImmutableMap.of("metadata", ImmutableMap.of("source", "x")))); + } + table.newRowDelta().addDeletes(writer.toDeleteFile()).commit(); + table.newDelete().deleteFile(file).commit(); + + // only the file deletion, whose rows are filtered by the earlier equality delete + IcebergScanConfig scanConfig = + baseConfigBuilder(table, tableId) + .setKeepFields(ImmutableList.of("id", "data", "metadata.region")) + .setFromSnapshotInclusive(table.currentSnapshot().snapshotId()) + .setToSnapshot(table.currentSnapshot().snapshotId()) + .build(); + + PAssert.that( + pipeline + .apply(new IncrementalChangelogSource(scanConfig)) + .apply(ParDo.of(new FormatValueKindAndRow()))) + .containsInAnyOrder("DELETE:1:a"); + + pipeline.run().waitUntilFinish(); + } + @Test public void watermarkColumnRestampsProjectedRows() throws Exception { TableIdentifier tableId = tableId(); @@ -632,6 +727,21 @@ public class IncrementalChangelogSourceTest { return file; } + private void commitPositionDelete(Table table, DataFile file, long position) throws IOException { + PositionDeleteWriter<Record> writer = + new GenericAppenderFactory(table.schema(), table.spec()) + .newPosDeleteWriter( + EncryptedFiles.plainAsEncryptedOutput( + table.io().newOutputFile(file.location() + ".deletes.parquet")), + FileFormat.PARQUET, + file.partition()); + try (writer) { + writer.write(PositionDelete.<Record>create().set(file.location(), position)); + } + table.newRowDelta().addDeletes(writer.toDeleteFile()).commit(); + table.refresh(); + } + private static List<Record> records(long id, String data) { return ImmutableList.of(record(id, data)); }
