Make the segment pre-processor pluggable via SegmentPreProcessorProvider

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java

@@ -282,7 +282,7 @@ public static boolean needPreprocess(SegmentDirectory segmentDirectory, IndexLoa
     if (indexLoadingConfig.getTableConfig() == null || indexLoadingConfig.getSchema() == null) {
       return false;
     }
-    return new SegmentPreProcessor(segmentDirectory, indexLoadingConfig).needProcess();
+    return SegmentPreProcessor.create(segmentDirectory, indexLoadingConfig).needProcess();
   }
 
   private static boolean needConvertSegmentFormat(IndexLoadingConfig indexLoadingConfig,
@@ -329,7 +329,7 @@ private static void preprocessSegment(File indexDir, String segmentName, String
         .build();
     SegmentDirectory segmentDirectory =
         SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader().load(indexDir.toURI(), segmentLoaderContext);
-    try (SegmentPreProcessor preProcessor = new SegmentPreProcessor(segmentDirectory, indexLoadingConfig)) {
+    try (SegmentPreProcessor preProcessor = SegmentPreProcessor.create(segmentDirectory, indexLoadingConfig)) {
       preProcessor.process(segmentOperationsThrottlerSet);
     }
   }

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java

@@ -22,7 +22,10 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.ServiceLoader;
+import java.util.Set;
 import javax.annotation.Nullable;
 import org.apache.commons.configuration2.PropertiesConfiguration;
 import org.apache.commons.configuration2.ex.ConfigurationException;
@@ -54,6 +57,7 @@
 import org.apache.pinot.spi.config.table.MultiColumnTextIndexConfig;
 import org.apache.pinot.spi.config.table.TableConfig;
 import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.plugin.PluginManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -68,6 +72,41 @@
 public class SegmentPreProcessor implements AutoCloseable {
   private static final Logger LOGGER = LoggerFactory.getLogger(SegmentPreProcessor.class);
 
+  // The highest-priority ServiceLoader-registered provider, or null to use this class directly. Resolved once, at
+  // first use (segment loading), by which point PluginManager has loaded the plugin classloaders.
+  @Nullable
+  private static final SegmentPreProcessorProvider PROVIDER = loadProvider();
+
+  @Nullable
+  private static SegmentPreProcessorProvider loadProvider() {
+    // Enumerate this class's own classloader plus every plugin classloader: new-style plugins live in isolated
+    // realms whose services a plain ServiceLoader.load() cannot see (see PluginManager#getPluginClassLoaders).
+    Set<ClassLoader> classLoaders = new LinkedHashSet<>();
+    classLoaders.add(SegmentPreProcessorProvider.class.getClassLoader());
+    classLoaders.addAll(PluginManager.get().getPluginClassLoaders());
+    SegmentPreProcessorProvider best = null;
+    for (ClassLoader classLoader : classLoaders) {
+      for (SegmentPreProcessorProvider provider : ServiceLoader.load(SegmentPreProcessorProvider.class,
+          classLoader)) {
+        if (best == null || provider.getPriority() > best.getPriority()) {
+          best = provider;
+        }
+      }
+    }
+    if (best != null) {
+      LOGGER.info("Using segment pre-processor provider: {}", best.getClass().getName());
+    }
+    return best;
+  }
+
+  /// Creates the segment pre-processor: the highest-priority [SegmentPreProcessorProvider]'s instance, or a plain
+  /// [SegmentPreProcessor] when no provider is registered.
+  public static SegmentPreProcessor create(SegmentDirectory segmentDirectory, IndexLoadingConfig indexLoadingConfig) {
+    return PROVIDER != null
+        ? PROVIDER.create(segmentDirectory, indexLoadingConfig)
+        : new SegmentPreProcessor(segmentDirectory, indexLoadingConfig);
+  }
+
   private final SegmentDirectory _segmentDirectory;
   private final IndexLoadingConfig _indexLoadingConfig;
   private final TableConfig _tableConfig;

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorProvider.java

@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.pinot.segment.local.segment.index.loader;
+
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+
+
+/// Pluggable factory for [SegmentPreProcessor] instances, letting plugins substitute a subclass — e.g. one that
+/// manages resources shared across the index handlers of a preprocess run, which no single handler can scope because
+/// handlers run in unspecified order relative to each other.
+///
+/// Implementations are discovered through [java.util.ServiceLoader] (register the implementation class in
+/// `META-INF/services`) — enumerated on the default classloader and on every plugin classloader
+/// ([org.apache.pinot.spi.plugin.PluginManager#getPluginClassLoaders]), so provider jars may live on the classpath
+/// or in the plugins directory — and the highest-priority one wins, the same convention as
+/// [org.apache.pinot.segment.spi.index.IndexPlugin]. When none is registered, [SegmentPreProcessor#create] falls
+/// back to the base [SegmentPreProcessor].
+public interface SegmentPreProcessorProvider {
+
+  /// Creates the pre-processor for one segment preprocess run.
+  SegmentPreProcessor create(SegmentDirectory segmentDirectory, IndexLoadingConfig indexLoadingConfig);
+
+  /// Priority used to choose between multiple registered providers: the highest wins. It plays no part in replacing
+  /// the base pre-processor — any registered provider does that, and the fallback to the base [SegmentPreProcessor]
+  /// applies only when no provider is registered at all.
+  default int getPriority() {
+    return 0;
+  }
+}