xiangfu0 commented on code in PR #19797:
URL: https://github.com/apache/pinot/pull/19797#discussion_r4229537268
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java:
##########
@@ -134,6 +141,66 @@ private static void
initForwardIndexConfig(PinotConfiguration instanceConfig) {
LOGGER.info("Setting forward index default target docs per chunk to:
{}", defaultTargetDocsPerChunk);
ForwardIndexConfig.setDefaultTargetDocsPerChunk(Integer.parseInt(defaultTargetDocsPerChunk));
}
+ String defaultCompressionCodec =
+
instanceConfig.getProperty(CommonConstants.ForwardIndexConfigs.CONFIG_OF_DEFAULT_COMPRESSION_CODEC);
+ if (defaultCompressionCodec != null) {
+ setDefaultCompressionCodec(defaultCompressionCodec);
+ }
+ }
+
+ /// Applies the cluster-wide default compression codec for raw forward
indexes.
+ ///
+ /// The value is a [CompressionCodec], the same spelling used by
`compressionCodec` in a table config,
+ /// so that an operator setting a cluster default and a table author setting
a column override write the
+ /// same word. Codecs that are not applicable to a raw forward index -- the
CLP family, `MV_ENTRY_DICT`,
+ /// `DELTA` -- are rejected rather than applied, since they describe
whole-index or dictionary formats
+ /// that cannot stand in for a chunk codec.
+ ///
+ /// Unlike the numeric defaults above, an unusable value is logged and
ignored instead of failing
+ /// startup: this one is routinely set as a *cluster* config, where throwing
would take down every
+ /// component that restarts after the bad value is saved. The warning names
the codecs that are
+ /// accepted, so a rejected value is self-correcting from the log.
+ ///
+ /// Scope, which this being a *cluster* config does not make as wide as it
sounds:
+ ///
+ /// - It is read once per JVM at startup, so after changing it a rolling
restart leaves restarted
+ /// instances writing the new codec while the rest keep the old one until
they restart too. Each
+ /// segment records the codec it was written with, so this mixes formats
without breaking reads.
+ /// - It reaches only components that call [#applyClusterConfig]: the
controller, broker, server and
+ /// minion. Segments built outside the cluster -- standalone, Spark or
Hadoop batch ingestion, and
+ /// `pinot-admin CreateSegment` -- keep the compiled-in default, and
because a reload ignores
+ /// defaults, nothing later reconciles the two.
+ /// - It is a process-global default, so in a single-JVM multi-role
deployment the last role to start
+ /// wins for the whole process.
+ ///
+ /// These all hold for the three forward index defaults above as well; they
are spelled out here
+ /// because this is the first of them an operator is routinely expected to
set.
+ @VisibleForTesting
+ static void setDefaultCompressionCodec(String defaultCompressionCodec) {
+ CompressionCodec codec;
+ try {
+ codec =
CompressionCodec.valueOf(defaultCompressionCodec.trim().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ LOGGER.warn("Unknown forward index default compression codec: {},
expected one of: {}, keeping: {}",
+ defaultCompressionCodec, rawIndexCompressionCodecs(),
ForwardIndexConfig.getDefaultCompressionType());
+ return;
+ }
+ if (!codec.isApplicableToRawIndex()) {
Review Comment:
**[severity: high] Allowing GZIP as a cluster-wide default exposes every raw
column to the non-thread-safe `GzipDecompressor`.**
`GzipDecompressor` wraps a single `java.util.zip.Inflater` and runs
`reset()` / `setInput()` / `inflate()` as an unsynchronized sequence.
`BaseChunkForwardIndexReader` stores one decompressor per reader (field
`_chunkDecompressor`, used directly in `decompress()`), and
`VarByteChunkForwardIndexReaderV4` shares one instance across all reader
contexts — while the reader contract (`ForwardIndexReaderContext`) requires
readers to be stateless because they are accessed by multiple threads. The
other codecs (LZ4, ZSTD, Snappy, pass-through) use stateless decompressors, so
GZIP is the only affected one.
Today this race is reachable only for columns that explicitly configure
GZIP. With `pinot.forward.index.default.compression.codec=GZIP`, every raw
non-metric column without an explicit codec is built with GZIP, and two
concurrent queries on the same segment column can interleave (A:
reset+setInput(chunkA); B: reset+setInput(chunkB); A: inflate -> gets chunkB's
bytes), silently returning wrong values or throwing `DataFormatException`/OOB.
Consider either excluding GZIP from the accepted defaults until the
decompressor is thread-safe, or fixing `GzipDecompressor` (ThreadLocal
`Inflater`, or a fresh decompressor per reader context) in/before this PR.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
