Match Flink's legacy window property contract

[Flanderzz]

Summary:

This PR adds native coverage for Flink’s deprecated but still widely used GROUP BY TUMBLE(...) and GROUP BY HOP(...) syntax.

Legacy group windows plan as StreamPhysicalGroupWindowAggregate, separately from window TVFs. The new matcher translates supported legacy plans onto StreamFusion’s existing native fixed-window operators. No new Rust operator, JNI interface, physical rel, or exec node is introduced.

Closes #9.

Changes:

• Routes event-time and processing-time legacy TUMBLE and HOP aggregates through the existing native window engine.
• Preserves the existing native event-time legacy SESSION path under the generalized matcher.
• Reuses the same aggregate, key, and value-type coverage as TVF window aggregates.
• Supports keyed, global, zero-aggregate, and no-window-property legacy plans.
• Reproduces Flink’s legacy output contract:
  • window_start and window_end as TIMESTAMP(3)
  • rowtime as window_end - 1ms
  • the internal null proctime marker, materialized by the outer Calc
• Keeps the existing windowAggregate configuration switch.
• Adds explicit fallback reasons for unsupported legacy semantics:
  • row-count windows
  • early or late firing and allowed lateness
  • retracting or updating input
  • processing-time SESSION
  • processing-time HOP whose slide does not divide its size
  • unsupported aggregate, key, value, or Arrow boundary types
• Conservatively falls back for LTZ fixed windows unless the session zone has a fixed post-1970 offset aligned with the slide. This avoids diverging from Flink’s DST-aware local-time assignment and firing rules.
• Updates the window coverage and unsupported-operator documentation

Testing

Validated locally with Java 17:

• Full Maven reactor:
  • mvn test
  • all 13 modules passing
• Focused window SQL and operator coverage:
  • 115 tests passing
• Release fixed-window benchmarks:
  • SF_BENCHMARK=true mvn -pl :streamfusion-runtime test -Pbench -Dtest='ThroughputBenchmark#tumblingThroughput+legacyTumblingThroughput+legacyHoppingThroughput'
• Release columnar-source benchmark:
  • SF_BENCHMARK=true mvn -pl :streamfusion-runtime test -Pbench -Dtest='ThroughputBenchmark#legacyWindowedColumnarSourceThroughput'
• Docker-backed Kafka/JSON Nexmark regression matrix for q4, q7, q8, and q11.
• Clean-main before/after benchmark comparison confirmed no Nexmark regression from this PR.

[jordepic]

