[AURON #2419] RowNullChecker may mis-detect null join keys for complex Arrow row types

[weimingdiit]

Which issue does this PR close?

Closes #2419

Rationale for this change
Auron's sort merge join uses encoded Arrow rows for join key comparison and uses RowNullChecker to skip rows with null join keys.

RowNullChecker previously inspected Arrow row bytes manually. This is fragile for complex row encodings such as struct, list, and dictionary, because their layout is controlled by Arrow RowConverter and should not be interpreted with fixed or estimated field lengths.

This can cause RowNullChecker to mis-detect null join keys for complex key types.

What changes are included in this PR?
Adds an Arrow RowConverter to RowNullChecker.
Updates RowNullChecker's has_nulls path to decode Rows through Arrow RowConverter before building the null-key mask.
Preserves conservative join key null semantics:
top-level null key values are treated as null join keys
nested null fields or elements inside non-null struct/list keys do not make the key null
Keeps the existing RowNullChecker public method signatures unchanged.
Keeps the existing byte-level helper path unchanged.
Adds unit coverage for complex struct, list, and dictionary key rows.

Are there any user-facing changes?
No user-facing API changes.

How was this patch tested?
UT.

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No
  Generated-by: OpenAI Codex (GPT-5)

[Copilot]

Pull request overview

This PR fixes incorrect join-key null detection for complex Arrow row encodings (e.g., struct/list/dictionary) by switching RowNullChecker::has_nulls from manual byte inspection to decoding rows via Arrow’s RowConverter, then deriving a null-key mask from the decoded top-level key arrays.

Changes:

• Add and retain an Arrow RowConverter inside RowNullChecker for row decoding.
• Rework has_nulls to decode Rows into key columns and compute validity based on top-level nulls.
• Add unit tests covering complex key types (struct, list, dictionary) and ensuring nested nulls don’t invalidate a non-null top-level key.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[slfan1989]

Thanks for fixing null detection for complex row types. However, has_nulls is called for every sort-merge join batch, and this change now fully decodes all key columns even for primitive keys. Could we preserve the existing fast path for simple types and only decode complex types to avoid a regression in the common join path?

[weimingdiit]

Good point. I updated the change to preserve the existing byte-level fast path for simple key types.

RowNullChecker now only decodes rows when the key schema contains complex types that need Arrow-level logical null handling, such as struct/list/dictionary. Primitive, boolean, variable-width, fixed-size binary, and NullType keys continue to use the existing encoded-row sentinel checks, so the common sort-merge join path should avoid the extra decode cost.

[weiqingy]

Array::is_null only reads the null buffer (arrow-array/src/array/mod.rs:250-252), and NullArray::nulls() returns None (.../null_array.rs:112-114). So for a DataType::Null key, which is what Arrow decodes that field to (arrow-row/src/lib.rs:1548), is_null is always false.

On master such a key is always null (row_null_checker.rs:240, DataTypeInfo::Null => true), so after this change the null-skip arms at sort_merge_join_exec.rs:382-383 stop firing and rows that never matched before would take part in the join. Spark's NullType does map to Arrow Null (ArrowUtils.scala:45), though I could not get Spark to actually plan an SMJ with one in on, so I cannot say it is reachable today.

Would logical_nulls() be the safer read here? Roughly this, in case it helps:

let logical_nulls: Vec<_> = key_columns.iter().map(|c| c.logical_nulls()).collect();
NullBuffer::from_iter((0..rows.num_rows()).map(|i| {
    logical_nulls.iter().all(|n| n.as_ref().is_none_or(|n| n.is_valid(i)))
}))

[weimingdiit]

Thanks for calling this out. I switched the decoded path from Array::is_null to Array::logical_nulls().

That preserves the old NullType behavior: when a NullType key reaches the decoded path, Arrow's NullArray logical null buffer marks all rows as null. I also added coverage for a NullType key combined with a complex key to force the decoded path and verify those rows are treated as null.

[weiqingy]

parse() re-wraps the bytes with this converter's fields, which is how they get past assert!(Arc::ptr_eq(...), "rows were not produced by this RowConverter") in convert_rows (arrow-row/src/lib.rs:701-704). That assert is what licenses the unsafe { convert_raw(...) } just below it (:710-713).

One consequence worth knowing before picking a fix for the panic already raised here: ? would not fully close it. RowParser forces validate_utf8: true (:844), and that path runs GenericStringArray::from(decoded) (arrow-row/src/variable.rs:325), whose From impl is try_from_binary(v).unwrap() (arrow-array/src/array/string_array.rs:74). So bad bytes panic inside Arrow before convert_rows can return Err.

Would reaching the producer's converter through the stream be an option? common/key_rows_output.rs:41-44 exposes only schema() and keys() today. With the converter available, convert_rows(rows.iter()) would work directly, no parse() round-trip, and the assert would be doing real work.

[weimingdiit]

Agreed. Re-parsing row bytes with RowParser is not the right ownership model and does not fully avoid Arrow panics either.

I changed the key-row pipeline to carry the RowConverter that produced each Rows value. RowNullChecker now uses that producer converter directly for the complex decoded path via convert_rows(rows.iter()), so Arrow's RowConverter identity check remains meaningful and we avoid the parse() round-trip.