You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2026/07/03 10:13:31 UTC
(iceberg) branch main updated: Flink: Backport: Hold back equality delete converter watermark until completion (#17038) (#17067)
This is an automated email from the ASF dual-hosted git repository.
nastra 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 11706a286b Flink: Backport: Hold back equality delete converter watermark until completion (#17038) (#17067)
11706a286b is described below
commit 11706a286bc94f852aa6046d2d1ec6c3b85039a5
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Fri Jul 3 12:13:24 2026 +0200
Flink: Backport: Hold back equality delete converter watermark until completion (#17038) (#17067)
---
.../operator/EqualityConvertCommitter.java | 44 ++++++++++++---------
.../operator/TestEqualityConvertCommitter.java | 45 ++++++++++++++++++++++
.../operator/EqualityConvertCommitter.java | 44 ++++++++++++---------
.../operator/TestEqualityConvertCommitter.java | 45 ++++++++++++++++++++++
4 files changed, 142 insertions(+), 36 deletions(-)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
index b3e8fb4d89..95c9ccd5de 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
@@ -50,6 +50,12 @@ import org.slf4j.LoggerFactory;
*
* <p>The commit is gated on the plan's done-timestamp watermark.
*
+ * <p>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.
+ *
* <p>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
* the Aggregator.
@@ -139,27 +145,29 @@ public class EqualityConvertCommitter extends AbstractStreamOperator<Trigger>
@Override
public void processWatermark(Watermark mark) throws Exception {
- if (planResult != null && mark.getTimestamp() >= planResult.doneTimestamp()) {
- try {
- commitIfNeeded();
- } catch (Exception e) {
- LOG.error(
- "Failed to commit equality convert result for table {} task {}",
- tableName,
- taskName,
- e);
- output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
- errorCounter.inc();
- }
-
- // Emit Trigger for the Aggregator (even on error or no-op).
- output.collect(new StreamRecord<>(Trigger.create(planResult.triggerTimestamp(), 0)));
+ 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.
+ return;
+ }
- bufferedResults.clear();
- planResult = null;
+ try {
+ commitIfNeeded();
+ } catch (Exception e) {
+ LOG.error(
+ "Failed to commit equality convert result for table {} task {}", tableName, taskName, e);
+ output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+ errorCounter.inc();
}
- // Always forward watermarks to prevent stalling downstream.
+ // Emit Trigger for the Aggregator (even on error or no-op).
+ output.collect(new StreamRecord<>(Trigger.create(planResult.triggerTimestamp(), 0)));
+
+ bufferedResults.clear();
+ planResult = null;
+
super.processWatermark(mark);
}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
index f74570c878..54f0b1758d 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
@@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.io.IOException;
import java.util.List;
+import java.util.stream.Collectors;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
@@ -83,6 +84,43 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
}
}
+ @Test
+ void holdsBackWatermarkUntilCommit() 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 doneTs = System.currentTimeMillis();
+ EqualityConvertPlan planResult =
+ new EqualityConvertPlan(
+ Lists.newArrayList(stagingDataFile),
+ Lists.newArrayList(),
+ 42L,
+ snapshotIdBefore,
+ doneTs - 2,
+ doneTs);
+
+ harness.processElement2(new StreamRecord<>(planResult, doneTs - 2));
+
+ // A phase watermark before the plan's done timestamp must not be forwarded: it would let the
+ // LockRemover release the maintenance lock before this cycle commits.
+ harness.processBothWatermarks(new Watermark(doneTs - 1));
+ assertThat(harness.extractOutputValues()).isEmpty();
+ assertThat(watermarks(harness)).isEmpty();
+
+ // The done-timestamp watermark commits the cycle; the watermark is forwarded only now.
+ harness.processBothWatermarks(new Watermark(doneTs));
+ assertThat(harness.extractOutputValues()).hasSize(1);
+ assertThat(watermarks(harness)).containsExactly(new Watermark(doneTs));
+ }
+ }
+
@Test
void skipsCommitForEmptyCycle() throws Exception {
Table table = createTable(3, FileFormat.PARQUET);
@@ -476,6 +514,13 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
SnapshotRef.MAIN_BRANCH));
}
+ private static List<Watermark> watermarks(TwoInputStreamOperatorTestHarness<?, ?, ?> harness) {
+ return harness.getOutput().stream()
+ .filter(Watermark.class::isInstance)
+ .map(Watermark.class::cast)
+ .collect(Collectors.toList());
+ }
+
private static List<DeleteFile> deletesForDataFile(Table table, String dataFilePath) {
List<DeleteFile> deletes = Lists.newArrayList();
for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) {
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
index b3e8fb4d89..95c9ccd5de 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java
@@ -50,6 +50,12 @@ import org.slf4j.LoggerFactory;
*
* <p>The commit is gated on the plan's done-timestamp watermark.
*
+ * <p>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.
+ *
* <p>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
* the Aggregator.
@@ -139,27 +145,29 @@ public class EqualityConvertCommitter extends AbstractStreamOperator<Trigger>
@Override
public void processWatermark(Watermark mark) throws Exception {
- if (planResult != null && mark.getTimestamp() >= planResult.doneTimestamp()) {
- try {
- commitIfNeeded();
- } catch (Exception e) {
- LOG.error(
- "Failed to commit equality convert result for table {} task {}",
- tableName,
- taskName,
- e);
- output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
- errorCounter.inc();
- }
-
- // Emit Trigger for the Aggregator (even on error or no-op).
- output.collect(new StreamRecord<>(Trigger.create(planResult.triggerTimestamp(), 0)));
+ 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.
+ return;
+ }
- bufferedResults.clear();
- planResult = null;
+ try {
+ commitIfNeeded();
+ } catch (Exception e) {
+ LOG.error(
+ "Failed to commit equality convert result for table {} task {}", tableName, taskName, e);
+ output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+ errorCounter.inc();
}
- // Always forward watermarks to prevent stalling downstream.
+ // Emit Trigger for the Aggregator (even on error or no-op).
+ output.collect(new StreamRecord<>(Trigger.create(planResult.triggerTimestamp(), 0)));
+
+ bufferedResults.clear();
+ planResult = null;
+
super.processWatermark(mark);
}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
index f74570c878..54f0b1758d 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java
@@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.io.IOException;
import java.util.List;
+import java.util.stream.Collectors;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
@@ -83,6 +84,43 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
}
}
+ @Test
+ void holdsBackWatermarkUntilCommit() 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 doneTs = System.currentTimeMillis();
+ EqualityConvertPlan planResult =
+ new EqualityConvertPlan(
+ Lists.newArrayList(stagingDataFile),
+ Lists.newArrayList(),
+ 42L,
+ snapshotIdBefore,
+ doneTs - 2,
+ doneTs);
+
+ harness.processElement2(new StreamRecord<>(planResult, doneTs - 2));
+
+ // A phase watermark before the plan's done timestamp must not be forwarded: it would let the
+ // LockRemover release the maintenance lock before this cycle commits.
+ harness.processBothWatermarks(new Watermark(doneTs - 1));
+ assertThat(harness.extractOutputValues()).isEmpty();
+ assertThat(watermarks(harness)).isEmpty();
+
+ // The done-timestamp watermark commits the cycle; the watermark is forwarded only now.
+ harness.processBothWatermarks(new Watermark(doneTs));
+ assertThat(harness.extractOutputValues()).hasSize(1);
+ assertThat(watermarks(harness)).containsExactly(new Watermark(doneTs));
+ }
+ }
+
@Test
void skipsCommitForEmptyCycle() throws Exception {
Table table = createTable(3, FileFormat.PARQUET);
@@ -476,6 +514,13 @@ class TestEqualityConvertCommitter extends OperatorTestBase {
SnapshotRef.MAIN_BRANCH));
}
+ private static List<Watermark> watermarks(TwoInputStreamOperatorTestHarness<?, ?, ?> harness) {
+ return harness.getOutput().stream()
+ .filter(Watermark.class::isInstance)
+ .map(Watermark.class::cast)
+ .collect(Collectors.toList());
+ }
+
private static List<DeleteFile> deletesForDataFile(Table table, String dataFilePath) {
List<DeleteFile> deletes = Lists.newArrayList();
for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) {