Skip to content

Correct FLOAT precision and column-less batch routing - #29

Open
Flanderzz wants to merge 2 commits into
datafusion-contrib:mainfrom
Flanderzz:fix/columnar-float-and-zero-column-batch
Open

Correct FLOAT precision and column-less batch routing#29
Flanderzz wants to merge 2 commits into
datafusion-contrib:mainfrom
Flanderzz:fix/columnar-float-and-zero-column-batch

Conversation

@Flanderzz

@Flanderzz Flanderzz commented Aug 30, 2026

Copy link
Copy Markdown
Contributor

Summary:

Found two native-execution defects found while validating Flink 2.1.3 support (future PR).

Both reproduce on main against Flink 2.2.1.

A projected FLOAT expression containing a FLOAT constant aborts the job. Such constants were encoded as double literals, so DataFusion promoted the whole expression to double precision and returned a DOUBLE column where the plan declared FLOAT. The columnar boundary selects a Flink column vector from the Arrow type it is handed rather than the type the plan declared, so the widened result arrived as a Double vector and threw ClassCastException the moment a FLOAT field was
read.

The scope is narrower than it first appears, which is why it went unnoticed: f4 * CAST(2 AS FLOAT) fails, but f4 * 2 does not (narrow integer literals already carry their declared width), nor does f4 + f4 (no literal to widen), nor a FLOAT literal confined to a filter predicate (the widening lands on the boolean mask, not a projected output column).

The width was wrong as well as the type. Flink evaluates FLOAT arithmetic in single precision, so computing in double and narrowing afterwards can differ in the last bits even where the cast succeeds.

The second is a task abort in the columnar exchange. Splitting a batch into per-channel sub-batches takes row subsets and rebuilds a batch from the resulting columns. A projection can legitimately leave no columns behind (a key-only shuffle carries routing information and nothing else), and Arrow cannot infer how many rows such a batch holds, so the split panicked.

Closes #28

Changes:

  • Encodes single-precision constants as FLOAT literals rather than double, so a FLOAT expression keeps both its declared result type and the host's evaluation width.
  • States the row count explicitly when building exchange sub-batches, covering both the destination and key-group splits, matching how the aggregate and calc paths already construct batches.
  • Adds FLOAT coverage to the Calc parity tests, which had none.

Testing

Validated locally with Java 17 (as always pls validate my parking here):

  • Calc SQL parity: mvn -pl :streamfusion-runtime test -Dtest=FlinkCalcSqlHarnessTest 68 tests passing, including three new FLOAT cases (literal arithmetic, column
    arithmetic, and a cast projection).
  • Native unit tests: cd native && cargo test --lib exchange:: two new cases covering zero-column destination and key-group splits.
  • Reproduced on clean main (c9f7905c) and confirmed fixed: a datagen table with a FLOAT column fails with ArrowDoubleColumnVector cannot be cast to FloatColumnVector before the change and returns correct values after. Verified across seven FLOAT query shapes; the three using a FLOAT literal in a projection fail before and pass after, the other four are unaffected either way.
  • Upstream Flink suite, CastFunctionITCase:
    • Flink 2.2.1 — 557 tests, 0 failures, 0 errors.

An exchange splits a batch into per-channel sub-batches by taking row subsets.
A projection can legitimately leave no columns behind — a key-only shuffle
carries routing information and nothing else — and Arrow cannot infer how many
rows such a batch holds without being told. Building the sub-batch without a row
count therefore aborted the task, and only for the shapes where the columnar
exchange is doing the least work.

State the row count explicitly, matching how the aggregate and calc paths already
construct batches, and cover both the destination and key-group splits.
Constants in a FLOAT expression were encoded as double literals, so the engine
evaluated the whole expression in double precision. That contradicted the plan
twice over: the result arrived as a DOUBLE column where the declared output type
was FLOAT, and the arithmetic itself was carried out at a width the host never
uses.

The type mismatch was the louder half. The columnar boundary picks a Flink vector
from the Arrow type it is handed rather than the declared one, so a widened
result reached the host as a Double column and failed the moment a FLOAT field
was read — an ordinary FLOAT projection was enough to abort the job.

Encode single-precision constants as such, mirroring the existing treatment of
narrow integer literals, which carry their declared width for the same reason.
Float coverage was missing from the calc parity tests entirely, which is why this
survived; add it.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

BUG: FLOAT projection with a float constant fails: ArrowDoubleColumnVector cannot be cast to FloatColumnVector

1 participant