[FLINK-38928] Implement an operator for handling DO ERROR/NOTHING (#2… by dawidwys · Pull Request #27602 · apache/flink

dawidwys · 2026-02-13T14:32:18Z

second attempt at #27502

flinkbot · 2026-02-13T14:40:15Z

CI report:

2dd2f0d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

dawidwys · 2026-02-16T10:34:36Z

@twalthr @pnowojski Could you take a look at this PR? I tried addressing @pnowojski concerns.

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

pnowojski · 2026-02-19T08:37:00Z

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

        buffer.put(timestamp, records);

-        timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, timestamp);
+        timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, timestamp + 1);


nit: could you explain in the comment how this +1 here and -1 in timer firing is supposed to work? I get it, but I think it would be good to explain it for the future.

I remember some code paths when we emit Long.MAX_VALUE on end_of_input; can we add a check that this increment doesn't result in overflow?

I don't see a point in that. When you emit MAX_VALUE there won't be any new records anyhow afterwards. So there will be no records with timestamp MAX_VALUE and we don't care if a timer fires or not.

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

pnowojski

LGTM % I would like @rkhachatryan to also take a look here before merging.

I also presume we don't need a feature toggle for this one, as users would have to manually change the conflict resolution strategy in their schemas/tables for this change to take the effect. Right?

dawidwys · 2026-02-19T13:26:41Z

I also presume we don't need a feature toggle for this one, as users would have to manually change the conflict resolution strategy in their schemas/tables for this change to take the effect. Right

Correct

rkhachatryan

Thanks for the PR!
I've left some comments - PTAL (sorry if some questions were already asked on the PR).

Meta remark:
During our previous discussions around SinkUpsertMaterializer, my understanding was that we'll implement compaction on watermark on top of the existing implementation (be it sum v1 or v2).

I'm fine with adding a 3rd one, but I must say it complicates not only the code, but also the operations for the user.

On testing:
if the bugs I described are real, we should probably plug the existing testing code for SUM V1/2 - it was extended significantly for FLIP-544

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java

...untime/src/main/java/org/apache/flink/table/runtime/operators/sink/SortedLongSerializer.java

...me/src/test/java/org/apache/flink/table/runtime/operators/sink/SortedLongSerializerTest.java

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

rkhachatryan · 2026-02-19T23:34:44Z

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

+        if (previousValue != null) {
+            records.add(previousValue);
+        }
+        Iterator<Map.Entry<Long, List<RowData>>> iterator = buffer.entries().iterator();


For every timer timestamp X, we should know exactly the time X-1 when the record was added, right?

Why do we need to iterate over the whole state here?
Can't we use point lookup (which is MUCH less expensive than iteration)

Theoretically you're correct. Still I'd say it's safer to iterate over the records. In a common scenario it should not matter much as there should not be many parallel watermarks flowing through a channel.

In a common scenario it should not matter much as there should not be many parallel watermarks flowing through a channel.

That's a happy path, but if one channel is idling for some reason, we might have a SUMv1-like performance problem.

rkhachatryan · 2026-02-19T23:35:40Z

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

