andygrove commented on code in PR #4053:
URL: https://github.com/apache/datafusion-comet/pull/4053#discussion_r3139438864
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala:
##########
@@ -208,8 +215,41 @@ case class CometNativeScanExec(
case _ =>
}
}
- // Extract common data from nativeOp
- val commonBytes = nativeOp.getNativeScan.getCommon.toByteArray
+ // Resolve scalar subqueries in dataFilters and push to the native Parquet
reader.
+ // supportedDataFilters excludes PlanExpression at planning time
(unresolved), so these
+ // aren't in the serialized native plan yet. We resolve them here and
append to the
+ // NativeScanCommon protobuf. Same approach as
FileSourceScanLike.pushedDownFilters
+ // (DataSourceScanExec.scala), which resolves ScalarSubquery -> Literal at
execution time.
+ val commonBytes = {
+ val base = nativeOp.getNativeScan.getCommon
+ val scalarSubqueryFilters = dataFilters
+
.filter(_.exists(_.isInstanceOf[org.apache.spark.sql.execution.ScalarSubquery]))
+ scalarSubqueryFilters.foreach { f =>
+ f.foreach {
+ case s: org.apache.spark.sql.execution.ScalarSubquery =>
+ s.updateResult()
+ case _ =>
+ }
+ }
+ val resolvedFilters = scalarSubqueryFilters
+ .map(_.transform { case s:
org.apache.spark.sql.execution.ScalarSubquery =>
+ Literal.create(s.eval(null), s.dataType)
+ })
+ if (resolvedFilters.nonEmpty) {
+ import org.apache.comet.serde.QueryPlanSerde.exprToProto
Review Comment:
can this be a file level import?
--
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]
