This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 10480976c0 [spark] Improve unresolved expression errors in MERGE INTO
(#9068)
10480976c0 is described below
commit 10480976c04d0f64e7eb215b8ab87028ff2e199c
Author: Zouxxyy <[email protected]>
AuthorDate: Thu Aug 6 22:14:14 2026 +0800
[spark] Improve unresolved expression errors in MERGE INTO (#9068)
---
.../analysis/expressions/ExpressionHelper.scala | 13 +++++----
.../paimon/spark/sql/MergeIntoTableTestBase.scala | 33 +++++++++++++++++++++-
2 files changed, 39 insertions(+), 7 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/expressions/ExpressionHelper.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/expressions/ExpressionHelper.scala
index 82dcb594a2..7a3fce7fe0 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/expressions/ExpressionHelper.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/expressions/ExpressionHelper.scala
@@ -124,12 +124,13 @@ trait ExpressionHelperBase extends PredicateHelper {
} else {
val newPlan = FakeLogicalPlan(Seq(expr), plan.children)
spark.sessionState.analyzer.execute(newPlan) match {
- case FakeLogicalPlan(resolvedExpr, _) =>
- resolvedExpr.foreach {
- expr =>
- if (!expr.resolved) {
- throw new RuntimeException(s"cannot resolve ${expr.sql} from
$plan")
- }
+ case analyzedPlan @ FakeLogicalPlan(resolvedExpr, _) =>
+ resolvedExpr.find(expr => !expr.resolved).foreach {
+ unresolvedExpr =>
+ // Let Spark report the concrete unresolved column or type error
with its structured
+ // AnalysisException before falling back to Paimon's generic
error.
+ spark.sessionState.analyzer.checkAnalysis(analyzedPlan)
+ throw new RuntimeException(s"cannot resolve
${unresolvedExpr.sql} from $plan")
}
resolvedExpr.head
case _ =>
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoTableTestBase.scala
index bc0924ac17..05804740a2 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoTableTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoTableTestBase.scala
@@ -21,7 +21,7 @@ package org.apache.paimon.spark.sql
import org.apache.paimon.Snapshot
import org.apache.paimon.spark.{PaimonAppendTable, PaimonPrimaryKeyTable,
PaimonSparkTestBase, PaimonTableTest}
-import org.apache.spark.sql.Row
+import org.apache.spark.sql.{AnalysisException, Row}
import java.util.UUID
import java.util.concurrent.Executors
@@ -270,6 +270,37 @@ abstract class MergeIntoTableTestBase extends
PaimonSparkTestBase with PaimonTab
}
}
+ test("Paimon MergeInto: unresolved target column reports Spark analysis
error") {
+ withTable("source", "target") {
+ Seq((1, "new-oneid", "new-customer"))
+ .toDF("id", "oneid", "customer_id")
+ .createOrReplaceTempView("source")
+
+ createTable("target", "id INT, oneid STRING", Seq("id"))
+
+ val error = intercept[AnalysisException] {
+ spark.sql(s"""
+ |MERGE INTO target
+ |USING source
+ |ON target.id = source.id
+ |WHEN MATCHED AND (
+ | NOT (target.oneid <=> source.oneid)
+ | OR NOT (target.customer_id <=> source.customer_id)
+ |) THEN UPDATE SET oneid = source.oneid
+ |""".stripMargin)
+ }
+
+ // Spark versions use different error classes for an unresolved column.
+ Option(error.getErrorClass).foreach {
+ errorClass =>
+ assert(Set("MISSING_COLUMN",
"UNRESOLVED_COLUMN.WITH_SUGGESTION").contains(errorClass))
+ }
+ val normalizedMessage = error.getMessage.replace("`", "")
+ assert(normalizedMessage.contains("target.customer_id"))
+ assert(normalizedMessage.contains("source.customer_id"))
+ }
+ }
+
test(s"Paimon MergeInto: conditional insert") {
withTable("source", "target") {
