[FLINK-40168][table] Thread the EARLY_FIRE hint into the interval join

[weiqingy]

Part of the FLIP-497 implementation stack under umbrella FLINK-36953. Landing order:

Step	Sub-task	Scope
PR-1a	FLINK-40167	EARLY_FIRE hint surface + option validation (#28353, merged)
PR-1b (this PR)	FLINK-40168	Thread the hint into the interval join (planner → ExecNode)
PR-2	FLINK-40169	target option
PR-3	FLINK-40170	Update-producing changelog mode + insert-only guard
PR-4	FLINK-40171	Runtime early-fire emit + retraction
PR-5	FLINK-40172	Processing-time early fire on an event-time join
PR-6	FLINK-40173	State restore coverage
PR-7	FLINK-40174	User-facing documentation

What is the purpose of the change

Threads the (already-registered) EARLY_FIRE hint through the planner to the ExecNode. StreamPhysicalIntervalJoinRule reads the hint, resolves the effective time mode from the join's time domain, validates the domain combinations, and threads the delay and time mode into StreamExecIntervalJoin as NON_NULL JSON fields. The operator receives the parameters but ignores them; runtime behavior lands in PR-4.

Brief change log

• StreamPhysicalIntervalJoinRule reads the hint, resolves the effective time mode, rejects row-time triggering on a processing-time join, and rejects (for now) processing-time triggering on an event-time join.
• Thread earlyFireDelay/earlyFireTimeMode through StreamPhysicalIntervalJoin into StreamExecIntervalJoin as NON_NULL JSON fields.

Verifying this change

This change added tests and can be verified as follows:

• EarlyFireJoinHintTest: earlyFireDelay/earlyFireTimeMode reach the exec plan for both a row-time (default ROWTIME) and a processing-time (default PROCTIME) interval join; row-time-on-proctime and processing-time-on-rowtime are rejected; a compiled-plan JSON round-trip covers serialization of the new ExecNode fields.

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: yes (compiled-plan ExecNode JSON only; no state serializer change)
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no (planner threading for the FLIP-497 hint)
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Anthropic)

[flinkbot]

CI report:

• a856401 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[weiqingy]

Hi @RocMarshal, opening PR-1b of the FLIP-497 stack ahead as a draft. It's stacked on #28353, so I'll rebase it onto master and take it out of draft for review once #28353 is merged. Thanks!

[weiqingy]

Hi @RocMarshal, #28353 (PR-1a) has merged, so I've rebased this onto master. The diff is now standalone: just the planner threading and its tests. Taking it out of draft. Ready for review when you have a chance, thanks!

[RocMarshal]

Thanks @weiqingy
LGTM +1 on the whole, just a comment.

[RocMarshal]

anchor-A

[RocMarshal]

At Anchor-A, the early fire parameters are now encapsulated within a dedicated class. That said, the call sites currently split them apart for invocation.

Given the semantics of this link and its subsequent stages, are there any specific downsides to propagating these parameters strictly via the EarlyFire(or EarlyFireConfig) wrapper?

Since these two parameters are almost always co-located, I’m merely questioning the trade-off. If using EarlyFire(or EarlyFireConfig) proves cumbersome, decoupling them is an acceptable compromise.

And I'd like to hear more ideas about it

[weiqingy]

Thanks @RocMarshal for the review and the +1.

Good observation. EarlyFire is a small planner-internal holder for the result of extractEarlyFire(), effectively Java's stand-in for a tuple, because that method resolves the effective time mode and performs the domain validation together. It is unpacked once at its single call site into the two values propagated downstream.

My reason for continuing to propagate them separately is the eventual ExecNode boundary, which is also the compiled-plan serde boundary. earlyFireDelay and earlyFireTimeMode are persisted as two optional scalar properties on StreamExecIntervalJoin. Keeping them flat follows existing ExecNode patterns for small optional properties and avoids adding a nullable nested spec to the plan format for two values. IntervalJoinSpec is nested, but it represents an always-present group of interval-join properties, whereas early fire itself is optional.

Regarding whether more configuration will join these values: the next PR adds a target option, but it remains a rule-level applicability gate. It determines whether the hint applies to this operator kind and is discarded after extraction, so the propagated state remains exactly delay plus timeMode.

My preference is therefore to keep the two flat fields for now. If another parameter eventually needs to reach the operator, promoting them to a small shared spec would make more sense. If you prefer establishing the wrapper now, though, I'm happy to change it.

[RocMarshal]

Thanks @weiqingy .
LGTM +1