ajsquared commented on code in PR #7343:
URL: https://github.com/apache/datafusion-comet/pull/7343#discussion_r4504797757
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -122,9 +126,89 @@ case class CometScanRule(session: SparkSession)
transformV2Scan(fullPlan, scanExec)
}
- plan.transform {
- case scan if isSupportedScanNode(scan) => transformScan(scan)
+ def transform(node: SparkPlan): SparkPlan = sparkScans.get(node.id) match {
+ // This guard precedes the contrib hook so Delta keeps its Spark reader
too.
+ case Some(scan) => withFallbackReason(scan, makeDateScanReason)
+ case None =>
+ val children = node.children.map(transform)
+ val rebuilt =
+ if (children == node.children) node
+ else {
+ // Re-plan native parents when a restored reader changes their
serialized input.
+ val restored = node match {
+ case _: CometSinkPlaceHolder => children.head
+ case native: CometExec => native.sparkFallback(children)
+ case exchange: CometBroadcastExchangeExec =>
+ exchange.originalPlan.withNewChildren(children)
+ case exchange: CometShuffleExchangeExec =>
+ exchange.originalPlan.withNewChildren(children)
+ case _ => node.withNewChildren(children)
+ }
+ restored.copyTagsFrom(node)
+ restored
+ }
+ if (isSupportedScanNode(rebuilt)) transformScan(rebuilt) else rebuilt
}
+ transform(plan)
+ }
+
+ /**
+ * Spark omits partition-column and nondeterministic predicates from a V1
scan's data filters.
+ * Keep its original reader when those filters can throw: native metadata
pruning need not visit
+ * the same rows as Spark's per-file statistics and dictionary filters.
+ */
+ private def scansRequiringSparkForMakeDate(plan: SparkPlan): Map[Int,
FileSourceScanExec] = {
+ val scans = mutable.Map.empty[Int, FileSourceScanExec]
+
+ def inspectScan(node: SparkPlan, scan: FileSourceScanExec, filters:
Seq[Expression]): Unit = {
+ // This pass owns only its own reason; an earlier parent context must
not prevent conversion.
+ scan.getTagValue(CometExplainInfo.FALLBACK_REASONS).foreach { reasons =>
+ val remaining = reasons - makeDateScanReason
+ if (remaining.isEmpty)
scan.unsetTagValue(CometExplainInfo.FALLBACK_REASONS)
+ else scan.setTagValue(CometExplainInfo.FALLBACK_REASONS, remaining)
+ }
+ if ((filters ++ scan.dataFilters).exists(_.exists {
+ case date: MakeDate =>
!MakeDateEvaluation.canEvaluateMakeDateEagerly(date)
+ case _ => false
+ })) {
+ scans(node.id) = scan
+ }
+ }
+
+ def throughProject(
+ expressions: Seq[NamedExpression],
+ filters: Seq[Expression]): Seq[Expression] = {
+ val aliases = AttributeMap(expressions.collect { case alias: Alias =>
+ alias.toAttribute -> alias.child
+ })
+ filters.map(_.transform { case attribute: Attribute =>
Review Comment:
[P2] A filter over stacked projections can take exponential time to plan
even when the query contains no `make_date`. For example, start with a numeric
Parquet column `x`, repeat `selectExpr("sin(x) + cos(x) + rand() AS x")` 20
times, then filter on `x > 0`. The nondeterministic projections stay separate.
`throughProject` substitutes each alias at both references to `x`, so
`inspectScan` traverses over a million occurrences of the original column
before reading a row. Trace alias dependencies by expression ID and reuse each
result when checking for a throwing `MakeDate`, so repeated references do not
expand the filter tree.
--
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]
