You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "mxm (via GitHub)" <gi...@apache.org> on 2026/06/18 04:57:40 UTC
[PR] Flink: Add equality delete conversion DV resolution and writing [iceberg]
mxm opened a new pull request, #16858:
URL: https://github.com/apache/iceberg/pull/16858
Adds EqualityConvertDVWriter, a two input operator keyed by data file path for the equality delete conversion task. It consumes the DVPositions emitted by EqualityConvertPKIndex and produces a DVWriteResult for the committer. Keying by data file path collects every position for a data file on one task, so the operator can write exactly one DV per file.
Positions are buffered until the watermark reaches the cycle's plan timestamp, broadcast on the second input, at which point the operator resolves and writes the DVs.
Existing DVs, including staging DVs for newly added data files, are folded into the rewrite because V3 allows only one DV per data file. Delete manifests are first pruned by partition summary to bound reads to the cycle's affected partitions rather than the table's full DV history.
The operator fails fast if the main branch snapshot changed since planning. On an upstream abort signal or a write failure it emits DVWriteResult.ABORT to ensure the committer doesn't commit partial results.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add equality delete conversion DV resolution and writing [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #16858:
URL: https://github.com/apache/iceberg/pull/16858#discussion_r3434364454
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java:
##########
@@ -0,0 +1,354 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.operator;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.BaseDeleteLoader;
+import org.apache.iceberg.data.DeleteLoader;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.ManifestEvaluator;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.StructLikeWrapper;
+import org.roaringbitmap.longlong.Roaring64Bitmap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Keyed parallel resolver that buffers {@link DVPosition}s per data-file path, then writes Puffin
+ * DV files directly via {@link BaseDVFileWriter}. Plan metadata arrives broadcast on input 2, so
+ * every parallel task sees the cycle's metadata and can validate against the main snapshot.
+ *
+ * <p>Each buffered {@link DVPosition} carries the data file's {@code specId} + encoded partition,
+ * so writing DVs needs no data-manifest scan. Existing DVs are folded into the rewrite (V3 allows
+ * one DV per data file): delete manifests are pruned by partition summary to the cycle's affected
+ * partitions, then filtered to entries referencing the affected data files. No cross-cycle state is
+ * kept; reads are bounded by the pruned manifest set, not the table's full DV history.
+ *
+ * <p>Buffered positions are transient per-task. On failure recovery, upstream replay rebuilds them.
+ */
+@Internal
+public class EqualityConvertDVWriter extends AbstractStreamOperator<DVWriteResult>
+ implements TwoInputStreamOperator<DVPosition, EqualityConvertPlan, DVWriteResult> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(EqualityConvertDVWriter.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final TableLoader tableLoader;
+ private final String targetBranch;
+
+ private transient Table table;
+ private transient OutputFileFactory fileFactory;
+ private transient DeleteLoader deleteLoader;
+ private transient Map<String, FilePositions> positionsByFile;
+ private transient EqualityConvertPlan planResult;
+ private transient boolean hasUpstreamError;
+ private transient int manifestsRead;
+
+ public EqualityConvertDVWriter(
+ String tableName, String taskName, TableLoader tableLoader, String targetBranch) {
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.tableLoader = tableLoader;
+ this.targetBranch = targetBranch;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ if (!tableLoader.isOpen()) {
+ tableLoader.open();
+ }
+
+ table = tableLoader.loadTable();
+ int subtaskIndex = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
+ fileFactory =
+ OutputFileFactory.builderFor(table, subtaskIndex, 0L).format(FileFormat.PUFFIN).build();
+ deleteLoader = new BaseDeleteLoader(deleteFile -> table.io().newInputFile(deleteFile));
+ positionsByFile = Maps.newHashMap();
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DVPosition> record) {
+ DVPosition pos = record.getValue();
+ if (pos.isAbort()) {
+ hasUpstreamError = true;
+ }
+
+ if (!hasUpstreamError) {
+ positionsByFile
+ .computeIfAbsent(
+ pos.dataFilePath(), k -> new FilePositions(pos.specId(), pos.partition()))
+ .positions
+ .addLong(pos.position());
+ }
+ }
+
+ @Override
+ public void processElement2(StreamRecord<EqualityConvertPlan> record) {
+ planResult = record.getValue();
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ if (planResult != null && mark.getTimestamp() >= planResult.doneTimestamp()) {
+ if (hasUpstreamError) {
+ output.collect(new StreamRecord<>(DVWriteResult.ABORT));
+ } else {
+ try {
+ resolveAndWrite();
+ } catch (Exception e) {
+ LOG.error("Error writing DVs for table {} task {}", tableName, taskName, e);
+ output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+ output.collect(new StreamRecord<>(DVWriteResult.ABORT));
+ }
+ }
+
+ positionsByFile.clear();
+ hasUpstreamError = false;
+ planResult = null;
+ }
+
+ super.processWatermark(mark);
+ }
+
+ private void resolveAndWrite() throws IOException {
+ if (positionsByFile.isEmpty()) {
+ return;
+ }
+
+ table.refresh();
+
+ Snapshot mainSnapshot = table.snapshot(targetBranch);
+
+ // Fail fast if the main branch changed since planning, to avoid writing DV files that the
+ // committer would reject via validateFromSnapshot. The next cycle will reindex.
+ if (mainSnapshot != null
+ && planResult.mainSnapshotId() != null
Review Comment:
Thanks, I added a dedicated test to verify this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add equality delete conversion DV resolution and writing [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #16858:
URL: https://github.com/apache/iceberg/pull/16858#issuecomment-4748450668
Thanks for reviewing / merging @wombatu-kun @huaxingao!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add equality delete conversion DV resolution and writing [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #16858:
URL: https://github.com/apache/iceberg/pull/16858#issuecomment-4748068615
Thanks @mxm for the PR! Thanks @wombatu-kun for the review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add equality delete conversion DV resolution and writing [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #16858:
URL: https://github.com/apache/iceberg/pull/16858#discussion_r3433539030
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java:
##########
@@ -0,0 +1,431 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.flink.SimpleDataUtil.createRecord;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.Test;
+
+class TestEqualityConvertDVWriter extends OperatorTestBase {
+
+ private static final byte[] EMPTY_PARTITION = new byte[0];
+
+ @Test
+ void writesDVFileForSinglePosition() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ List<DVWriteResult> output = harness.extractOutputValues();
+ assertThat(output).hasSize(1);
+ assertThat(output.get(0).hasError()).isFalse();
+ assertThat(output.get(0).dvFiles()).hasSize(1);
+ }
+ }
+
+ @Test
+ void writesSingleDVFileForMultiplePositionsOnSameDataFile() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+ insert(table, 3, "c");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 1, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 2, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ List<DVWriteResult> output = harness.extractOutputValues();
+ assertThat(output).hasSize(1);
+ assertThat(output.get(0).hasError()).isFalse();
+ assertThat(output.get(0).dvFiles()).hasSize(1);
+ assertThat(output.get(0).dvFiles().get(0).recordCount()).isEqualTo(3);
+ assertThat(output.get(0).rewrittenDvFiles()).isEmpty();
+ }
+ }
+
+ @Test
+ void emptyPositionsProducesNoOutput() throws Exception {
+ createTableWithDelete(3);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ assertThat(harness.extractOutputValues()).isEmpty();
+ }
+ }
+
+ @Test
+ void noOutputWithoutPlanResult() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ assertThat(harness.extractOutputValues()).isEmpty();
+ }
+ }
+
+ @Test
+ void writesDVFilesForMultipleDataFiles() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ List<String> dataFilePaths = getDataFilePaths(table);
+ assertThat(dataFilePaths).hasSize(2);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(
+ new DVPosition(dataFilePaths.get(0), 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement1(
+ new StreamRecord<>(
+ new DVPosition(dataFilePaths.get(1), 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ List<DVWriteResult> output = harness.extractOutputValues();
+ assertThat(output).hasSize(1);
+ assertThat(output.get(0).dvFiles()).hasSize(2);
+ }
+ }
+
+ @Test
+ void routesErrorToErrorStream() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ dropTable();
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ assertThat(harness.extractOutputValues()).hasSize(1);
+ assertThat(harness.extractOutputValues().get(0).hasError()).isTrue();
+ assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ }
+ }
+
+ @Test
+ void writesDVForStagingDataFile() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+ DataFile stagingDataFile = getFirstDataFile(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ EqualityConvertPlan planResult =
+ new EqualityConvertPlan(
+ Lists.newArrayList(stagingDataFile), Lists.newArrayList(), 1L, null, 0L, 0L);
Review Comment:
This passes the data file as the plan's `dataFiles` (arg 1) and leaves `stagingDVFiles` (arg 2) empty, but the operator folds staging DVs by iterating `planResult.stagingDVFiles()` (EqualityConvertDVWriter.java:240) and never reads `dataFiles`. So this case exercises no staging-DV merge and is behaviorally identical to writesDVFileForSinglePosition. To cover the fold, put a real DV (a PUFFIN DeleteFile whose referencedDataFile is the data file path) in `stagingDVFiles` and assert it is merged, e.g. it shows up in rewrittenDvFiles or the written DV's recordCount reflects both the staged and new positions.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java:
##########
@@ -0,0 +1,354 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.operator;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.BaseDeleteLoader;
+import org.apache.iceberg.data.DeleteLoader;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.ManifestEvaluator;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.StructLikeWrapper;
+import org.roaringbitmap.longlong.Roaring64Bitmap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Keyed parallel resolver that buffers {@link DVPosition}s per data-file path, then writes Puffin
+ * DV files directly via {@link BaseDVFileWriter}. Plan metadata arrives broadcast on input 2, so
+ * every parallel task sees the cycle's metadata and can validate against the main snapshot.
+ *
+ * <p>Each buffered {@link DVPosition} carries the data file's {@code specId} + encoded partition,
+ * so writing DVs needs no data-manifest scan. Existing DVs are folded into the rewrite (V3 allows
+ * one DV per data file): delete manifests are pruned by partition summary to the cycle's affected
+ * partitions, then filtered to entries referencing the affected data files. No cross-cycle state is
+ * kept; reads are bounded by the pruned manifest set, not the table's full DV history.
+ *
+ * <p>Buffered positions are transient per-task. On failure recovery, upstream replay rebuilds them.
+ */
+@Internal
+public class EqualityConvertDVWriter extends AbstractStreamOperator<DVWriteResult>
+ implements TwoInputStreamOperator<DVPosition, EqualityConvertPlan, DVWriteResult> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(EqualityConvertDVWriter.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final TableLoader tableLoader;
+ private final String targetBranch;
+
+ private transient Table table;
+ private transient OutputFileFactory fileFactory;
+ private transient DeleteLoader deleteLoader;
+ private transient Map<String, FilePositions> positionsByFile;
+ private transient EqualityConvertPlan planResult;
+ private transient boolean hasUpstreamError;
+ private transient int manifestsRead;
+
+ public EqualityConvertDVWriter(
+ String tableName, String taskName, TableLoader tableLoader, String targetBranch) {
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.tableLoader = tableLoader;
+ this.targetBranch = targetBranch;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ if (!tableLoader.isOpen()) {
+ tableLoader.open();
+ }
+
+ table = tableLoader.loadTable();
+ int subtaskIndex = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
+ fileFactory =
+ OutputFileFactory.builderFor(table, subtaskIndex, 0L).format(FileFormat.PUFFIN).build();
+ deleteLoader = new BaseDeleteLoader(deleteFile -> table.io().newInputFile(deleteFile));
+ positionsByFile = Maps.newHashMap();
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DVPosition> record) {
+ DVPosition pos = record.getValue();
+ if (pos.isAbort()) {
+ hasUpstreamError = true;
+ }
+
+ if (!hasUpstreamError) {
+ positionsByFile
+ .computeIfAbsent(
+ pos.dataFilePath(), k -> new FilePositions(pos.specId(), pos.partition()))
+ .positions
+ .addLong(pos.position());
+ }
+ }
+
+ @Override
+ public void processElement2(StreamRecord<EqualityConvertPlan> record) {
+ planResult = record.getValue();
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ if (planResult != null && mark.getTimestamp() >= planResult.doneTimestamp()) {
+ if (hasUpstreamError) {
+ output.collect(new StreamRecord<>(DVWriteResult.ABORT));
+ } else {
+ try {
+ resolveAndWrite();
+ } catch (Exception e) {
+ LOG.error("Error writing DVs for table {} task {}", tableName, taskName, e);
+ output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+ output.collect(new StreamRecord<>(DVWriteResult.ABORT));
+ }
+ }
+
+ positionsByFile.clear();
+ hasUpstreamError = false;
+ planResult = null;
+ }
+
+ super.processWatermark(mark);
+ }
+
+ private void resolveAndWrite() throws IOException {
+ if (positionsByFile.isEmpty()) {
+ return;
+ }
+
+ table.refresh();
+
+ Snapshot mainSnapshot = table.snapshot(targetBranch);
+
+ // Fail fast if the main branch changed since planning, to avoid writing DV files that the
+ // committer would reject via validateFromSnapshot. The next cycle will reindex.
+ if (mainSnapshot != null
+ && planResult.mainSnapshotId() != null
Review Comment:
This snapshot-changed fail-fast only runs when `planResult.mainSnapshotId() != null`, but every test builds its plan with a null mainSnapshotId (emptyEqualityConvertPlan and the staging case at TestEqualityConvertDVWriter.java:210), so this branch and its IllegalStateException never execute under test. Add a case that sets a non-null mainSnapshotId, commits a new snapshot on the target branch after planning, and asserts the writer emits DVWriteResult.ABORT and routes the error to ERROR_STREAM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add equality delete conversion DV resolution and writing [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #16858:
URL: https://github.com/apache/iceberg/pull/16858#discussion_r3434366995
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java:
##########
@@ -0,0 +1,431 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.flink.SimpleDataUtil.createRecord;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.Test;
+
+class TestEqualityConvertDVWriter extends OperatorTestBase {
+
+ private static final byte[] EMPTY_PARTITION = new byte[0];
+
+ @Test
+ void writesDVFileForSinglePosition() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ List<DVWriteResult> output = harness.extractOutputValues();
+ assertThat(output).hasSize(1);
+ assertThat(output.get(0).hasError()).isFalse();
+ assertThat(output.get(0).dvFiles()).hasSize(1);
+ }
+ }
+
+ @Test
+ void writesSingleDVFileForMultiplePositionsOnSameDataFile() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+ insert(table, 3, "c");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 1, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 2, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ List<DVWriteResult> output = harness.extractOutputValues();
+ assertThat(output).hasSize(1);
+ assertThat(output.get(0).hasError()).isFalse();
+ assertThat(output.get(0).dvFiles()).hasSize(1);
+ assertThat(output.get(0).dvFiles().get(0).recordCount()).isEqualTo(3);
+ assertThat(output.get(0).rewrittenDvFiles()).isEmpty();
+ }
+ }
+
+ @Test
+ void emptyPositionsProducesNoOutput() throws Exception {
+ createTableWithDelete(3);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ assertThat(harness.extractOutputValues()).isEmpty();
+ }
+ }
+
+ @Test
+ void noOutputWithoutPlanResult() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ assertThat(harness.extractOutputValues()).isEmpty();
+ }
+ }
+
+ @Test
+ void writesDVFilesForMultipleDataFiles() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ List<String> dataFilePaths = getDataFilePaths(table);
+ assertThat(dataFilePaths).hasSize(2);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(
+ new DVPosition(dataFilePaths.get(0), 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement1(
+ new StreamRecord<>(
+ new DVPosition(dataFilePaths.get(1), 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ List<DVWriteResult> output = harness.extractOutputValues();
+ assertThat(output).hasSize(1);
+ assertThat(output.get(0).dvFiles()).hasSize(2);
+ }
+ }
+
+ @Test
+ void routesErrorToErrorStream() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ harness.processElement1(
+ new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time));
+ harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time));
+
+ dropTable();
+
+ harness.processBothWatermarks(new Watermark(time));
+
+ assertThat(harness.extractOutputValues()).hasSize(1);
+ assertThat(harness.extractOutputValues().get(0).hasError()).isTrue();
+ assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ }
+ }
+
+ @Test
+ void writesDVForStagingDataFile() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+ DataFile stagingDataFile = getFirstDataFile(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan, DVWriteResult> harness =
+ createHarness()) {
+ harness.open();
+
+ long time = System.currentTimeMillis();
+ EqualityConvertPlan planResult =
+ new EqualityConvertPlan(
+ Lists.newArrayList(stagingDataFile), Lists.newArrayList(), 1L, null, 0L, 0L);
Review Comment:
Thanks! I modified the test to verify that the merging works as expected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Add equality delete conversion DV resolution and writing [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao merged PR #16858:
URL: https://github.com/apache/iceberg/pull/16858
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org