[SPARK-55024][SQL][FOLLOWUP] Delay namespace length check to v1 identifier creation

### What changes were proposed in this pull request?

This is a followup to #53788 which moved the namespace length check from individual command handlers to the `CatalogAndIdentifier` extractor. That approach is too aggressive: users can extend the session catalog via `CatalogExtension` and support multi-part namespaces through v2 APIs. The check should only happen when we actually create v1 identifiers like `TableIdentifier`, not at the shared name-resolution layer.

### Changes

- **Remove the namespace check from `CatalogAndIdentifier.unapply`** so it remains a pure name-resolution mechanism, preserving `CatalogExtension` flexibility.
- **Tighten `CatalogV2Implicits.IdentifierHelper.asTableIdentifier` and `asFunctionIdentifier`** to require exactly one namespace part and throw `REQUIRES_SINGLE_PART_NAMESPACE` (instead of the less precise `IDENTIFIER_TOO_MANY_NAME_PARTS`). This centralizes the validation.
- **Remove the now-redundant `V2SessionCatalog.TableIdentifierHelper`** and use the unified `CatalogV2Implicits` conversion everywhere.
- **Simplify `ResolveSessionCatalog` extractors** (`ResolvedV1Identifier`, `ResolvedIdentifierInSessionCatalog`, `ResolvedViewIdentifier`) to delegate to `ident.asTableIdentifier`.
- **Fix `SparkSqlParser` temp view creation** to check name length before calling `asTableIdentifier`, so the user always sees `notAllowedToAddDBPrefixForTempViewError` instead of a generic error.

### Why are the changes needed?

`CatalogExtension` allows users to extend the built-in session catalog and potentially support multi-part namespaces for v2 operations. The early check in `CatalogAndIdentifier` would block such extensions. The namespace length should only be validated when we actually need to create v1 identifiers (e.g. `TableIdentifier`), which inherently require a single-part namespace.

Additionally, this PR unifies the scattered namespace validation into a single point (`CatalogV2Implicits.IdentifierHelper`), reducing code duplication and ensuring consistent `REQUIRES_SINGLE_PART_NAMESPACE` errors.

### Does this PR introduce _any_ user-facing change?

Yes. Multi-part namespace identifiers now flow through `CatalogAndIdentifier` without error, allowing `CatalogExtension` implementations to handle them. The error is only thrown when the session catalog actually needs to convert to a v1 identifier.

### How was this patch tested?

Existing tests updated:
- `LookupCatalogSuite` — removed the early-rejection test, added multi-part namespace cases back.
- `V2SessionCatalogSuite` — updated to expect `REQUIRES_SINGLE_PART_NAMESPACE`.
- `DataSourceV2SQLSuiteV1Filter` and `DataSourceV2FunctionSuite` — all 313 tests pass.

### Was this patch authored or co-authored using generative AI tooling?

Yes.

Made with [Cursor](https://cursor.com)

Closes #54247 from cloud-fan/follow.

Authored-by: Wenchen Fan <wenchen@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: cloud-fan

common/utils/src/main/resources/error/error-conditions.json

@@ -5366,7 +5366,7 @@
   },
   "REQUIRES_SINGLE_PART_NAMESPACE" : {
     "message" : [
-      "<sessionCatalog> requires a single-part namespace, but got <namespace>."
+      "<sessionCatalog> requires a single-part namespace, but got identifier <identifier>."
     ],
     "sqlState" : "42K05"
   },

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Implicits.scala

