lgbo-ustc opened a new issue, #12340:
URL: https://github.com/apache/gluten/issues/12340
### Description
# Improve Watermark Semantics in Gluten Flink
## Background
Gluten Flink currently supports the basic propagation of watermark
timestamps through the native execution path, but it does not yet fully match
Flink's runtime watermark semantics. The current implementation can emit native
`StatefulWatermark` events and translate them into Flink `Watermark` events,
but several parts of the event-time runtime contract are either incomplete or
missing.
This issue tracks the work required to gradually improve watermark support
in Gluten Flink while minimizing risk. The proposed approach is to start with
low-risk Java-side runtime fixes, then move to native operator semantics, and
finally address source-level coordination and watermark alignment.
## Goals
- Preserve existing data record routing semantics unless a task explicitly
changes them.
- Improve watermark observability and runtime behavior step by step.
- Align Gluten Flink watermark behavior with Flink's event-time semantics.
- Add unit and end-to-end tests for each milestone.
## Tasks
- [ ] Complete the basic watermark event path
- [ ] Update `GlutenOutputCollector.emitWatermark(...)` to update
`WatermarkGauge`.
- [ ] Update `GlutenOutputCollector.collect(...)` to update the gauge
before forwarding watermark elements.
- [ ] Add unit tests to verify watermark gauge advancement for
source/operator watermark output.
- [ ] Keep existing semantics unchanged; this task should only fix
observability and basic runtime behavior.
- [ ] Implement `WatermarkStatus` propagation
- [ ] Implement Java-side `emitWatermarkStatus(WatermarkStatus)` and
broadcast it to all downstream outputs.
- [ ] Add `processWatermarkStatus` support in `GlutenOneInputOperator` and
`GlutenTwoInputOperator`.
- [ ] Add `notifyWatermarkStatus(id, status[, inputIndex])` to velox4j JNI.
- [ ] Add `processWatermarkStatus` to native `StatefulTask` and
`StatefulOperator`.
- [ ] Extend `CombinedWatermarkStatus` to support active/idle state
updates.
- [ ] Ensure Flink `IDLE` and `ACTIVE` statuses can pass through the
Gluten operator chain.
- [ ] Implement idle source handling
- [ ] Use `idleTimeout_` in native `WatermarkAssigner`.
- [ ] Detect idle input through either a native timer or a Java mailbox
timer.
- [ ] Emit `WatermarkStatus.IDLE` when the source becomes idle.
- [ ] Emit `WatermarkStatus.ACTIVE` when new data arrives after an idle
period.
- [ ] Exclude idle inputs from min-watermark calculation in multi-input
operators.
- [ ] Emit idle status downstream when all inputs are idle.
- [ ] Ensure idle sources do not block global watermark progress.
- [ ] Implement periodic watermark emission
- [ ] Align with Flink's `AUTO_WATERMARK_INTERVAL` semantics.
- [ ] Add periodic emission because the current native path only emits
watermarks when data arrives and crosses a threshold.
- [ ] Evaluate implementation option A: register a Java processing-time
timer and periodically call native `onPeriodicEmit`.
- [ ] Evaluate implementation option B: use a native timer service to
trigger the watermark assigner periodically.
- [ ] Track `maxTimestampSeen` and `lastEmittedWatermark` in
`WatermarkAssigner`.
- [ ] Emit the current advanceable watermark on each interval to avoid
stalls with sparse input.
- [ ] Match Flink `WatermarkGenerator.onPeriodicEmit` behavior.
- [ ] Fix null semantics for watermark expressions
- [ ] Check nulls in the watermark expression output `timestampVector`.
- [ ] Skip null watermark results instead of reading raw values or failing
unexpectedly.
- [ ] Clarify whether the rowtime field itself may be null; do not
conflate rowtime nullability with watermark expression nullability.
- [ ] Add SQL tests such as `WATERMARK FOR ts AS CASE WHEN ... THEN ts -
INTERVAL ... ELSE NULL END`.
- [ ] Align expression-based watermark behavior with Flink planner/runtime
semantics.
- [ ] Complete end-of-input and final watermark handling
- [ ] Implement bounded-input finish interfaces in Java Gluten operators.
- [ ] Add native `finishInput(inputIndex)` and `finishInput()` APIs.
- [ ] When all inputs finish, advance the final watermark.
- [ ] Flush event-time windows, joins, and timers.
- [ ] Emit `Watermark.MAX_WATERMARK`.
- [ ] Drain pending output before closing operators.
- [ ] Fix the commented-out max-watermark flush in
`WindowAggregator::close()`.
- [ ] Unify end-of-input behavior across `LocalWindowAggregator`,
`WindowAggregator`, and `WindowJoin`.
- [ ] Ensure bounded streams and batch-like streams produce complete final
window results.
- [ ] Complete source-level watermark support
- [ ] Define the intended source support scope.
- [ ] Short term: support source-level watermarks in legacy
`GlutenSourceFunction`.
- [ ] Medium term: maintain per-split watermarks inside native connector
splits.
- [ ] Long term: restore or emulate Flink `SourceCoordinator` semantics.
- [ ] For Kafka source, add partition/split-level watermark support.
- [ ] For Kafka source, add split idle support.
- [ ] For Kafka source, add source idle support.
- [ ] Ensure multi-partition sources do not block watermark progress when
one partition is idle.
- [ ] Implement watermark alignment
- [ ] Treat this as a separate large milestone and implement it after the
previous tasks.
- [ ] Decide the architecture first.
- [ ] Option A: preserve Flink `SourceCoordinator` and only offload
reader-side data reading to native.
- [ ] Option B: implement alignment in a Gluten-owned source coordinator.
- [ ] Support source reporting of current watermark.
- [ ] Support coordinator-side group minimum watermark calculation.
- [ ] Support pausing sources/splits that are too far ahead.
- [ ] Support resuming paused sources/splits.
- [ ] Support checkpoint and restore of alignment state.
- [ ] Prefer option A if possible, because the current Gluten source path
has been rewritten to legacy source and reimplementing Source API semantics
would be costly.
- [ ] Align behavior with Flink's new Source API watermark alignment
semantics.
- [ ] Complete control event and multi-output semantics
- [ ] Implement or explicitly reject `LatencyMarker` forwarding.
- [ ] Implement or explicitly reject `RecordAttributes` forwarding.
- [ ] Implement side-output watermark/status propagation, or reject
unsupported side-output plans during planning/offload.
- [ ] Avoid runtime `throw new RuntimeException("Not implemented for
gluten")` for unsupported stream events when the plan can be rejected earlier.
- [ ] Prefer planner-time fail-fast behavior for unsupported events.
- [ ] Add tests and acceptance coverage
- [ ] Java unit test: watermark gauge advancement.
- [ ] Java unit test: watermark status forwarding.
- [ ] Java unit test: one-input and two-input watermark merging.
- [ ] Java unit test: idle input does not block watermark progress.
- [ ] Java unit test: final watermark handling.
- [ ] Native unit test: `WatermarkAssigner` periodic emission.
- [ ] Native unit test: null watermark expression handling.
- [ ] Native unit test: `CombinedWatermarkStatus` active/idle behavior.
- [ ] Native unit test: window final flush.
- [ ] E2E SQL test: Kafka/Nexmark idle source.
- [ ] E2E SQL test: tumble, hop, and cumulate windows.
- [ ] E2E SQL test: two-input join.
- [ ] E2E SQL test: bounded source final results.
- [ ] E2E SQL test: watermark progress after checkpoint restore.
### Gluten version
None
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
