mxm commented on code in PR #16889: URL: https://github.com/apache/iceberg/pull/16889#discussion_r3459441731
########## flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java: ########## @@ -0,0 +1,1122 @@ +/* + * 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); + assertThat(planner(harness).processedStagingSnapshotNum()).isEqualTo(1); + assertThat(planner(harness).processedEqDeleteFileNum()).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 emitsDataFileFromInsertOnlyStagingSnapshot() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + table.manageSnapshots().createBranch(STAGING_BRANCH).commit(); + table.refresh(); + + // S1: eq delete targeting main data + 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 still be indexed for the configured + // equality fields + 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) + sendTrigger(harness); + int afterFirst = harness.extractOutputValues().size(); + assertThat(countEqDeleteTasks(harness.extractOutputValues())).isEqualTo(1); + + // Record S1's conversion on main, so the planner walks past S1 before processing S2. + simulateConvertCommit(table, s1SnapshotId); + + // Trigger 2: processes S2 (insert-only). Must emit its data file so the configured equality + // fields stay indexed. + 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); + assertThat(planner(harness).skippedNoOpCycles()).isZero(); + + // 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); + assertThat(planner(harness).skippedNoOpCycles()).isEqualTo(1); + } + } + + @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); + assertThat(planner(harness).skippedNoOpCycles()).isEqualTo(1); + assertThat(planner(harness).reindexCount()).isZero(); + } + } + + @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 { Review Comment: Yes, it's better to use explicit checks. 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]
