[spark] Remove useless Factories for bitmap and lucene

Author: JingsongLi

…dex/bitmap/BitmapGlobalIndexBuilder.java …obalindex/DefaultGlobalIndexBuilder.javapaimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/bitmap/BitmapGlobalIndexBuilder.java renamed to paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/bitmap/BitmapGlobalIndexBuilder.java renamed to paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java

@@ -16,20 +16,11 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.spark.globalindex.bitmap;
+package org.apache.paimon.spark.globalindex;
 
-import org.apache.paimon.spark.globalindex.GlobalIndexBuilder;
-import org.apache.paimon.spark.globalindex.GlobalIndexBuilderContext;
-
-/**
- * Builder for creating bitmap-based global indexes.
- *
- * <p>This implementation does not apply any custom transformations to the input dataset, allowing
- * the data to be processed as-is for bitmap index creation.
- */
-public class BitmapGlobalIndexBuilder extends GlobalIndexBuilder {
-
-    protected BitmapGlobalIndexBuilder(GlobalIndexBuilderContext context) {
+/** Default {@link GlobalIndexBuilder}. */
+public class DefaultGlobalIndexBuilder extends GlobalIndexBuilder {
+    public DefaultGlobalIndexBuilder(GlobalIndexBuilderContext context) {
         super(context);
     }
 }

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexBuilderFactory.java

@@ -25,7 +25,7 @@ public interface GlobalIndexBuilderFactory {
 
     GlobalIndexBuilder create(GlobalIndexBuilderContext context);
 
-    default GlobalIndexTopoBuilder createTopoBulder() {
+    default GlobalIndexTopoBuilder createTopoBuilder() {
         return null;
     }
 }

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexBuilderFactoryUtils.java

@@ -21,6 +21,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.util.HashMap;
 import java.util.Map;
 import java.util.ServiceLoader;
@@ -52,22 +54,20 @@ public class GlobalIndexBuilderFactoryUtils {
         }
     }
 
-    /**
-     * Loads the global index builder factory for the specified type.
-     *
-     * @param indexType The type of index (e.g., "bitmap")
-     * @return The corresponding factory
-     * @throws IllegalArgumentException If no factory is found for the specified type
-     */
-    public static GlobalIndexBuilderFactory load(String indexType) {
+    public static GlobalIndexBuilder createIndexBuilder(GlobalIndexBuilderContext context) {
+        GlobalIndexBuilderFactory factory = FACTORIES.get(context.indexType());
+        if (factory == null) {
+            return new DefaultGlobalIndexBuilder(context);
+        }
+        return factory.create(context);
+    }
+
+    @Nullable
+    public static GlobalIndexTopoBuilder createTopoBuilder(String indexType) {
         GlobalIndexBuilderFactory factory = FACTORIES.get(indexType);
         if (factory == null) {
-            throw new IllegalArgumentException(
-                    String.format(
-                            "No GlobalIndexBuilderFactory found for index type '%s'. "
-                                    + "Available types: %s",
-                            indexType, FACTORIES.keySet()));
+            return null;
         }
-        return factory;
+        return factory.createTopoBuilder();
     }
 }

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/bitmap/BitmapGlobalIndexBuilderFactory.java

@@ -27,7 +27,6 @@
 import org.apache.paimon.partition.PartitionPredicate;
 import org.apache.paimon.spark.globalindex.GlobalIndexBuilder;
 import org.apache.paimon.spark.globalindex.GlobalIndexBuilderContext;
-import org.apache.paimon.spark.globalindex.GlobalIndexBuilderFactory;
 import org.apache.paimon.spark.globalindex.GlobalIndexBuilderFactoryUtils;
 import org.apache.paimon.spark.globalindex.GlobalIndexTopoBuilder;
 import org.apache.paimon.spark.utils.SparkProcedureUtils;
@@ -127,10 +126,6 @@ public InternalRow[] call(InternalRow args) {
 
         String finalWhere = partitions != null ? SparkProcedureUtils.toWhere(partitions) : null;
 
-        // Early validation: check if the index type is supported
-        GlobalIndexBuilderFactory globalIndexBuilderFactory =
-                GlobalIndexBuilderFactoryUtils.load(indexType);
-
         LOG.info("Starting to build index for table " + tableIdent + " WHERE: " + finalWhere);
 
         return modifyPaimonTable(
@@ -184,7 +179,7 @@ public InternalRow[] call(InternalRow args) {
                         List<CommitMessage> indexResults;
                         // Step 2: build index by certain index system
                         GlobalIndexTopoBuilder topoBuildr =
-                                globalIndexBuilderFactory.createTopoBulder();
+                                GlobalIndexBuilderFactoryUtils.createTopoBuilder(indexType);
                         if (topoBuildr != null) {
                             indexResults =
                                     topoBuildr.buildIndex(
@@ -266,11 +261,9 @@ private List<CommitMessage> buildIndex(
                                             InstantiationUtil.deserializeObject(
                                                     dataSplitBytes,
                                                     GlobalIndexBuilder.class.getClassLoader());
-                                    GlobalIndexBuilderFactory globalIndexBuilderFactory =
-                                            GlobalIndexBuilderFactoryUtils.load(
-                                                    builderContext.indexType());
                                     GlobalIndexBuilder globalIndexBuilder =
-                                            globalIndexBuilderFactory.create(builderContext);
+                                            GlobalIndexBuilderFactoryUtils.createIndexBuilder(
+                                                    builderContext);
                                     return commitMessageSerializer.serialize(
                                             globalIndexBuilder.build(split));
                                 })