diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index ada16ec09384..34d996c3d9c3 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -143,12 +143,21 @@ public void processElement1(StreamRecord record) { @Override public void processElement2(StreamRecord record) { + // Includes the cycle-start marker. Its maximal done timestamp holds watermarks until the real + // plan replaces it. planResult = record.getValue(); } @Override public void processWatermark(Watermark mark) throws Exception { - if (planResult == null || mark.getTimestamp() < planResult.doneTimestamp()) { + if (planResult == null) { + // No cycle in flight: the watermark comes from another task's trigger or a recovery trigger + // sharing the maintenance lock. Holding it back would keep that lock forever. + super.processWatermark(mark); + return; + } + + if (mark.getTimestamp() < planResult.doneTimestamp()) { // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance lock // for the whole cycle. Forwarding the planner's mid-cycle phase watermarks will release the // lock early and could let the next trigger run a concurrent cycle on the same staging diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java index 191bf8322039..4aa326d19a19 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java @@ -34,6 +34,9 @@ * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to construct one and {@link * #noOp()} to check. * + *

A cycle-start marker is a no-op plan whose done timestamp is never reached. Use {@link + * #cycleStart(long)} to construct one. + * * @param dataFiles new staging data files committed in the cycle * @param stagingDVFiles staging DVs passed through to main, used by DVWriter to merge with * newly-created DVs @@ -78,4 +81,12 @@ public static EqualityConvertPlan noOp( triggerTimestamp, doneTimestamp); } + + /** + * Cycle-start marker: a no-op plan whose done timestamp is never reached, so consumers hold it + * until the cycle's actual plan replaces it. + */ + public static EqualityConvertPlan cycleStart(long triggerTimestamp) { + return noOp(null, triggerTimestamp, Long.MAX_VALUE); + } } diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java index 82791bae4bc1..96e5861f1c6f 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java @@ -82,7 +82,10 @@ * {@link #advancePhase()}. * *

An {@link EqualityConvertPlan} with the current cycle's metadata is emitted via the {@link - * #METADATA_STREAM} side output after the read commands. + * #METADATA_STREAM} side output after the read commands. The planner emits phase watermarks before + * that plan. A {@link EqualityConvertPlan#cycleStart} marker is emitted first, before any phase + * watermark. The marker tells the committer a cycle has started, and the committer holds watermarks + * until the commit. * *

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 @@ -264,6 +267,12 @@ public void processElement(StreamRecord element) throws Exception { long triggerTs = element.getTimestamp(); nextPhaseTs = Math.max(triggerTs, nextPhaseTs + 1); + // Mark cycle start before the first phase watermark. The committer holds back watermarks from + // here until the plan arrives; without the marker it cannot tell a phase watermark emitted + // before the plan from an idle watermark that must be forwarded, and forwarding the former + // would release the maintenance lock before the commit. + output.collect(METADATA_STREAM, new StreamRecord<>(EqualityConvertPlan.cycleStart(triggerTs))); + Long currentMainSnapshotId = lastMainSnapshotId; try { table.refresh(); diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java index f3bdaa8fe1c6..e39d05b66f00 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java @@ -19,13 +19,22 @@ package org.apache.iceberg.flink.maintenance.api; import static org.apache.iceberg.flink.SimpleDataUtil.createRecord; +import static org.apache.iceberg.flink.maintenance.api.TableMaintenance.LOCK_REMOVER_OPERATOR_NAME; +import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER; import static org.assertj.core.api.Assertions.assertThat; import java.io.File; import java.io.IOException; +import java.io.Serializable; import java.nio.file.Path; import java.time.Duration; +import java.util.Collections; +import java.util.List; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.typeutils.ResultTypeQueryable; import org.apache.flink.core.execution.JobClient; +import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; @@ -42,12 +51,16 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericAppenderHelper; import org.apache.iceberg.flink.SimpleDataUtil; +import org.apache.iceberg.flink.maintenance.operator.ManualSource; +import org.apache.iceberg.flink.maintenance.operator.MetricsReporterFactoryForTests; import org.apache.iceberg.flink.maintenance.operator.OperatorTestBase; +import org.apache.iceberg.flink.maintenance.operator.TableChange; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.util.ContentFileUtil; import org.awaitility.Awaitility; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -60,6 +73,7 @@ */ class TestConvertEqualityDeletesE2E extends OperatorTestBase { private static final String STAGING_BRANCH = "staging"; + private static final String PROBE_TASK_NAME = "SharedLockProbe"; @TempDir private Path tempDir; private StreamExecutionEnvironment env; @@ -136,6 +150,76 @@ void testConvertEqualityDeletesE2E(String stagingBranch) throws Exception { } } + /** + * ConvertEqualityDeletes shares the Flink-maintained maintenance lock with another task. The lock + * is released only when the unioned watermark reaches the lock remover, so a watermark from the + * other task has to pass through an idle converter. + */ + @Test + void releasesSharedLockAfterEachCycle() throws Exception { + Table table = createTable(3, FileFormat.PARQUET); + String tableName = table.name(); + + env.setParallelism(1); + ManualSource changes = + new ManualSource<>(env, TypeInformation.of(TableChange.class)); + + TableMaintenance.forChangeStream(changes.dataStream(), tableLoader()) + .uidSuffix("ConvertEqualityDeletesSharedLock") + .rateLimit(Duration.ofMillis(50)) + .lockCheckDelay(Duration.ofMillis(50)) + .add(new CompletingTaskBuilder().scheduleOnCommitCount(1)) + .add( + ConvertEqualityDeletes.builder() + .scheduleOnCommitCount(1) + .stagingBranch(SnapshotRef.MAIN_BRANCH) + .targetBranch(SnapshotRef.MAIN_BRANCH) + .equalityFieldColumns(ImmutableList.of("id")) + .parallelism(1)) + .append(); + + List probeSucceeded = + ImmutableList.of( + LOCK_REMOVER_OPERATOR_NAME, + tableName, + PROBE_TASK_NAME + " [0]", + "0", + SUCCEEDED_TASK_COUNTER); + List converterSucceeded = + ImmutableList.of( + LOCK_REMOVER_OPERATOR_NAME, + tableName, + "ConvertEqualityDeletes [1]", + "1", + SUCCEEDED_TASK_COUNTER); + + JobClient jobClient = null; + try { + jobClient = env.executeAsync(); + + // The probe runs first, while the converter has no plan. Its watermark must leave the + // converter or the shared lock stays held and the converter never starts. + changes.sendRecord(TableChange.builder().commitCount(1).build()); + awaitSucceeded(probeSucceeded, 1L); + awaitSucceeded(converterSucceeded, 1L); + + changes.sendRecord(TableChange.builder().commitCount(1).build()); + awaitSucceeded(probeSucceeded, 2L); + awaitSucceeded(converterSucceeded, 2L); + } finally { + closeJobClient(jobClient); + } + } + + private static void awaitSucceeded(List counterKey, long expected) { + Awaitility.await() + .atMost(Duration.ofMinutes(3)) + .pollInterval(Duration.ofMillis(200)) + .untilAsserted( + () -> + assertThat(MetricsReporterFactoryForTests.counter(counterKey)).isEqualTo(expected)); + } + private DataFile writeDataFile(Table table, Integer id, String data) throws IOException { return new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir) .writeFile(Lists.newArrayList(SimpleDataUtil.createRecord(id, data))); @@ -193,4 +277,36 @@ private static long eqDeleteCountOnMain(Table table) throws IOException { return count; } + + private static class CompletingTaskBuilder extends MaintenanceTaskBuilder { + + @Override + String maintenanceTaskName() { + return PROBE_TASK_NAME; + } + + @Override + DataStream append(DataStream trigger) { + return setSlotSharingGroup( + trigger + .map(new CompletingTask()) + .name(maintenanceTaskName()) + .uid(uidSuffix() + "-shared-lock-probe") + .forceNonParallel()); + } + } + + private static class CompletingTask + implements MapFunction, ResultTypeQueryable, Serializable { + + @Override + public TaskResult map(Trigger trigger) { + return new TaskResult(trigger.taskId(), trigger.timestamp(), true, Collections.emptyList()); + } + + @Override + public TypeInformation getProducedType() { + return TypeInformation.of(TaskResult.class); + } + } } diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java index 53e98a06b478..35aa9908bce7 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java @@ -133,6 +133,67 @@ void holdsBackWatermarkUntilCommit() throws Exception { } } + @Test + void holdsBackPhaseWatermarkBeforePlanArrives() throws Exception { + Table table = createTable(3, FileFormat.PARQUET); + insert(table, 1, "a"); + + DataFile stagingDataFile = getFirstDataFile(table); + long snapshotIdBefore = table.currentSnapshot().snapshotId(); + + try (TwoInputStreamOperatorTestHarness harness = + createHarness()) { + harness.open(); + + long triggerTs = System.currentTimeMillis(); + long doneTs = triggerTs + 10; + + // The planner marks cycle start, then emits phase watermarks before the plan. The pre-plan + // phase watermark must not escape: forwarding it would let the LockRemover release the + // maintenance lock before this cycle commits. + harness.processElement2(new StreamRecord<>(EqualityConvertPlan.cycleStart(triggerTs))); + harness.processBothWatermarks(new Watermark(triggerTs + 1)); + assertThat(watermarks(harness)).isEmpty(); + assertThat(harness.extractOutputValues()).isEmpty(); + + // The plan arrives; the done-timestamp watermark then commits the cycle and forwards. + EqualityConvertPlan planResult = + new EqualityConvertPlan( + Lists.newArrayList(stagingDataFile), + Lists.newArrayList(), + Lists.newArrayList(), + 42L, + snapshotIdBefore, + triggerTs, + doneTs); + harness.processElement2(new StreamRecord<>(planResult)); + harness.processBothWatermarks(new Watermark(doneTs)); + + assertThat(harness.extractOutputValues()).hasSize(1); + assertThat(watermarks(harness)).containsExactly(new Watermark(doneTs)); + table.refresh(); + assertThat(table.currentSnapshot().snapshotId()).isNotEqualTo(snapshotIdBefore); + } + } + + @Test + void forwardsWatermarkWithoutActiveCycle() throws Exception { + Table table = createTable(3, FileFormat.PARQUET); + insert(table, 1, "a"); + + try (TwoInputStreamOperatorTestHarness harness = + createHarness()) { + harness.open(); + + // No plan was received, so no conversion cycle is active. A watermark here originates + // from another task's trigger sharing the maintenance lock; dropping it would stall + // the unioned watermark and the lock would never be released. + long markTs = System.currentTimeMillis(); + harness.processBothWatermarks(new Watermark(markTs)); + assertThat(watermarks(harness)).containsExactly(new Watermark(markTs)); + } + } + @Test void skipsCommitForEmptyCycle() throws Exception { Table table = createTable(3, FileFormat.PARQUET); diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java index 56c6a2f98b33..93614fe8ab11 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java @@ -137,6 +137,37 @@ void noOutputWithoutPlanResult() throws Exception { } } + @Test + void ignoresCycleStartMarker() throws Exception { + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + String dataFilePath = getFirstDataFilePath(table); + + try (TwoInputStreamOperatorTestHarness harness = + createHarness()) { + harness.open(); + + long time = System.currentTimeMillis(); + harness.processElement1( + new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time)); + + // The cycle-start marker is not a plan: it must not trigger a write. + harness.processElement2(new StreamRecord<>(EqualityConvertPlan.cycleStart(time), time)); + harness.processBothWatermarks(new Watermark(time)); + assertThat(harness.extractOutputValues()).isEmpty(); + + // The real plan that follows still drives the write. The watermark must advance past the + // previous one for the operator to process it. + harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time)); + harness.processBothWatermarks(new Watermark(time + 1)); + + List output = harness.extractOutputValues(); + assertThat(output).hasSize(1); + assertThat(output.get(0).hasError()).isFalse(); + } + } + @Test void writesDVFilesForMultipleDataFiles() throws Exception { Table table = createTableWithDelete(3); diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java index ed243695bced..1482ce872661 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPlanner.java @@ -84,7 +84,7 @@ void bootstrapsIndexFromBuilderEqualityFieldIds() throws Exception { assertThat(countDataFileTasks(commands)).isEqualTo(2); assertThat(countEqDeleteTasks(commands)).isZero(); - assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(1); + assertThat(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(2); } } @@ -202,8 +202,9 @@ void emitsReadCommandsForEqualityDeletes() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); - assertThat(metadata).hasSize(1); - assertThat(metadata.get(0).getValue().dataFiles()).isEmpty(); + assertThat(metadata).hasSize(2); + assertThat(metadata.get(0).getValue().doneTimestamp()).isEqualTo(Long.MAX_VALUE); + assertThat(metadata.get(1).getValue().dataFiles()).isEmpty(); } } @@ -280,8 +281,9 @@ void includesNewDataFilesFromStaging() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); - assertThat(metadata).hasSize(1); - assertThat(metadata.get(0).getValue().dataFiles()).hasSize(1); + assertThat(metadata).hasSize(2); + assertThat(metadata.get(0).getValue().doneTimestamp()).isEqualTo(Long.MAX_VALUE); + assertThat(metadata.get(1).getValue().dataFiles()).hasSize(1); } } @@ -371,7 +373,7 @@ void noOutputWhenStagingBranchEmpty() throws Exception { // 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(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)).hasSize(2); assertThat(planner(harness).skippedNoOpCycles()).isEqualTo(1); assertThat(planner(harness).reindexCount()).isZero(); } @@ -405,8 +407,9 @@ void propagatesStagingOnlyPositionalDeletes() throws Exception { List> metadata = Lists.newArrayList(harness.getSideOutput(EqualityConvertPlanner.METADATA_STREAM)); - assertThat(metadata).hasSize(1); - EqualityConvertPlan result = metadata.get(0).getValue(); + assertThat(metadata).hasSize(2); + assertThat(metadata.get(0).getValue().doneTimestamp()).isEqualTo(Long.MAX_VALUE); + EqualityConvertPlan result = metadata.get(1).getValue(); assertThat(result.dataFiles()).isEmpty(); assertThat(result.stagingDVFiles()).hasSize(1); assertThat(result.stagingDVFiles().get(0).location()).isEqualTo(stagingDV.location());