sunchao commented on code in PR #57443:
URL: https://github.com/apache/spark/pull/57443#discussion_r3661734288
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala:
##########
@@ -242,6 +242,18 @@ trait LeafNode extends LogicalPlan with
LeafLike[LogicalPlan] {
throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3114")
}
+/** A materialized leaf whose output can safely be scanned again to build a
runtime filter. */
+private[sql] trait LeafNodeWithAccurateStats extends LeafNode {
+ /** Whether the current materialized output has complete, accurate
statistics. */
+ def statsAvailable: Boolean
+
+ /** Whether scanning the materialized output again returns the same rows. */
+ def isOutputRepeatable: Boolean
+
+ /** Whether the original plan contains a predicate that is likely to be
selective. */
+ def hasSelectivePredicate: Boolean
+}
Review Comment:
Addressed in `e783e16ff6a`. I renamed the trait to `MaterializedLeafNode`
and expanded its class-level documentation to state that `InjectRuntimeFilter`
uses this contract to decide whether another scan is safe and profitable. The
`statsAvailable` documentation now explicitly requires durable materialized
storage for another scan and explains why memory-only caches are excluded even
when their current statistics are exact.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala:
##########
@@ -392,11 +448,106 @@ case class CachedRDDBuilder(
}
}.persist(storageLevel)
cached.setName(cachedName)
+ isCachedRDDRepeatable = hasStrictFileSourceReads &&
+ cached.outputDeterministicLevel != DeterministicLevel.INDETERMINATE &&
+ InMemoryRelation.hasRepeatablePhysicalPlan(cachedPlan)
cached
}
}
-object InMemoryRelation {
+object InMemoryRelation extends PredicateHelper {
+
+ private val trustedFileFormatClasses: Set[Class[_ <: FileFormat]] = Set(
+ classOf[BinaryFileFormat],
+ classOf[CSVFileFormat],
+ classOf[JsonFileFormat],
+ classOf[OrcFileFormat],
+ classOf[ParquetFileFormat],
+ classOf[TextFileFormat])
+
+ private val trustedExternalFileFormatNames = Set(
+ "org.apache.spark.sql.avro.AvroFileFormat",
+ "org.apache.spark.sql.hive.orc.OrcFileFormat")
+
+ private def hasSafeExpressions(plan: QueryPlan[_]): Boolean = {
+ plan.expressions.forall { expression =>
+ !expression.exists {
+ case _: AesEncrypt | _: NonSQLExpression | _: UserDefinedExpression =>
true
+ case value => !value.deterministic ||
value.containsPattern(CURRENT_LIKE) ||
+
!value.getClass.getName.startsWith("org.apache.spark.sql.catalyst.expressions.")
Review Comment:
Good point—the package check is an intentional namespace trust boundary, not
an independent proof of repeatability. For expressions whose class is in the
Catalyst namespace, this code relies on the documented
`Expression.deterministic` contract: an expression must return the same result
for fixed child inputs, and expressions using mutable state or implicit inputs
must report themselves as non-deterministic. It separately rejects
`CURRENT_LIKE`, `NonSQLExpression`, and user-defined expressions.
`AesEncrypt` contains a known pre-existing exception when the IV is omitted
(for example, default GCM): evaluation generates a fresh random IV even though
the wrapper inherits `deterministic` from its children. This change rejects
`AesEncrypt` in the analyzed plan before optimization replaces it with
`StaticInvoke`; the optimized `StaticInvoke` is also rejected through
`NonSQLExpression`. Commit `e783e16ff6a` adds a source comment clarifying this
trust boundary.
I did a targeted scan of current built-ins involving randomness, current
time, `TaskContext`, reflection, or runtime replacement and did not find a
second current expression that passes these checks while producing
non-repeatable output, though that is not an absolute proof against future
additions. An exact-class allowlist would duplicate the evolving Catalyst
expression set and disable safe expressions by default, so I kept
`deterministic` as the core contract. Do you know of another concrete current
case we should cover here?
--
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]
