[SPARK-54130][SQL] Add detailed error messages for catalog assertion failures

Author: ganeshas-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InMemoryCatalog.scala

@@ -192,7 +192,10 @@ class InMemoryCatalog(
   override def createTable(
       tableDefinition: CatalogTable,
       ignoreIfExists: Boolean): Unit = synchronized {
-    assert(tableDefinition.identifier.database.isDefined)
+    assert(tableDefinition.identifier.database.isDefined,
+      "Table identifier " + tableDefinition.identifier.quotedString +
+      " is missing database name. " +
+      "Cannot create table without a database specified.")
     val db = tableDefinition.identifier.database.get
     requireDbExists(db)
     val table = tableDefinition.identifier.table
@@ -313,7 +316,10 @@ class InMemoryCatalog(
   }
 
   override def alterTable(tableDefinition: CatalogTable): Unit = synchronized {
-    assert(tableDefinition.identifier.database.isDefined)
+    assert(tableDefinition.identifier.database.isDefined,
+      "Table identifier " + tableDefinition.identifier.quotedString +
+      " is missing database name. " +
+      "Cannot alter table without a database specified.")
     val db = tableDefinition.identifier.database.get
     requireTableExists(db, tableDefinition.identifier.table)
     val updatedProperties = tableDefinition.properties.filter(kv => kv._1 != "comment")

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SQLFunction.scala

@@ -61,8 +61,15 @@ case class SQLFunction(
     owner: Option[String] = None,
     createTimeMs: Long = System.currentTimeMillis) extends UserDefinedFunction {
 
-  assert(exprText.nonEmpty || queryText.nonEmpty)
-  assert((isTableFunc && returnType.isRight) || (!isTableFunc && returnType.isLeft))
+  assert(exprText.nonEmpty || queryText.nonEmpty,
+    "SQL function '" + name + "' is missing function body. " +
+    "Either exprText or queryText must be defined. " +
+    "Found: exprText=" + exprText + ", queryText=" + queryText + ".")
+  assert((isTableFunc && returnType.isRight) || (!isTableFunc && returnType.isLeft),
+    "SQL function '" + name + "' has mismatched function type and return type. " +
+    "isTableFunc=" + isTableFunc + ", returnType.isRight=" + returnType.isRight + ", " +
+    "returnType.isLeft=" + returnType.isLeft + ". " +
+    "Table functions require Right[StructType] and scalar functions require Left[DataType].")
 
   import SQLFunction._
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -1663,7 +1663,9 @@ class SessionCatalog(
         .putString("__funcInputAlias", "true")
         .build()
     }
-    assert(!function.isTableFunc)
+    assert(!function.isTableFunc,
+      "Function '" + function.name + "' is a table function. " +
+      "Use makeSQLTableFunctionPlan() instead of makeSQLFunctionPlan().")
     val funcName = function.name.funcName
 
     // Use captured SQL configs when parsing a SQL function.
@@ -1674,7 +1676,10 @@ class SessionCatalog(
       val inputParam = function.inputParam
       val returnType = function.getScalarFuncReturnType
       val (expression, query) = function.getExpressionAndQuery(parser, isTableFunc = false)
-      assert(expression.isDefined || query.isDefined)
+      assert(expression.isDefined || query.isDefined,
+        "SQL function '" + function.name + "' could not be parsed. " +
+        "Neither expression nor query could be extracted from function body. " +
+        "exprText=" + function.exprText + ", queryText=" + function.queryText + ".")
 
       // Check function arguments
       val paramSize = inputParam.map(_.size).getOrElse(0)
@@ -1763,12 +1768,17 @@ class SessionCatalog(
       function: SQLFunction,
       input: Seq[Expression],
       outputAttrs: Seq[Attribute]): LogicalPlan = {
-    assert(function.isTableFunc)
+    assert(function.isTableFunc,
+      "Function '" + function.name + "' is a scalar function. " +
+      "Use makeSQLFunctionPlan() instead of makeSQLTableFunctionPlan().")
     val funcName = function.name.funcName
     val inputParam = function.inputParam
     val returnParam = function.getTableFuncReturnCols
     val (_, query) = function.getExpressionAndQuery(parser, isTableFunc = true)
-    assert(query.isDefined)
+    assert(query.isDefined,
+      "SQL table function '" + function.name + "' could not be parsed. " +
+      "Query could not be extracted from function body. " +
+      "queryText=" + function.queryText + ".")
 
     // Check function arguments
     val paramSize = inputParam.map(_.size).getOrElse(0)
@@ -1807,7 +1817,12 @@ class SessionCatalog(
       query.get
     }
 
-    assert(returnParam.length == outputAttrs.length)
+    assert(returnParam.length == outputAttrs.length,
+      "SQL table function '" + function.name + "' has mismatched return columns. " +
+      "Expected " + outputAttrs.length + " output attributes but found " +
+      returnParam.length + " return parameters. " +
+      "Return parameters: [" + returnParam.fieldNames.mkString(", ") + "], " +
+      "Output attributes: [" + outputAttrs.map(_.name).mkString(", ") + "].")
     val output = returnParam.fields.zipWithIndex.map { case (param, i) =>
       // Since we cannot get the output of a unresolved logical plan, we need
       // to reference the output column of the lateral join by its position.
@@ -2390,7 +2405,9 @@ class SessionCatalog(
     requireTableNotExists(newName)
     val oldTable = getTableMetadata(oldName)
     if (oldTable.tableType == CatalogTableType.MANAGED) {
-      assert(oldName.database.nonEmpty)
+      assert(oldName.database.nonEmpty,
+        "Table identifier " + oldName.quotedString + " is missing database name. " +
+        "Managed tables must have a database defined.")
       val databaseLocation =
         externalCatalog.getDatabase(oldName.database.get).locationUri
       val newTableLocation = new Path(new Path(databaseLocation), format(newName.table))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala

@@ -1070,7 +1070,9 @@ case class UnresolvedCatalogRelation(
     tableMeta: CatalogTable,
     options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty(),
     override val isStreaming: Boolean = false) extends UnresolvedLeafNode {
-  assert(tableMeta.identifier.database.isDefined)
+  assert(tableMeta.identifier.database.isDefined,
+    "Table identifier " + tableMeta.identifier.quotedString + " is missing database name. " +
+    "UnresolvedCatalogRelation requires a fully qualified table identifier with database.")
 }
 
 /**
@@ -1097,7 +1099,9 @@ case class HiveTableRelation(
     tableStats: Option[Statistics] = None,
     @transient prunedPartitions: Option[Seq[CatalogTablePartition]] = None)
   extends LeafNode with MultiInstanceRelation with NormalizeableRelation {
-  assert(tableMeta.identifier.database.isDefined)
+  assert(tableMeta.identifier.database.isDefined,
+    "Table identifier " + tableMeta.identifier.quotedString + " is missing database name. " +
+    "HiveTableRelation requires a fully qualified table identifier with database.")
   assert(DataTypeUtils.sameType(tableMeta.partitionSchema, partitionCols.toStructType))
   assert(DataTypeUtils.sameType(tableMeta.dataSchema, dataCols.toStructType))
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala

@@ -34,6 +34,7 @@ import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAM
 import org.apache.spark.sql.connector.catalog.SupportsNamespaces.PROP_OWNER
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.sql.types._
+import org.apache.spark.util.Utils
 
 class InMemorySessionCatalogSuite extends SessionCatalogSuite {
   protected val utils = new CatalogTestUtils {
@@ -2091,4 +2092,179 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
       assert(exception.getMessage.matches(expectedPattern))
     }
   }
+
+  test("UnresolvedCatalogRelation requires database in identifier") {
+    val catalog = new SessionCatalog(newBasicCatalog())
+    val db = "test_db"
+    catalog.createDatabase(newDb(db), ignoreIfExists = false)
+
+    // Create a table with database
+    val validTable = CatalogTable(
+      identifier = TableIdentifier("test_table", Some(db)),
+      tableType = CatalogTableType.MANAGED,
+      storage = CatalogStorageFormat.empty,
+      schema = new StructType().add("id", IntegerType)
+    )
+    catalog.createTable(validTable, ignoreIfExists = false)
+
+    // Try to create UnresolvedCatalogRelation without database - should fail
+    val tableMetaWithoutDb = validTable.copy(
+      identifier = TableIdentifier("test_table", None)
+    )
+
+    val exception = intercept[AssertionError] {
+      UnresolvedCatalogRelation(tableMetaWithoutDb)
+    }
+
+    val expectedMessage =
+      "assertion failed: Table identifier `test_table` is missing database name. " +
+      "UnresolvedCatalogRelation requires a fully qualified table identifier with database."
+    assert(exception.getMessage === expectedMessage)
+  }
+
+  test("HiveTableRelation requires database in identifier") {
+    val catalog = new SessionCatalog(newBasicCatalog())
+    val db = "test_db"
+    catalog.createDatabase(newDb(db), ignoreIfExists = false)
+
+    // Create a table with database
+    val validTable = CatalogTable(
+      identifier = TableIdentifier("test_table", Some(db)),
+      tableType = CatalogTableType.MANAGED,
+      storage = CatalogStorageFormat.empty,
+      schema = new StructType()
+        .add("id", IntegerType)
+        .add("name", StringType)
+    )
+
+    // Try to create HiveTableRelation without database - should fail
+    val tableMetaWithoutDb = validTable.copy(
+      identifier = TableIdentifier("test_table", None)
+    )
+
+    val exception = intercept[AssertionError] {
+      HiveTableRelation(
+        tableMetaWithoutDb,
+        Seq(AttributeReference("id", IntegerType)()),
+        Seq.empty
+      )
+    }
+
+    val expectedMessage =
+      "assertion failed: Table identifier `test_table` is missing database name. " +
+      "HiveTableRelation requires a fully qualified table identifier with database."
+    assert(exception.getMessage === expectedMessage)
+  }
+
+  test("SQLFunction requires either exprText or queryText") {
+    // Test case 1: Neither exprText nor queryText provided
+    val exception1 = intercept[AssertionError] {
+      SQLFunction(
+        name = FunctionIdentifier("test_func"),
+        inputParam = None,
+        returnType = scala.util.Left(IntegerType),
+        exprText = None,
+        queryText = None,
+        comment = None,
+        deterministic = Some(true),
+        containsSQL = Some(false),
+        isTableFunc = false,
+        properties = Map.empty
+      )
+    }
+
+    val expectedMessage = "assertion failed: SQL function 'test_func' is missing function body. " +
+      "Either exprText or queryText must be defined. " +
+      "Found: exprText=None, queryText=None."
+    assert(exception1.getMessage === expectedMessage)
+  }
+
+  test("SQLFunction return type must match function type") {
+    // Test case: isTableFunc=true but returnType is Left (scalar type)
+    val exception = intercept[AssertionError] {
+      SQLFunction(
+        name = FunctionIdentifier("test_func"),
+        inputParam = None,
+        returnType = scala.util.Left(IntegerType),  // Scalar return type
+        exprText = Some("SELECT 1"),
+        queryText = None,
+        comment = None,
+        deterministic = Some(true),
+        containsSQL = Some(true),
+        isTableFunc = true,  // But marked as table function
+        properties = Map.empty
+      )
+    }
+
+    val expectedMessage =
+      "assertion failed: SQL function 'test_func' has mismatched function type " +
+      "and return type. " +
+      "isTableFunc=true, returnType.isRight=false, returnType.isLeft=true. " +
+      "Table functions require Right[StructType] and scalar functions require Left[DataType]."
+    assert(exception.getMessage === expectedMessage)
+  }
+
+  test("InMemoryCatalog.createTable requires database in identifier") {
+    val catalog = new InMemoryCatalog()
+    val db = "test_db"
+    val dbDefinition = CatalogDatabase(
+      name = db,
+      description = "test database",
+      locationUri = Utils.createTempDir().toURI,
+      properties = Map.empty
+    )
+    catalog.createDatabase(dbDefinition, ignoreIfExists = false)
+
+    // Try to create table without database - should fail
+    val tableWithoutDb = CatalogTable(
+      identifier = TableIdentifier("test_table", None),
+      tableType = CatalogTableType.MANAGED,
+      storage = CatalogStorageFormat.empty,
+      schema = new StructType().add("id", IntegerType)
+    )
+
+    val exception = intercept[AssertionError] {
+      catalog.createTable(tableWithoutDb, ignoreIfExists = false)
+    }
+
+    val expectedMessage =
+      "assertion failed: Table identifier `test_table` is missing database name. " +
+      "Cannot create table without a database specified."
+    assert(exception.getMessage === expectedMessage)
+  }
+
+  test("InMemoryCatalog.alterTable requires database in identifier") {
+    val catalog = new InMemoryCatalog()
+    val db = "test_db"
+    val dbDefinition = CatalogDatabase(
+      name = db,
+      description = "test database",
+      locationUri = Utils.createTempDir().toURI,
+      properties = Map.empty
+    )
+    catalog.createDatabase(dbDefinition, ignoreIfExists = false)
+
+    // First create a valid table
+    val validTable = CatalogTable(
+      identifier = TableIdentifier("test_table", Some(db)),
+      tableType = CatalogTableType.MANAGED,
+      storage = CatalogStorageFormat.empty,
+      schema = new StructType().add("id", IntegerType)
+    )
+    catalog.createTable(validTable, ignoreIfExists = false)
+
+    // Try to alter table with identifier without database - should fail
+    val tableWithoutDb = validTable.copy(
+      identifier = TableIdentifier("test_table", None)
+    )
+
+    val exception = intercept[AssertionError] {
+      catalog.alterTable(tableWithoutDb)
+    }
+
+    val expectedMessage =
+      "assertion failed: Table identifier `test_table` is missing database name. " +
+      "Cannot alter table without a database specified."
+    assert(exception.getMessage === expectedMessage)
+  }
 }