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..8186269e85ce 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 @@ -50,11 +50,9 @@ * *

The commit is gated on the plan's done-timestamp watermark. * - *

Watermarks are forwarded only after the cycle commits, never mid-cycle. The {@link - * LockRemover} releases the maintenance lock once a watermark past the trigger's start epoch - * reaches it. The planner emits phase watermarks in the middle of a cycle; forwarding those would - * release the lock before this commit, letting the TriggerManager start a concurrent cycle that - * re-processes the same uncommitted staging snapshot. + *

Watermarks are forwarded while no cycle is active. After the planner marks cycle start, + * watermarks are held back until the cycle commits; forwarding a pre-plan phase watermark would + * let the {@link LockRemover} release the maintenance lock before this commit. * *

Emits a {@link Trigger} after each cycle (commit, no-op, or error) so the downstream {@link * TaskResultAggregator} can track task completion. This is the sole source of Trigger records for @@ -93,6 +91,7 @@ public class EqualityConvertCommitter extends AbstractStreamOperator private transient Table table; private transient List bufferedResults; private transient EqualityConvertPlan planResult; + private transient boolean cycleInFlight; private transient Counter errorCounter; private transient Counter addedDataFileNumCounter; @@ -143,16 +142,38 @@ public void processElement1(StreamRecord record) { @Override public void processElement2(StreamRecord record) { + if (record.getValue().cycleStart()) { + // The planner started a conversion cycle. Hold back watermarks until the plan arrives; the + // planner emits phase watermarks before the plan, and forwarding those would release the + // maintenance lock before the commit. + cycleInFlight = true; + return; + } + planResult = record.getValue(); + cycleInFlight = true; } @Override public void processWatermark(Watermark mark) throws Exception { - if (planResult == null || 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 - // snapshot. + if (!cycleInFlight) { + // No conversion cycle is active: forward the watermark. It may originate from another + // task's trigger sharing the maintenance lock, and holding it back would stall lock release + // for that task. + super.processWatermark(mark); + return; + } + + if (planResult == null) { + // The cycle started but the plan has not arrived yet. The planner emits phase watermarks + // before the plan; forwarding them would release the maintenance lock before the commit, + // letting the next trigger run a concurrent cycle on the same staging snapshot. + return; + } + + if (mark.getTimestamp() < planResult.doneTimestamp()) { + // Hold back watermarks until the cycle commits so the LockRemover keeps the maintenance + // lock for the whole cycle. return; } @@ -170,6 +191,7 @@ public void processWatermark(Watermark mark) throws Exception { bufferedResults.clear(); planResult = null; + cycleInFlight = false; super.processWatermark(mark); } diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java index 486b933ee6d7..f506fba8cf2f 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertDVWriter.java @@ -133,6 +133,10 @@ public void processElement1(StreamRecord record) { @Override public void processElement2(StreamRecord record) { + if (record.getValue().cycleStart()) { + return; + } + planResult = record.getValue(); } 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..33561ce54300 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,10 @@ * #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to construct one and {@link * #noOp()} to check. * + *

A cycle-start marker is encoded by {@link #stagingSnapshotId()} == {@link + * #CYCLE_START_STAGING_SNAPSHOT_ID}. Use {@link #cycleStart(long)} to construct one and {@link + * #cycleStart()} to check. It is never a plan: consumers must not treat it as 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 @@ -60,12 +64,18 @@ public record EqualityConvertPlan( implements Serializable { public static final long NO_OP_STAGING_SNAPSHOT_ID = -1L; + public static final long CYCLE_START_STAGING_SNAPSHOT_ID = -2L; /** Returns {@code true} if this cycle has nothing to commit. */ public boolean noOp() { return stagingSnapshotId == NO_OP_STAGING_SNAPSHOT_ID; } + /** Returns {@code true} if this is a cycle-start marker rather than a plan. */ + public boolean cycleStart() { + return stagingSnapshotId == CYCLE_START_STAGING_SNAPSHOT_ID; + } + /** No-op cycle: empty file lists, sentinel staging snapshot id. */ public static EqualityConvertPlan noOp( Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) { @@ -78,4 +88,19 @@ public static EqualityConvertPlan noOp( triggerTimestamp, doneTimestamp); } + + /** + * Cycle-start marker. The done timestamp is maximal so no watermark can act on it if a consumer + * fails to filter it out. + */ + public static EqualityConvertPlan cycleStart(long triggerTimestamp) { + return new EqualityConvertPlan( + ImmutableList.of(), + ImmutableList.of(), + ImmutableList.of(), + CYCLE_START_STAGING_SNAPSHOT_ID, + 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..3af1b705fd31 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,9 @@ * {@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. A {@link EqualityConvertPlan#cycleStart} + * marker is emitted first, before any phase watermark, so the committer can hold back watermarks + * until the plan arrives. * *

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 +266,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/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..3f49ad3fa9b6 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,36 @@ 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. + harness.processElement2(new StreamRecord<>(emptyEqualityConvertPlan(), time)); + harness.processBothWatermarks(new Watermark(time)); + + 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..08e536388adf 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().cycleStart()).isTrue(); + 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().cycleStart()).isTrue(); + 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().cycleStart()).isTrue(); + EqualityConvertPlan result = metadata.get(1).getValue(); assertThat(result.dataFiles()).isEmpty(); assertThat(result.stagingDVFiles()).hasSize(1); assertThat(result.stagingDVFiles().get(0).location()).isEqualTo(stagingDV.location());