[SPARK-55645][SQL][FOLLOWUP] Move serdeName to last parameter and filter empty strings

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

Two followup improvements to #54467 (SPARK-55645):

1. Move `serdeName` to the last parameter of `CatalogStorageFormat` with a default value of `None`, so that existing callers that construct `CatalogStorageFormat` positionally remain source-compatible without code changes.

2. Filter empty strings when reading `serdeName` from the Hive Metastore API — Hive returns `""` for tables without an explicit serde name, which should map to `None` rather than `Some("")`.

### Why are the changes needed?

1. Adding `serdeName` as a required positional parameter in the middle of the parameter list breaks source compatibility for all external callers (e.g., third-party connectors) that construct `CatalogStorageFormat` positionally. Moving it to the last position with a default value avoids this.

2. The Hive Metastore returns an empty string for `SerDeInfo.name` when no serde name is explicitly set. Without filtering, `Option("")` produces `Some("")` instead of the semantically correct `None`, which could cause unexpected behavior in downstream code that checks `serdeName.isDefined` or pattern-matches on it.

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

No.

### How was this patch tested?

Added a new test `serdeName should be None for tables without an explicit serde name` that verifies the empty string filtering. Existing tests cover the parameter reordering.

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

Yes

Closes #54860 from cloud-fan/SPARK-55645-followup.

Lead-authored-by: Wenchen Fan <wenchen@databricks.com>
Co-authored-by: Wenchen Fan <cloud0fan@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: cloud-fan

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

