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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