Reviewed against Flink 2.2's planner/runtime sources. The core contract work checks out — verified against AggsHandlerCodeGenerator.getWindowExpressions that Flink emits the proctime property as a null marker, rowtime as toEpochMills(localEnd − 1, shiftZone), and start/end as shifted-local TIMESTAMP(3); and against WindowPropertiesRules that stream plans carry either zero properties or exactly (start, end[, rowtime][, proctime]) in that order, which is what supportedProperties assumes. The LTZ gate is genuinely required: native assignment is a pure epoch grid (window_agg.rs, rem_euclid(slide)), while Flink's legacy WindowOperator shifts every timestamp into session-local time before assigning, so "fixed post-1970 offset, aligned with the slide" is the right admission condition — under it the rendering, rowtime, and firing math all reduce to the native epoch grid. The emit-strategy / insert-only / row-count gates are correct and conservative (WindowEmitStrategy forces allowLateness to 0 unless late-fire is enabled, so that check can't misfire), and the parity tests assert substitutions > 0, so they can't pass via silent fallback.

A few things to address or answer before merge:

1. Legacy LTZ SESSION skips the zone gate but has the same DST exposure. WindowOperator shifts timestamps before assignment for all window types, sessions included — Flink merges sessions in shifted-local time, while the native session operator merges in epoch time. A fixed offset cancels out of the gap arithmetic, so no alignment condition is needed, but a zone with post-1970 transitions distorts gaps around each transition and can merge differently (and the emitted bounds then disagree too). The session branch in GroupWindowAggregateMatcher.unsupportedReason returns before sessionZoneAlignsWithSlide is ever consulted. This hole predates the PR (the old GroupWindowSessionMatcher had it too), but since this PR owns and generalizes the matcher, either apply the "no post-1970 transitions" half of the check (without the slide-alignment half) to LTZ sessions, or record the divergence and its reasoning in divergences/.

2. Run the upstream Flink suite on this branch and confirm the feature engaged. bin/flink-suite.sh runtime runs Flink's own GroupWindowITCase (both stream.sql and stream.table) — the exact upstream tests that pin legacy group-window semantics — but that suite only runs weekly on main, so this PR hasn't been through it. Two asks:

• Run it on this branch (at minimum FLINK_SUITE_TEST=GroupWindowITCase bin/flink-suite.sh runtime) and post the summary here.
• Note that GroupWindowITCase pins Asia/Shanghai as the session zone, which has post-1970 DST transitions — so every LTZ and proctime variant will (correctly) fall back, and a green suite proves parity only for the plain-TIMESTAMP event-time half. Please check the -Dstreamfusion.logFallbackReasons=true output to confirm the non-LTZ cases actually substituted, since the runtime suite has no engagement markers the way the state suite does.

3. The TVF path looks exposed to the same LTZ grid divergence — follow-up issue? This PR makes the legacy path stricter than the TVF path: WindowAggregateMatcher admits LTZ time attributes with no session-zone check, yet Flink's slicing assigners take the same shift timezone and the native side assigns on the same epoch grid. If that's right, a TVF TUMBLE/HOP over TIMESTAMP_LTZ with, say, an Asia/Kolkata session zone and a 1-hour window already diverges today. That belongs in a follow-up issue (and the new gate probably wants to live in shared code both matchers call), not in this PR — but please confirm the reading.

4. Benchmark edits to pre-existing tests. Changing tumblingThroughput's sink from TIMESTAMP_LTZ to TIMESTAMP(3) and pinning the shared tumblingEnvironment to UTC changes what the existing TVF benchmark measures (drops the sink-side cast, changes the zone). The UTC pin is needed for portability — on a machine in a DST zone the legacy benches would otherwise throw "did not engage", which is the right failure — but please confirm the sink-type change was intentional and that old numbers aren't compared against new runs. Also, per repo convention, the routing commit message should carry the measured legacy-window numbers (native vs. Flink) from the -Pbench runs.

Nits:

• GroupWindowAggregateMatcher.java is missing its trailing newline.
• legacyLtzRowtimePropertyMatchesHostInSessionZone / legacyLtzWindowWithMisalignedSessionZoneFallsBack in FlinkWindowSqlHarnessTest are mis-indented relative to the rest of the file.
• size() / slide() in the matcher repeat the interval-extraction expression three times; a small helper taking the size/slide ValueLiteralExpression would collapse them.

[Flanderzz]

Hey, thanks for the review. Good catch on the upstream Flink tests and LTZ session behavior.

Running FLINK_SUITE_TEST=GroupWindowITCase bin/flink-suite.sh runtime exposed a checkpoint bug in StreamFusion that i would have introduced. Canonical-state serialization temporarily changes the active key, but only restored the key afterward. Flink then hashed that synthetic key again and selected a key group outside the subtask’s range. I fixed this by preserving and restoring the exact key and key-group context, including handling an initially unset context. The pre-fix control reproduced all 10 errors (83/93). the fixed version passes 93/93.

Before results:

# StreamFusion upstream Flink suite
- Reports: 4
- Tests: 93
- Passed: 83
- Failures: 0
- Errors: 10
- Expected upstream failures: 0
- Skipped: 0

After results:

# StreamFusion upstream Flink suite
- Reports: 4
- Tests: 93
- Passed: 93
- Failures: 0
- Errors: 0
- Expected upstream failures: 0
- Skipped: 0

(I as always, encourage running it on your PC cause there is a chance mine is messed up....)

For the LTZ sessions point, fixed offsets are safe because they cancel out in gap arithmetic, but DST transitions can change which records merge. Fixed-offset zones remain native, while zones with post-1970 transitions now fall back explicitly. I added parity and fallback coverage and updated the docs.

The benchmark sink change was intentional. TVF window properties are TIMESTAMP(3); the old TIMESTAMP_LTZ(3) sink inserted a cast that prevented native substitution, so there are no valid old native TVF numbers being compared. UTC is pinned so results and legacy admission do not depend on the host timezone.

I also updated the window-aggregate.md file along side this.
and also since the fix to the bug touched canonical state i also updated those docs too.

Release benchmark results (I encourage you to verify on your machine):

• Legacy TUMBLE: Flink 3,298,662, native 2,593,014 rows/s (0.79x)
• Legacy HOP: Flink 2,451,344, native 2,574,562 rows/s (1.05x)
• Columnar legacy TUMBLE: Flink 3,578,066, native 8,729,686 rows/s (2.44x)

I also fixed the formatting/newline/helper nits.

TL;DR:

• Full upstream Flink runtime suite: 8,619 tests, no failures or errors with new fixes
• Upstream GroupWindowITCase: 93/93, before it was 83/93
• Full StreamFusion reactor: all 13 modules passed
• Focused window/state tests: 66/66