andygrove opened a new pull request, #6364: URL: https://github.com/apache/datafusion-comet/pull/6364
## Which issue does this PR close? Closes #6291. ## Rationale for this change `AlignedArrowStreamReader` existed because arrow 58's `from_ffi_and_data_type` passed JVM-allocated `Decimal128` buffers through under-aligned ([apache/arrow-rs#10028](https://github.com/apache/arrow-rs/issues/10028)). Arrow 59 fixed that ([apache/arrow-rs#10030](https://github.com/apache/arrow-rs/pull/10030)), and on 59.3.0 `from_ffi` and `from_ffi_and_data_type` call `align_buffers` themselves, so the reader's own call was a second pass over every buffer of every input batch. The dictionary unpack in `ScanExec`'s `import_column` cannot be reached by real input, and the Native to JVM half of `ffi.md` documented an API that has never existed. ## What changes are included in this PR? - `ScanExec` reads the JVM stream with `arrow::ffi_stream::ArrowArrayStreamReader`, and `aligned_stream_reader.rs` is removed. `realigns_under_aligned_decimal128` moves to `scan.rs` and now pushes the under-aligned batch through the stock reader, using a small hand-written one-batch C stream, as a guard against an arrow downgrade. - `import_column` only decodes invalid UTF-8. No input stream carries a dictionary, because `ColumnarBatchArrowReader` decodes them on the JVM. A dictionary that did arrive would still be unpacked by the cast in `build_record_batch`, since the scan schema never has a dictionary type. `test_unpack_dictionary_primitive` and `test_unpack_dictionary_string` already cover that cast: they seed their batch with `set_input_batch`, which bypasses `import_column`, so they stay. With no callers left, `copy.rs` (`copy_or_unpack_array`, `copy_array` and `CopyMode`) is removed. - `NativeUtil.takeRows`, which had no callers, is removed. - `ffi.md`: the Native to JVM section is rewritten from the code (`NativeUtil.getNextBatch`, `Native.executePlan`, `prepare_output` and `move_to_spark`, then `ArrowImporter.importVector`). A new Offset Normalization section covers the `take` in `prepare_output` from #2051 and the gaps tracked in #6288. The Buffer Alignment section is replaced by a note on where alignment now happens, and the stale "copy only to unpack dictionaries" ownership row is fixed. - `memory_management.md` and the `review-comet-ffi-pr` and `review-comet-pr` skills are updated to match. There is one observable difference, in an error path. When `get_next` fails, the stock reader `unwrap()`s the stream's last error. arrow-java leaves that error empty only when it cannot build the message, for example under OOM. In that case the task still fails, but with a caught panic message rather than `get_next returned 5`. ## How are these changes tested? - `realigns_under_aligned_decimal128` passes through the stock reader. Building the same under-aligned struct as a typed array without going through `from_ffi_and_data_type` panics with "Memory pointer is not aligned with the specified scalar type", so the test does depend on arrow's realignment. - The full native `cargo test` on Linux, plus `cargo fmt --check` and `cargo clippy --all-targets --workspace -- -D warnings` on Rust 1.98.1. - The JVM build with scalastyle, `spotless:check`, and the Lint Java scalafix check on the Spark 3.5 profile. - JVM suites: `CometArrowStreamSuite`, `CometExecIteratorLifecycleSuite`, `CometNativeSuite`, the `SparkToColumnar` and dictionary tests in `CometExecSuite`, and `CometInMemoryCacheSuite`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