@@ -144,10 +144,10 @@ case class CatalogStorageFormat(
     locationUri: Option[URI],
     inputFormat: Option[String],
     outputFormat: Option[String],
-    serdeName: Option[String],
     serde: Option[String],
     compressed: Boolean,
-    properties: Map[String, String]) extends MetadataMapSupport {
+    properties: Map[String, String],
+    serdeName: Option[String] = None) extends MetadataMapSupport {
 
   override def toString: String = {
     toLinkedHashMap.map { case (key, value) =>
@@ -181,7 +181,7 @@ case class CatalogStorageFormat(
 object CatalogStorageFormat {
   /** Empty storage format for default values and copies. */
   val empty = CatalogStorageFormat(locationUri = None, inputFormat = None, outputFormat = None,
-    serdeName = None, serde = None, compressed = false, properties = Map.empty)
+    serde = None, compressed = false, properties = Map.empty)
 }
 
 /**
@@ -616,11 +616,11 @@ case class CatalogTable(
       inputFormat: Option[String] = storage.inputFormat,
       outputFormat: Option[String] = storage.outputFormat,
       compressed: Boolean = false,
-      serdeName: Option[String] = storage.serdeName,
       serde: Option[String] = storage.serde,
-      properties: Map[String, String] = storage.properties): CatalogTable = {
+      properties: Map[String, String] = storage.properties,
+      serdeName: Option[String] = storage.serdeName): CatalogTable = {
     copy(storage = CatalogStorageFormat(
-      locationUri, inputFormat, outputFormat, serdeName, serde, compressed, properties))
+      locationUri, inputFormat, outputFormat, serde, compressed, properties, serdeName))
   }
 
   def toJsonLinkedHashMap: mutable.LinkedHashMap[String, JValue] = {

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

@@ -247,7 +247,6 @@ class ExternalCatalogEventSuite extends SparkFunSuite {
       locationUri = Some(tableUri),
       inputFormat = Some("tableInputFormat"),
       outputFormat = Some("tableOutputFormat"),
-      serdeName = None,
       serde = None,
       compressed = false,
       properties = Map.empty)

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

@@ -911,7 +911,7 @@ abstract class ExternalCatalogSuite extends SparkFunSuite {
       tableType = CatalogTableType.EXTERNAL,
       storage = CatalogStorageFormat(
         Some(Utils.createTempDir().toURI),
-        None, None, None, None, false, Map.empty),
+        None, None, None, false, Map.empty),
       schema = new StructType().add("a", "int").add("b", "string"),
       provider = Some(defaultProvider)
     )
@@ -959,7 +959,7 @@ abstract class ExternalCatalogSuite extends SparkFunSuite {
       Map("partCol1" -> "7", "partCol2" -> "8"),
       CatalogStorageFormat(
         Some(tempPath.toURI),
-        None, None, None, None, false, Map.empty))
+        None, None, None, false, Map.empty))
     catalog.createPartitions("db1", "tbl", Seq(partWithExistingDir), ignoreIfExists = false)
 
     tempPath.delete()
@@ -968,7 +968,7 @@ abstract class ExternalCatalogSuite extends SparkFunSuite {
       Map("partCol1" -> "9", "partCol2" -> "10"),
       CatalogStorageFormat(
         Some(tempPath.toURI),
-        None, None, None, None, false, Map.empty))
+        None, None, None, false, Map.empty))
     catalog.createPartitions("db1", "tbl", Seq(partWithNonExistingDir), ignoreIfExists = false)
     assert(tempPath.exists())
   }
@@ -1030,7 +1030,6 @@ abstract class CatalogTestUtils {
     locationUri = None,
     inputFormat = Some(tableInputFormat),
     outputFormat = Some(tableOutputFormat),
-    serdeName = None,
     serde = None,
     compressed = false,
     properties = Map.empty)

sql/core/src/test/scala/org/apache/spark/sql/AlwaysPersistedConfigsSuite.scala

@@ -158,7 +158,7 @@ class AlwaysPersistedConfigsSuite extends QueryTest with SharedSparkSession {
     val catalogTable = new CatalogTable(
       identifier = TableIdentifier(testViewName),
       tableType = CatalogTableType.VIEW,
-      storage = CatalogStorageFormat(None, None, None, None, None, false, Map.empty),
+      storage = CatalogStorageFormat(None, None, None, None, false, Map.empty),
       schema = new StructType(),
       properties = Map.empty[String, String]
     )

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -317,7 +317,7 @@ trait DDLSuiteBase extends SQLTestUtils {
     spec: TablePartitionSpec,
     tableName: TableIdentifier): Unit = {
     val part = CatalogTablePartition(
-      spec, CatalogStorageFormat(None, None, None, None, None, false, Map()))
+      spec, CatalogStorageFormat(None, None, None, None, false, Map()))
     catalog.createPartitions(tableName, Seq(part), ignoreIfExists = false)
   }
 

sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala

@@ -944,7 +944,6 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
           locationUri = None,
           inputFormat = None,
           outputFormat = None,
-          serdeName = None,
           serde = None,
           compressed = false,
           properties = Map.empty),
@@ -973,7 +972,6 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
           locationUri = None,
           inputFormat = None,
           outputFormat = None,
-          serdeName = None,
           serde = None,
           compressed = false,
           properties = Map.empty),

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -177,23 +177,23 @@ private[hive] class HiveMetastoreCatalog(sparkSession: SparkSession) extends Log
       val options = storage.properties + (ParquetOptions.MERGE_SCHEMA ->
         SQLConf.get.getConf(HiveUtils.CONVERT_METASTORE_PARQUET_WITH_SCHEMA_MERGING).toString)
       storage.copy(
-        serdeName = None,
         serde = None,
-        properties = options
+        properties = options,
+        serdeName = None
       )
     } else {
       val options = storage.properties
       if (SQLConf.get.getConf(SQLConf.ORC_IMPLEMENTATION) == "native") {
         storage.copy(
-          serdeName = None,
           serde = None,
-          properties = options
+          properties = options,
+          serdeName = None
         )
       } else {
         storage.copy(
-          serdeName = None,
           serde = None,
-          properties = options
+          properties = options,
+          serdeName = None
         )
       }
     }

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -557,11 +557,11 @@ private[hive] class HiveClientImpl(
         outputFormat = Option(h.getTTable.getSd.getOutputFormat).orElse {
           Option(h.getStorageHandler).map(_.getOutputFormatClass.getName)
         },
-        serdeName = Option(h.getTTable.getSd.getSerdeInfo.getName),
         serde = Option(h.getSerializationLib),
         compressed = h.getTTable.getSd.isCompressed,
         properties = Option(h.getTTable.getSd.getSerdeInfo.getParameters)
-          .map(_.asScala.toMap).orNull
+          .map(_.asScala.toMap).orNull,
+        serdeName = Option(h.getTTable.getSd.getSerdeInfo.getName).filter(_.nonEmpty)
       ),
       // For EXTERNAL_TABLE, the table properties has a particular field "EXTERNAL". This is added
       // in the function toHiveTable.
@@ -1202,7 +1202,7 @@ private[hive] object HiveClientImpl extends Logging {
       hiveTable.getTTable.getSd.setLocation(loc)}
     table.storage.inputFormat.map(toInputFormat).foreach(hiveTable.setInputFormatClass)
     table.storage.outputFormat.map(toOutputFormat).foreach(hiveTable.setOutputFormatClass)
-    table.storage.serdeName.foreach(hiveTable.getSd.getSerdeInfo.setName)
+    table.storage.serdeName.foreach(hiveTable.getTTable.getSd.getSerdeInfo.setName)
     hiveTable.setSerializationLib(
       table.storage.serde.getOrElse("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
     table.storage.properties.foreach { case (k, v) => hiveTable.setSerdeParam(k, v) }
@@ -1283,11 +1283,11 @@ private[hive] object HiveClientImpl extends Logging {
         locationUri = Option(CatalogUtils.stringToURI(apiPartition.getSd.getLocation)),
         inputFormat = Option(apiPartition.getSd.getInputFormat),
         outputFormat = Option(apiPartition.getSd.getOutputFormat),
-        serdeName = Option(apiPartition.getSd.getSerdeInfo.getName),
         serde = Option(apiPartition.getSd.getSerdeInfo.getSerializationLib),
         compressed = apiPartition.getSd.isCompressed,
         properties = Option(apiPartition.getSd.getSerdeInfo.getParameters)
-          .map(_.asScala.toMap).orNull),
+          .map(_.asScala.toMap).orNull,
+        serdeName = Option(apiPartition.getSd.getSerdeInfo.getName).filter(_.nonEmpty)),
       createTime = apiPartition.getCreateTime.toLong * 1000,
       lastAccessTime = apiPartition.getLastAccessTime.toLong * 1000,
       parameters = properties,

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetastoreCatalogSuite.scala

@@ -513,4 +513,16 @@ class DataSourceWithHiveMetastoreCatalogSuite
       assert(tableWithSerdeName.storage.serdeName === Some("testSerdeName"))
     }
   }
+
+  test("SPARK-55645: serdeName should be None for tables without an explicit serde name") {
+    withTable("t") {
+      sql("CREATE TABLE t (d1 DECIMAL(10,3), d2 STRING) STORED AS TEXTFILE")
+
+      val hiveTable =
+        sessionState.catalog.getTableMetadata(TableIdentifier("t", Some("default")))
+      // Hive Metastore returns "" for tables without an explicit serde name.
+      // This should be mapped to None, not Some("").
+      assert(hiveTable.storage.serdeName === None)
+    }
+  }
 }

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveSchemaInferenceSuite.scala

@@ -107,7 +107,6 @@ class HiveSchemaInferenceSuite
           locationUri = Option(dir.toURI),
           inputFormat = serde.inputFormat,
           outputFormat = serde.outputFormat,
-          serdeName = None,
           serde = serde.serde,
           compressed = false,
           properties = Map("serialization.format" -> "1")),