diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java index b8e293e301d4..53405a689449 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java +++ b/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); } } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java index e686d40e763a..e4d74d336b28 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java +++ b/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 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; diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorProvider.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorProvider.java new file mode 100644 index 000000000000..8bee2e711599 --- /dev/null +++ b/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; + } +}