goutamadwant opened a new pull request, #5538: URL: https://github.com/apache/datafusion-comet/pull/5538
## Which issue does this PR close? Closes #3762. ## Rationale for this change Broadcast exchange input currently crosses the JVM twice before native execution: the JVM decompresses and deserializes Arrow IPC, then exports the resulting arrays through Arrow FFI. Comet can read the codec-prefixed IPC blocks directly and avoid that work, as it already does for direct shuffle reads. ## What changes are included in this PR? - Add the experimental `spark.comet.exec.broadcast.directRead.enabled` configuration, disabled by default. - Serialize direct-read broadcast batches as codec-prefixed LZ4 Arrow IPC while keeping them decodable by JVM consumers. - Add a `BroadcastScan` protobuf operator and native `BroadcastScanExec` planning path that reuse the existing direct block protocol. - Wire broadcast blocks through AQE and exchange-reuse boundaries without creating Arrow stream inputs. - Keep direct-read mode in broadcast exchange equality and canonicalization so exchanges using incompatible serialization modes cannot be reused together. - Preserve independently serialized blocks instead of coalescing a native broadcast beyond the JNI protocol's per-block `Integer.MAX_VALUE` limit. - Preserve the legacy broadcast serialization and Arrow FFI path when the configuration is disabled. - Close any already resolved streams and block iterators when input-slot resolution or native plan construction fails. ## How are these changes tested? - `make core` - `cargo check -p datafusion-comet` - `cargo fmt --all -- --check` - `cargo clippy -p datafusion-comet --all-targets -- -D warnings` - `cargo test -p datafusion-comet test_broadcast_scan_has_distinct_plan_name` - `./mvnw test -Dtest=none -Dsuites=org.apache.spark.sql.comet.util.UtilsSuite -DfailIfNoTests=false` - `./mvnw test -Dtest=none -Dsuites=org.apache.comet.exec.CometNativeShuffleSuite -DfailIfNoTests=false` - `./mvnw test -Dtest=none -Dsuites=org.apache.comet.exec.CometExecSuite -DfailIfNoTests=false` The regression coverage checks legacy serialization with the feature disabled, native codec framing with the feature enabled, JVM decoding compatibility, AQE enabled and disabled, exchange identity across serialization modes, the direct block-size boundary without a large allocation, native-plan failure cleanup, empty build-side handling, and Spark-equivalent join answers. -- 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]
