wombatu-kun commented on code in PR #16889: URL: https://github.com/apache/iceberg/pull/16889#discussion_r3446593825
########## flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java: ########## @@ -0,0 +1,741 @@ +/* + * 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.Objects; +import java.util.Set; +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.state.ListState; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.runtime.state.StateInitializationContext; +import org.apache.flink.runtime.state.StateSnapshotContext; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.util.OutputTag; +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotChanges; +import org.apache.iceberg.SnapshotSummary; +import org.apache.iceberg.Table; +import org.apache.iceberg.flink.TableLoader; +import org.apache.iceberg.flink.maintenance.api.Trigger; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.ContentFileUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Planner for the equality delete conversion pipeline. For each trigger, it picks the oldest + * staging snapshot that hasn't been converted yet and emits {@link ReadCommand}s describing the + * files its downstream readers and workers must process. + * + * <p>Each trigger runs two steps in order: + * + * <ol> + * <li>{@link #ensureIndexCurrent}: updates {@link #lastStagingSnapshotId} from main's history, + * bootstraps the worker index from main on first run, and reindexes when external commits + * (e.g. compaction) have advanced main past the currently-indexed snapshot. + * <li>{@link #processStagingSnapshot}: resolve the chosen staging snapshot's eq deletes against + * the (now-current) index, pass through any DV files, and index the snapshot's new data files + * for the next cycle. + * </ol> + * + * Watermarks separate phases that gate the worker's keyed state. The contract is documented on + * {@link #advancePhase()}. + * + * <p>An {@link EqualityConvertPlan} with the current cycle's metadata is emitted via the {@link + * #METADATA_STREAM} side output after the read commands. + * + * <p>Assumes a single equality-field set supplied via the builder; staging eq-deletes with a + * different {@code equalityFieldIds} fail fast in {@link #retrieveStagingFiles}. Concurrent writes + * on the target branch are handled by {@link #ensureIndexCurrent} reindexing from the new main + * snapshot; commit-time conflicts are caught by {@code RowDelta.validateFromSnapshot}. + */ +@Internal +public class EqualityConvertPlanner extends AbstractStreamOperator<ReadCommand> + implements OneInputStreamOperator<Trigger, ReadCommand> { + + private static final Logger LOG = LoggerFactory.getLogger(EqualityConvertPlanner.class); + + public static final OutputTag<EqualityConvertPlan> METADATA_STREAM = + new OutputTag<>("metadata-stream") {}; + + public static final OutputTag<IndexCommand> CLEAR_BROADCAST_STREAM = + new OutputTag<>("clear-broadcast-stream") {}; + + private static final String PROCESSED_EQ_DELETE_FILE_NUM_METRIC = "processedEqDeleteFileNum"; + private static final String PROCESSED_STAGING_SNAPSHOT_NUM_METRIC = "processedStagingSnapshotNum"; + private static final String SKIPPED_NO_OP_CYCLES_METRIC = "skippedNoOpCycles"; + private static final String REINDEX_COUNT_METRIC = "reindexCount"; + + private final String tableName; + private final String taskName; + private final TableLoader tableLoader; + private final String stagingBranch; + private final String targetBranch; + private final boolean stagingOnTargetBranch; + // Equality-field-id set the worker keys on. Supplied via the builder; every staging + // eq-delete's equalityFieldIds() must match exactly. + private final Set<Integer> eqFieldIds; + + // Main snapshot id the worker's index reflects. + private transient ListState<Long> indexSnapshotState; + // Main sequence number the worker's index reflects. + private transient ListState<Long> indexedSequenceNumberState; + // Equality field IDs the index was built with, allows to detect reconfiguration. + private transient ListState<Integer> eqFieldIdsState; + + private transient Table table; + + private transient Long lastMainSnapshotId; + private transient Long lastStagingSnapshotId; + private transient Long indexSnapshotId; + private transient Long indexedSequenceNumber; + + private transient long nextPhaseTs; + + private transient Counter processedEqDeleteFileNumCounter; + private transient Counter processedStagingSnapshotNumCounter; + private transient Counter skippedNoOpCyclesCounter; + private transient Counter reindexCounter; + private transient Counter errorCounter; + + public EqualityConvertPlanner( + String tableName, + String taskName, + TableLoader tableLoader, + String stagingBranch, + String targetBranch, + Set<Integer> eqFieldIds) { + this.tableName = tableName; + this.taskName = taskName; + this.tableLoader = tableLoader; + this.stagingBranch = stagingBranch; + this.targetBranch = targetBranch; + this.stagingOnTargetBranch = stagingBranch.equals(targetBranch); + Preconditions.checkArgument( + eqFieldIds != null && !eqFieldIds.isEmpty(), "eqFieldIds must not be null or empty"); + this.eqFieldIds = ImmutableSet.copyOf(eqFieldIds); + } + + @Override + public void open() throws Exception { + super.open(); + if (!tableLoader.isOpen()) { + tableLoader.open(); + } + + table = tableLoader.loadTable(); + + MetricGroup taskMetricGroup = + TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, 0); + this.processedEqDeleteFileNumCounter = + taskMetricGroup.counter(PROCESSED_EQ_DELETE_FILE_NUM_METRIC); + this.processedStagingSnapshotNumCounter = + taskMetricGroup.counter(PROCESSED_STAGING_SNAPSHOT_NUM_METRIC); + this.skippedNoOpCyclesCounter = taskMetricGroup.counter(SKIPPED_NO_OP_CYCLES_METRIC); + this.reindexCounter = taskMetricGroup.counter(REINDEX_COUNT_METRIC); + this.errorCounter = taskMetricGroup.counter(TableMaintenanceMetrics.ERROR_COUNTER); + } + + @Override + public void initializeState(StateInitializationContext context) throws Exception { + super.initializeState(context); + indexSnapshotState = + context + .getOperatorStateStore() + .getListState(new ListStateDescriptor<>("indexSnapshotId", Types.LONG)); + + indexSnapshotId = null; + for (Long stateValue : indexSnapshotState.get()) { + Preconditions.checkState( + indexSnapshotId == null, "indexSnapshotId state should hold at most one value"); + indexSnapshotId = stateValue; + } + + indexedSequenceNumberState = + context + .getOperatorStateStore() + .getListState(new ListStateDescriptor<>("indexedSequenceNumber", Types.LONG)); + + indexedSequenceNumber = null; + for (Long stateValue : indexedSequenceNumberState.get()) { + Preconditions.checkState( + indexedSequenceNumber == null, + "indexedSequenceNumber state should hold at most one value"); + indexedSequenceNumber = stateValue; + } + + eqFieldIdsState = + context + .getOperatorStateStore() + .getListState(new ListStateDescriptor<>("eqFieldIds", Types.INT)); + Set<Integer> restoredEqFieldIds = Sets.newHashSet(eqFieldIdsState.get()); + Preconditions.checkState( + restoredEqFieldIds.isEmpty() || restoredEqFieldIds.equals(eqFieldIds), + "Equality field IDs changed across restart: restored=%s, configured=%s. " + + "Reconfiguring equality-field columns is not supported; " + + "restart from a clean state (no savepoint).", + restoredEqFieldIds, + eqFieldIds); + } + + @Override + public void snapshotState(StateSnapshotContext context) throws Exception { + super.snapshotState(context); + indexSnapshotState.clear(); + if (indexSnapshotId != null) { + indexSnapshotState.add(indexSnapshotId); + } + + indexedSequenceNumberState.clear(); + if (indexedSequenceNumber != null) { + indexedSequenceNumberState.add(indexedSequenceNumber); + } + + eqFieldIdsState.clear(); + for (int id : eqFieldIds) { + eqFieldIdsState.add(id); + } + } + + @Override + public void processElement(StreamRecord<Trigger> element) throws Exception { + long triggerTs = element.getTimestamp(); + nextPhaseTs = Math.max(triggerTs, nextPhaseTs + 1); + + Long currentMainSnapshotId = lastMainSnapshotId; + try { + table.refresh(); + Snapshot mainSnapshot = table.snapshot(targetBranch); + currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId() : null; + + ensureIndexCurrent(mainSnapshot); + + Snapshot nextToProcess = + nextUnprocessedStagingSnapshot(table.snapshot(stagingBranch), mainSnapshot); + + if (nextToProcess == null) { + LOG.info("Nothing new to convert on staging branch '{}'.", stagingBranch); + emitNoOpResult(triggerTs, currentMainSnapshotId); + return; + } + + processStagingSnapshot(nextToProcess, triggerTs, currentMainSnapshotId); + } catch (Exception e) { + LOG.error("Error processing equality deletes for table {} task {}", tableName, taskName, e); + output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e)); + errorCounter.inc(); + emitDrainResult(triggerTs, currentMainSnapshotId); + } + } + + /** + * Brings the worker's index up to date with the current state of the target branch: + * + * <ul> + * <li>Updates {@link #lastStagingSnapshotId} from the most recent committer marker on main. + * <li>Bootstraps the index from main on the first trigger with a non-null main snapshot. + * <li>Reindexes from main when external commits (e.g. compaction or direct writes) have + * advanced main past the currently-indexed snapshot. + * </ul> + * + * <p>No-op when main hasn't moved since the last trigger. Otherwise the history walk is bounded + * to commits added since {@link #lastMainSnapshotId}. + */ + private void ensureIndexCurrent(Snapshot mainSnapshot) { + Long currentMainSnapshotId = mainSnapshot != null ? mainSnapshot.snapshotId() : null; + + if (Objects.equals(lastMainSnapshotId, currentMainSnapshotId)) { + return; + } + + LastCommittedWork info = discoverLastCommittedWork(mainSnapshot); + updateLastStagingSnapshotId(info); + + boolean bootstrap = mainSnapshot != null && indexSnapshotId == null; + boolean reindex = indexSnapshotId != null && info.externalCommitCount() > 0; + if (bootstrap || reindex) { + LOG.info( + "{} worker index from main snapshot {} for field IDs {}.", + bootstrap ? "Bootstrapping" : "Reindexing", + currentMainSnapshotId, + eqFieldIds); + if (reindex) { + // Evict keyed entries the reindex will not re-add (e.g. data file removed by CoW). + output.collect( + CLEAR_BROADCAST_STREAM, + new StreamRecord<>( + IndexCommand.clearBeforeReindex( + currentMainSnapshotId, mainSnapshot.sequenceNumber()))); + reindexCounter.inc(); + } + + indexSnapshotId = currentMainSnapshotId; + indexedSequenceNumber = mainSnapshot.sequenceNumber(); + emitMainDataReadCommands(mainSnapshot); + } + + lastMainSnapshotId = currentMainSnapshotId; + } + + private void updateLastStagingSnapshotId(LastCommittedWork info) { + if (info.lastCommittedStaging() != null) { + lastStagingSnapshotId = info.lastCommittedStaging(); + return; + } + + Preconditions.checkState( + lastMainSnapshotId == null || lastStagingSnapshotId == null || info.reachedLastInspected(), Review Comment: This guard - a prior staging marker was seen but none is now reachable on the target branch (rollback, replace_main, or snapshot expiration) - is the only protection against silently reprocessing already-committed data after a target rewrite, and it has no test. The other three fail-fast paths in `retrieveStagingFiles` (removed data files, eq-field-id mismatch, V2 positional delete) are each covered. Suggest adding a test that seeds `lastStagingSnapshotId` via a committed marker, then rewrites or rolls back the target so the history walk reaches root without a marker, and asserts the IllegalStateException. ########## flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java: ########## @@ -0,0 +1,1081 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Files; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RewriteFiles; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.Table; +import org.apache.iceberg.data.FileHelpers; +import org.apache.iceberg.data.GenericAppenderHelper; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.PositionDelete; +import org.apache.iceberg.flink.maintenance.api.Trigger; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class TestEqualityConvertPlanner extends OperatorTestBase { + + private static final String STAGING_BRANCH = "__flink_staging_test"; + + @TempDir private Path tempDir; + + @Test + void bootstrapsIndexFromBuilderEqualityFieldIds() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + + // Staging branch exists but contains no eq-delete files yet. The planner still populates the + // worker index from main using the builder-configured equality field set so the index is ready + // when the first delete arrives. + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH, Lists.newArrayList(1, 2))) { + harness.open(); + sendTrigger(harness); + + List<ReadCommand> commands = harness.extractOutputValues(); + assertThat(countDataFileTasks(commands)).isEqualTo(2); + assertThat(countEqDeleteTasks(commands)).isZero(); + + assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(1); + } + } + + @Test + void failsWhenStagingEqDeleteFieldIdsMismatchBuilder() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + // Builder configured for [1, 2], but writer produces an eq delete with [1] only. + DeleteFile mismatched = writeIdOnlyEqualityDelete(table, 1); + table.newRowDelta().addDeletes(mismatched).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH, Lists.newArrayList(1, 2))) { + harness.open(); + sendTrigger(harness); + + assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1); + } + } + + @Test + void doesNotDuplicateNewDataFilesWhenStagingEqualsTarget() throws Exception { + // When stagingBranch == targetBranch, the writer commits new data files directly to main. + // Bootstrap scans the main snapshot (which already includes those files) and indexes them. + // The planner must NOT also emit the staging-data phase for the same files — that would + // duplicate ADD_DATA_ROW commands for the same (PK, file, position) in the worker's index. + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + + // Writer commits (new-data id=3) + (eq-delete id=1) in one RowDelta directly to main. + DataFile newDataFile = writeDataFile(table, createRecord(3, "c")); + DeleteFile eqDelete = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addRows(newDataFile).addDeletes(eqDelete).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(SnapshotRef.MAIN_BRANCH)) { + harness.open(); + sendTrigger(harness); + + List<ReadCommand> commands = harness.extractOutputValues(); + + // Bootstrap from main emits one data-file command per file (id=1, id=2, id=3) = 3. + // Without the same-branch guard, emitSnapshotDataPhase would also emit newDataFile → 4. + assertThat(countDataFileTasks(commands)).isEqualTo(3); + assertThat(countEqDeleteTasks(commands)).isEqualTo(1); + + long newDataFileCount = + commands.stream() + .filter(c -> c.task() instanceof FileScanTask) + .filter(c -> c.task().file().location().equals(newDataFile.location())) + .count(); + assertThat(newDataFileCount).isEqualTo(1); + } + } + + @Test + void emitsReadCommandsForEqualityDeletes() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + insert(table, 3, "c"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + DeleteFile eqDelete = writeEqualityDelete(table, 2, "b"); + table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + sendTrigger(harness); + + List<ReadCommand> commands = harness.extractOutputValues(); + // 3 DATA_FILE (from main) + 1 EQ_DELETE_FILE + assertThat(countDataFileTasks(commands)).isEqualTo(3); + assertThat(countEqDeleteTasks(commands)).isEqualTo(1); + + List<StreamRecord<EqualityConvertPlan>> metadata = + Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); + assertThat(metadata).hasSize(1); + assertThat(metadata.get(0).getValue().dataFiles()).isEmpty(); + } + } + + @Test + void emitsDataFileFromInsertOnlySnapshotForActiveFieldSets() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + // S1: eq delete targeting main data (activates a field set in the worker index) + DeleteFile eqDelete = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + long s1SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId(); + + // S2: insert-only (no eq deletes, but its data must be indexed for the active field set) + DataFile insertS2 = writeDataFile(table, createRecord(2, "b")); + table.newAppend().appendFile(insertS2).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + + // Trigger 1: processes S1 (eq delete), activates the field set + sendTrigger(harness); + int afterFirst = harness.extractOutputValues().size(); + assertThat(countEqDeleteTasks(harness.extractOutputValues())).isEqualTo(1); + + // Simulate the committer writing the S1 conversion to main so the planner + // promotes its pending cursor before processing S2. + simulateConvertCommit(table, s1SnapshotId); + + // Trigger 2: processes S2 (insert-only). Must emit its data file for the active field set. + sendTrigger(harness); + List<ReadCommand> allCommands = harness.extractOutputValues(); + List<ReadCommand> trigger2Commands = allCommands.subList(afterFirst, allCommands.size()); + + Set<String> dataFilePaths = + trigger2Commands.stream() + .filter(c -> c.task() instanceof FileScanTask) + .map(TestEqualityConvertPlanner::filePath) + .collect(Collectors.toSet()); + + assertThat(dataFilePaths).contains(insertS2.location()); + } + } + + @Test + void includesNewDataFilesFromStaging() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + DataFile newDataFile = writeDataFile(table, createRecord(2, "b")); + DeleteFile eqDelete = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addRows(newDataFile).addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + sendTrigger(harness); + + List<ReadCommand> commands = harness.extractOutputValues(); + // 1 main data file + 1 eq delete + 1 staging data file + assertThat(countDataFileTasks(commands)).isEqualTo(2); + assertThat(countEqDeleteTasks(commands)).isEqualTo(1); + + List<StreamRecord<EqualityConvertPlan>> metadata = + Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); + assertThat(metadata).hasSize(1); + assertThat(metadata.get(0).getValue().dataFiles()).hasSize(1); + } + } + + @Test + void findsIntersectionWhenMainAdvancedAfterStagingFork() throws Exception { + Table table = createTableWithDelete(3); + // Pre-fork main: two snapshots. + insert(table, 1, "a"); + insert(table, 2, "b"); + + // Fork staging from current main head. + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + // Main advances with two more snapshots after the fork. These are on main's + // lineage but not on staging's. + insert(table, 3, "c"); + insert(table, 4, "d"); + + // Staging gets one new snapshot containing an equality delete. + DeleteFile eqDelete = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + sendTrigger(harness); + + // Planner identifies the fork point and processes only the staging-only + // commit (the eq delete), and emits all four current main data files + // for the index refresh. + List<ReadCommand> commands = harness.extractOutputValues(); + assertThat(countEqDeleteTasks(commands)).isEqualTo(1); + assertThat(countDataFileTasks(commands)).isEqualTo(4); + } + } + + @Test + void skipsAlreadyProcessedStagingSnapshot() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + DeleteFile eqDelete = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + long stagingSnapshotId = table.snapshot(STAGING_BRANCH).snapshotId(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + + sendTrigger(harness); + int firstTriggerCount = harness.extractOutputValues().size(); + assertThat(firstTriggerCount).isGreaterThan(0); + + // Simulate the committer committing to main with the staging snapshot property + // (the planner promotes pending only after confirming the commit landed on main). + simulateConvertCommit(table, stagingSnapshotId); + + sendTrigger(harness); + assertThat(harness.extractOutputValues()).hasSize(firstTriggerCount); + } + } + + @Test + void noOutputWhenStagingBranchEmpty() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + sendTrigger(harness); + + // Bootstrap from main for the configured field set: 1 main DATA_FILE; no-op metadata still + // emitted because there's nothing on staging to convert. + assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(1); + assertThat(countEqDeleteTasks(harness.extractOutputValues())).isZero(); + assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(1); + } + } + + @Test + void propagatesStagingOnlyPositionalDeletes() throws Exception { + Table table = createTableWithDelete(3); + DataFile mainData = writeDataFile(table, createRecord(1, "a")); + table.newAppend().appendFile(mainData).commit(); + table.refresh(); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + // Staging snapshot has only a DV (no eq deletes, no new data files). Without the + // planner forwarding stagingDVFiles in this case, the committer would never + // see the DV and it would be lost. + DeleteFile stagingDV = writeStagingDV(table, mainData.location(), 0L); + table.newRowDelta().addDeletes(stagingDV).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + sendTrigger(harness); + + List<ReadCommand> commands = harness.extractOutputValues(); + assertThat(countDataFileTasks(commands)).isEqualTo(1); + assertThat(countEqDeleteTasks(commands)).isZero(); + + List<StreamRecord<EqualityConvertPlan>> metadata = + Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); + assertThat(metadata).hasSize(1); + EqualityConvertPlan result = metadata.get(0).getValue(); + assertThat(result.dataFiles()).isEmpty(); + assertThat(result.stagingDVFiles()).hasSize(1); + assertThat(result.stagingDVFiles().get(0).location()).isEqualTo(stagingDV.location()); + } + } + + @Test + void phaseTimestampsAreMonotonicallyIncreasing() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + // Staging: eq delete + new data file (triggers 3 phases: main data, eq delete, staging data) + DataFile newDataFile = writeDataFile(table, createRecord(2, "b")); + DeleteFile eqDelete = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addRows(newDataFile).addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + sendTrigger(harness); + + List<StreamRecord<ReadCommand>> records = Lists.newArrayList(harness.getRecordOutput()); + // Verify timestamps are non-decreasing and phases are distinct + long prevTs = Long.MIN_VALUE; + for (StreamRecord<ReadCommand> record : records) { + assertThat(record.getTimestamp()).isGreaterThanOrEqualTo(prevTs); + prevTs = record.getTimestamp(); + } + + // DATA_FILE commands should have different timestamps than EQ_DELETE_FILE commands + Set<Long> dataFileTimestamps = + records.stream() + .filter(r -> r.getValue().task() instanceof FileScanTask) + .map(StreamRecord::getTimestamp) + .collect(Collectors.toSet()); + Set<Long> eqDeleteTimestamps = + records.stream() + .filter(r -> isEqDelete(r.getValue())) + .map(StreamRecord::getTimestamp) + .collect(Collectors.toSet()); + + // Main data and eq delete should be in different phases + for (long eqTs : eqDeleteTimestamps) { + boolean hasLowerDataTs = dataFileTimestamps.stream().anyMatch(ts -> ts < eqTs); + assertThat(hasLowerDataTs).isTrue(); + } + } + } + + @Test + void routesExceptionToErrorStream() throws Exception { + createTableWithDelete(3); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + + dropTable(); + + assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull(); + sendTrigger(harness); + assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1); + } + } + + @Test + void failsOnRemovedDataFilesOnStagingBranch() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + Set<DataFile> oldDataFiles = Sets.newHashSet(); + for (ManifestFile manifest : table.currentSnapshot().dataManifests(table.io())) { + try (ManifestReader<DataFile> reader = + ManifestFiles.read(manifest, table.io(), table.specs())) { + for (DataFile df : reader) { + oldDataFiles.add(df.copy()); + } + } + } + + assertThat(oldDataFiles).hasSize(2); + + // Rewrite on the staging branch (not main). Equality delete conversion does not support + // rewrites on staging; the planner must fail the cycle instead of silently dropping the + // removed files. + DataFile compactedFile = + new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir) + .writeFile(Lists.newArrayList(createRecord(1, "a"), createRecord(2, "b"))); + RewriteFiles rewrite = table.newRewrite(); + for (DataFile old : oldDataFiles) { + rewrite.deleteFile(old); + } + + rewrite.addFile(compactedFile); + rewrite.toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + sendTrigger(harness); + + assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1); + // Bootstrap runs before processCycle's failure, so the main data commands are already on + // the wire. Only the cycle itself fails. + assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(2); + assertThat(countEqDeleteTasks(harness.extractOutputValues())).isZero(); + } + } + + @Test + void reEmitsMainDataAfterCompaction() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + DeleteFile eqDelete1 = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete1).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + + sendTrigger(harness); + int firstTriggerCount = harness.extractOutputValues().size(); + // 2 DATA_FILE (main) + 1 EQ_DELETE_FILE + assertThat(firstTriggerCount).isEqualTo(3); + + // Compact data files on main: rewrite 2 files into 1 + Set<DataFile> oldDataFiles = Sets.newHashSet(); + for (ManifestFile manifest : table.currentSnapshot().dataManifests(table.io())) { + try (ManifestReader<DataFile> reader = + ManifestFiles.read(manifest, table.io(), table.specs())) { + for (DataFile df : reader) { + oldDataFiles.add(df.copy()); + } + } + } + + assertThat(oldDataFiles).hasSize(2); + + DataFile compactedFile = + new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir) + .writeFile(Lists.newArrayList(createRecord(1, "a"), createRecord(2, "b"))); + RewriteFiles rewrite = table.newRewrite(); + for (DataFile old : oldDataFiles) { + rewrite.deleteFile(old); + } + + rewrite.addFile(compactedFile); + rewrite.commit(); + table.refresh(); + + DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b"); + table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + sendTrigger(harness); + List<ReadCommand> allCommands = harness.extractOutputValues(); + List<ReadCommand> trigger2Commands = + allCommands.subList(firstTriggerCount, allCommands.size()); + + // 1 DATA_FILE (compacted main) + 1 EQ_DELETE_FILE + assertThat(countDataFileTasks(trigger2Commands)).isEqualTo(1); + assertThat(countEqDeleteTasks(trigger2Commands)).isEqualTo(1); + + // DATA_FILE should reference the compacted file + List<ReadCommand> dataCmds = + trigger2Commands.stream() + .filter(c -> c.task() instanceof FileScanTask) + .collect(Collectors.toList()); + for (ReadCommand cmd : dataCmds) { + assertThat(cmd.task().file().location()).isEqualTo(compactedFile.location()); + } + } + } + + @Test + void refreshesIndexBeforeFirstCycleProcessesEqDeletes() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + // Stage MULTIPLE eq-delete snapshots before the first trigger arrives. Without bootstrap we + // would create the index entry lazily over multiple cycles. With bootstrap, the first trigger + // discovers every unprocessed schema (just one in this case, since both deletes use the same + // schema) and creates it once before any cycle processes. + DeleteFile eqDelete1 = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete1).toBranch(STAGING_BRANCH).commit(); + DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b"); + table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + long s2SnapshotId = table.snapshot(STAGING_BRANCH).snapshotId(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + + sendTrigger(harness); + int afterFirst = harness.extractOutputValues().size(); + + // First trigger processes the OLDER staging snapshot (eqDelete1). Bootstrap created the + // index entry for the eq-delete schema once: 2 main DATA_FILE + 1 EQ_DELETE_FILE = 3 + // commands. + assertThat(afterFirst).isEqualTo(3); + assertThat(countDataFileTasks(harness.extractOutputValues())).isEqualTo(2); + assertThat(countEqDeleteTasks(harness.extractOutputValues())).isEqualTo(1); + + simulateConvertCommit(table, table.snapshot(STAGING_BRANCH).parentId()); + + // Second trigger processes the newer staging snapshot. The index entry already exists — no + // main re-emission, just the new eq delete. + sendTrigger(harness); + List<ReadCommand> allCommands = harness.extractOutputValues(); + List<ReadCommand> trigger2Commands = allCommands.subList(afterFirst, allCommands.size()); + assertThat(countDataFileTasks(trigger2Commands)).isZero(); + assertThat(countEqDeleteTasks(trigger2Commands)).isEqualTo(1); + + simulateConvertCommit(table, s2SnapshotId); + } + } + + @Test + void noMainReEmitWhenUnchanged() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + DeleteFile eqDelete1 = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete1).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + + sendTrigger(harness); + int firstTriggerCount = harness.extractOutputValues().size(); + // 2 DATA_FILE (main) + 1 EQ_DELETE_FILE + assertThat(firstTriggerCount).isEqualTo(3); + assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(1); + + DeleteFile eqDelete2 = writeEqualityDelete(table, 2, "b"); + table.newRowDelta().addDeletes(eqDelete2).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + sendTrigger(harness); + List<ReadCommand> allCommands = harness.extractOutputValues(); + List<ReadCommand> trigger2Commands = + allCommands.subList(firstTriggerCount, allCommands.size()); + + // Only 1 EQ_DELETE_FILE, no main re-emission + assertThat(countDataFileTasks(trigger2Commands)).isEqualTo(0); + assertThat(countEqDeleteTasks(trigger2Commands)).isEqualTo(1); + + assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(2); + } + } + + @Test + void noMainReEmitAfterOwnCommit() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + insert(table, 2, "b"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + DeleteFile eqDelete1 = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete1).toBranch(STAGING_BRANCH).commit(); + table.refresh(); + + long indexedStagingSnapshotId = table.snapshot(STAGING_BRANCH).snapshotId(); + + try (OneInputStreamOperatorTestHarness<Trigger, ReadCommand> harness = + createHarness(STAGING_BRANCH)) { + harness.open(); + + sendTrigger(harness); + int firstTriggerCount = harness.extractOutputValues().size(); + + // Simulate converter's own commit referencing the staging snapshot that the + // first trigger processed. The planner promotes pendingStagingSnapshotId on the Review Comment: This comment, and the same wording at lines 209, 316, 575, 866, 871, and 875-876, describes a `pendingStagingSnapshotId` field, a "promote pending cursor" step, and an "index high-water mark" that this planner does not have. The factored-out planner derives its position by walking the target branch for `COMMITTED_STAGING_SNAPSHOT_PROPERTY` (`EqualityConvertPlanner.java:316`, `updateLastStagingSnapshotId`) and sets `lastStagingSnapshotId` straight from that marker; nothing is promoted. Line 575's "discovers every unprocessed schema" is also stale - the planner keys on the single builder-configured `eqFieldIds` set and processes one snapshot per trigger. Line 876's "which is the state that gets checkpointed" is incorrect: only `indexSnapshotId`, `indexedSequenceNumber`, and `eqFieldIds` are persisted, not `lastStagingSnapshotId`. Suggest rewording these to the marker-walk behavior. -- 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]
