peterxcli commented on code in PR #5538:
URL: https://github.com/apache/datafusion-comet/pull/5538#discussion_r3999119823
##########
native/core/src/execution/operators/shuffle_scan.rs:
##########
@@ -47,11 +47,35 @@ use std::{
use super::scan::InputBatch;
-/// ShuffleScanExec reads compressed shuffle blocks from JVM via JNI and
decodes them natively.
-/// Unlike ScanExec which receives Arrow arrays via FFI, ShuffleScanExec
receives raw compressed
-/// bytes from CometShuffleBlockIterator and decodes them using
read_ipc_compressed().
+/// Identifies the JVM source using the shared direct-block protocol.
+#[derive(Debug, Clone, Copy)]
+enum BlockScanKind {
+ Shuffle,
+ Broadcast,
+}
+
+impl BlockScanKind {
+ fn execution_plan_name(self) -> &'static str {
+ match self {
+ Self::Shuffle => "ShuffleScanExec",
+ Self::Broadcast => "BroadcastScanExec",
+ }
+ }
+
+ fn input_name(self) -> &'static str {
+ match self {
+ Self::Shuffle => "shuffle",
+ Self::Broadcast => "broadcast",
+ }
+ }
+}
+
+/// Reads codec-prefixed compressed Arrow IPC blocks from JVM via JNI and
decodes them natively.
+/// Shuffle and broadcast scans share the same iterator protocol while
retaining distinct plan
+/// names for explain output and diagnostics.
#[derive(Debug, Clone)]
pub struct ShuffleScanExec {
Review Comment:
```suggestion
pub struct BlockScanExec {
```
rename to `BlockScanExec` as it now reads codec-prefixed IPC blocks from
either shuffle or broadcast.
##########
native/core/src/execution/operators/shuffle_scan.rs:
##########
@@ -79,6 +103,33 @@ impl ShuffleScanExec {
exec_context_id: i64,
input_source: Option<Arc<Global<JObject<'static>>>>,
data_types: Vec<DataType>,
+ ) -> Result<Self, CometError> {
+ Self::new_with_kind(
+ exec_context_id,
+ input_source,
+ data_types,
+ BlockScanKind::Shuffle,
+ )
+ }
+
+ pub fn new_broadcast(
+ exec_context_id: i64,
+ input_source: Option<Arc<Global<JObject<'static>>>>,
+ data_types: Vec<DataType>,
+ ) -> Result<Self, CometError> {
+ Self::new_with_kind(
+ exec_context_id,
+ input_source,
+ data_types,
+ BlockScanKind::Broadcast,
+ )
+ }
+
+ fn new_with_kind(
+ exec_context_id: i64,
+ input_source: Option<Arc<Global<JObject<'static>>>>,
+ data_types: Vec<DataType>,
+ kind: BlockScanKind,
Review Comment:
```suggestion
pub fn new(
exec_context_id: i64,
input_source: Option<Arc<Global<JObject<'static>>>>,
data_types: Vec<DataType>,
kind: BlockScanKind,
```
remove the selected constructor and only keep new\_with_kind and rename
new_with_kind to new?
##########
native/core/src/execution/operators/shuffle_scan.rs:
##########
Review Comment:
```suggestion
pub fn new_with_shuffle(
```
--
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]
