[SPARK-52404][SQL] Give a separate error class to failures in BindReferences

ala · hvanhovell · commit 1dfcd8f010ad · 2025-06-10T09:29:56.000-04:00
### What changes were proposed in this pull request? We add a new error class to cover a distinct portion of the `INTERNAL_ERROR` class, that is, a situation where an attribute reference in the query plan cannot be bound. The error massage, once hardcoded in `BoundAttribute.scala` is now moved the the JSON file. ### Why are the changes needed? The error messages should be stored in the JSON files, and not hard-coded into the source. Failures in `BoundAttribute.scala` are a distinct and don't share much with the other `INTERNAL_ERROR` failures, yet they are currently bundled under the same error class. ### Does this PR introduce _any_ user-facing change? Yes. Change of error class. ### How was this patch tested? New test is added. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#51066 from ala/attr_not_found. Authored-by: ala <ala.luszczak@gmail.com> Signed-off-by: Herman van Hovell <herman@databricks.com>
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
@@ -2227,6 +2227,12 @@
     ],
     "sqlState" : "XX000"
   },
+  "INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND" : {
+    "message" : [
+      "Could not find <missingAttr> in <inputAttrs>."
+    ],
+    "sqlState" : "XX000"
+  },
   "INTERNAL_ERROR_BROADCAST" : {
     "message" : [
       "<message>"
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala
@@ -77,8 +77,7 @@ object BindReferences extends Logging {
         if (allowFailures) {
           a
         } else {
-          throw SparkException.internalError(
-            s"Couldn't find $a in ${input.attrs.mkString("[", ",", "]")}")
+          throw attributeNotFoundException(a, input.attrs)
         }
       } else {
         BoundReference(ordinal, a.dataType, input(ordinal).nullable)
@@ -94,4 +93,18 @@ object BindReferences extends Logging {
       input: AttributeSeq): Seq[A] = {
     expressions.map(BindReferences.bindReference(_, input))
   }
+
+  /**
+   * A helper function to produce an exception when binding a reference fails.
+   * Public for testing.
+   */
+  def attributeNotFoundException(
+      missingAttr: AttributeReference,
+      inputAttrs: Seq[Attribute]): SparkException =
+    new SparkException(
+      errorClass = "INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND",
+      messageParameters = Map(
+        "missingAttr" -> missingAttr.toString,
+        "inputAttrs" -> inputAttrs.mkString("[", ",", "]")),
+      cause = null)
 }
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/BindReferencesSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/BindReferencesSuite.scala
@@ -0,0 +1,43 @@
+/*
+ * 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.expressions
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.types.IntegerType
+
+class BindReferencesSuite extends SparkFunSuite{
+  test("attribute not found exception") {
+    val missing = AttributeReference("missing", IntegerType)(qualifier = Seq.empty)
+    val input1 = AttributeReference("one", IntegerType)(qualifier = Seq.empty)
+    val input2 = AttributeReference("two", IntegerType)(qualifier = Seq.empty)
+    val input3 = AttributeReference("three", IntegerType)(qualifier = Seq.empty)
+
+    val e = BindReferences.attributeNotFoundException(
+        missingAttr = missing,
+        inputAttrs = Seq(input1, input2, input3))
+
+    checkError(e,
+      condition = "INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND",
+      sqlState = "XX000",
+      parameters = Map(
+        "missingAttr" -> s"$missing",
+        "inputAttrs" -> s"[$input1,$input2,$input3]"
+      )
+    )
+  }
+}
