parthchandra commented on code in PR #1746:
URL: https://github.com/apache/datafusion-comet/pull/1746#discussion_r2096122217
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -2376,12 +2375,26 @@ object QueryPlanSerde extends Logging with
CometExprShim {
val cond = exprToProto(condition, child.output)
if (cond.isDefined && childOp.nonEmpty) {
+ // We need to determine whether to use DataFusion's FilterExec or
Comet's
+ // FilterExec. The difference is that DataFusion's implementation
will sometimes pass
+ // batches through whereas the Comet implementation guarantees that
a copy is always
+ // made, which is critical when using `native_comet` scans due to
buffer re-use
+
+ // TODO this could be optimized more to stop walking the tree on
hitting
+ // certain operators such as join or aggregate which will copy
batches
+ def containsNativeCometScan(plan: SparkPlan): Boolean = {
+ plan match {
+ case w: CometScanWrapper =>
containsNativeCometScan(w.originalPlan)
+ case scan: CometScanExec => scan.scanImpl ==
CometConf.SCAN_NATIVE_COMET
Review Comment:
Do we need `CometConf.COMET_NATIVE_SCAN_IMPL.get() ==
CometConf.SCAN_NATIVE_ICEBERG_COMPAT` as well?
`NATIVE_ICEBERG_COMPAT` does not have any buffer reuse so I think this
change is correct, but would just like to confirm that it does not affect
Filter.
--
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]
