fix(smj): handle floating-point keys across batch boundaries - #25134

…mark joins (apache#26830)

## Which issue does this PR close?

Closes apache#26003.

## Rationale for this change

`EXISTS`, `IN` and `NOT EXISTS` queries planned as sort-merge joins fail
with "Unsupported data type in sort merge join comparator" on
timezone-aware timestamp, `Dictionary`, `Decimal256` and
`Time32`/`Time64` keys, once a key group runs to the end of a batch.
Inner sort-merge join and hash join already work on these keys.

## What changes are included in this PR?

The bitwise stream compares keys with `JoinKeyComparator` everywhere
except in `keys_match`, the check for whether a key group continues into
the next batch. That function still called `compare_join_arrays` for
non-float keys, and `compare_join_arrays` only knows a fixed list of
types. apache#25134 moved float keys over to `JoinKeyComparator`; this moves
the rest.

`keys_match` now builds a one-row comparator each time a key group
reaches the end of a batch. The scan already rebuilds its comparators on
every new batch, so I don't expect a measurable cost.

One edge case changes: `compare_join_arrays` treated `DataType::Null`
keys as always equal, while `JoinKeyComparator` treats them as unequal
under `NullEqualsNothing`. As far as I can tell this can't change any
output, because such keys never compare equal in the scan either, so the
code paths that call `keys_match` don't run for them.

`compare_join_arrays` has no callers left, but it's public in
`joins::utils`, so I left it alone. Happy to deprecate it here or in a
follow-up if you prefer.

## What is the testing strategy for this PR?

- `sort_merge_join.slt`: the reproducer from the issue, plus `batch_size
= 2` queries for `Dictionary`, `Decimal256` and `Time64` keys.
- `join_key_type_batch_boundaries` in `sort_merge_join/tests.rs`: the
same matrix as `join_float_key_batch_boundaries` (join type, null
equality, sort direction, 1- and 2-row batches) for the five key types
above. I moved the body of the float test into a shared helper so the
two don't duplicate it. Like apache#25134, it's a physical-plan test, so
batches split at exact rows regardless of what the optimizer does.

Without the fix, all of the new cases fail with the error from the
issue.

The change also covers other types `compare_join_arrays` didn't handle
(e.g. `Duration`, `Interval`, `Decimal32`/`Decimal64`, nested types),
but I haven't added tests for those.

## Are there any user-facing changes?

Those queries now return results instead of an error. No API changes.

Push float comparisons to Iceberg as exact predicates. Arrow and
Iceberg order floats by IEEE totalOrder, while PostgreSQL treats -0 and
0 as equal and orders every NaN above every number. The new
`predicate::float` rewrites each comparison so it accepts exactly the
rows PostgreSQL accepts:

- A zero constant gets the sign that puts both zeros on the correct
  side.
- `>` and `>=` also accept `isnan`, and `<` and `<=` reject it.
- A double value compared with a real column is rounded to the
  neighbouring real on the side each operator keeps.

The planner costs `>` and `>=` as the OR with `isnan` that they bind
to.

Query offload now accepts float comparisons, GROUP BY keys and join
keys. It no longer rejects float join keys. DataFusion folds -0 into +0
and keeps NaN payloads apart, which is an accepted engine difference.
Float ORDER BY stays in PostgreSQL.

Comments now record the upstream defects that are not worked around
here:

- DataFusion 55's float IN-list, and the IN-list dynamic filter built
  from a small hash-join build side, compare float bits and miss -0
  rows (apache/datafusion#25186).
- The sort-merge semi/anti/mark fallback panics on a NaN key at a batch
  boundary (apache/datafusion#25134).
- Iceberg filters on files written before a type promotion round the
  literal down to the old type (apache/iceberg-rust#3122, #3540).

SUM(int8) now casts its input to Decimal128(38,0) before DataFusion's
native SUM, because the native Int64 SUM wraps on overflow. The sum
stays exact and can be compared in HAVING. Remove the now-unused
`NumericFromInt64` output codec.

…ies (#25134) (#27353)

## Which issue does this PR close?

- Part of #26868 (the 55.3.0 patch release tracking issue).
- Backport of #25134 to `branch-55`. #25134 fixed #25133.

## Rationale for this change

On 55.2.0, a mark, semi or anti sort-merge join on a floating-point key
fails when a key group ends at the end of an input batch: a Float32 or
Float64 key that is NaN panics with ``called `Option::unwrap()` on a
`None` value``, and a Float16 key returns `Unsupported data type in sort
merge join comparator: Float16`. The main merge scan handles the same
keys. These are failed queries, which makes this eligible for backport
as a correctness fix.

#25134 fixed it on `main` (labelled `v56.0.0`); it is on `branch-56` but
not on `branch-55`. It was requested for 55.3.0 on #26868
(#26868 (comment)).

## What changes are included in this PR?

- A `git cherry-pick -x` of 2abf69d
(#25134, by @jeffw13). It applied without conflicts and this PR changes
nothing else.
-
`datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs`:
`keys_match` uses the existing `JoinKeyComparator` for floating-point
keys, matching the merge scan's equality (including signed zero),
instead of the scalar comparator whose `partial_cmp` panics on NaN.
Integer-only comparisons keep the scalar path.
- `datafusion/physical-plan/src/joins/sort_merge_join/tests.rs`:
`join_float_key_batch_boundaries`, 144 cases (Float16/32/64, left/right
mark/semi/anti, both null-equality modes, ascending/descending, one- and
two-row input batches).

## Are these changes tested?

Yes, by the 144 cases that #25134 added. On this branch (Rust 1.95.0) I
ran:

- `cargo test -p datafusion-physical-plan --lib`: 1844 passed.
- With the change in `bitwise_stream.rs` reverted and the tests kept,
`cargo test -p datafusion-physical-plan --lib sort_merge_join` fails
exactly the 144 new cases (96 NaN panics and 48 Float16 errors).
- `cargo fmt --all -- --check` and `cargo clippy -p
datafusion-physical-plan --all-targets --all-features -- -D warnings`
pass.

## Are there any user-facing changes?

Affected sort-merge joins complete instead of failing at floating-point
batch boundaries. No public API change.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Jeff Wang <[email protected]>
Co-authored-by: Claude Sonnet 5.5 <[email protected]>

…ies (#25134) (#27357)

## Which issue does this PR close?

- Part of #23805 (the 54.2.0 patch release tracking issue).
- Backport of #25134 to `branch-54`. #25134 fixed #25133.

## Rationale for this change

On `branch-54`, a mark, semi or anti sort-merge join on a floating-point
key fails when a key group ends at the end of an input batch: a Float32
or Float64 key that is NaN panics with ``called `Option::unwrap()` on a
`None` value``, and a Float16 key returns `Unsupported data type in sort
merge join comparator: Float16`. The main merge scan handles the same
keys. These are failed queries, which makes this eligible for backport
as a correctness fix.

#25134 fixed it on `main` (labelled `v56.0.0`); it is on `branch-56` but
not on `branch-54`. It was requested for `branch-54` on #26868
(#26868 (comment)),
"to prevent regressions on upgrade to 54".

## What changes are included in this PR?

- A `git cherry-pick -x` of 2abf69d
(#25134, by @jeffw13). It applied without conflicts and this PR changes
nothing else.
-
`datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs`:
`keys_match` uses the existing `JoinKeyComparator` for floating-point
keys, matching the merge scan's equality (including signed zero),
instead of the scalar comparator whose `partial_cmp` panics on NaN.
Integer-only comparisons keep the scalar path.
- `datafusion/physical-plan/src/joins/sort_merge_join/tests.rs`:
`join_float_key_batch_boundaries`, 144 cases (Float16/32/64, left/right
mark/semi/anti, both null-equality modes, ascending/descending, one- and
two-row input batches).

## Are these changes tested?

Yes, by the 144 cases that #25134 added. On this branch (Rust 1.95.0) I
ran:

- `cargo test -p datafusion-physical-plan --lib`: 1583 passed.
- With the change in `bitwise_stream.rs` reverted and the tests kept,
`cargo test -p datafusion-physical-plan --lib sort_merge_join` fails
exactly the 144 new cases (96 NaN panics and 48 Float16 errors).
- `cargo fmt --all -- --check` and `cargo clippy -p
datafusion-physical-plan --all-targets --all-features -- -D warnings`
pass.

## Are there any user-facing changes?

Affected sort-merge joins complete instead of failing at floating-point
batch boundaries. No public API change.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Jeff Wang <[email protected]>
Co-authored-by: Claude Sonnet 5.5 <[email protected]>