[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:

6903b2c Azure: FAILURE

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

…ache#27502)

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

rkhachatryan · 2026-02-19T22:45:30Z

...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);
+        }


Can we validate somehow that the watermarks are generated at all?

The plan was to add it in a subsequent PR.

rkhachatryan · 2026-02-19T22:46:49Z

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

+                    pkFieldNames);
+        }
+
+        // Use existing logic for DEDUPLICATE (legacy behavior)


Why don't we apply compaction on watermark to the default behavior?

Why would we?

rkhachatryan · 2026-02-19T22:47:19Z

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

+        if (isErrorOrNothingConflictStrategy()) {
+            RowType keyType = RowTypeUtils.projectRowType(physicalRowType, primaryKeys);
+
+            return WatermarkCompactingSinkMaterializer.create(


Is this mode compatible with ReducingUpsertSink?

Yes, why wouldn't it?

rkhachatryan · 2026-02-19T22:56:46Z

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

+
+    @Override
+    public void serialize(Long record, DataOutputView target) throws IOException {
+        target.writeLong(MathUtils.flipSignBit(record));


This places positive watermarks before negative ones, right? (because we flip + to -)

For example:

original: [9223372036854775807, -3, 0, -9223372036854775808, 3, 2, 1, -1, -2] output: 0 [-128, 0, 0, 0, 0, 0, 0, 0] 1 [-128, 0, 0, 0, 0, 0, 0, 1] 2 [-128, 0, 0, 0, 0, 0, 0, 2] 3 [-128, 0, 0, 0, 0, 0, 0, 3] 9223372036854775807 [-1, -1, -1, -1, -1, -1, -1, -1] -9223372036854775808 [0, 0, 0, 0, 0, 0, 0, 0] -3 [127, -1, -1, -1, -1, -1, -1, -3] -2 [127, -1, -1, -1, -1, -1, -1, -2] -1 [127, -1, -1, -1, -1, -1, -1, -1]

The result is sorted using Arrays.sort(outs, Arrays::compare); where outs is byte[][] and holds serialization results.

This sounds like a bug or a potential bug.

Are you suggesting timers don't work in Flink? This is copied from org.apache.flink.streaming.api.operators.TimerSerializer

Yeah, I think timers may not work correctly in Flink for negative time (which we probably didn't have so far).

rkhachatryan · 2026-02-19T22:57:52Z

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

+import java.util.Random;
+
+/** A test for the {@link SortedLongSerializer}. */
+class SortedLongSerializerTest extends SerializerTestBase<Long> {


Please add some sorting tests

It's copied from org.apache.flink.streaming.api.operators.TimerSerializer

rkhachatryan · 2026-02-19T23:33:22Z

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

+        } else if (result.isEmpty()) {
+            if (previousValue != null) {
+                emit(previousValue, DELETE);
+                currentValue.clear();


So the next time we get +I, we'll emit it again. This makes sense, but I think is inconsistent with NOTHING strategy description:

Keep the first record that arrives for a given primary key, discard subsequent conflicts.

No, it is not. If there was a deletion it is deleted. There is a point in time, where there was NO value. NOTHING strategy is applied when there are two ACTIVE values at the same time.

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.

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.

rkhachatryan · 2026-02-19T23:36:53Z

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

+                            break;
+                        case UPDATE_BEFORE:
+                        case DELETE:
+                            retractRow(records, pendingRecord);


if +I and -D arrive within the same watermark then we won't emit anything (which makes sense).
But strategy NOTHING says that we'll emit the first arrived record - should we clarify that in the java/docs?

I don't think that is true what you're saying.

NOTHING is applied when there is a conflict in active keys. If, as you said, a value is retracted, which means non-existing, no value is appropriate.

rkhachatryan · 2026-02-19T23:37:52Z

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

+
+    @Override
+    public void processElement(StreamRecord<RowData> element) throws Exception {
+        element.setTimestamp(currentWatermark);


Is this timestamp emitted (to kafka or other sinks)?
If so, should we let the users know that the timestamp is adjusted in these modes?

No, it's not.

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.

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

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

51392e5

…ache#27502)

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 added 2 commits February 20, 2026 11:27

register a timer for MIN_VALUE + 1 to protect against overflow

d9615a4

remove duplicated fields

6903b2c

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!

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!

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!

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!

flinkbot commented Feb 13, 2026 •

edited

Loading

rkhachatryan left a comment •

edited

Loading

dawidwys Feb 20, 2026 •

edited

Loading