@@ -166,9 +166,9 @@ private[sql] object CatalogV2Implicits {
     def asMultipartIdentifier: Seq[String] = (ident.namespace :+ ident.name).toImmutableArraySeq
 
     def asTableIdentifier: TableIdentifier = ident.namespace match {
-      case ns if ns.isEmpty => TableIdentifier(ident.name)
       case Array(dbName) => TableIdentifier(ident.name, Some(dbName))
-      case _ => throw QueryCompilationErrors.identifierTooManyNamePartsError(original)
+      case _ =>
+        throw QueryCompilationErrors.requiresSinglePartNamespaceError(asMultipartIdentifier)
     }
 
     /**
@@ -192,9 +192,9 @@ private[sql] object CatalogV2Implicits {
     }
 
     def asFunctionIdentifier: FunctionIdentifier = ident.namespace() match {
-      case ns if ns.isEmpty => FunctionIdentifier(ident.name())
       case Array(dbName) => FunctionIdentifier(ident.name(), Some(dbName))
-      case _ => throw QueryCompilationErrors.identifierTooManyNamePartsError(original)
+      case _ =>
+        throw QueryCompilationErrors.requiresSinglePartNamespaceError(asMultipartIdentifier)
     }
 
     def toQualifiedNameParts(catalog: CatalogPlugin): Seq[String] = {

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/LookupCatalog.scala

@@ -19,9 +19,7 @@ package org.apache.spark.sql.connector.catalog
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
-import org.apache.spark.util.ArrayImplicits._
 
 /**
  * A trait to encapsulate catalog lookup function and helpful extractors.
@@ -109,29 +107,24 @@ private[sql] trait LookupCatalog extends Logging {
 
     def unapply(nameParts: Seq[String]): Option[(CatalogPlugin, Identifier)] = {
       assert(nameParts.nonEmpty)
-      val (catalog, ident) = if (nameParts.length == 1) {
-        (currentCatalog, Identifier.of(catalogManager.currentNamespace, nameParts.head))
+      if (nameParts.length == 1) {
+        Some((currentCatalog, Identifier.of(catalogManager.currentNamespace, nameParts.head)))
       } else if (nameParts.head.equalsIgnoreCase(globalTempDB)) {
         // Conceptually global temp views are in a special reserved catalog. However, the v2 catalog
         // API does not support view yet, and we have to use v1 commands to deal with global temp
         // views. To simplify the implementation, we put global temp views in a special namespace
         // in the session catalog. The special namespace has higher priority during name resolution.
         // For example, if the name of a custom catalog is the same with `GLOBAL_TEMP_DATABASE`,
         // this custom catalog can't be accessed.
-        (catalogManager.v2SessionCatalog, nameParts.asIdentifier)
+        Some((catalogManager.v2SessionCatalog, nameParts.asIdentifier))
       } else {
         try {
-          (catalogManager.catalog(nameParts.head), nameParts.tail.asIdentifier)
+          Some((catalogManager.catalog(nameParts.head), nameParts.tail.asIdentifier))
         } catch {
           case _: CatalogNotFoundException =>
-            (currentCatalog, nameParts.asIdentifier)
+            Some((currentCatalog, nameParts.asIdentifier))
         }
       }
-      if (CatalogV2Util.isSessionCatalog(catalog) && ident.namespace().length != 1) {
-        throw QueryCompilationErrors.requiresSinglePartNamespaceError(
-          ident.namespace().toImmutableArraySeq)
-      }
-      Some((catalog, ident))
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -1542,12 +1542,12 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
     new TableAlreadyExistsException(ident.asMultipartIdentifier)
   }
 
-  def requiresSinglePartNamespaceError(namespace: Seq[String]): Throwable = {
+  def requiresSinglePartNamespaceError(identifier: Seq[String]): Throwable = {
     new AnalysisException(
       errorClass = "REQUIRES_SINGLE_PART_NAMESPACE",
       messageParameters = Map(
         "sessionCatalog" -> CatalogManager.SESSION_CATALOG_NAME,
-        "namespace" -> toSQLId(namespace)))
+        "identifier" -> toSQLId(identifier)))
   }
 
   def namespaceAlreadyExistsError(namespace: Array[String]): Throwable = {

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/LookupCatalogSuite.scala

@@ -57,15 +57,15 @@ class LookupCatalogSuite extends SparkFunSuite with LookupCatalog with Inside {
 
   test("catalog and identifier") {
     Seq(
-      // Session catalog with single-part namespace
       ("tbl", sessionCatalog, Seq("default"), "tbl"),
       ("db.tbl", sessionCatalog, Seq("db"), "tbl"),
       (s"$globalTempDB.tbl", sessionCatalog, Seq(globalTempDB), "tbl"),
+      (s"$globalTempDB.ns1.ns2.tbl", sessionCatalog, Seq(globalTempDB, "ns1", "ns2"), "tbl"),
+      ("ns1.ns2.tbl", sessionCatalog, Seq("ns1", "ns2"), "tbl"),
       ("`db.tbl`", sessionCatalog, Seq("default"), "db.tbl"),
       ("parquet.`file:/tmp/db.tbl`", sessionCatalog, Seq("parquet"), "file:/tmp/db.tbl"),
       ("`org.apache.spark.sql.json`.`s3://buck/tmp/abc.json`", sessionCatalog,
         Seq("org.apache.spark.sql.json"), "s3://buck/tmp/abc.json"),
-      // Non-session catalogs (no namespace restriction)
       ("prod.func", catalogs("prod"), Seq.empty, "func"),
       ("prod.db.tbl", catalogs("prod"), Seq("db"), "tbl"),
       ("test.db.tbl", catalogs("test"), Seq("db"), "tbl"),
@@ -79,21 +79,6 @@ class LookupCatalogSuite extends SparkFunSuite with LookupCatalog with Inside {
     }
   }
 
-  test("session catalog requires single-part namespace") {
-    // Multi-part namespaces are not allowed for session catalog
-    Seq(
-      "ns1.ns2.tbl",                  // two-part namespace
-      s"$globalTempDB.ns1.ns2.tbl"    // three-part namespace
-    ).foreach { sql =>
-      val e = intercept[org.apache.spark.sql.AnalysisException] {
-        parseMultipartIdentifier(sql) match {
-          case CatalogAndIdentifier(_, _) =>
-        }
-      }
-      assert(e.getCondition === "REQUIRES_SINGLE_PART_NAMESPACE")
-    }
-  }
-
   test("table identifier") {
     Seq(
       ("tbl", "tbl", None),

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -37,7 +37,6 @@ import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils
 import org.apache.spark.sql.internal.{HiveSerDe, SQLConf}
 import org.apache.spark.sql.internal.connector.V1Function
 import org.apache.spark.sql.types.{DataType, MetadataBuilder, StringType, StructField, StructType}
-import org.apache.spark.util.ArrayImplicits._
 import org.apache.spark.util.SparkStringUtils
 
 /**
@@ -727,10 +726,10 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
     def unapply(resolved: LogicalPlan): Option[TableIdentifier] = resolved match {
       case ResolvedPersistentView(catalog, ident, _) =>
         assert(isSessionCatalog(catalog))
-        assert(ident.namespace().length == 1)
-        Some(TableIdentifier(ident.name, Some(ident.namespace.head), Some(catalog.name)))
+        Some(ident.asTableIdentifier.copy(catalog = Some(catalog.name)))
 
-      case ResolvedTempView(ident, _) => Some(ident.asTableIdentifier)
+      case ResolvedTempView(ident, _) =>
+        Some(TableIdentifier(ident.name(), ident.namespace().headOption))
 
       case _ => None
     }
@@ -763,24 +762,16 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
   object ResolvedV1Identifier {
     def unapply(resolved: LogicalPlan): Option[TableIdentifier] = resolved match {
       case ResolvedIdentifier(catalog, ident) if supportsV1Command(catalog) =>
-        if (ident.namespace().length != 1) {
-          throw QueryCompilationErrors
-            .requiresSinglePartNamespaceError(ident.namespace().toImmutableArraySeq)
-        }
-        Some(TableIdentifier(ident.name, Some(ident.namespace.head), Some(catalog.name)))
+        Some(ident.asTableIdentifier.copy(catalog = Some(catalog.name)))
       case _ => None
     }
   }
 
   // Use this object to help match commands that do not have a v2 implementation.
-  object ResolvedIdentifierInSessionCatalog{
+  object ResolvedIdentifierInSessionCatalog {
     def unapply(resolved: LogicalPlan): Option[TableIdentifier] = resolved match {
       case ResolvedIdentifier(catalog, ident) if isSessionCatalog(catalog) =>
-        if (ident.namespace().length != 1) {
-          throw QueryCompilationErrors
-            .requiresSinglePartNamespaceError(ident.namespace().toImmutableArraySeq)
-        }
-        Some(TableIdentifier(ident.name, Some(ident.namespace.head), Some(catalog.name)))
+        Some(ident.asTableIdentifier.copy(catalog = Some(catalog.name)))
       case _ => None
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -698,12 +698,12 @@ class SparkSqlAstBuilder extends AstBuilder {
       }
 
       withIdentClause(ctx.identifierReference(), Seq(qPlan), (ident, otherPlans) => {
-        val tableIdentifier = ident.asTableIdentifier
-        if (tableIdentifier.database.isDefined) {
+        if (ident.length > 1) {
           // Temporary view names should NOT contain database prefix like "database.table"
           throw QueryParsingErrors
-            .notAllowedToAddDBPrefixForTempViewError(tableIdentifier.nameParts, ctx)
+            .notAllowedToAddDBPrefixForTempViewError(ident, ctx)
         }
+        val tableIdentifier = TableIdentifier(ident.head)
 
         CreateViewCommand(
           tableIdentifier,

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala

@@ -24,7 +24,7 @@ import scala.collection.mutable
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.SparkUnsupportedOperationException
-import org.apache.spark.sql.catalyst.{FunctionIdentifier, QualifiedTableName, SQLConfHelper, TableIdentifier}
+import org.apache.spark.sql.catalyst.{QualifiedTableName, SQLConfHelper}
 import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogStorageFormat, CatalogTable, CatalogTableType, CatalogUtils, ClusterBySpec, SessionCatalog}
 import org.apache.spark.sql.catalyst.util.TypeUtils._
@@ -45,6 +45,7 @@ import org.apache.spark.util.Utils
  */
 class V2SessionCatalog(catalog: SessionCatalog)
   extends TableCatalog with FunctionCatalog with SupportsNamespaces with SQLConfHelper {
+  import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
   import V2SessionCatalog._
 
   override val defaultNamespace: Array[String] = Array(conf.defaultDatabase)
@@ -367,26 +368,6 @@ class V2SessionCatalog(catalog: SessionCatalog)
     catalog.renameTable(oldIdent.asTableIdentifier, newIdent.asTableIdentifier)
   }
 
-  implicit class TableIdentifierHelper(ident: Identifier) {
-    def asTableIdentifier: TableIdentifier = {
-      ident.namespace match {
-        case Array(db) =>
-          TableIdentifier(ident.name, Some(db))
-        case other =>
-          throw QueryCompilationErrors.requiresSinglePartNamespaceError(other.toImmutableArraySeq)
-      }
-    }
-
-    def asFunctionIdentifier: FunctionIdentifier = {
-      ident.namespace match {
-        case Array(db) =>
-          FunctionIdentifier(ident.name, Some(db))
-        case other =>
-          throw QueryCompilationErrors.requiresSinglePartNamespaceError(other.toImmutableArraySeq)
-      }
-    }
-  }
-
   override def namespaceExists(namespace: Array[String]): Boolean = namespace match {
     case Array(db) =>
       catalog.databaseExists(db)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala

@@ -40,7 +40,6 @@ class JDBCTableCatalog extends TableCatalog
   with FunctionCatalog
   with DataTypeErrorsBase
   with Logging {
-  import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 
   private var catalogName: String = null
   private var options: JDBCOptions = _
@@ -434,7 +433,7 @@ class JDBCTableCatalog extends TableCatalog
 
   override def loadFunction(ident: Identifier): UnboundFunction = {
     if (ident.namespace().nonEmpty) {
-      throw QueryCompilationErrors.noSuchFunctionError(ident.asFunctionIdentifier)
+      throw new NoSuchFunctionException(ident)
     }
     functions.get(ident.name()) match {
       case Some(func) =>

sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause-legacy.sql.out

@@ -883,7 +883,7 @@ org.apache.spark.sql.AnalysisException
   "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
   "sqlState" : "42K05",
   "messageParameters" : {
-    "namespace" : "`a`.`b`",
+    "identifier" : "`a`.`b`.`c`",
     "sessionCatalog" : "spark_catalog"
   }
 }
@@ -897,7 +897,7 @@ org.apache.spark.sql.AnalysisException
   "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
   "sqlState" : "42K05",
   "messageParameters" : {
-    "namespace" : "`a`.`b`",
+    "identifier" : "`a`.`b`.`c`",
     "sessionCatalog" : "spark_catalog"
   }
 }
@@ -911,7 +911,7 @@ org.apache.spark.sql.AnalysisException
   "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
   "sqlState" : "42K05",
   "messageParameters" : {
-    "namespace" : "`a`.`b`",
+    "identifier" : "`a`.`b`.`c`",
     "sessionCatalog" : "spark_catalog"
   }
 }
@@ -925,7 +925,7 @@ org.apache.spark.sql.AnalysisException
   "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
   "sqlState" : "42K05",
   "messageParameters" : {
-    "namespace" : "`a`.`b`",
+    "identifier" : "`a`.`b`.`c`",
     "sessionCatalog" : "spark_catalog"
   }
 }
@@ -939,7 +939,7 @@ org.apache.spark.sql.AnalysisException
   "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
   "sqlState" : "42K05",
   "messageParameters" : {
-    "namespace" : "`a`.`b`.`c`",
+    "identifier" : "`a`.`b`.`c`.`d`",
     "sessionCatalog" : "spark_catalog"
   }
 }
@@ -975,7 +975,7 @@ org.apache.spark.sql.AnalysisException
   "errorClass" : "REQUIRES_SINGLE_PART_NAMESPACE",
   "sqlState" : "42K05",
   "messageParameters" : {
-    "namespace" : "`a`.`b`.`c`",
+    "identifier" : "`a`.`b`.`c`.`d`",
     "sessionCatalog" : "spark_catalog"
   },
   "queryContext" : [ {