You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ah...@apache.org on 2026/10/20 03:52:17 UTC
(beam) branch iceberg-mor-projection-fixes created (now 11a635cf4cb)
This is an automated email from the ASF dual-hosted git repository.
ahmedabu98 pushed a change to branch iceberg-mor-projection-fixes
in repository https://gitbox.apache.org/repos/asf/beam.git
at 11a635cf4cb m-o-r fixes
This branch includes the following new commits:
new 11a635cf4cb m-o-r fixes
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(beam) 01/01: m-o-r fixes
Posted by ah...@apache.org.
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 <a....@gmail.com>
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));
}