Add reflection-based auto-registration for ServerSidePlanningClientFactory

build.sbt

@@ -1105,6 +1105,8 @@ lazy val testDeltaIcebergJar = (project in file("testDeltaIcebergJar"))
 val deltaIcebergSparkIncludePrefixes = Seq(
   // We want everything from this package
   "org/apache/spark/sql/delta/icebergShaded",
+  // Server-side planning support
+  "org/apache/spark/sql/delta/serverSidePlanning",
 
   // We only want the files in this project from this package. e.g. we want to exclude
   // org/apache/spark/sql/delta/commands/convert/ConvertTargetFile.class (from delta-spark project).

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

@@ -70,17 +70,48 @@ private[serverSidePlanning] trait ServerSidePlanningClientFactory {
 }
 
 /**
- * Registry for client factories. Can be configured for testing or to provide
- * production implementations (e.g., IcebergRESTCatalogPlanningClientFactory).
- *
- * By default, no factory is registered. Production code should register an appropriate
- * factory implementation before attempting to create clients.
+ * Registry for client factories. Automatically discovers and registers implementations
+ * using reflection-based auto-discovery on first access to the factory. Manual registration
+ * using setFactory() is only needed for testing or to override the auto-discovered factory.
  */
 private[serverSidePlanning] object ServerSidePlanningClientFactory {
   @volatile private var registeredFactory: Option[ServerSidePlanningClientFactory] = None
+  @volatile private var autoRegistrationAttempted: Boolean = false
+
+  // Lazy initialization - only runs when getFactory() is called and no factory is set.
+  // Uses reflection to load the hardcoded IcebergRESTCatalogPlanningClientFactory class.
+  private def tryAutoRegisterFactory(): Unit = {
+    // Double-checked locking pattern to ensure initialization happens only once
+    if (!autoRegistrationAttempted) {
+      synchronized {
+        if (!autoRegistrationAttempted) {
+          autoRegistrationAttempted = true
+
+          try {
+            // Use reflection to load the Iceberg factory class
+            // scalastyle:off classforname
+            val clazz = Class.forName(
+              "org.apache.spark.sql.delta.serverSidePlanning." +
+              "IcebergRESTCatalogPlanningClientFactory")
+            // scalastyle:on classforname
+            val factory = clazz.getConstructor().newInstance()
+              .asInstanceOf[ServerSidePlanningClientFactory]
+            registeredFactory = Some(factory)
+          } catch {
+            case e: Exception =>
+              throw new IllegalStateException(
+                "Unable to load IcebergRESTCatalogPlanningClientFactory " +
+                "for server-side planning. Ensure the delta-iceberg JAR is on the " +
+                "classpath and compatible with this Delta version.",
+                e)
+          }
+        }
+      }
+    }
+  }
 
   /**
-   * Set a factory for production use or testing.
+   * Set a factory, overriding any auto-registered factory.
    */
   private[serverSidePlanning] def setFactory(factory: ServerSidePlanningClientFactory): Unit = {
     registeredFactory = Some(factory)
@@ -91,20 +122,40 @@ private[serverSidePlanning] object ServerSidePlanningClientFactory {
    */
   private[serverSidePlanning] def clearFactory(): Unit = {
     registeredFactory = None
+    autoRegistrationAttempted = false
   }
 
   /**
    * Get the currently registered factory.
-   * Throws IllegalStateException if no factory has been registered.
+   * Throws IllegalStateException if no factory has been registered (either via reflection-based
+   * auto-discovery or explicit setFactory() call).
    */
   def getFactory(): ServerSidePlanningClientFactory = {
+    // Try auto-registration if not already attempted and no factory is manually set
+    if (registeredFactory.isEmpty) {
+      tryAutoRegisterFactory()
+    }
+
     registeredFactory.getOrElse {
       throw new IllegalStateException(
         "No ServerSidePlanningClientFactory has been registered. " +
-        "Call ServerSidePlanningClientFactory.setFactory() to register an implementation.")
+        "Ensure delta-iceberg JAR is on the classpath for auto-registration, " +
+        "or call ServerSidePlanningClientFactory.setFactory() to register manually.")
     }
   }
 
+  /**
+   * Check if a factory is currently registered (either via ServiceLoader or setFactory()).
+   * Useful for testing or conditional logic.
+   */
+  def isFactoryRegistered(): Boolean = registeredFactory.isDefined
+
+  /**
+   * Get information about the currently registered factory for debugging/logging.
+   * Returns None if no factory is registered.
+   */
+  def getFactoryInfo(): Option[String] = registeredFactory.map(_.getClass.getName)
+
   /**
    * Convenience method to create a client from metadata using the registered factory.
    */

testDeltaIcebergJar/src/test/scala/JarSuite.scala

@@ -32,6 +32,8 @@ class JarSuite extends AnyFunSuite {
     "scala/",
     // e.g. org/apache/spark/sql/delta/icebergShaded/IcebergTransactionUtils.class
     "org/apache/spark/sql/delta/icebergShaded/",
+    // Server-side planning support
+    "org/apache/spark/sql/delta/serverSidePlanning/",
     // We explicitly include all the /delta/commands/convert classes we want, to ensure we don't
     // accidentally pull in some from delta-spark package.
     "org/apache/spark/sql/delta/commands/convert/IcebergFileManifest",