[FLINK-39436][table] Allow late data in PTFs

[fhueske]

What is the purpose of the change

Previously, late events (rowtime <= watermark) were silently dropped before reaching PTF eval(), and timer registrations for times <= watermark were also silently dropped. This change removes both restrictions.
These are breaking changes that are part of FLIP-565 which was approved.

I checked that the docs don't need to be adjusted because the earlier behavior of PTFs dropping late data was not documented. The new behavior is aligned with other functions in Flink.

Brief change log

• ProcessTableRunner: remove the early-return guard in processEval() so that late events are passed to the PTF's eval() method.

• WritableInternalTimeContext: remove the watermark check in registerOnTimeInternal() so that timers can be registered for past times. Such timers fire immediately at the next watermark advance, including when registered from within onTimer(). The previous guard also had an unintended side effect: any call to replaceNamedTimer() with a past time would delete the existing timer entry but then silently drop the new registration, leaving the named timer in a state where it appeared un-registered but the old timer was gone.

• AbstractProcessTableOperator: remove the fired named timer's state entry before invoking onTimer() to prevent stale entries from accumulating in the named timers map state.

Verifying this change

Tests are updated to reflect the new semantics:

• PROCESS_LATE_EVENTS: demonstrates that late events enter eval(), can register timers (including for past times), and that such timers fire immediately at the next watermark advance.

• PROCESS_ROW_LATE_EVENTS (new): verifies the same for row-semantics PTFs.

• PROCESS_OPTIONAL_ON_TIME / PROCESS_NAMED_TIMERS: updated to reflect that timer registrations for past times are no longer dropped. Previously, once the watermark passed the registered time, the timer was silently discarded; now it fires immediately.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? n/a

[flinkbot]

CI report:

• 1781123 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[gustavodemorais]

The PR looks good in general to me!

Two nits: can you rename the PR title and commit title to use this format [FLINK-39436][table]?

[gustavodemorais]

Nice, this is basically the bugfix, right? We were incorrectly firing timeout2 at 2 and now we fire corrently at 5

[fhueske]

It's not really a bug fix. It's an artifact of allowing late timer registration.

[gustavodemorais]

One thing that was a bit confusing for me is why we have watermark 9223372036854775807 here? Do we emit this max watermark only for testing or do our sources actually emit this when all values have been read? I wondered if that basically makes the user not know if something is late anymore since

[twalthr]

Sources emit this when all values have been read. It will flush all remaining timers and windows and mark the job as complete.

[gustavodemorais]

Makes sense. This is an unrelated topic to the PR so definitely not a blocker: but I guess it makes sense for users to be aware of this so they can write their logic accordingly as well? Maybe it's already documented somewhere and then there's nothing to do 🙂

[gustavodemorais]

And for this PR: maybe add a short comment Sources emit 9223372036854775807 as a MAX watermark after all values have been read. It makes understanding the test a bit easier

[fhueske]

AFAICT this is a property of the testing framework.
I can see that this looks surprising, but I'm not sure if it makes sense to put a comment everywhere the framework is being used.

[gustavodemorais]

Suggested change

	"test that late events enter a row-level PTF")
	"test that late events enter a PTF with row semantics")

[gustavodemorais]

One thing worth noting: removing the watermark guard means past-time timers registered from onTimer() fire immediately (within the same advanceWatermark loop). But it also means something like this lead to infinite loops:

 public void onTimer(OnTimerContext ctx) {
      // "remind me again in the past" = fires immediately = infinite loop
      ctx.timeContext(Long.class).registerOnTime(0L);
  }

AI says this is the same behavior as standard Flink KeyedProcessFunction, so the design is consistent - just wanted to flag that PTF users were previously protected from this and they may shoot themselves in the foot more easily now. Maybe worth a short note in the timers section of ptfs.md along the lines of "timers registered at or below the current watermark fire immediately at the next watermark advance; avoid unconditional re-registration of past-time timers from onTimer() as this will cause an infinite loop."

[twalthr]

Thanks for the PR @fhueske. I added some minor comments. Could we update the JavaDoc in ProcessTableFunction and the documentation for late events. Late events deserve a dedicated section, before * <h2>Efficiency and Design Principles</h2>.

[twalthr]

good catch! some nit comments

Suggested change

	// Remove the fired timer's state entry immediately to prevent stale entries from
	// accumulating. Without this, entries for fired timers would persist until the same
	// timer name is re-registered or the state is explicitly cleared.
	if (namedTimersMapState != null && namedTimer != VoidNamespace.INSTANCE) {
	namedTimersMapState.remove((StringData) namedTimer);
	}
	processTableRunner.ingestTimerEvent(
	timer.getKey(),
	namedTimer == VoidNamespace.INSTANCE ? null : (StringData) namedTimer,
	boolean isNamedTimer = namedTimer != VoidNamespace.INSTANCE;
	// Remove the fired timer's state entry immediately to prevent stale entries from
	// accumulating. Without this, entries for fired timers would persist until the same
	// timer name is re-registered or the state is explicitly cleared.
	if (isNamedTimer) {
	namedTimersMapState.remove((StringData) namedTimer);
	}
	processTableRunner.ingestTimerEvent(
	timer.getKey(),
	isNamedTimer ? (StringData) namedTimer: null,

[twalthr]

wouldn't this be more correct and allowing for watermarks before epoch.

Suggested change

	long timer = wm == null || wm < 0 ? 1 : wm + 1;
	long timer = wm == null ? Long.MIN_VALUE + 1: wm + 1;

[fhueske]

This is just a function for testing purposes and we assert against the behavior that we implement.
So I think it doesn't really matter.
Or is there an (edge) case that would be covered with your suggestion that isn't with the current version?

[fhueske]

Thanks for your reviews @gustavodemorais and @twalthr.
I've addressed your comments and updated the PR.

[gustavodemorais]

The PR looks good to me! Thanks, @fhueske

• I think you have to run spotless to make the CI green

[twalthr]

Thanks for the update @fhueske, I had two last comments. Should be good in the next iteration.

[twalthr]

copy this to the Chinese docs as well

[twalthr]

keep docs in sync:

Suggested change

	watermark. PTFs handle late records just like non-late records by calling the `eval()` method.
	watermark. PTFs handle late records just like non-late records by calling the `eval()` method. If the {@code on_time} argument is specified, the late timestamp is preserved in
	* the output