s-maling-celonis commented on code in PR #58856:
URL: https://github.com/apache/spark/pull/58856#discussion_r4105880653
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/Resolver.scala:
##########
@@ -78,6 +78,8 @@ import org.apache.spark.sql.errors.{QueryCompilationErrors,
QueryErrorsBase}
* relation operators in [[MetadataResolver]].
* @param externalRelationResolution An optional [[RelationResolution]] with
to override the default
* one. The default is constructed using
[[Resolver.createRelationResolution]].
+ * @param hintResolutionRules Rules of the fixed-point Analyzer's "Hints"
batch that are injected
Review Comment:
Addressed in b6e1d99b190: the scaladoc now describes this as the
independently overridable single-pass sequence, with the
`injectHintResolutionRule` list only as the default.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/HintResolutionRunner.scala:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.ResolveHints
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper,
LogicalPlan, UnresolvedWith}
+import org.apache.spark.sql.catalyst.rules.{Rule, RuleExecutor}
+import org.apache.spark.sql.catalyst.trees.CurrentOrigin
+import org.apache.spark.sql.catalyst.trees.TreePattern.{PLAN_EXPRESSION,
UNRESOLVED_WITH}
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Utility wrapper on top of [[RuleExecutor]], used to apply the hint
resolution rules injected
+ * through `SparkSessionExtensions.injectHintResolutionRule` on the unresolved
plan.
+ *
+ * The fixed-point Analyzer runs these rules in its "Hints" batch, which
completes before the
+ * "Resolution" batch, so the rules always observe an unresolved plan. The
single-pass Resolver
+ * has to run them before the [[MetadataResolver]] pre-pass for the same
reason: the pre-pass
+ * resolves relation metadata eagerly (which, for path-based relations, lists
files), therefore
+ * rules that rewrite [[UnresolvedRelation]]s have to run ahead of it.
+ *
+ * The "Disable Hints" batch precedes the "Hints" batch, exactly like in the
fixed-point Analyzer,
+ * so that `spark.sql.optimizer.disableHints` keeps hiding the hints from the
injected rules.
+ *
+ * The rules are applied on the main plan, the CTE definitions and the
subquery plans. The
+ * fixed-point Analyzer covers those same subtrees, through two different
mechanisms:
+ * - `CTESubstitution` merges the CTE definitions into the main plan in the
"Substitution" batch,
+ * which runs before the "Hints" batch, so the rules observe them as part
of the main plan;
+ * - subquery plans are resolved by `ResolveSubquery` re-entering the whole
Analyzer through
+ * `executeSameContext`, which replays every batch, "Hints" included.
+ *
+ * View bodies are covered for the same reason: the fixed-point Analyzer
resolves them by
+ * re-entering the Analyzer (`ViewResolution.resolve` is called with
`executeSameContext`), and the
+ * single-pass Resolver calls [[Resolver.lookupMetadataAndResolve]], and
therefore this runner, once
+ * per unresolved [[View]].
+ */
+class HintResolutionRunner(hintResolutionRules: Seq[Rule[LogicalPlan]])
+ extends ResolverMetricTracker
+ with SQLConfHelper {
+
+ private val hintResolver = new RuleExecutor[LogicalPlan] {
+ override def batches: Seq[Batch] =
+ Seq(
+ Batch("Disable Hints", Once, new ResolveHints.DisableHints),
+ Batch(
+ "Hints",
+ FixedPoint(
+ conf.analyzerMaxIterations,
+ errorOnExceed = true,
+ maxIterationsSetting = SQLConf.ANALYZER_MAX_ITERATIONS.key
+ ),
+ hintResolutionRules: _*
+ )
+ )
+ }
+
+ /**
+ * Applies the hint resolution rules by first recursing into the CTE
definitions and the
+ * subqueries and then applying the rules on the entire plan. No rules are
injected by default,
+ * in which case the plan is returned as is, without constructing a
[[RuleExecutor]]. That also
+ * skips the "Disable Hints" batch, which only matters for plans that still
carry an
+ * [[UnresolvedHint]] - a shape the single-pass Resolver does not support in
the first place.
+ *
+ * The recursion needs [[AnalysisHelper.allowInvokingTransformsInAnalyzer]]
to rewrite the inner
+ * plans, so the rules are invoked within its scope, like the rewrite rules
in [[PlanRewriter]].
+ */
Review Comment:
Addressed in b6e1d99b190: `hintResolver` is now a `lazy val`, so the
empty-rule path never constructs the `RuleExecutor`. The scaladoc matches that.
##########
sql/core/src/test/scala/org/apache/spark/sql/SparkSessionExtensionSuite.scala:
##########
@@ -250,6 +271,199 @@ class SparkSessionExtensionSuite extends PlanTest with
AdaptiveSparkPlanHelper {
}
}
+ test("SPARK-59574: inject custom hint rule with single-pass resolver") {
+ withSession(Seq(_.injectHintResolutionRule(MyHintRule))) { session =>
+ session.conf.set(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key,
"true")
+
session.conf.set(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key,
"false")
+ assert(
+
session.range(1).hint("CONVERT_TO_EMPTY").logicalPlan.isInstanceOf[LocalRelation],
+ "plan is expected to be a local relation"
+ )
+ }
+ }
+
+ /**
+ * [[Deduplicate]] is not among the operators the single-pass Resolver can
resolve, and it is not
+ * listed in `ExplicitlyUnsupportedResolverFeature.OPERATORS` either, so a
rule that introduces
+ * it makes the single-pass resolution fail rather than fall back.
+ */
+ case class DeduplicatingRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ case _: Deduplicate => plan
+ case _ => Deduplicate(Nil, plan)
+ }
+ }
+
+ test("SPARK-59574: rule introducing an operator unsupported by single-pass
fails resolution") {
+ withSession(Seq(_.injectHintResolutionRule(DeduplicatingRule))) { session
=>
+ session.conf.set(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key,
"true")
+
session.conf.set(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key,
"false")
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ session.sql("SELECT id FROM VALUES (1), (2) AS
t(id)").queryExecution.analyzed
+ },
+ condition = "UNSUPPORTED_SINGLE_PASS_ANALYZER_FEATURE",
+ parameters = Map(
+ "feature" ->
+ ("class org.apache.spark.sql.catalyst.plans.logical.Deduplicate
operator resolution")
+ )
+ )
+ }
+
+ // In dual-run the failure is reported against the fixed-point result
rather than falling
+ // back, because the plan only becomes unsupported after the guard has
already accepted it.
+ // Before this change the same query failed too, with
LOGICAL_PLAN_COMPARISON_MISMATCH, as
+ // only the fixed-point half applied the rule.
+ withSession(Seq(_.injectHintResolutionRule(DeduplicatingRule))) { session
=>
+ session.conf.set(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key,
"false")
+
session.conf.set(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key,
"true")
+
+ val error = intercept[AnalysisException] {
+ session.sql("SELECT id FROM VALUES (1), (2) AS
t(id)").queryExecution.analyzed
+ }
+ assert(
+ error.getCondition ===
+ "HYBRID_ANALYZER_EXCEPTION.SINGLE_PASS_FAILED_FIXED_POINT_SUCCEEDED"
+ )
+ }
+ }
+
+ test("SPARK-59574: injected hint rules are bound by
spark.sql.analyzer.maxIterations") {
+ Seq(false, true).foreach { singlePass =>
+ withSession(Seq(_.injectHintResolutionRule(_ => AddLimitAlways))) {
session =>
+ session.conf.set(SQLConf.ANALYZER_MAX_ITERATIONS.key, "5")
+ session.conf.set(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key,
singlePass.toString)
+
session.conf.set(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key,
"false")
+
+ // The rule never reaches a fixed point, so both analyzers are
expected to stop at
+ // spark.sql.analyzer.maxIterations and to point at that config in the
error message.
+ val error = intercept[RuntimeException] {
+ session.range(1).logicalPlan
+ }
+ assert(
+ error.getMessage.contains(SQLConf.ANALYZER_MAX_ITERATIONS.key),
Review Comment:
Addressed in b6e1d99b190: the test now asserts `Max iterations (5) reached
for batch Hints` as well as the config key, for both analyzer modes.
--
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]
