This is an automated email from the ASF dual-hosted git repository.
xiangfu0 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 3a0ebd94728 Make the default forward index compression codec
configurable (#19797)
3a0ebd94728 is described below
commit 3a0ebd9472866a212c03fb8c5c9ad0b049baae6d
Author: Arunkumar Saravanan <[email protected]>
AuthorDate: Mon Oct 12 12:06:45 2026 +0530
Make the default forward index compression codec configurable (#19797)
The raw forward index writer version, target chunk size and target docs per
chunk are all settable cluster-wide through pinot.forward.index.default.*,
but
the compression codec is hardcoded:
ForwardIndexType#getDefaultCompressionType
returns PASS_THROUGH for metrics and LZ4 for everything else, with no way to
change it short of naming a codec in every table config.
Add pinot.forward.index.default.compression.codec, applied at startup by
ServiceStartableUtils alongside the existing forward index defaults. Metrics
keep PASS_THROUGH: they are usually small fixed-width values where
compression
costs more than it saves, so a cluster-wide default should not start
compressing
them.
The value names a CompressionCodec -- the same vocabulary as
compressionCodec in
a table config -- and ForwardIndexConfig decides what is acceptable, so the
class that owns the default owns its invariant and an embedder setting it
directly gets the same guard. Rejected are the CLP family and MV_ENTRY_DICT,
which name whole-index or dictionary-id formats rather than chunk codecs;
DELTA
and DELTADELTA, which are chunk codecs but are rejected by table config
validation on every column and whose compressors throw on chunks that are
not a
multiple of 4 bytes; and GZIP, whose decompressor keeps a single mutable
Inflater that a chunk reader shares across all of a column's reader
contexts.
Naming GZIP on one column is an existing opt-in exposure; making it the
default
would put every raw non-metric column on that path.
This matters because the default is injected in ForwardIndexCreatorFactory,
after table config validation has run, so a codec the table config path
would
have rejected has nothing left to catch it.
The CompressionCodec-to-ChunkCompressionType mapping moves into one
authoritative ForwardIndexConfig helper used by both the @JsonCreator and
the
new setter, rather than a second, name-based copy that would compile
through a
rename and fail at runtime.
An unusable value is logged, naming the codecs that are accepted, and
ignored
rather than failing startup -- matching initFieldSpecConfig rather than the
numeric forward index defaults above it, which throw. This one is routinely
set
as a cluster config, where throwing would take down every component that
restarts after a bad value is saved.
Naming a codec in a table config still wins, and because this is a default
rather than a table config value, changing it does not rewrite existing
segments: ForwardIndexHandler ignores the default when deciding whether a
reload must rewrite a forward index, though a rewrite triggered for another
reason does write the reloading component's current default. Its comment
naming
LZ4 as the fixed non-metric default is updated too, as is the OpenStruct
splitter's, whose LZ4 pin covers its build path but not a dictionary-to-raw
conversion on reload.
---
.../pinot/common/utils/ServiceStartableUtils.java | 68 ++++++++-
.../common/utils/ServiceStartableUtilsTest.java | 152 +++++++++++++++++++++
.../impl/openstruct/OpenStructColumnSplitter.java | 11 +-
.../segment/index/forward/ForwardIndexType.java | 17 ++-
.../segment/index/loader/ForwardIndexHandler.java | 7 +-
.../index/openstruct/OpenStructIndexType.java | 7 +-
.../openstruct/OpenStructColumnSplitterTest.java | 2 +-
.../forward/ForwardIndexCreatorFactoryTest.java | 52 +++++++
.../ForwardIndexDefaultCompressionTypeTest.java | 100 ++++++++++++++
.../index/openstruct/OpenStructIndexTypeTest.java | 5 +-
.../segment/spi/index/ForwardIndexConfig.java | 128 +++++++++++------
.../apache/pinot/spi/utils/CommonConstants.java | 2 +
12 files changed, 498 insertions(+), 53 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java
index ca5de5d609a..a3fcb7b68ff 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java
@@ -18,6 +18,8 @@
*/
package org.apache.pinot.common.utils;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.Locale;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.TimeUnit;
@@ -26,6 +28,7 @@ import
org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.impl.client.ZkClient;
import org.apache.pinot.common.utils.request.RequestUtils;
import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
+import org.apache.pinot.spi.config.table.FieldConfig.CompressionCodec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.services.ServiceRole;
@@ -115,7 +118,8 @@ public class ServiceStartableUtils {
LOGGER.info("Timezone: {}", _timeZone);
}
- private static void initForwardIndexConfig(PinotConfiguration
instanceConfig) {
+ @VisibleForTesting
+ static void initForwardIndexConfig(PinotConfiguration instanceConfig) {
String defaultRawIndexWriterVersion =
instanceConfig.getProperty(CommonConstants.ForwardIndexConfigs.CONFIG_OF_DEFAULT_RAW_INDEX_WRITER_VERSION);
if (defaultRawIndexWriterVersion != null) {
@@ -134,6 +138,68 @@ public class ServiceStartableUtils {
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. [ForwardIndexConfig#setDefaultCompressionCodec] decides what
is acceptable and this method
+ /// only parses, so the rejected set stays with the class that owns the
default: the CLP family and
+ /// `MV_ENTRY_DICT`, which name whole-index or dictionary-id formats rather
than chunk codecs;
+ /// `DELTA` and `DELTADELTA`, which are chunk codecs but are rejected by
table config validation on every
+ /// column and whose compressors throw on chunks that are not a multiple of
4 bytes; and `GZIP`, whose
+ /// decompressor is not thread-safe.
+ ///
+ /// 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. A reload
ignores the default when
+ /// *deciding* whether to rewrite a forward index, so nothing reconciles
the two on its own; but a
+ /// rewrite triggered for another reason -- a dictionary-to-raw
conversion, codecSpec removal,
+ /// re-enabling a disabled forward index, a new derived or default column
-- does write the reloading
+ /// component's current default.
+ /// - It is a process-global default, so in a single-JVM multi-role
deployment the last role to start
+ /// wins for the whole process.
+ ///
+ /// The first two 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. The third is worse for
+ /// those: under a parallel bootstrap the roles race, and unlike this one
their statics are not
+ /// `volatile`.
+ @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,
ForwardIndexConfig.getAcceptableDefaultCompressionCodecs(),
+ ForwardIndexConfig.getDefaultCompressionType());
+ return;
+ }
+ try {
+ ForwardIndexConfig.setDefaultCompressionCodec(codec);
+ } catch (IllegalArgumentException e) {
+ LOGGER.warn("Rejected forward index default compression codec: {} ({}),
expected one of: {}, keeping: {}",
+ codec, e.getMessage(),
ForwardIndexConfig.getAcceptableDefaultCompressionCodecs(),
+ ForwardIndexConfig.getDefaultCompressionType());
+ return;
+ }
+ LOGGER.info("Set forward index default compression codec to: {}",
ForwardIndexConfig.getDefaultCompressionType());
}
public static void initFieldSpecConfig(PinotConfiguration instanceConfig) {
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/ServiceStartableUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ServiceStartableUtilsTest.java
new file mode 100644
index 00000000000..fb444182357
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ServiceStartableUtilsTest.java
@@ -0,0 +1,152 @@
+/**
+ * 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.common.utils;
+
+import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
+import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
+import org.apache.pinot.spi.config.table.FieldConfig.CompressionCodec;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/// Covers parsing of the `pinot.forward.index.default.compression.codec`
cluster config.
+///
+/// The value names a
[org.apache.pinot.spi.config.table.FieldConfig.CompressionCodec], the same
+/// vocabulary a table config uses, and anything that cannot serve as a raw
forward index chunk codec is
+/// ignored rather than applied: it would otherwise be injected after
table-config validation has run and
+/// fail at segment build time, across every table at once.
+///
+/// Not thread-safe: every method mutates the process-global
[ForwardIndexConfig] default, so this class
+/// must not run in parallel with other tests, or with anything that builds
segments.
+public class ServiceStartableUtilsTest {
+
+ /// Every assertion starts from a baseline that is NOT the value it expects,
so neither "applied" nor
+ /// "kept" can be satisfied by a no-op or by something that wrongly resolves
to LZ4.
+ private static final ChunkCompressionType BASELINE =
ChunkCompressionType.SNAPPY;
+
+ @BeforeMethod
+ public void setUp() {
+ ForwardIndexConfig.setDefaultCompressionCodec(CompressionCodec.SNAPPY);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void tearDown() {
+ ForwardIndexConfig.setDefaultCompressionCodec(CompressionCodec.LZ4);
+ }
+
+ @DataProvider
+ public Object[][] applicableCodecs() {
+ return new Object[][]{
+ {"ZSTANDARD", ChunkCompressionType.ZSTANDARD},
+ {"zstandard", ChunkCompressionType.ZSTANDARD},
+ {" ZSTANDARD ", ChunkCompressionType.ZSTANDARD},
+ {"SNAPPY", ChunkCompressionType.SNAPPY},
+ {"PASS_THROUGH", ChunkCompressionType.PASS_THROUGH},
+ {"LZ4", ChunkCompressionType.LZ4}
+ };
+ }
+
+ @Test(dataProvider = "applicableCodecs")
+ public void testApplicableCodecIsApplied(String configured,
ChunkCompressionType expected) {
+ if (expected == BASELINE) {
+ ForwardIndexConfig.setDefaultCompressionCodec(CompressionCodec.LZ4);
+ }
+
+ ServiceStartableUtils.setDefaultCompressionCodec(configured);
+
+ assertEquals(ForwardIndexConfig.getDefaultCompressionType(), expected);
+ }
+
+ /// `DELTA` and `DELTADELTA` matter most here: they are rejected by
table-config validation on every
+ /// column, and their compressors throw on any chunk whose size is not a
multiple of 4 bytes, so every
+ /// raw var-width column in the cluster would fail to build if one slipped
through as a default.
+ @DataProvider
+ public Object[][] codecsNotApplicableToRawIndexes() {
+ return new Object[][]{
+ {"DELTA"}, {"DELTADELTA"}, {"MV_ENTRY_DICT"}, {"CLP"}, {"CLPV2"},
{"CLPV2_ZSTD"}, {"CLPV2_LZ4"}
+ };
+ }
+
+ @Test(dataProvider = "codecsNotApplicableToRawIndexes")
+ public void testCodecNotApplicableToRawIndexIsIgnored(String configured) {
+ ServiceStartableUtils.setDefaultCompressionCodec(configured);
+
+ assertEquals(ForwardIndexConfig.getDefaultCompressionType(), BASELINE);
+ }
+
+ /// GZIP is applicable to a raw index, so the applicability guard does not
cover it; it is rejected
+ /// because its decompressor keeps one mutable Inflater that a column's
reader contexts share, which
+ /// makes it unsafe to put every raw non-metric column on by default.
+ @Test
+ public void testGzipIsRejectedAsADefault() {
+ ServiceStartableUtils.setDefaultCompressionCodec("GZIP");
+
+ assertEquals(ForwardIndexConfig.getDefaultCompressionType(), BASELINE);
+ }
+
+ /// `ZSTD` is the spelling the `codecSpec` DSL uses, and
`LZ4_LENGTH_PREFIXED` is an internal
+ /// ChunkCompressionType with no table-config counterpart; neither is a
CompressionCodec, so both are
+ /// ignored rather than silently accepted.
+ @DataProvider
+ public Object[][] unrecognisedValues() {
+ return new Object[][]{
+ {"ZSTD"}, {"LZ4_LENGTH_PREFIXED"}, {"not-a-codec"}, {""}, {" "}
+ };
+ }
+
+ @Test(dataProvider = "unrecognisedValues")
+ public void testUnrecognisedValueKeepsCurrentDefault(String configured) {
+ ServiceStartableUtils.setDefaultCompressionCodec(configured);
+
+ assertEquals(ForwardIndexConfig.getDefaultCompressionType(), BASELINE);
+ }
+
+ /// Covers the wiring from the config key to the setter: a refactor that
read the wrong constant, or
+ /// dropped the lookup, would otherwise pass every other test here while the
knob did nothing.
+ @Test
+ public void testTheConfigKeyReachesTheDefault() {
+ PinotConfiguration config = new PinotConfiguration();
+
config.setProperty(CommonConstants.ForwardIndexConfigs.CONFIG_OF_DEFAULT_COMPRESSION_CODEC,
"ZSTANDARD");
+
+ ServiceStartableUtils.initForwardIndexConfig(config);
+
+ assertEquals(ForwardIndexConfig.getDefaultCompressionType(),
ChunkCompressionType.ZSTANDARD);
+ }
+
+ @Test
+ public void testAbsentConfigKeyLeavesTheDefaultAlone() {
+ ServiceStartableUtils.initForwardIndexConfig(new PinotConfiguration());
+
+ assertEquals(ForwardIndexConfig.getDefaultCompressionType(), BASELINE);
+ }
+
+ @Test
+ public void testAnIgnoredValueDoesNotUndoAnEarlierValidOne() {
+ ServiceStartableUtils.setDefaultCompressionCodec("ZSTANDARD");
+ ServiceStartableUtils.setDefaultCompressionCodec("DELTA");
+
+ assertEquals(ForwardIndexConfig.getDefaultCompressionType(),
ChunkCompressionType.ZSTANDARD);
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
index e0efc2aa0a0..1ca8783bb4a 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
@@ -562,12 +562,17 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
// below discards any per-key codecSpec, so refuse to silently drop one
that slipped past validation.
ForwardIndexConfig configuredForwardIndex =
configsForDecision.getConfig(StandardIndexes.forward());
Preconditions.checkState(configuredForwardIndex.getCodecSpec() == null,
- "OPEN_STRUCT column '%s': codecSpec is not supported for key '%s';
materialized keys always use a "
- + "dictionary-encoded or LZ4 raw forward index", _columnName, key);
+ "OPEN_STRUCT column '%s': codecSpec is not supported for key '%s'; the
splitter fixes materialized "
+ + "keys to a dictionary-encoded or LZ4 raw forward index",
_columnName, key);
// Reconcile dictionary + forward encoding with the final decision
(mirrors BaseSegmentCreator.adaptConfig);
// ForwardIndexCreatorFactory selects dict-vs-raw from the forward
config's EncodingType. A compression codec
- // applies only to the raw forward format (LZ4 preserves the dense child's
current on-disk layout); attaching
+ // applies only to the raw forward format (LZ4 preserves the dense child's
current on-disk layout, and is
+ // pinned here regardless of the cluster-wide default compression codec).
Note the pin covers this build
+ // path only: a materialized key with no explicit indexes.forward node
resolves through
+ // ForwardIndexConfig.getDefault(RAW) with a null codec, so a
dictionary-to-raw conversion on reload
+ // writes the reloading component's default instead, and the same key can
end up with different codecs
+ // across segments. Attaching
// one to a dictionary-encoded forward is rejected by
ForwardIndexType.validate.
ForwardIndexConfig.Builder forwardBuilder = new ForwardIndexConfig.Builder(
useDictionary ? FieldConfig.EncodingType.DICTIONARY :
FieldConfig.EncodingType.RAW);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java
index b1f48d74c2c..77bfcb34d6d 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java
@@ -335,11 +335,26 @@ public class ForwardIndexType extends
AbstractIndexType<ForwardIndexConfig, Forw
return builder.build();
}
+ /// The compression type used for a raw column whose table config does not
name one.
+ ///
+ /// Metrics stay uncompressed: they are usually small fixed-width values
where compression costs more
+ /// than it saves. Everything else uses
[ForwardIndexConfig#getDefaultCompressionType()], which is LZ4
+ /// unless `pinot.forward.index.default.compression.codec` says otherwise.
That config is applied at
+ /// startup by components that read cluster config, so segments built
outside the cluster -- standalone,
+ /// Spark or Hadoop batch ingestion -- keep LZ4.
+ ///
+ /// This also decides the codec for a column listed in the legacy
`noDictionaryConfig`, whose per-column
+ /// codec value is discarded: only its key set is read, here and in
`DictionaryIndexType`. Such a column
+ /// used to land on a hardcoded LZ4 and now follows the configured default.
+ ///
+ /// This is a default, not a table config value, so changing it does not
rewrite existing segments:
+ /// `ForwardIndexHandler` deliberately ignores it when deciding whether a
reload must rewrite a forward
+ /// index.
public static ChunkCompressionType
getDefaultCompressionType(FieldSpec.FieldType fieldType) {
if (fieldType == FieldSpec.FieldType.METRIC) {
return ChunkCompressionType.PASS_THROUGH;
} else {
- return ChunkCompressionType.LZ4;
+ return ForwardIndexConfig.getDefaultCompressionType();
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java
index 920fd98d3ae..88355ba28de 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java
@@ -569,9 +569,10 @@ public class ForwardIndexHandler extends BaseIndexHandler {
"Legacy raw forward index for column=%s returned null
ChunkCompressionType", column);
ChunkCompressionType newCompressionType =
newConfig.getChunkCompressionType();
- // Note that default compression type (PASS_THROUGH for metric and LZ4 for
dimension) is not considered if the
- // compressionType is not explicitly provided in tableConfig. This is to
avoid incorrectly rewriting all the
- // forward indexes during segmentReload when the default compressionType
changes.
+ // Note that the default compression type (PASS_THROUGH for metric, and
otherwise whatever
+ // ForwardIndexConfig#getDefaultCompressionType() is set to) is not
considered if the compressionType is not
+ // explicitly provided in tableConfig. This is to avoid incorrectly
rewriting all the forward indexes during
+ // segmentReload when the default compressionType changes.
return newCompressionType != null && existingCompressionType !=
newCompressionType;
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
index dfb66045868..2144aaadc55 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
@@ -59,7 +59,7 @@ public class OpenStructIndexType
public static final String INDEX_DISPLAY_NAME = "open_struct";
private static final List<String> EXTENSIONS = List.of(".open_struct.idx");
private static final String FIXED_FORWARD_INDEX =
- "materialized keys always use a dictionary-encoded or LZ4 raw forward
index";
+ "the splitter fixes materialized keys to a dictionary-encoded or LZ4 raw
forward index";
private static final String KEY_ENCODING_RULE = "a key is dictionary-encoded
when its encodingType is "
+ "DICTIONARY, or indexes.dictionary enables one, or its inverted index
is enabled, and raw with LZ4 "
+ "otherwise";
@@ -164,6 +164,8 @@ public class OpenStructIndexType
JsonNode forwardIndex =
indexes.get(StandardIndexes.forward().getPrettyName());
// The OPEN_STRUCT splitter builds its own per-key forward-index configs
(dict-vs-raw decision plus a
// fixed LZ4 raw compression), so a per-key codecSpec would be silently
discarded. Reject it explicitly.
+ // The LZ4 pin is the splitter's: a key rebuilt by a dictionary-to-raw
conversion on reload instead takes
+ // the reloading component's default compression codec.
Preconditions.checkState(forwardIndex == null ||
!forwardIndex.hasNonNull("codecSpec"),
"OPEN_STRUCT column '%s': codecSpec is not supported for %s; " +
FIXED_FORWARD_INDEX, column, target);
Iterator<String> indexNames = indexes.fieldNames();
@@ -192,7 +194,8 @@ public class OpenStructIndexType
}
/// The splitter replaces a key's forward-index config with a
dictionary-encoded or LZ4 raw one, so
- /// `indexes.forward` may only restate that encoding or keep the index
enabled.
+ /// `indexes.forward` may only restate that encoding or keep the index
enabled. The LZ4 here is what the
+ /// splitter writes; a dictionary-to-raw conversion on reload uses the
configured default codec instead.
private static void validatePerKeyForwardIndex(JsonNode forwardIndex,
boolean dictionaryEncoded, String column,
String target) {
Preconditions.checkState(forwardIndex.isObject(), "OPEN_STRUCT column
'%s': indexes.forward of %s must be a JSON "
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
index 22057b34f98..d6891edab13 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
@@ -434,7 +434,7 @@ public class OpenStructColumnSplitterTest {
IllegalStateException exception =
expectThrows(IllegalStateException.class, splitter::seal);
assertEquals(exception.getMessage(), "OPEN_STRUCT column 'metrics':
codecSpec is not supported for key 'clicks'; "
- + "materialized keys always use a dictionary-encoded or LZ4 raw
forward index");
+ + "the splitter fixes materialized keys to a dictionary-encoded or LZ4
raw forward index");
}
private static FieldConfig rawCodecSpecFieldConfig(String name) {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexCreatorFactoryTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexCreatorFactoryTest.java
index 8da8be2c939..588d3029d58 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexCreatorFactoryTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexCreatorFactoryTest.java
@@ -47,6 +47,7 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.MetricFieldSpec;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -78,6 +79,40 @@ public class ForwardIndexCreatorFactoryTest {
}
}
+ /// The producer side of the configurable default: the factory fills a
codec-less config from
+ /// [ForwardIndexConfig#getDefaultCompressionType()], so this is what
decides the codec a segment is
+ /// actually written with. The test above pins the shipped LZ4; this one
proves the fill-in follows the
+ /// configured value rather than a constant.
+ @Test
+ public void testCreatorFollowsTheConfiguredDefaultCompressionType()
+ throws Exception {
+ File indexDir =
Files.createTempDirectory("ForwardIndexCreatorFactoryTest").toFile();
+
ForwardIndexConfig.setDefaultCompressionCodec(FieldConfig.CompressionCodec.ZSTANDARD);
+ try (ForwardIndexCreator creator =
ForwardIndexCreatorFactory.createIndexCreator(newContext(indexDir, true),
+ new ForwardIndexConfig.Builder(FieldConfig.EncodingType.RAW).build()))
{
+ assertEquals(creator.getRawForwardIndexChunkCompressionType(),
ChunkCompressionType.ZSTANDARD);
+ } finally {
+
ForwardIndexConfig.setDefaultCompressionCodec(FieldConfig.CompressionCodec.LZ4);
+ FileUtils.deleteQuietly(indexDir);
+ }
+ }
+
+ /// Metrics are exempt from the configured default, so a metric column keeps
PASS_THROUGH even when the
+ /// cluster asks for something else.
+ @Test
+ public void testMetricColumnIgnoresTheConfiguredDefaultCompressionType()
+ throws Exception {
+ File indexDir =
Files.createTempDirectory("ForwardIndexCreatorFactoryTest").toFile();
+
ForwardIndexConfig.setDefaultCompressionCodec(FieldConfig.CompressionCodec.ZSTANDARD);
+ try (ForwardIndexCreator creator =
ForwardIndexCreatorFactory.createIndexCreator(
+ newMetricContext(indexDir), new
ForwardIndexConfig.Builder(FieldConfig.EncodingType.RAW).build())) {
+ assertEquals(creator.getRawForwardIndexChunkCompressionType(),
ChunkCompressionType.PASS_THROUGH);
+ } finally {
+
ForwardIndexConfig.setDefaultCompressionCodec(FieldConfig.CompressionCodec.LZ4);
+ FileUtils.deleteQuietly(indexDir);
+ }
+ }
+
@Test
public void testCreatorReportsExplicitCompressionType()
throws Exception {
@@ -557,4 +592,21 @@ public class ForwardIndexCreatorFactoryTest {
.build();
return new IndexCreationContext.Builder(indexDir, tableConfig,
metadata).build();
}
+
+ /// A METRIC column, to check the exemption the configurable default must
not override.
+ private static IndexCreationContext newMetricContext(File indexDir) {
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").build();
+ FieldSpec fieldSpec = new MetricFieldSpec(COLUMN_NAME, DataType.INT);
+ ColumnMetadataImpl metadata = new ColumnMetadataImpl.Builder()
+ .setFieldSpec(fieldSpec)
+ .setTotalDocs(1)
+ .setCardinality(2)
+ .setHasDictionary(true)
+ .setLengthOfShortestElement(DataType.INT.size())
+ .setLengthOfLongestElement(DataType.INT.size())
+ .setTotalNumberOfEntries(1)
+ .setMaxNumberOfMultiValues(0)
+ .build();
+ return new IndexCreationContext.Builder(indexDir, tableConfig,
metadata).build();
+ }
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexDefaultCompressionTypeTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexDefaultCompressionTypeTest.java
new file mode 100644
index 00000000000..21ca4872dcf
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexDefaultCompressionTypeTest.java
@@ -0,0 +1,100 @@
+/**
+ * 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.forward;
+
+import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
+import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
+import org.apache.pinot.spi.config.table.FieldConfig.CompressionCodec;
+import org.apache.pinot.spi.data.FieldSpec.FieldType;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/// Tests that the configurable default compression type is honoured for raw
columns whose table config
+/// does not name a codec, and that metrics keep their uncompressed default
regardless.
+///
+/// Not thread-safe: every method mutates the process-global
[ForwardIndexConfig] default, so this class
+/// must not run in parallel with other tests, or with anything that builds
segments.
+public class ForwardIndexDefaultCompressionTypeTest {
+
+ /// Captured before any @BeforeMethod runs, so it observes the value
compiled into
+ /// [ForwardIndexConfig] rather than one a test set. The module runs one JVM
per test class
+ /// (`reuseForks=false`), so nothing else can have moved it first.
+ private ChunkCompressionType _compiledInDefault;
+
+ @BeforeClass
+ public void captureCompiledInDefault() {
+ _compiledInDefault = ForwardIndexConfig.getDefaultCompressionType();
+ }
+
+ /// Every "segments built outside the cluster keep LZ4" statement in this
feature's docs rests on this
+ /// value, and no other test asserts it: the rest set a default first.
+ @Test
+ public void testCompiledInDefaultIsLZ4() {
+ assertEquals(_compiledInDefault, ChunkCompressionType.LZ4);
+ }
+
+ /// Baseline is deliberately not LZ4: every assertion below expects
something else, so none of them can
+ /// pass on a no-op or on a regression that resolves to the old hardcoded
LZ4.
+ @BeforeMethod
+ public void setUp() {
+ ForwardIndexConfig.setDefaultCompressionCodec(CompressionCodec.SNAPPY);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void tearDown() {
+ ForwardIndexConfig.setDefaultCompressionCodec(CompressionCodec.LZ4);
+ }
+
+ @Test
+ public void testMetricIsUncompressedByDefault() {
+ assertEquals(ForwardIndexType.getDefaultCompressionType(FieldType.METRIC),
ChunkCompressionType.PASS_THROUGH);
+ }
+
+ /// Each case configures a codec the @BeforeMethod baseline is not, so a
regression that went back to
+ /// returning a hardcoded LZ4 -- or that ignored the configured default --
fails here. LZ4 is covered as a
+ /// configured value too, from a non-LZ4 baseline, for the same reason.
+ @DataProvider
+ public Object[][] nonMetricFieldTypes() {
+ return new Object[][]{{FieldType.DIMENSION}, {FieldType.DATE_TIME},
{FieldType.COMPLEX}};
+ }
+
+ @Test(dataProvider = "nonMetricFieldTypes")
+ public void testConfiguredDefaultAppliesToNonMetric(FieldType fieldType) {
+ ForwardIndexConfig.setDefaultCompressionCodec(CompressionCodec.ZSTANDARD);
+ assertEquals(ForwardIndexType.getDefaultCompressionType(fieldType),
ChunkCompressionType.ZSTANDARD);
+
+ ForwardIndexConfig.setDefaultCompressionCodec(CompressionCodec.LZ4);
+ assertEquals(ForwardIndexType.getDefaultCompressionType(fieldType),
ChunkCompressionType.LZ4);
+ }
+
+ /// Metrics are deliberately excluded: compressing small fixed-width values
usually costs more than it
+ /// saves, so raising the cluster-wide default must not start compressing
them.
+ @Test
+ public void testConfiguredDefaultDoesNotAffectMetric() {
+ ForwardIndexConfig.setDefaultCompressionCodec(CompressionCodec.ZSTANDARD);
+
+ assertEquals(ForwardIndexType.getDefaultCompressionType(FieldType.METRIC),
ChunkCompressionType.PASS_THROUGH);
+ }
+}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java
index 139f6e7a747..bdb43be7460 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java
@@ -213,7 +213,7 @@ public class OpenStructIndexTypeTest {
IllegalStateException exception = expectThrows(IllegalStateException.class,
() -> validateTableWithOpenStruct(openStructConfig));
assertEquals(exception.getMessage(), "OPEN_STRUCT column 'payload':
codecSpec is not supported for "
- + expectedTarget + "; materialized keys always use a
dictionary-encoded or LZ4 raw forward index");
+ + expectedTarget + "; the splitter fixes materialized keys to a
dictionary-encoded or LZ4 raw forward index");
}
/// Per-key settings that a materialized key would silently ignore. JSON
uses single quotes for readability.
@@ -344,7 +344,8 @@ public class OpenStructIndexTypeTest {
IllegalStateException exception =
expectThrows(IllegalStateException.class, () -> validateTableWithOpenStruct(
new OpenStructIndexConfig(false, null, -1, null, 0.5,
List.of(ignoredKey), null, null, null, null)));
assertEquals(exception.getMessage(), "OPEN_STRUCT column 'payload':
indexes.forward.targetDocsPerChunk is not "
- + "supported for key 'clicks'; materialized keys always use a
dictionary-encoded or LZ4 raw forward index");
+ + "supported for key 'clicks'; the splitter fixes materialized keys to
a dictionary-encoded or LZ4 "
+ + "raw forward index");
}
private static FieldConfig parseFieldConfig(String singleQuotedJson)
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java
index 9545225b1a7..0899e808d73 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java
@@ -22,9 +22,12 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.pinot.segment.spi.codec.CodecSpecParser;
import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
@@ -46,6 +49,7 @@ public class ForwardIndexConfig extends IndexConfig {
private static String _defaultTargetMaxChunkSize = "1MB";
private static int _defaultTargetMaxChunkSizeBytes = 1024 * 1024;
private static int _defaultTargetDocsPerChunk = 1000;
+ private static volatile ChunkCompressionType _defaultCompressionType =
ChunkCompressionType.LZ4;
public static int getDefaultRawWriterVersion() {
return _defaultRawIndexWriterVersion;
@@ -76,6 +80,87 @@ public class ForwardIndexConfig extends IndexConfig {
_defaultTargetDocsPerChunk = defaultTargetDocsPerChunk;
}
+ /// The chunk compression type for a raw column whose table config does not
name one. Defaults to LZ4.
+ ///
+ /// This is the raw value: the exemption that keeps metrics uncompressed is
applied by the caller, in
+ /// `ForwardIndexType` (pinot-segment-local), so a caller resolving a
default for a specific column must
+ /// go through that rather than reading this directly.
+ ///
+ /// `volatile` because it is written by a starter thread and read by
segment-build threads, and the
+ /// thread-start happens-before does not cover every launch path:
`startBootstrapServicesInParallel`
+ /// initialises roles on sibling threads, and the PinotServiceManager REST
endpoint adds a role to a JVM
+ /// that is already serving. A stale read here silently changes what gets
written to disk.
+ public static ChunkCompressionType getDefaultCompressionType() {
+ return _defaultCompressionType;
+ }
+
+ /// Sets the default compression type from the codec vocabulary a table
config uses.
+ ///
+ /// The codec must be applicable to a raw forward index, because this
default is applied in
+ /// `ForwardIndexCreatorFactory` after table config validation has run: a
codec the table config path
+ /// would have rejected has nothing left to catch it, and `DELTA` /
`DELTADELTA` in particular throw on
+ /// any chunk whose size is not a multiple of 4 bytes.
+ ///
+ /// `GZIP` is rejected as well, despite being applicable to a raw index. Its
decompressor keeps a single
+ /// mutable `Inflater`, and a chunk reader shares one decompressor across
all of a column's reader
+ /// contexts, so concurrent queries on a GZIP column already race. Naming it
per column is an existing,
+ /// opt-in exposure; making it the default would put every raw non-metric
column on that path.
+ public static void setDefaultCompressionCodec(CompressionCodec codec) {
+ Preconditions.checkArgument(codec != null, "Compression codec must not be
null");
+ Preconditions.checkArgument(codec.isApplicableToRawIndex(),
+ "Compression codec: %s is not applicable to raw forward indexes",
codec);
+ Preconditions.checkArgument(codec != CompressionCodec.GZIP,
+ "Compression codec: GZIP cannot be a default because its decompressor
is not thread-safe; it can "
+ + "still be set on an individual column");
+ ChunkCompressionType compressionType = toChunkCompressionType(codec);
+ // Unreachable today -- the only codec mapping to null is MV_ENTRY_DICT,
which the applicability check
+ // above already rejects -- but kept so a future raw-applicable codec
without a chunk type cannot leave
+ // the default null and NPE in ChunkCompressorFactory instead.
+ Preconditions.checkArgument(compressionType != null, "Compression codec:
%s has no chunk compression type", codec);
+ _defaultCompressionType = compressionType;
+ }
+
+ /// The codecs [#setDefaultCompressionCodec] accepts. Exposed so callers can
name them when reporting a
+ /// rejection, instead of restating the rule and letting the two drift.
+ public static List<CompressionCodec> getAcceptableDefaultCompressionCodecs()
{
+ return Arrays.stream(CompressionCodec.values())
+ .filter(c -> c.isApplicableToRawIndex() && c !=
CompressionCodec.GZIP).collect(Collectors.toList());
+ }
+
+ /// The chunk compression type a [CompressionCodec] maps to, or `null` when
the codec names a
+ /// dictionary-id format rather than a chunk codec. This is the one
authoritative mapping; note it is
+ /// deliberately not name-based, since the CLP family maps onto
`PASS_THROUGH`.
+ ///
+ /// Private on purpose: it maps codecs that [#setDefaultCompressionCodec]
exists to reject, so exposing it
+ /// would offer a way around those guards.
+ @Nullable
+ private static ChunkCompressionType toChunkCompressionType(CompressionCodec
codec) {
+ switch (codec) {
+ case PASS_THROUGH:
+ case CLP:
+ case CLPV2:
+ case CLPV2_ZSTD:
+ case CLPV2_LZ4:
+ return ChunkCompressionType.PASS_THROUGH;
+ case SNAPPY:
+ return ChunkCompressionType.SNAPPY;
+ case ZSTANDARD:
+ return ChunkCompressionType.ZSTANDARD;
+ case LZ4:
+ return ChunkCompressionType.LZ4;
+ case GZIP:
+ return ChunkCompressionType.GZIP;
+ case DELTA:
+ return ChunkCompressionType.DELTA;
+ case DELTADELTA:
+ return ChunkCompressionType.DELTADELTA;
+ case MV_ENTRY_DICT:
+ return null;
+ default:
+ throw new IllegalStateException("Unsupported compression codec: " +
codec);
+ }
+ }
+
public static ForwardIndexConfig getDefault(EncodingType encodingType) {
return new Builder(encodingType).build();
}
@@ -135,46 +220,9 @@ public class ForwardIndexConfig extends IndexConfig {
_targetDocsPerChunk = targetDocsPerChunk == null ?
_defaultTargetDocsPerChunk : targetDocsPerChunk;
_configs = configs != null ? configs : new HashMap<>();
if (_compressionCodec != null) {
- switch (_compressionCodec) {
- case PASS_THROUGH:
- case CLP:
- case CLPV2:
- case CLPV2_ZSTD:
- case CLPV2_LZ4:
- _chunkCompressionType = ChunkCompressionType.PASS_THROUGH;
- _dictIdCompressionType = null;
- break;
- case SNAPPY:
- _chunkCompressionType = ChunkCompressionType.SNAPPY;
- _dictIdCompressionType = null;
- break;
- case ZSTANDARD:
- _chunkCompressionType = ChunkCompressionType.ZSTANDARD;
- _dictIdCompressionType = null;
- break;
- case LZ4:
- _chunkCompressionType = ChunkCompressionType.LZ4;
- _dictIdCompressionType = null;
- break;
- case GZIP:
- _chunkCompressionType = ChunkCompressionType.GZIP;
- _dictIdCompressionType = null;
- break;
- case DELTA:
- _chunkCompressionType = ChunkCompressionType.DELTA;
- _dictIdCompressionType = null;
- break;
- case DELTADELTA:
- _chunkCompressionType = ChunkCompressionType.DELTADELTA;
- _dictIdCompressionType = null;
- break;
- case MV_ENTRY_DICT:
- _dictIdCompressionType = DictIdCompressionType.MV_ENTRY_DICT;
- _chunkCompressionType = null;
- break;
- default:
- throw new IllegalStateException("Unsupported compression codec: " +
_compressionCodec);
- }
+ _chunkCompressionType = toChunkCompressionType(_compressionCodec);
+ _dictIdCompressionType =
+ _compressionCodec == CompressionCodec.MV_ENTRY_DICT ?
DictIdCompressionType.MV_ENTRY_DICT : null;
} else {
_dictIdCompressionType = null;
_chunkCompressionType = null;
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 4b31ac6e346..5a2a7abc8d1 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -3049,6 +3049,8 @@ public class CommonConstants {
"pinot.forward.index.default.target.max.chunk.size";
public static final String CONFIG_OF_DEFAULT_TARGET_DOCS_PER_CHUNK =
"pinot.forward.index.default.target.docs.per.chunk";
+ public static final String CONFIG_OF_DEFAULT_COMPRESSION_CODEC =
+ "pinot.forward.index.default.compression.codec";
}
public static class FieldSpecConfigs {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
