Abhishek-Mohan commented on code in PR #7151: URL: https://github.com/apache/datafusion-comet/pull/7151#discussion_r4385080429
########## docs/source/contributor-guide/plugin_overview.md: ########## @@ -127,10 +140,26 @@ In the native code there is a `PhysicalPlanner` struct (in `planner.rs`) which c Apache DataFusion `ExecutionPlan`. In some cases, Comet provides specialized physical operators and expressions to override the DataFusion versions to ensure compatibility with Apache Spark. -The leaf nodes in the physical plan are always `ScanExec`. Each JVM-sourced input is exported once as an +Not every expression in a Rust plan runs in Rust. When an expression has no Rust implementation, or its Rust +implementation does not match Spark for the given input, its serde can route it through the JVM codegen dispatcher +instead of falling back to Spark (see the +[Implementation legend](../user-guide/latest/expressions.md#implementation-legend)). The Rust plan then evaluates the +expression with `JvmScalarUdfExpr`, which passes each batch's input columns through JNI and the Arrow C Data Interface +to `CometUdfBridge`, which hands them to `CometScalaUDFCodegen` on the JVM. There, Spark's own generated code for the expression, compiled into a batch +kernel by `CometBatchKernelCodegen`, runs over the Arrow vectors and returns an Arrow column. Scala and Java UDFs take +the same path. The operator around the expression stays Rust-implemented, so this JVM code runs inside the Comet +pipeline. `spark.comet.exec.scalaUDF.codegen.enabled`, which is on by default, turns the dispatcher on. When it is off, +these expressions fall back to Spark. + +A leaf node in the physical plan that takes Arrow batches from the JVM is a `ScanExec`. Each such input is exported once as an [Arrow C Stream](https://arrow.apache.org/docs/format/CStreamInterface.html) (`org.apache.arrow.c.ArrowArrayStream`), and `ScanExec` pulls each input batch through a single C callback rather than making a JNI call per batch. The -input could be a Comet native Parquet scan, a Spark exchange, or another native plan. +input could be a broadcast, a shuffle read decoded through JNI and imported into the JVM, the output of +`CometSparkToColumnarExec`, or the output of an earlier Comet block. When direct shuffle read is enabled and the input +is a Comet shuffle whose output types Comet supports, the input is instead a `ShuffleScanExec`, which pulls compressed Review Comment: Fixed in 3c982c8cf. ########## docs/source/user-guide/latest/understanding-comet-plans.md: ########## @@ -94,6 +94,8 @@ They serve different purposes and produce output in different places. | `spark.comet.explain.codegen.enabled` | `ExtendedExplainInfo` / SQL UI | A single `[COMET-INFO: JVM codegen dispatcher: <name1>, <name2>, ...]` segment naming each expression on the operator that was routed through the JVM codegen dispatcher. | | `spark.comet.explain.format` | Spark SQL UI (Spark 4.0 and newer) | Annotated plan or fallback-reason list, depending on `verbose` (default) or `fallback` value. | | `spark.comet.explain.native.enabled` | Executor logs, per task | The DataFusion plan with metrics, useful for inspecting Rust execution. | +| `spark.comet.explain.planOnly.enabled` | Driver log | The Comet plan and coverage summary, prefixed `[Comet plan-only]`, while Spark runs the original plan. | +| `spark.comet.explain.rules` | Driver log | Each plan transformation that Comet's planner rules make. | Review Comment: Fixed in 3c982c8cf. ########## docs/source/contributor-guide/jvm_shuffle.md: ########## @@ -65,7 +65,7 @@ returns an `RDD[InternalRow]`. The rows flow directly to the JVM shuffle writers ### Comet Columnar Input -When the child plan is a Comet native operator (e.g., `CometHashAggregate`) but JVM shuffle is selected +When the child plan is a Comet operator (e.g., `CometHashAggregate`) but JVM shuffle is selected Review Comment: Fixed in 3c982c8cf. ########## docs/source/contributor-guide/native_shuffle.md: ########## @@ -25,33 +25,35 @@ see [JVM Shuffle](jvm_shuffle.md). ## Overview -Native shuffle takes columnar input directly from Comet native operators and performs partitioning, +Native shuffle takes Arrow batches directly from Comet operators and performs partitioning, encoding, and writing in native Rust code. This avoids the columnar-to-row-to-columnar conversion overhead that JVM shuffle incurs. ``` -Comet Native (columnar) → Native Shuffle → Arrow IPC → columnar +Comet operator (Arrow batches) → Native Shuffle → Arrow IPC → columnar ``` Compare this to JVM shuffle's data path: ``` -Comet Native (columnar) → ColumnarToRowExec → rows → JVM Shuffle → Arrow IPC → columnar +Comet operator (Arrow batches) → ColumnarToRowExec → rows → JVM Shuffle → Arrow IPC → columnar ``` When `RevertNativeForTransitionHeavyStages` restores the map stage to Spark execution, the -native exchange stays in place. Its input still needs Arrow-backed Comet vectors, even if the -restored Spark operator supports columnar output. The rule adds `CometSparkToColumnarExec` to -convert either Spark rows or Spark columnar batches to Arrow before the native shuffle consumes -them. Spark's `RowToColumnarExec` alone does not satisfy this input contract. +native shuffle exchange usually stays in place. Its input still needs Arrow-backed Comet vectors, +even if the restored Spark operator supports columnar output. The rule adds `CometSparkToColumnarExec` +to convert either Spark rows or Spark columnar batches to Arrow before the native shuffle consumes +them. Spark's `RowToColumnarExec` alone does not satisfy this input contract. The exception is a +native shuffle whose input was converted by `spark.comet.convert.shuffleInput.enabled`: the rule +switches it back to JVM columnar shuffle, which reads the restored stage's rows. ## When Native Shuffle is Used Native shuffle (`CometExchange`) is selected when all of the following conditions are met: -1. **Shuffle mode allows native**: `spark.comet.shuffle.mode` is `native` or `auto`. +1. **Shuffle mode allows native shuffle**: `spark.comet.shuffle.mode` is `native` or `auto`. -2. **Child plan is a Comet native operator**: The child must be a `CometPlan` that produces +2. **Child plan is a Comet operator**: The child must be a `CometPlan` that produces Review Comment: Fixed in 3c982c8cf. -- 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]
