Conversation
Casts are evaluated through Flink's own cast rules, and the code generated for a non-nullable input dereferences the value with no null guard. Priming that code with a null therefore aborted operator startup rather than surfacing the per-row error the query expects, turning every cast of a non-nullable string into a job failure. Prime with a value the declared input type admits instead. This was invisible locally because the harnesses only ever exercised nullable columns; non-nullable types reach the engine through SQL literals, so only the upstream suite covered the shape that broke.
Flink carries day-time intervals as integer milliseconds. DataFusion compiles interval literals as Arrow day/millisecond pairs instead, which the existing output-type check rejected and downstream keyed consumers could not encode. Accept that inferred type only for a top-level SQL day-time interval, never for BIGINT. Convert Calc's materialized interval columns to nullable integer milliseconds before they reach exchanges or keyed operators, while retaining the original inferred type for the planner's SQL-type check. The host reader also accepts Arrow's day-time encoding. Nested fields retain their existing checks, without positional pairing of Flink and Arrow child layouts. Apply the same plan-time compilation check to filter-only Calcs and join residual predicates. Otherwise a separate filter can encode a comparison between a materialized interval and an interval literal yet fail to coerce it on its first batch. Unsupported comparisons now fall back with their compilation reason. Timestamp differences producing Duration remain on the host: Flink subtracts the operands' millisecond values, not a narrowed nanosecond difference. Year-month literals and Arrow calendar-month interval output are not added. Boundary diagnostics name the Arrow vector and SQL type when a reader or writer cannot handle a representation.
Flink validates a forced delta-join strategy after our substitution pass. Removing an ordinary join can erase the evidence that makes Flink reject a statement, allowing a query that the host intentionally refuses. Leave an optimizer block unchanged under FORCE when it contains an ordinary join and no delta join. A delta join in the same block bypasses this guard only: it remains unsupported natively, and ordinary admission and island checks still apply. The host validates across all statement roots, while this pass sees one optimizer block at a time. A regular-join block therefore still falls back when the only delta join is in another block, even if the host accepts the statement. Record this conservative scope in the fallback reason and coverage docs. Test actual regular and delta rel nodes through the guard, rather than a separate boolean policy that bypasses traversal.
57213a7 to
aea3e04
Compare
Flanderzz
changed the title
jordepic
left a comment
jordepic
left a comment
There was a problem hiding this comment.
Thanks for tracking these down. All three fixes are real: I reproduced both crash premises against main and confirmed the branch resolves them, ran the PR's test classes (103/103), the two native tests plus cargo fmt --check, and added a dozen parity probes of my own. Requesting changes for one correctness gap in the interval work and one design point; the rest is polish.
Confirmed
- Cast warm-up —
SELECT CAST(COALESCE(s, '0') AS INT) FROM tfails the job onmainat operator open (NullPointerException: BinaryStringData.trim()wrapped inError when casting STRING NOT NULL to INT NOT NULL). Flink'sAbstractNullAwareCodeGeneratorCastRuleemits no null guard for a NOT NULL input, so the diagnosis is exactly right. Native parity on the branch. - Interval filter —
WHERE dur > INTERVAL '1' HOURon a source interval column panics natively onmain(failed to coerce predicate … has type Int64 … has type Interval). The branch falls back at plan time with that reason. - FORCE guard — mirrors
StreamPhysicalDeltaJoinForceValidator's finder, conservative per optimizer block, documented, and tested with real rel nodes.
Blocking: interval arithmetic followed by a comparison is wrong natively
arrow-rs IntervalDayTime derives Ord lexicographically on (days, milliseconds) and = compares the two components, so a non-normalized pair produced by +/- compares incorrectly. With t(k BIGINT), k = 1..4:
| Query | Flink | Native |
|---|---|---|
SELECT k FROM t WHERE (CASE WHEN k > 2 THEN INTERVAL '1' DAY ELSE INTERVAL '1' HOUR END) + INTERVAL -'2' HOUR < INTERVAL '23' HOUR |
1, 2, 3, 4 | 1, 2 |
same with = INTERVAL '22' HOUR |
3, 4 | (none) |
Native holds (1 day, −7 200 000 ms) and compares it against (0, 82 800 000) component-wise. This predates the PR (the same queries diverge on main), but this PR is the one codifying interval handling in filter-only Calcs and documenting it, and projecting the same expression is correct only because Calc normalizes afterwards. Either normalize before the comparison, or have the encoder decline comparison and arithmetic between two interval operands (keeping timestamp ± interval).
Design: one representation, decided once
The batch is converted in calc.rs, but inference still reports Interval(DAY_TIME), so Java grows a readsAs exemption plus ArrowIntervalDayColumnVector — and no runtime path can reach that vector (Calc converts, nothing else emits an interval array). The days * 86_400_000 + ms conversion now lives in three places (calc.rs, over_agg.rs::normalize_interval_columns, expr.rs::IntervalScale), and the OVER helper is now dead code with a comment that describes behaviour Calc no longer has.
Doing it once in the expression compiler — wrap each interval-typed projection root and each interval comparison operand in the to-millis conversion IntervalScale already contains — fixes the blocking item, makes infer_calc_output_schema agree with evaluate by construction, and lets you delete the Java vector, the readsAs exemption, the per-batch downcast probe, and the OVER helper. The stated reason for keeping inference on Interval (guarding a BIGINT-declared projection) doesn't hold: nothing the encoder emits can yield an interval array for a Calcite BIGINT expression — interval literals and CASEs over them are INTERVAL-typed, and IntervalScale already returns Int64.
Docs
The residual-predicate paragraph is pasted verbatim onto four join pages, and the FORCE guard is explained three times (index.md, regular-join.md, unsupported.md) with overlapping sentences. Repo convention is one explanation under Global switches on docs/operators/index.md and a one-line bullet linking to it from each page. The calc-filter.md addition narrates implementation ("Inference retains the expression's interval type so this exemption cannot admit…") rather than stating what is admitted; the coverage page should just say day-time interval expressions are native, timestamp differences and year-month intervals fall back, and why.
Nits
FilterCalcMatcher.matchesnow encodes the Calc twice (encodedConditionthenencodeCalc).- The
BOOLEANarm inwarmupValueis unreachable —hostCastSupportednever admits a boolean source. - The three identical
"Unsupported type %s (Arrow vector %s, arrow type %s)"blocks inArrowConversionwant a helper. - Commit 2's message has a stray line break ("Otherwise a\nseparate filter").
- The branch is 12 commits behind
main; it rebases cleanly.
Integrate the cast initialization and FORCE-join repairs from PR #30, retaining its authorship in the preceding commits. Current temporal lowering already carries day-time milliseconds and year-month integers through the scalar bridge, so its interval cases need regression coverage rather than the older Arrow interval conversion and conservative fallback implementation. The ten interval cases now assert actual native execution for signed literals, nullable grouping keys, filters after Top-N, outer-join residuals, nanosecond timestamp differences, and year-month CASE results. Add the audited runtime CHAR(8) NOT NULL to DOUBLE query. Two focused runs pass 32 and 81 tests against released Flink 2.2.1, including 52 dynamic collection cases and decimal/binary UDF ownership checks. Fixes #84. Fixes #89. Fixes #90. Co-authored-by: Devine Chinemere <Fredfres1@gmail.com>
|
Closing this cause it seems these changes landed yesterday 603f7c1 carries commit 1 (cast warm-up) 🎉🎉🎉🎉🎉🎉 |
2 participants
Summary:
Found three issues in the shared code on
mainwhile validating the Flink 2.1.3 backport: cast initialization uses an invalid null input, day-time interval expressions cannot safely pass through the native pipeline, and join substitution can hide an error Flink is supposed to raise. These fixes belong onmainindependently of 2.1 support. They need to land before the backport can build on them; the version-specific changes remain in a separate PR.Changes:
main, the output-type check otherwise rejects this shape.table.optimizer.delta-join.strategy=FORCE. The guard is conservative per optimizer block; it does not add native delta-join support.Testing
Validated locally:
mvn -B -ntp -pl :streamfusion-runtime test— 1,074 tests, 0 failures/errors, 33 skipped.cargo test --manifest-path native/Cargo.toml --lib— 400 passed.