This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 3c315c8df4 Flink: Forward watermarks in EqualityConvertCommitter with
no active cycle (#18439)
3c315c8df4 is described below
commit 3c315c8df478713f12346df37b590e1e33a0f5e2
Author: Rakesh Pai <[email protected]>
AuthorDate: Tue Oct 20 05:41:45 2026 -0700
Flink: Forward watermarks in EqualityConvertCommitter with no active cycle
(#18439)
Co-authored-by: Maximilian Michels <[email protected]>
Fixes #18435.
---
.../operator/EqualityConvertCommitter.java | 11 +-
.../maintenance/operator/EqualityConvertPlan.java | 11 ++
.../operator/EqualityConvertPlanner.java | 11 +-
.../api/TestConvertEqualityDeletesE2E.java | 116 +++++++++++++++++++++
.../operator/TestEqualityConvertCommitter.java | 61 +++++++++++
.../operator/TestEqualityConvertDVWriter.java | 31 ++++++
.../operator/TestEqualityConvertPlanner.java | 19 ++--
7 files changed, 250 insertions(+), 10 deletions(-)
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 ada16ec093..34d996c3d9 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 class EqualityConvertCommitter extends
AbstractStreamOperator<Trigger>
@Override
public void processElement2(StreamRecord<EqualityConvertPlan> 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 191bf83220..4aa326d19a 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 @@ import
org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
* #NO_OP_STAGING_SNAPSHOT_ID}. Use {@link #noOp(Long, long, long)} to
construct one and {@link
* #noOp()} to check.
*
+ * <p>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 record EqualityConvertPlan(
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 82791bae4b..96e5861f1c 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 @@ import org.slf4j.LoggerFactory;
* {@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.
+ * #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.
*
* <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
@@ -264,6 +267,12 @@ public class EqualityConvertPlanner extends
AbstractStreamOperator<ReadCommand>
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 f3bdaa8fe1..e39d05b66f 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.Table;
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 @@ import org.junit.jupiter.params.provider.ValueSource;
*/
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 @@ class TestConvertEqualityDeletesE2E extends
OperatorTestBase {
}
}
+ /**
+ * 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<TableChange> 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<String> probeSucceeded =
+ ImmutableList.of(
+ LOCK_REMOVER_OPERATOR_NAME,
+ tableName,
+ PROBE_TASK_NAME + " [0]",
+ "0",
+ SUCCEEDED_TASK_COUNTER);
+ List<String> 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<String> 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 @@ class TestConvertEqualityDeletesE2E extends
OperatorTestBase {
return count;
}
+
+ private static class CompletingTaskBuilder extends
MaintenanceTaskBuilder<CompletingTaskBuilder> {
+
+ @Override
+ String maintenanceTaskName() {
+ return PROBE_TASK_NAME;
+ }
+
+ @Override
+ DataStream<TaskResult> append(DataStream<Trigger> trigger) {
+ return setSlotSharingGroup(
+ trigger
+ .map(new CompletingTask())
+ .name(maintenanceTaskName())
+ .uid(uidSuffix() + "-shared-lock-probe")
+ .forceNonParallel());
+ }
+ }
+
+ private static class CompletingTask
+ implements MapFunction<Trigger, TaskResult>,
ResultTypeQueryable<TaskResult>, Serializable {
+
+ @Override
+ public TaskResult map(Trigger trigger) {
+ return new TaskResult(trigger.taskId(), trigger.timestamp(), true,
Collections.emptyList());
+ }
+
+ @Override
+ public TypeInformation<TaskResult> 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 53e98a06b4..35aa9908bc 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 @@ class TestEqualityConvertCommitter extends
OperatorTestBase {
}
}
+ @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<DVWriteResult, EqualityConvertPlan,
Trigger> 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<DVWriteResult, EqualityConvertPlan,
Trigger> 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 56c6a2f98b..93614fe8ab 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 @@ class TestEqualityConvertDVWriter extends OperatorTestBase
{
}
}
+ @Test
+ void ignoresCycleStartMarker() throws Exception {
+ Table table = createTableWithDelete(3);
+ insert(table, 1, "a");
+
+ String dataFilePath = getFirstDataFilePath(table);
+
+ try (TwoInputStreamOperatorTestHarness<DVPosition, EqualityConvertPlan,
DVWriteResult> 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<DVWriteResult> 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 ed243695bc..1482ce8726 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 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
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 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
List<StreamRecord<EqualityConvertPlan>> 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 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
List<StreamRecord<EqualityConvertPlan>> 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 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
// 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 @@ class TestEqualityConvertPlanner extends OperatorTestBase {
List<StreamRecord<EqualityConvertPlan>> 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());
