You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "rseetham (via GitHub)" <gi...@apache.org> on 2026/04/16 04:54:09 UTC
[PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
rseetham opened a new pull request, #18233:
URL: https://github.com/apache/pinot/pull/18233
**Why**
ProtoBufCodeGenMessageDecoder.init() is called once per consuming segment creation — once per topic partition every time a segment rolls over, and once per partition on server restart. Each call unconditionally fetched the protobuf schema JAR from remote storage (S3, HDFS, etc.) via ProtoBufUtils.getFileCopiedToLocal(), which copies the JAR into a new timestamped temp directory every time. The JAR only changes when the table's decoder config is updated, so in normal operation every fetch after the first is unnecessary network I/O.
Additionally, if jar is fetched by object store and that connection is broken, ingestion stops right now. With this fix, ingestion will continue based on the cached copy.
**What**
Introduce a JVM-level ConcurrentHashMap<String, CachedJar> keyed by topicName. CachedJar stores the remote JAR path and the local File it was copied to. On every init():
- Cache hit (same jarPath): return the cached local file immediately — no network call. Codegen and Janino compilation still run fresh per init(),
which is correct because fieldsToRead can differ between decoder instances for the same topic.
- jarPath changed (config update): fetch the new JAR, replace the cache entry.
- Fetch failure with a stale entry: log an error and return the previously cached local file so segment creation succeeds rather than failing on a
transient network issue. Rows decoded with a stale schema during this window are made explicit in the error log.
The URLClassLoader created to load the proto class is closed after the compiled Method is extracted, releasing the file handle immediately rather than accumulating them across segment rollovers.
How it behaves in each lifecycle event
Normal segment rollover: init() hits the cache, skips the remote fetch, runs codegen + Janino in memory (~ms), and returns. Each segment manager thread runs its own init() in parallel — no serialization across topics.
New table creation: First init() for that topicName — cache miss, JAR is fetched and cached. Subsequent segments for the same table hit the fast path.
Decoder config update (new JAR deployed): Next init() sees cached._jarPath != newJarPath, fetches the new JAR, replaces the cache entry.
Server restart: All cache entries are gone (JVM-level cache). Each partition's first init() after restart fetches the JAR once; subsequent rollovers hit the cache.
Tests
- Existing behavioral tests are unchanged and continue to pass.
- Added testCacheHit: two decoders initialized for the same topic and JAR both decode correctly, exercising the cache-hit path.
- Added testStaleFallbackOnFetchFailure: a decoder initialized with an unreachable JAR path falls back to the previously cached local file and
decodes correctly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "rseetham (via GitHub)" <gi...@apache.org>.
rseetham commented on code in PR #18233:
URL: https://github.com/apache/pinot/pull/18233#discussion_r3092439300
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
+ return cached._localFile;
Review Comment:
For a given kafka topic, the data can only be in 1 format. So the jar is the same to read from it. It the schema of different tables is different only those fields will be extracted per table. Here we are only caching the jar so that we don't keep fetching all the time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "rseetham (via GitHub)" <gi...@apache.org>.
rseetham commented on code in PR #18233:
URL: https://github.com/apache/pinot/pull/18233#discussion_r3094281585
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
Review Comment:
I dabbled with having a background job refresh the jar fetch every hour. The issue is the plugin module does not have access to the server executor service so I'll have to create and manage it here. So I don't think this is a good solution.
Another solution is having a cache that expires after an hour/ some configured value (server property) That would also force a periodic fetch. The issue here is if you set the segment completion time to the same time as the cache expiration, all segments completed at the same time so they would wait for the jar fetch anyway. Users would have to create this more carefully. But this solves the problem. I'll add this and address the other smaller comments that were brought up.
Is there another solution you'd suggest?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "rseetham (via GitHub)" <gi...@apache.org>.
rseetham commented on code in PR #18233:
URL: https://github.com/apache/pinot/pull/18233#discussion_r3094281585
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
Review Comment:
I dabbled with having a background job refresh the jar fetch every hour. The issue is the plugin module does not have access to the server executor service so I'll have to create and manage it here. So I don't think this is a good solution.
Another solution is having a cache with a ttl of an hour/ some configured value (server property) That would also force a periodic fetch. The issue here is if you set the segment completion time to the same time as the cache expiration, all segments completed at the same time so they would wait for the jar fetch anyway. Users would have to create this more carefully. But this solves the problem. I'll add this and address the other smaller comments that were brought up.
Still the fundamental issue with both of these is the only way to force a fetch is a restart in case the jar was changed in place. At the moment a table force commit will force a fetch. During incidents saying this will take 1 hr will be an issue.
Is there another solution you'd suggest?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "rseetham (via GitHub)" <gi...@apache.org>.
rseetham commented on code in PR #18233:
URL: https://github.com/apache/pinot/pull/18233#discussion_r3094281585
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
Review Comment:
I dabbled with having a background job refresh the jar fetch every hour. The issue is the plugin module does not have access to the server executor service so I'll have to create and manage it here. So I don't think this is a good solution.
Another solution is having a cache with a ttl of an hour/ some configured value (server property) That would also force a periodic fetch. The issue here is if you set the segment completion time to the same time as the cache expiration, all segments completed at the same time so they would wait for the jar fetch anyway. Users would have to set this more carefully. But this solves the problem. I'll add this and address the other smaller comments that were brought up.
Still the fundamental issue with both of these is the only way to force a fetch is a restart in case the jar was changed in place. At the moment a table force commit will force a fetch. During incidents saying this will take 1 hr will be an issue.
Is there another solution you'd suggest?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #18233:
URL: https://github.com/apache/pinot/pull/18233#discussion_r3090983321
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
Review Comment:
`URLClassLoader loader` is only closed on the happy path. If `getDescriptorForProtoClass()`, `codegen()`, `compileClass()`, or `getMethod()` throws, the classloader (and underlying JAR handle) will leak. Use try-with-resources (or a `finally`) so the loader is always closed.
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
+ return cached._localFile;
+ }
+ try {
+ File localFile = ProtoBufUtils.getFileCopiedToLocal(jarPath);
+ localFile.getParentFile().deleteOnExit();
Review Comment:
`localFile.getParentFile().deleteOnExit()` won’t delete the temp directory at JVM exit because the directory is non-empty (the JAR file remains). If you want cleanup-on-exit, register `localFile.deleteOnExit()` first and then register the directory, or otherwise perform recursive cleanup.
```suggestion
localFile.deleteOnExit();
File parentFile = localFile.getParentFile();
if (parentFile != null) {
parentFile.deleteOnExit();
}
```
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
+ return cached._localFile;
Review Comment:
The cache is keyed only by `topicName`. Pinot can have multiple tables consuming the same topic (note other components key by `(table, topic)`), and in that case a fetch failure could cause this decoder to fall back to a cached JAR from a different table/config, risking silent schema mismatch and bad ingestion. Consider including at least `protoClassName` (and/or `jarPath`) in the cache key, or restricting stale fallback so it cannot reuse an entry from a different configuration scope.
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
+ return cached._localFile;
+ }
+ try {
+ File localFile = ProtoBufUtils.getFileCopiedToLocal(jarPath);
+ localFile.getParentFile().deleteOnExit();
+ JAR_CACHE.put(topicName, new CachedJar(jarPath, localFile));
+ return localFile;
+ } catch (Exception e) {
+ if (cached != null && cached._localFile.exists()) {
+ LOGGER.error("Failed to fetch JAR for topic '{}' from '{}', reusing stale local copy from '{}'. "
+ + "Rows decoded with the stale schema may be incorrect if the schema has changed.",
+ topicName, jarPath, cached._jarPath, e);
Review Comment:
The error log says it is "reusing stale local copy from '…'" but the value being logged is `cached._jarPath` (remote path), not the local file path. This makes troubleshooting difficult; log `cached._localFile` (and optionally also log the stale remote path separately).
```suggestion
topicName, jarPath, cached._localFile, e);
```
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtilsTest.java:
##########
@@ -54,7 +55,7 @@ public void testGetTypeStrFromProto(String fieldName, String javaType) {
@Test
public void testGetTypeStrFromProto() throws Exception {
URL jarFile = getClass().getClassLoader().getResource("complex_types.jar");
- ClassLoader clsLoader = ProtoBufCodeGenMessageDecoder.loadClass(jarFile.getPath());
+ ClassLoader clsLoader = new URLClassLoader(new URL[]{jarFile});
Descriptors.Descriptor desc = ProtoBufCodeGenMessageDecoder.getDescriptorForProtoClass(clsLoader,
"org.apache.pinot.plugin.inputformat.protobuf.ComplexTypes$TestMessage");
Review Comment:
`URLClassLoader` implements `Closeable`, but this test never closes it. Use try-with-resources to avoid leaking an open JAR handle across the test suite (especially since this PR is explicitly addressing classloader/JAR handle churn).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #18233:
URL: https://github.com/apache/pinot/pull/18233#issuecomment-4257634099
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18233?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 55.29%. Comparing base ([`87e09fd`](https://app.codecov.io/gh/apache/pinot/commit/87e09fd31f1e322e585e14acb732cea0f916e0d0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`e9f3f4f`](https://app.codecov.io/gh/apache/pinot/commit/e9f3f4f57c8027ecca0e1abe15757a418d7a17cd?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
> :exclamation: There is a different number of reports uploaded between BASE (87e09fd) and HEAD (e9f3f4f). Click for more details.
>
> <details><summary>HEAD has 36 uploads less than BASE</summary>
>
>| Flag | BASE (87e09fd) | HEAD (e9f3f4f) |
>|------|------|------|
>|java-21|5|1|
>|unittests1|2|1|
>|unittests|4|1|
>|temurin|10|1|
>|java-11|5|0|
>|unittests2|2|0|
>|integration|6|0|
>|integration1|2|0|
>|integration2|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18233 +/- ##
============================================
- Coverage 63.36% 55.29% -8.07%
+ Complexity 1627 837 -790
============================================
Files 3243 2548 -695
Lines 197038 145989 -51049
Branches 30466 23471 -6995
============================================
- Hits 124845 80724 -44121
+ Misses 62195 58300 -3895
+ Partials 9998 6965 -3033
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.29% <ø> (-8.04%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.29% <ø> (-8.07%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.29% <ø> (-8.07%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.29% <ø> (-0.04%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18233/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/18233?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18233:
URL: https://github.com/apache/pinot/pull/18233#discussion_r3106114937
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
Review Comment:
hmm, then another solution might be just keep the crc for the jar as a map then you can still perform a periodically fetch, but only replace when jar crc mismatch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18233:
URL: https://github.com/apache/pinot/pull/18233#discussion_r3093036405
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
Review Comment:
This changes the rollout semantics from 're-fetch the protobuf JAR on every init' to 'reuse it indefinitely while the URI string stays the same'. Pinot deployments often replace the decoder JAR in place at the same S3/HDFS path during schema rollouts; after this cache hits once, long-lived servers will keep decoding with the old generated classes until restart, which can silently ingest rows with the wrong schema. We need either a freshness/version check here or an explicit versioned-URI contract before making the cached file authoritative.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Cache local JAR file in ProtoBufCodeGenMessageDecoder to eliminate redundant remote fetches [pinot]
Posted by "rseetham (via GitHub)" <gi...@apache.org>.
rseetham commented on code in PR #18233:
URL: https://github.com/apache/pinot/pull/18233#discussion_r3094281585
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -47,14 +92,50 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
"Protocol Buffer schema jar file must be provided");
Preconditions.checkState(props.containsKey(PROTO_CLASS_NAME),
"Protocol Buffer Message class name must be provided");
- String protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- String jarPath = props.getOrDefault(PROTOBUF_JAR_FILE_PATH, "");
- ClassLoader protoMessageClsLoader = loadClass(jarPath);
- Descriptors.Descriptor descriptor = getDescriptorForProtoClass(protoMessageClsLoader, protoClassName);
+ String jarPath = props.get(PROTOBUF_JAR_FILE_PATH);
+ String protoClassName = props.get(PROTO_CLASS_NAME);
+
+ File localFile = resolveJar(topicName, jarPath);
+ URLClassLoader loader = new URLClassLoader(new URL[]{localFile.toURI().toURL()});
+ Descriptors.Descriptor descriptor = getDescriptorForProtoClass(loader, protoClassName);
String codeGenCode = new MessageCodeGen().codegen(descriptor, fieldsToRead);
- Class<?> recordExtractor = compileClass(protoMessageClsLoader,
+ Class<?> recordExtractor = compileClass(loader,
MessageCodeGen.EXTRACTOR_PACKAGE_NAME + "." + MessageCodeGen.EXTRACTOR_CLASS_NAME, codeGenCode);
_decodeMethod = recordExtractor.getMethod(MessageCodeGen.EXTRACTOR_METHOD_NAME, byte[].class, GenericRow.class);
+ loader.close();
+ }
+
+ /**
+ * Returns the local {@link File} for the given {@code jarPath}, fetching it from remote storage if needed.
+ *
+ * <p>Fast path: if the cache already holds a local copy for this topic with the same remote path, return it
+ * immediately with no network I/O.
+ *
+ * <p>Slow path: fetch the JAR, update the cache, register the temp dir for deletion on JVM exit.
+ *
+ * <p>Stale fallback: if the fetch fails and a prior local copy exists (e.g. from an older {@code jarPath}),
+ * log a warning and return the stale file so segment creation can proceed.
+ */
+ private static File resolveJar(String topicName, String jarPath)
+ throws Exception {
+ CachedJar cached = JAR_CACHE.get(topicName);
+ if (cached != null && cached._jarPath.equals(jarPath) && cached._localFile.exists()) {
Review Comment:
I dabbled with having a background job refresh the jar fetch every hour. The issue is the plugin module does not have access to the server executor service so I'll have to create and manage it here. So I don't think this is a good solution.
Another solution is having a cache with a ttl of an hour/ some configured value (server property) That would also force a periodic fetch. The issue here is if you set the segment completion time to the same time as the cache expiration, all segments completed at the same time so they would wait for the jar fetch anyway. Users would have to create this more carefully. But this solves the problem. I'll add this and address the other smaller comments that were brought up.
Is there another solution you'd suggest?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org