andygrove commented on code in PR #1209:
URL: https://github.com/apache/datafusion-comet/pull/1209#discussion_r1900473667
##########
spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala:
##########
@@ -788,68 +788,82 @@ class CometSparkSessionExtensions
}
// Native shuffle for Comet operators
- case s: ShuffleExchangeExec
- if isCometShuffleEnabled(conf) &&
+ case s: ShuffleExchangeExec =>
+ val nativeShuffle: Option[SparkPlan] =
+ if (isCometShuffleEnabled(conf) &&
isCometNativeShuffleMode(conf) &&
- QueryPlanSerde.supportPartitioning(s.child.output,
s.outputPartitioning)._1 =>
- logInfo("Comet extension enabled for Native Shuffle")
-
- val newOp = transform1(s)
- newOp match {
- case Some(nativeOp) =>
- // Switch to use Decimal128 regardless of precision, since Arrow
native execution
- // doesn't support Decimal32 and Decimal64 yet.
- conf.setConfString(CometConf.COMET_USE_DECIMAL_128.key, "true")
- val cometOp = CometShuffleExchangeExec(s, shuffleType =
CometNativeShuffle)
- CometSinkPlaceHolder(nativeOp, s, cometOp)
- case None =>
- s
- }
+ QueryPlanSerde.supportPartitioning(s.child.output,
s.outputPartitioning)._1) {
+ val newOp = transform1(s)
+ newOp match {
+ case Some(nativeOp) =>
+ // Switch to use Decimal128 regardless of precision, since
Arrow native execution
+ // doesn't support Decimal32 and Decimal64 yet.
+ conf.setConfString(CometConf.COMET_USE_DECIMAL_128.key,
"true")
+ val cometOp = CometShuffleExchangeExec(s, shuffleType =
CometNativeShuffle)
+ Some(CometSinkPlaceHolder(nativeOp, s, cometOp))
+ case None =>
+ None
+ }
+ } else {
+ None
+ }
- // Columnar shuffle for regular Spark operators (not Comet) and Comet
operators
- // (if configured).
- // If the child of ShuffleExchangeExec is also a ShuffleExchangeExec,
we should not
- // convert it to CometColumnarShuffle,
- case s: ShuffleExchangeExec
- if isCometShuffleEnabled(conf) && isCometJVMShuffleMode(conf) &&
Review Comment:
This part is triggered, but in the following code, `transform1(s)` returns
`None` so we return the original shuffle.
```scala
val newOp = transform1(s)
newOp match {
case Some(nativeOp) =>
// Switch to use Decimal128 regardless of precision, since
Arrow native execution
// doesn't support Decimal32 and Decimal64 yet.
conf.setConfString(CometConf.COMET_USE_DECIMAL_128.key, "true")
val cometOp = CometShuffleExchangeExec(s, shuffleType =
CometNativeShuffle)
CometSinkPlaceHolder(nativeOp, s, cometOp)
case None =>
s
}
```
--
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]
