You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2024/07/18 22:21:13 UTC
(spark) branch branch-3.5 updated: [SPARK-48921][SQL][3.5] ScalaUDF encoders in subquery should be resolved for MergeInto
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new e2479159ee85 [SPARK-48921][SQL][3.5] ScalaUDF encoders in subquery should be resolved for MergeInto
e2479159ee85 is described below
commit e2479159ee85ffdbadf5caf2fbf7b5e28add5f1f
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Thu Jul 18 15:21:01 2024 -0700
[SPARK-48921][SQL][3.5] ScalaUDF encoders in subquery should be resolved for MergeInto
### What changes were proposed in this pull request?
We got a customer issue that a `MergeInto` query on Iceberg table works earlier but cannot work after upgrading to Spark 3.4.
The error looks like
```
Caused by: org.apache.spark.SparkRuntimeException: Error while decoding: org.apache.spark.sql.catalyst.analysis.UnresolvedException: Invalid call to nullable on unresolved object
upcast(getcolumnbyordinal(0, StringType), StringType, - root class: java.lang.String).toString.
```
The source table of `MergeInto` uses `ScalaUDF`. The error happens when Spark invokes the deserializer of input encoder of the `ScalaUDF` and the deserializer is not resolved yet.
The encoders of ScalaUDF are resolved by the rule `ResolveEncodersInUDF` which will be applied at the end of analysis phase.
During rewriting `MergeInto` to `ReplaceData` query, Spark creates an `Exists` subquery and `ScalaUDF` is part of the plan of the subquery. Note that the `ScalaUDF` is already resolved by the analyzer.
Then, in `ResolveSubquery` rule which resolves the subquery, it will resolve the subquery plan if it is not resolved yet. Because the subquery containing `ScalaUDF` is resolved, the rule skips it so `ResolveEncodersInUDF` won't be applied on it. So the analyzed `ReplaceData` query contains a `ScalaUDF` with encoders unresolved that cause the error.
This patch modifies `ResolveSubquery` so it will resolve subquery plan if it is not analyzed to make sure subquery plan is fully analyzed.
This patch moves `ResolveEncodersInUDF` rule before rewriting `MergeInto` to make sure the `ScalaUDF` in the subquery plan is fully analyzed.
### Why are the changes needed?
Fixing production query error.
### Does this PR introduce _any_ user-facing change?
Yes, fixing user-facing issue.
### How was this patch tested?
Manually test with `MergeInto` query and add an unit test.
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #47406 from viirya/fix_subquery_resolve_3.5.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Dongjoon Hyun <dh...@apple.com>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 11 ++-
.../analysis/ResolveEncodersInUDFSuite.scala | 95 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 3 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index e63621d6a236..4388740a409d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -325,9 +325,6 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
new ResolveIdentifierClause(earlyBatches) ::
ResolveUnion ::
ResolveRowLevelCommandAssignments ::
- RewriteDeleteFromTable ::
- RewriteUpdateTable ::
- RewriteMergeIntoTable ::
typeCoercionRules ++
Seq(
ResolveWithCTE,
@@ -351,6 +348,14 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
UpdateAttributeNullability),
Batch("UDF", Once,
ResolveEncodersInUDF),
+ // The rewrite rules might move resolved query plan into subquery. Once the resolved plan
+ // contains ScalaUDF, their encoders won't be resolved if `ResolveEncodersInUDF` is not
+ // applied before the rewrite rules. So we need to apply `ResolveEncodersInUDF` before the
+ // rewrite rules.
+ Batch("DML rewrite", fixedPoint,
+ RewriteDeleteFromTable,
+ RewriteUpdateTable,
+ RewriteMergeIntoTable),
Batch("Subquery", Once,
UpdateOuterReferences),
Batch("Cleanup", fixedPoint,
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveEncodersInUDFSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveEncodersInUDFSuite.scala
new file mode 100644
index 000000000000..c7391ad9a430
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveEncodersInUDFSuite.scala
@@ -0,0 +1,95 @@
+/*
+ * 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
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Exists, ScalaUDF}
+import org.apache.spark.sql.catalyst.plans.logical.{Assignment, Filter, MergeIntoTable, ReplaceData, UpdateAction}
+import org.apache.spark.sql.catalyst.trees.TreePattern
+import org.apache.spark.sql.connector.catalog.InMemoryRowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+class ResolveEncodersInUDFSuite extends AnalysisTest {
+ test("SPARK-48921: ScalaUDF encoders in subquery should be resolved for MergeInto") {
+ val table = new InMemoryRowLevelOperationTable("table",
+ StructType(StructField("a", IntegerType) ::
+ StructField("b", DoubleType) ::
+ StructField("c", StringType) :: Nil),
+ Array.empty,
+ new java.util.HashMap[String, String]()
+ )
+ val relation = DataSourceV2Relation(table,
+ Seq(AttributeReference("a", IntegerType)(),
+ AttributeReference("b", DoubleType)(),
+ AttributeReference("c", StringType)()),
+ None,
+ None,
+ CaseInsensitiveStringMap.empty()
+ )
+
+
+ val string = relation.output(2)
+ val udf = ScalaUDF((_: String) => "x", StringType, string :: Nil,
+ Option(ExpressionEncoder[String]()) :: Nil)
+
+ val mergeIntoSource =
+ relation
+ .where($"c" === udf)
+ .select($"a", $"b")
+ .limit(1)
+ val cond = mergeIntoSource.output(0) == relation.output(0) &&
+ mergeIntoSource.output(1) == relation.output(1)
+
+ val mergePlan = MergeIntoTable(
+ relation,
+ mergeIntoSource,
+ cond,
+ Seq(UpdateAction(None,
+ Seq(Assignment(relation.output(0), relation.output(0)),
+ Assignment(relation.output(1), relation.output(1)),
+ Assignment(relation.output(2), relation.output(2))))),
+ Seq.empty,
+ Seq.empty)
+
+ val replaceData = mergePlan.analyze.asInstanceOf[ReplaceData]
+
+ val existsPlans = replaceData.groupFilterCondition.map(_.collect {
+ case e: Exists =>
+ e.plan.collect {
+ case f: Filter if f.containsPattern(TreePattern.SCALA_UDF) => f
+ }
+ }.flatten)
+
+ assert(existsPlans.isDefined)
+
+ val udfs = existsPlans.get.map(_.expressions.flatMap(e => e.collect {
+ case s: ScalaUDF =>
+ assert(s.inputEncoders.nonEmpty)
+ val encoder = s.inputEncoders.head
+ assert(encoder.isDefined)
+ assert(encoder.get.objDeserializer.resolved)
+
+ s
+ })).flatten
+ assert(udfs.size == 1)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org