[LuciferYang]

Purpose

close #10831

LocalMergeOperator holds back the watermark and emits it in flushBuffer() after the buffered rows. flushBuffer() returned early when the buffer was empty, so the watermark was not emitted. When records are filtered out before the sink while the source watermark advances, the buffer is empty at every checkpoint, and the snapshot watermark stays where it was when the last record arrived.

flushBuffer() now only skips the row output when the buffer is empty and still emits the watermark.

Tests

• LocalMergeWatermarkITCase#testWatermarkAdvancesWhileRecordsAreFilteredOut: a streaming job writes to a primary key table with local-merge-buffer-size through a filter that only lets the first record through, while the source watermark keeps advancing. It waits for the snapshot watermark to advance past its first value. On master it stays at 0 and the test times out; without local-merge-buffer-size it passes on master too.
• LocalMergeOperatorTest#testEmptyFlushStillEmitsWatermark: a watermark followed by prepareSnapshotPreBarrier with no records emits the watermark and no rows.

API and Format

no

Documentation

no