cloud-fan commented on code in PR #58856:
URL: https://github.com/apache/spark/pull/58856#discussion_r4105697767
##########
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:
**Nit (P3):** This argument is not necessarily the fixed-point Analyzer's
injected sequence: HybridAnalyzer passes `singlePassHintResolutionRules`, and
the propagation test deliberately overrides it with a sequence distinct from
`hintResolutionRules`. Could this describe the single-pass rule sequence and
mention the injected fixed-point rules only as its default?
##########
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:
**Non-blocking (P2):** This assertion does not verify that the configured
value `5` was used: `AddLimitAlways` fails for every positive limit, and
`RuleExecutor` can still report this config key with a hard-coded or default
bound. Please add a value-sensitive signal, such as counting rule invocations
or asserting an exact limit-bearing error, for both analyzer modes.
##########
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:
**Nit (P3):** `hintResolver` is an eager `val`, so the `RuleExecutor` has
already been constructed before this method reaches the empty-rule check. The
fast path skips execution, not construction; please narrow this guarantee
accordingly.
--
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]