+                    switch (pendingRecord.getRowKind()) {
+                        case INSERT:
+                        case UPDATE_AFTER:
+                            addRow(records, pendingRecord);


This call is O(N), so the innermost loop is O(N^N).
Why don't we use a hashmap instead of linear findFirst?

This bit is copied over from SUM v1.

Yes, but in SUMv1 this is scattered over time; here, it happens at once for all the buffered records.

It is also scattered over time in here. We eagerly try to apply it when processing a single record.

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

.../src/main/java/org/apache/flink/table/runtime/operators/sink/WatermarkTimestampAssigner.java

dawidwys · 2026-02-20T08:51:00Z

Meta remark:
During our previous discussions around SinkUpsertMaterializer, my understanding was that we'll implement compaction on watermark on top of the existing implementation (be it sum v1 or v2).
I'm fine with adding a 3rd one, but I must say it complicates not only the code, but also the operations for the user.

First time that I hear that. I can't find any such comments on the FLIP discussion. Moreover I can't think how that could be possible since we're changing the semantics slightly. Lastly adding watermark compaction to the existing SUM would not help with the state size. It still needs to keep the entire history.

rkhachatryan · 2026-02-20T22:15:19Z

The latest test failure seems to be caused by FLINK-39103 - which is now fixed in master.

rkhachatryan · 2026-02-24T14:47:49Z

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

+        if (ttlConfig.isEnabled()) {
+            bufferDescriptor.enableTimeToLive(
+                    StateTtlConfig.newBuilder(ttlConfig)
+                            .setTimeToLive(ttlConfig.getTimeToLive().plus(Duration.ofSeconds(1)))
+                            .build());
+            currentValueDescriptor.enableTimeToLive(ttlConfig);
+        }


Should consolidatedCheckpointId also be covered?
Although the value is small, with unbounded key space growth it can be a problem as well (and I guess unbounded key space is a common reason to enable TTL).

NIT: enforce stateVisibility = NeverReturnExpired or document potential issues of ReturnExpiredIfNotCleanedUp?
NIT: document the motivation for +1 second
NIT: increase this to 5-10 seconds

Also would be good have test coverage for TTL; but I think that's NIT

rkhachatryan · 2026-02-24T15:37:53Z

...a/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializerTest.java

+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link WatermarkCompactingSinkMaterializer}. */
+class WatermarkCompactingSinkMaterializerTest {


It would be great to parameterize this test with the type of (heap / rocksdb).
Or at least make it use rocksdb as the most widely used state backend.

rkhachatryan · 2026-02-24T15:38:46Z

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

+        if (ttlConfig.isEnabled()) {
+            bufferDescriptor.enableTimeToLive(
+                    StateTtlConfig.newBuilder(ttlConfig)
+                            .setTimeToLive(ttlConfig.getTimeToLive().plus(Duration.ofSeconds(1)))
+                            .build());
+            currentValueDescriptor.enableTimeToLive(ttlConfig);
+        }


Also would be good have test coverage for TTL; but I think that's NIT

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java

rkhachatryan · 2026-02-24T16:29:18Z

...a/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializerTest.java

+    void testDeleteAfterInsert(ConflictBehavior behavior) throws Exception {
+        try (KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData> harness =
+                createHarness(behavior)) {
+            harness.open();
+
+            // Insert and compact
+            harness.processElement(insertRecord(1L, 1, "a1"));


Many tests in this class cover the case before the 1st watermark.
For example, here both currentWatermark is null and record timestamp is null.
There are tests that cover after-watermark processing as well, but the coverage seems to be lower.

Maybe we can parameterize the class with something like nullable initialWatermark.
If it's set, then send it the harness before the test; and set on created record unless a specific timestamp is provided.

WDYT?

rkhachatryan · 2026-02-24T16:30:51Z

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java

+        if (isErrorOrNothingConflictStrategy()) {
+            // Use input parallelism to preserve watermark semantics
+            transformForMaterializer =
+                    ExecNodeUtil.createOneInputTransformation(
+                            inputTransform,
+                            createTransformationMeta(
+                                    WATERMARK_TIMESTAMP_ASSIGNER_TRANSFORMATION,
+                                    "WatermarkTimestampAssigner",
+                                    "WatermarkTimestampAssigner",
+                                    config),
+                            new WatermarkTimestampAssigner(),
+                            inputTransform.getOutputType(),
+                            inputTransform.getParallelism(),
+                            false);


Is adding WatermarkTimestampAssigner covered by any tests?

I think it would be also good to have an ITCase with more than one input channels.

…ache#27502)

rkhachatryan

LGTM

I still have some concerns (around validation, performance, test coverage) but I think we can address those in the follow-up PRs.

pnowojski reviewed Feb 16, 2026

View reviewed changes

.../java/org/apache/flink/table/runtime/operators/sink/WatermarkCompactingSinkMaterializer.java Outdated Show resolved Hide resolved

dawidwys force-pushed the flink38928-2 branch from 4645e69 to e3acc7e Compare February 17, 2026 12:15

pnowojski reviewed Feb 18, 2026

View reviewed changes

pnowojski reviewed Feb 19, 2026

View reviewed changes

dawidwys force-pushed the flink38928-2 branch from d12d81c to b784674 Compare February 19, 2026 12:55

pnowojski reviewed Feb 19, 2026

View reviewed changes

dawidwys force-pushed the flink38928-2 branch from b784674 to 51392e5 Compare February 19, 2026 15:03

rkhachatryan reviewed Feb 19, 2026

View reviewed changes

dawidwys force-pushed the flink38928-2 branch 5 times, most recently from 9a22e3c to 61146ea Compare February 24, 2026 14:31

rkhachatryan reviewed Feb 24, 2026

View reviewed changes

dawidwys force-pushed the flink38928-2 branch from a7b5181 to 0c79f05 Compare February 24, 2026 15:08

rkhachatryan reviewed Feb 24, 2026

View reviewed changes

[FLINK-38928] Implement an operator for handling DO ERROR/NOTHING (ap…

2dd2f0d

…ache#27502)

dawidwys force-pushed the flink38928-2 branch from c3dbb16 to 2dd2f0d Compare February 25, 2026 14:27

rkhachatryan approved these changes Feb 25, 2026

View reviewed changes

dawidwys merged commit 3ecbbd1 into apache:master Feb 26, 2026

Conversation

dawidwys commented Feb 13, 2026

Uh oh!

flinkbot commented Feb 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

dawidwys commented Feb 16, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

pnowojski left a comment

Choose a reason for hiding this comment

Uh oh!

dawidwys commented Feb 19, 2026

Uh oh!

rkhachatryan left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

dawidwys commented Feb 20, 2026

Uh oh!

rkhachatryan commented Feb 20, 2026

Uh oh!

rkhachatryan Feb 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rkhachatryan Feb 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

flinkbot commented Feb 13, 2026 •

edited

Loading

rkhachatryan left a comment •

edited

Loading

rkhachatryan Feb 24, 2026 •

edited

Loading

rkhachatryan Feb 24, 2026 •

edited

Loading