This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 4e12f1e1773 Make the segment pre-processor pluggable via
SegmentPreProcessorProvider (#19369)
4e12f1e1773 is described below
commit 4e12f1e1773cf8fdea49f2c5472c9bc2e5a9e4f1
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Aug 26 11:27:33 2026 -0700
Make the segment pre-processor pluggable via SegmentPreProcessorProvider
(#19369)
---
.../immutable/ImmutableSegmentLoader.java | 4 +-
.../segment/index/loader/SegmentPreProcessor.java | 39 +++++++++++++++++++
.../index/loader/SegmentPreProcessorProvider.java | 45 ++++++++++++++++++++++
3 files changed, 86 insertions(+), 2 deletions(-)
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 b8e293e301d..53405a68944 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 class ImmutableSegmentLoader {
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 @@ public class ImmutableSegmentLoader {
.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 e686d40e763..e4d74d336b2 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 com.google.common.base.Preconditions;
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.segment.spi.utils.SegmentMetadataUtils;
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 @@ import org.slf4j.LoggerFactory;
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;
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 00000000000..8bee2e71159
--- /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;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
