mxm commented on code in PR #16889: URL: https://github.com/apache/iceberg/pull/16889#discussion_r3447955932
########## 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: Thanks! Those comments got stale in the course of several review rounds. Updated. -- 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]
