[Server-Side Planning] Add metadata abstraction and factory pattern

This PR refactors ServerSidePlannedTable to use a metadata abstraction layer
instead of directly depending on Unity Catalog parameters. The changes include:

- ServerSidePlanningMetadata trait with UnityCatalogMetadata and DefaultMetadata
- Factory pattern with buildClient(metadata) instead of buildForCatalog()
- TestMetadata for unit testing with injectable values
- Comprehensive test coverage for all metadata implementations

The metadata abstraction enables future extensibility for different catalog
implementations while maintaining backwards compatibility.

Spark module only - Iceberg test infrastructure changes will be in Row 7 & 8.

Files modified (7 spark module files):
- ServerSidePlanningMetadata.scala (new)
- UnityCatalogMetadata.scala (new)
- TestMetadata.scala (new)
- ServerSidePlannedTable.scala (modified)
- ServerSidePlanningClient.scala (modified)
- ServerSidePlannedTableSuite.scala (modified)
- TestServerSidePlanningClient.scala (modified)

Author: murali-db

spark/src/main/scala/org/apache/spark/sql/delta/serverSidePlanning/ServerSidePlannedTable.scala

@@ -105,38 +105,49 @@ object ServerSidePlannedTable extends DeltaLogging {
       val namespace = ident.namespace().mkString(".")
       val tableName = ident.name()
 
-      // Extract catalog name from identifier namespace, or default to spark_catalog
-      //
-      // Spark Identifier structure:
-      // - For "catalog.database.table": namespace() = ["catalog", "database"], name() = "table"
-      // - For "database.table":          namespace() = ["database"], name() = "table"
-      // - For "table":                   namespace() = [], name() = "table"
-      //
-      // Note: We check namespace().length > 1 (not >= 1) because a single-element namespace
-      // represents just the database name without an explicit catalog, so we use the default.
-      // See Spark's LookupCatalog, CatalogAndIdentifier and ResolveSessionCatalog.
-      val catalogName = if (ident.namespace().length > 1) {
-        ident.namespace().head
-      } else {
-        "spark_catalog"
-      }
+      // Create metadata from table
+      val metadata = ServerSidePlanningMetadata.fromTable(table, spark, ident, isUnityCatalog)
 
       // Try to create ServerSidePlannedTable with server-side planning
-      try {
-        val client = ServerSidePlanningClientFactory.getClient(spark, catalogName)
-        Some(new ServerSidePlannedTable(spark, namespace, tableName, table.schema(), client))
-      } catch {
-        case _: IllegalStateException =>
-          // Factory not registered - fall through to normal path
-          logWarning(s"Server-side planning not available for catalog $catalogName. " +
+      val plannedTable = tryCreate(spark, namespace, tableName, table.schema(), metadata)
+      if (plannedTable.isEmpty) {
+        logWarning(
+          s"Server-side planning not available for catalog ${metadata.catalogName}. " +
             "Falling back to normal table loading.")
-          None
       }
+      plannedTable
     } else {
       None
     }
   }
 
+  /**
+   * Try to create a ServerSidePlannedTable with server-side planning.
+   * Returns None if the planning client factory is not available.
+   *
+   * @param spark The SparkSession
+   * @param database The database name (may include catalog prefix)
+   * @param tableName The table name
+   * @param tableSchema The table schema
+   * @param metadata Metadata extracted from loadTable response
+   * @return Some(ServerSidePlannedTable) if successful, None if factory not registered
+   */
+  private def tryCreate(
+      spark: SparkSession,
+      databaseName: String,
+      tableName: String,
+      tableSchema: StructType,
+      metadata: ServerSidePlanningMetadata): Option[ServerSidePlannedTable] = {
+    try {
+      val client = ServerSidePlanningClientFactory.buildClient(spark, metadata)
+      Some(new ServerSidePlannedTable(spark, databaseName, tableName, tableSchema, client))
+    } catch {
+      case _: IllegalStateException =>
+        // Factory not registered - this shouldn't happen in production but could during testing
+        None
+    }
+  }
+
   /**
    * Check if a table has credentials available.
    * Unity Catalog tables may lack credentials when accessed without proper permissions.
@@ -168,7 +179,7 @@ class ServerSidePlannedTable(
     extends Table with SupportsRead with DeltaLogging {
 
   // Returns fully qualified name (e.g., "catalog.database.table").
-  // The databaseName parameter receives ident.namespace().mkString(".") from DeltaCatalog,
+  // The database parameter receives ident.namespace().mkString(".") from DeltaCatalog,
   // which includes the catalog name when present, similar to DeltaTableV2's name() method.
   override def name(): String = s"$databaseName.$tableName"
 
@@ -212,10 +223,10 @@ class ServerSidePlannedScan(
 
   override def toBatch: Batch = this
 
-  override def planInputPartitions(): Array[InputPartition] = {
-    // Call the server-side planning API to get the scan plan
-    val scanPlan = planningClient.planScan(databaseName, tableName)
+  // Call the server-side planning API once and store the result
+  private val scanPlan = planningClient.planScan(databaseName, tableName)
 
+  override def planInputPartitions(): Array[InputPartition] = {
     // Convert each file to an InputPartition
     scanPlan.files.map { file =>
       ServerSidePlannedFileInputPartition(file.filePath, file.fileSizeInBytes, file.fileFormat)
@@ -241,29 +252,27 @@ case class ServerSidePlannedFileInputPartition(
  */
 class ServerSidePlannedFilePartitionReaderFactory(
     spark: SparkSession,
-    tableSchema: StructType)
+    schema: StructType)
     extends PartitionReaderFactory {
 
   import org.apache.spark.util.SerializableConfiguration
 
   // scalastyle:off deltahadoopconfiguration
   // We use sessionState.newHadoopConf() here instead of deltaLog.newDeltaHadoopConf().
   // This means DataFrame options (like custom S3 credentials) passed by users will NOT be
-  // included in the Hadoop configuration. This is intentional:
-  // - Server-side planning uses server-provided credentials, not user-specified credentials
-  // - ServerSidePlannedTable is NOT a Delta table, so we don't want Delta-specific options
-  //   from deltaLog.newDeltaHadoopConf()
-  // - General Spark options from spark.hadoop.* are included and work for all tables
+  // included in the Hadoop configuration. This would fail if users specify credentials in
+  // DataFrame read options expecting them to be used when accessing the underlying files.
+  // However, for now we accept this limitation to avoid requiring a DeltaLog parameter.
   private val hadoopConf = new SerializableConfiguration(spark.sessionState.newHadoopConf())
   // scalastyle:on deltahadoopconfiguration
 
   // Pre-build reader function for Parquet on the driver
   // This function will be serialized and sent to executors
   private val parquetReaderBuilder = new ParquetFileFormat().buildReaderWithPartitionValues(
     sparkSession = spark,
-    dataSchema = tableSchema,
+    dataSchema = schema,
     partitionSchema = StructType(Nil),
-    requiredSchema = tableSchema,
+    requiredSchema = schema,
     filters = Seq.empty,
     options = Map(
       FileFormat.OPTION_RETURNING_BATCH -> "false"

spark/src/main/scala/org/apache/spark/sql/delta/serverSidePlanning/ServerSidePlanningClient.scala

@@ -57,17 +57,17 @@ trait ServerSidePlanningClient {
  * Factory for creating ServerSidePlanningClient instances.
  * This allows for configurable implementations (REST, mock, Spark-based, etc.)
  */
-private[serverSidePlanning] trait ServerSidePlanningClientFactory {
+trait ServerSidePlanningClientFactory {
   /**
-   * Create a client for a specific catalog by reading catalog-specific configuration.
-   * This method reads configuration from spark.sql.catalog.<catalogName>.uri and
-   * spark.sql.catalog.<catalogName>.token.
+   * Create a client using metadata from catalog's loadTable.
    *
    * @param spark The SparkSession
-   * @param catalogName The name of the catalog (e.g., "spark_catalog", "unity")
-   * @return A ServerSidePlanningClient configured for the specified catalog
+   * @param metadata Metadata extracted from loadTable response
+   * @return A ServerSidePlanningClient configured with the metadata
    */
-  def buildForCatalog(spark: SparkSession, catalogName: String): ServerSidePlanningClient
+  def buildClient(
+      spark: SparkSession,
+      metadata: ServerSidePlanningMetadata): ServerSidePlanningClient
 }
 
 /**
@@ -77,37 +77,41 @@ private[serverSidePlanning] trait ServerSidePlanningClientFactory {
  * By default, no factory is registered. Production code should register an appropriate
  * factory implementation before attempting to create clients.
  */
-private[serverSidePlanning] object ServerSidePlanningClientFactory {
+object ServerSidePlanningClientFactory {
   @volatile private var registeredFactory: Option[ServerSidePlanningClientFactory] = None
 
   /**
    * Set a factory for production use or testing.
    */
-  private[serverSidePlanning] def setFactory(factory: ServerSidePlanningClientFactory): Unit = {
+  def setFactory(factory: ServerSidePlanningClientFactory): Unit = {
     registeredFactory = Some(factory)
   }
 
   /**
    * Clear the registered factory.
    */
-  private[serverSidePlanning] def clearFactory(): Unit = {
+  def clearFactory(): Unit = {
     registeredFactory = None
   }
 
   /**
-   * Get a client for a specific catalog using the registered factory.
-   * This is the single public entry point for obtaining a ServerSidePlanningClient.
-   *
-   * @param spark The SparkSession
-   * @param catalogName The name of the catalog (e.g., "spark_catalog", "unity")
-   * @return A ServerSidePlanningClient configured for the specified catalog
-   * @throws IllegalStateException if no factory has been registered
+   * Get the currently registered factory.
+   * Throws IllegalStateException if no factory has been registered.
    */
-  def getClient(spark: SparkSession, catalogName: String): ServerSidePlanningClient = {
+  def getFactory(): ServerSidePlanningClientFactory = {
     registeredFactory.getOrElse {
       throw new IllegalStateException(
         "No ServerSidePlanningClientFactory has been registered. " +
         "Call ServerSidePlanningClientFactory.setFactory() to register an implementation.")
-    }.buildForCatalog(spark, catalogName)
+    }
+  }
+
+  /**
+   * Convenience method to create a client from metadata using the registered factory.
+   */
+  def buildClient(
+      spark: SparkSession,
+      metadata: ServerSidePlanningMetadata): ServerSidePlanningClient = {
+    getFactory().buildClient(spark, metadata)
   }
 }

spark/src/main/scala/org/apache/spark/sql/delta/serverSidePlanning/ServerSidePlanningMetadata.scala

@@ -0,0 +1,91 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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.delta.serverSidePlanning
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connector.catalog.{Identifier, Table}
+
+/**
+ * Metadata required for creating a server-side planning client.
+ *
+ * This interface captures all information from the catalog's loadTable response
+ * that is needed to create and configure a ServerSidePlanningClient.
+ */
+trait ServerSidePlanningMetadata {
+  /**
+   * The base URI for the planning endpoint.
+   */
+  def planningEndpointUri: String
+
+  /**
+   * Authentication token for the planning endpoint.
+   */
+  def authToken: Option[String]
+
+  /**
+   * Catalog name for configuration lookups.
+   */
+  def catalogName: String
+
+  /**
+   * Additional table properties that may be needed.
+   * For example, table UUID, credential hints, etc.
+   */
+  def tableProperties: Map[String, String]
+}
+
+/**
+ * Default metadata for non-UC catalogs.
+ * Used when server-side planning is force-enabled for testing/development.
+ */
+case class DefaultMetadata(
+    catalogName: String,
+    tableProps: Map[String, String] = Map.empty) extends ServerSidePlanningMetadata {
+  override def planningEndpointUri: String = ""
+  override def authToken: Option[String] = None
+  override def tableProperties: Map[String, String] = tableProps
+}
+
+object ServerSidePlanningMetadata {
+  /**
+   * Create metadata from a loaded table.
+   *
+   * Returns UnityCatalogMetadata for UC catalogs, DefaultMetadata for non-UC catalogs.
+   */
+  def fromTable(
+      table: Table,
+      spark: SparkSession,
+      ident: Identifier,
+      isUnityCatalog: Boolean): ServerSidePlanningMetadata = {
+
+    if (isUnityCatalog) {
+      UnityCatalogMetadata.fromTable(table, spark, ident)
+    } else {
+      // For non-UC catalogs, return default metadata
+      val catalogName = extractCatalogName(ident)
+      DefaultMetadata(catalogName, Map.empty)
+    }
+  }
+
+  private def extractCatalogName(ident: Identifier): String = {
+    if (ident.namespace().length > 1) {
+      ident.namespace().head
+    } else {
+      "spark_catalog"
+    }
+  }
+}