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 866bb59cb9e Report V7 compression stats and reject per-key OPEN_STRUCT
codecSpec (#19308)
866bb59cb9e is described below
commit 866bb59cb9e16f87ecbaad30717c7382ff35c480
Author: Xiang Fu <[email protected]>
AuthorDate: Mon Sep 28 23:12:13 2026 +0700
Report V7 compression stats and reject per-key OPEN_STRUCT codecSpec
(#19308)
* Support codecSpec on segment reload and enable codecSpec end to end
* Harden codec reload and compression stats
* Report uncompressed value size for V7 codec-pipeline columns
This PR enabled compression statistics for V7 columns (the metadata and
reader no longer require a legacy ChunkCompressionType), but the creator
still returned -1 for the V7 writer's uncompressed value size. That -1 maps
to UNAVAILABLE, so codecSpec columns never reported compression stats at
creation or on reload-to-V7, and testV7SegmentCreationPersistsAndReports-
CompressionStats failed.
The V7 fixed-byte layout stores exactly one value per doc, so the
uncompressed size is totalDocs * valueType.size(). Report it when tracking
is enabled, mirroring the legacy writer's contract. Update the V7 round-trip
assertion in ForwardIndexCreatorFactoryTest, which encoded the old -1.
* Address review feedback on V7 compression stats and OPEN_STRUCT codecSpec
- Report the V7 uncompressed value size from the values actually written
(new FixedByteChunkForwardIndexWriterV7.getUncompressedValueSizeInBytes)
instead of the declared doc count, matching the legacy writer.
- Scope the codecSpec guidance in the CompressionCodec Javadoc to
single-value INT/LONG raw columns on V7-capable clusters, and correct
the DELTA wording for columns without a forward index.
- Name the OPEN_STRUCT column, and the key or defaultValueFieldConfig, in
codecSpec rejection errors; document which per-key settings OPEN_STRUCT
keys honor.
- Add V7 reload compression-stats tests to
ForwardIndexHandlerCompressionStatsTest.
- Fold the duplicate codecSpec validation cases into the existing
TableConfigUtilsTest case, and drop the no-op noDictionaryColumns case
and its TableConfigUtils comment.
- Use static Mockito imports and the DataType import in
ForwardIndexTypeTest, and the non-deprecated OpenStructIndexConfig
constructor in the new OPEN_STRUCT tests.
---------
Co-authored-by: intentlab-ai[bot]
<intentlab-ai[bot]@users.noreply.github.com>
---
.../ColumnCompressionStatsContribution.java | 3 +-
.../resources/ColumnCompressionStatsInfo.java | 3 +-
.../impl/FixedByteChunkForwardIndexWriterV7.java | 6 +
...ompressionStatsTrackingForwardIndexCreator.java | 3 +-
.../fwd/SingleValueFixedByteRawIndexCreator.java | 13 +-
.../impl/openstruct/OpenStructColumnSplitter.java | 8 ++
.../index/forward/CompressionStatsMetadata.java | 6 +-
.../index/openstruct/OpenStructIndexType.java | 47 ++++---
.../openstruct/OpenStructColumnSplitterTest.java | 39 ++++++
.../forward/CompressionStatsMetadataTest.java | 6 +
.../forward/ForwardIndexCreatorFactoryTest.java | 13 +-
.../index/forward/ForwardIndexTypeTest.java | 35 +++++
.../ForwardIndexHandlerCompressionStatsTest.java | 145 ++++++++++++++++++++-
.../index/openstruct/OpenStructIndexTypeTest.java | 50 +++++++
.../segment/local/utils/TableConfigUtilsTest.java | 18 +++
.../resources/SegmentCompressionStatsReader.java | 2 +-
.../SegmentCompressionStatsReaderTest.java | 77 +++++++++++
.../apache/pinot/spi/config/table/FieldConfig.java | 34 ++++-
.../spi/config/table/OpenStructIndexConfig.java | 4 +-
19 files changed, 475 insertions(+), 37 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ColumnCompressionStatsContribution.java
b/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ColumnCompressionStatsContribution.java
index cce98a1ab5f..240df22bc5c 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ColumnCompressionStatsContribution.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ColumnCompressionStatsContribution.java
@@ -119,7 +119,8 @@ public final class ColumnCompressionStatsContribution {
return _encoding;
}
- /// Returns the raw forward-index chunk-compression type, or null for
dictionary encoding.
+ /// Returns the raw forward-index chunk-compression type, or null for
dictionary encoding and raw codec-pipeline
+ /// formats that cannot be represented by one legacy
[ChunkCompressionType].
@Nullable
@JsonInclude(JsonInclude.Include.NON_NULL)
public ChunkCompressionType getChunkCompressionType() {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ColumnCompressionStatsInfo.java
b/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ColumnCompressionStatsInfo.java
index 5ea44700c35..943876cbae7 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ColumnCompressionStatsInfo.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/ColumnCompressionStatsInfo.java
@@ -187,7 +187,8 @@ public final class ColumnCompressionStatsInfo {
return _encoding;
}
- /// Returns the raw forward-index chunk-compression type, or null for
dictionary encoding.
+ /// Returns the raw forward-index chunk-compression type, or null for
dictionary encoding and raw codec-pipeline
+ /// formats that cannot be represented by one legacy
[ChunkCompressionType].
@Nullable
@JsonInclude(JsonInclude.Include.NON_NULL)
public ChunkCompressionType getChunkCompressionType() {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/writer/impl/FixedByteChunkForwardIndexWriterV7.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/writer/impl/FixedByteChunkForwardIndexWriterV7.java
index a35dbf6687c..ca254bea315 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/writer/impl/FixedByteChunkForwardIndexWriterV7.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/writer/impl/FixedByteChunkForwardIndexWriterV7.java
@@ -255,6 +255,12 @@ public class FixedByteChunkForwardIndexWriterV7 implements
FixedByteChunkWriter
throw new UnsupportedOperationException("V7 codec-pipeline writer does not
support DOUBLE");
}
+ /// Returns the value bytes written so far (`docsWritten * sizeOfEntry`),
including values in the chunk that has
+ /// not been flushed yet. These are the bytes presented to the codec
pipeline before any transform or compression.
+ public long getUncompressedValueSizeInBytes() {
+ return (long) _docsWritten * _sizeOfEntry;
+ }
+
/// Fail fast at write time if the caller would exceed the declared
`totalDocs`. Without this
/// guard the writer keeps producing chunks past the declared length and
only `close()` catches
/// the mismatch, leaving a semantically-invalid partial file behind.
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/fwd/CompressionStatsTrackingForwardIndexCreator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/fwd/CompressionStatsTrackingForwardIndexCreator.java
index 6550f6553e6..8fc541dfb44 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/fwd/CompressionStatsTrackingForwardIndexCreator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/fwd/CompressionStatsTrackingForwardIndexCreator.java
@@ -34,7 +34,8 @@ public interface CompressionStatsTrackingForwardIndexCreator
extends ForwardInde
@Override
long getRawForwardIndexUncompressedValueSizeInBytes();
- /// Returns the raw forward-index chunk-compression type, or null when
tracking is disabled.
+ /// Returns the raw forward-index chunk-compression type, or null when
tracking is disabled or the
+ /// raw format cannot be represented by one legacy [ChunkCompressionType].
@Override
@Nullable
ChunkCompressionType getRawForwardIndexChunkCompressionType();
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/fwd/SingleValueFixedByteRawIndexCreator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/fwd/SingleValueFixedByteRawIndexCreator.java
index 77bc481de89..b7e9a712e00 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/fwd/SingleValueFixedByteRawIndexCreator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/fwd/SingleValueFixedByteRawIndexCreator.java
@@ -20,6 +20,7 @@ package
org.apache.pinot.segment.local.segment.creator.impl.fwd;
import java.io.File;
import java.io.IOException;
+import javax.annotation.Nullable;
import org.apache.pinot.segment.local.io.codec.CodecPipelineExecutor;
import
org.apache.pinot.segment.local.io.writer.impl.FixedByteChunkForwardIndexWriter;
import
org.apache.pinot.segment.local.io.writer.impl.FixedByteChunkForwardIndexWriterV7;
@@ -40,7 +41,11 @@ import org.apache.pinot.spi.data.FieldSpec.DataType;
public class SingleValueFixedByteRawIndexCreator implements
CompressionStatsTrackingForwardIndexCreator {
private final FixedByteChunkWriter _indexWriter;
private final DataType _valueType;
+ @Nullable
private final ChunkCompressionType _chunkCompressionType;
+ // Whether compression-statistics tracking was requested. Mirrors the legacy
writer's contract of
+ // reporting an uncompressed value size only when tracking is enabled.
+ private boolean _trackUncompressedValueSize;
/// Constructor for the class
///
@@ -138,20 +143,26 @@ public class SingleValueFixedByteRawIndexCreator
implements CompressionStatsTrac
@Override
public long getRawForwardIndexUncompressedValueSizeInBytes() {
- // Compression-statistics metadata supports only the legacy
single-compressor format.
if (_indexWriter instanceof FixedByteChunkForwardIndexWriter legacyWriter)
{
return legacyWriter.getRawForwardIndexUncompressedValueSizeInBytes();
}
+ // V7 codec-pipeline writer: like the legacy writer, report the value
bytes actually written, or -1 unless
+ // tracking was requested.
+ if (_trackUncompressedValueSize && _indexWriter instanceof
FixedByteChunkForwardIndexWriterV7 v7Writer) {
+ return v7Writer.getUncompressedValueSizeInBytes();
+ }
return -1;
}
@Override
+ @Nullable
public ChunkCompressionType getRawForwardIndexChunkCompressionType() {
return _chunkCompressionType;
}
@Override
public void enableRawForwardIndexUncompressedValueSizeTracking() {
+ _trackUncompressedValueSize = true;
if (_indexWriter instanceof FixedByteChunkForwardIndexWriter legacyWriter)
{
legacyWriter.enableRawForwardIndexUncompressedValueSizeTracking();
}
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 d86cd641b9f..79664a04f22 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
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.segment.local.segment.creator.impl.openstruct;
+import com.google.common.base.Preconditions;
import com.google.common.base.Utf8;
import java.io.File;
import java.io.IOException;
@@ -543,6 +544,13 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
boolean useDictionary = resolveUseDictionary(childFieldSpec,
configsForDecision, statsCollector);
+ // Defense-in-depth mirror of OpenStructIndexType.validatePerKeyIndexes:
the child forward config built
+ // 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);
+
// 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
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadata.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadata.java
index 42329efa424..95fe70bb1af 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadata.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadata.java
@@ -47,10 +47,12 @@ public final class CompressionStatsMetadata {
_dictionaryEncodedUncompressedValueSizeInBytes =
dictionaryUncompressedValueSizeInBytes;
}
- /// Creates metadata for a raw forward index, or unavailable metadata when
either required value is absent.
+ /// Creates metadata for a raw forward index, or unavailable metadata when
the size was not tracked. The
+ /// compression type is nullable because codec-pipeline V7 indexes do not
have one legacy
+ /// [ChunkCompressionType]; their uncompressed size is still valid and must
remain reportable.
public static CompressionStatsMetadata forRawForwardIndex(long
uncompressedValueSizeInBytes,
@Nullable ChunkCompressionType chunkCompressionType) {
- return uncompressedValueSizeInBytes >= 0 && chunkCompressionType != null
+ return uncompressedValueSizeInBytes >= 0
? new CompressionStatsMetadata(uncompressedValueSizeInBytes,
chunkCompressionType, null) : UNAVAILABLE;
}
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 ebdad5e4afe..5dfcd8c72e8 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
@@ -20,7 +20,6 @@ package
org.apache.pinot.segment.local.segment.index.openstruct;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.base.Preconditions;
-import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -86,7 +85,7 @@ public class OpenStructIndexType
Preconditions.checkState(fieldSpec.isSingleValueField(),
"OPEN_STRUCT index can only be created on single-value columns, but
column '%s' is multi-value",
fieldSpec.getName());
- validatePerKeyIndexes(config);
+ validatePerKeyIndexes(config, fieldSpec.getName());
validateIgnoredKeys(config, fieldSpec);
if (fieldSpec instanceof ComplexFieldSpec) {
validateChildFieldSpecTypes((ComplexFieldSpec) fieldSpec);
@@ -124,26 +123,36 @@ public class OpenStructIndexType
}
}
- private void validatePerKeyIndexes(OpenStructIndexConfig config) {
- List<FieldConfig> fieldConfigs = new ArrayList<>();
+ private void validatePerKeyIndexes(OpenStructIndexConfig config, String
column) {
if (config.getValueFieldConfigs() != null) {
- fieldConfigs.addAll(config.getValueFieldConfigs());
+ for (FieldConfig fieldConfig : config.getValueFieldConfigs()) {
+ validatePerKeyFieldConfig(fieldConfig, column, "key '" +
fieldConfig.getName() + "'");
+ }
}
- if (config.getDefaultValueFieldConfig() != null) {
- fieldConfigs.add(config.getDefaultValueFieldConfig());
+ FieldConfig defaultValueFieldConfig = config.getDefaultValueFieldConfig();
+ if (defaultValueFieldConfig != null) {
+ validatePerKeyFieldConfig(defaultValueFieldConfig, column,
"defaultValueFieldConfig");
}
- for (FieldConfig fieldConfig : fieldConfigs) {
- JsonNode indexes = fieldConfig.getIndexes();
- if (indexes == null) {
- continue;
- }
- Iterator<String> indexNames = indexes.fieldNames();
- while (indexNames.hasNext()) {
- String indexName = indexNames.next();
-
Preconditions.checkState(OpenStructSupportedIndexes.ALLOWED_PRETTY_NAMES.contains(indexName),
- "OPEN_STRUCT key '%s' declares unsupported index '%s'; supported
indexes are %s",
- fieldConfig.getName(), indexName,
OpenStructSupportedIndexes.ALLOWED_PRETTY_NAMES);
- }
+ }
+
+ /// Validates one per-key [FieldConfig] of OPEN_STRUCT `column`; `target`
names it in error messages.
+ private static void validatePerKeyFieldConfig(FieldConfig fieldConfig,
String column, String target) {
+ JsonNode indexes = fieldConfig.getIndexes();
+ if (indexes == null) {
+ return;
+ }
+ 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.
+ Preconditions.checkState(forwardIndex == null ||
!forwardIndex.hasNonNull("codecSpec"),
+ "OPEN_STRUCT column '%s': codecSpec is not supported for %s;
materialized keys always use a "
+ + "dictionary-encoded or LZ4 raw forward index", column, target);
+ Iterator<String> indexNames = indexes.fieldNames();
+ while (indexNames.hasNext()) {
+ String indexName = indexNames.next();
+
Preconditions.checkState(OpenStructSupportedIndexes.ALLOWED_PRETTY_NAMES.contains(indexName),
+ "OPEN_STRUCT key '%s' declares unsupported index '%s'; supported
indexes are %s",
+ fieldConfig.getName(), indexName,
OpenStructSupportedIndexes.ALLOWED_PRETTY_NAMES);
}
}
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 03f97a38213..22057b34f98 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
@@ -20,6 +20,7 @@ package
org.apache.pinot.segment.local.segment.creator.impl.openstruct;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.File;
import java.math.BigDecimal;
import java.nio.file.Files;
@@ -51,6 +52,7 @@ import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.buffer.MutableRoaringBitmap;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -65,6 +67,7 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
public class OpenStructColumnSplitterTest {
@@ -97,6 +100,16 @@ public class OpenStructColumnSplitterTest {
perKeyMetricsEnabled);
}
+ @DataProvider(name = "codecSpecOpenStructConfigs")
+ public Object[][] codecSpecOpenStructConfigs() {
+ FieldConfig valueFieldConfig = rawCodecSpecFieldConfig("clicks");
+ FieldConfig defaultValueFieldConfig = rawCodecSpecFieldConfig("default");
+ return new Object[][]{
+ {new OpenStructIndexConfig(false, null, -1, null, 0.5,
List.of(valueFieldConfig), null, null, null)},
+ {new OpenStructIndexConfig(false, defaultValueFieldConfig, -1, null,
0.5, null, null, null, null)}
+ };
+ }
+
@Test
public void testClassifyByFillRate()
throws Exception {
@@ -410,6 +423,32 @@ public class OpenStructColumnSplitterTest {
denseCol +
V1Constants.Indexes.RAW_SV_FORWARD_INDEX_FILE_EXTENSION).exists());
}
+ @Test(dataProvider = "codecSpecOpenStructConfigs")
+ public void
testSealRejectsCodecSpecBeforeReplacingChildForwardConfig(OpenStructIndexConfig
openStructConfig)
+ throws Exception {
+ OpenStructColumnSplitter splitter = new OpenStructColumnSplitter(
+ _tempDir, "metrics", "testTable_OFFLINE", spec(), openStructConfig);
+ for (int docId = 0; docId < 10; docId++) {
+ splitter.add(Map.of("clicks", docId), docId);
+ }
+
+ 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");
+ }
+
+ private static FieldConfig rawCodecSpecFieldConfig(String name) {
+ ObjectNode forward = JsonUtils.newObjectNode();
+ forward.put("encodingType", FieldConfig.EncodingType.RAW.name());
+ forward.put("codecSpec", "LZ4");
+ ObjectNode indexes = JsonUtils.newObjectNode();
+ indexes.set("forward", forward);
+ return new FieldConfig.Builder(name)
+ .withEncodingType(FieldConfig.EncodingType.RAW)
+ .withIndexes(indexes)
+ .build();
+ }
+
@Test
public void testRangeAndBloomIndexesWrittenForKey()
throws Exception {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadataTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadataTest.java
index 88aea24a722..e9cba4ca115 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadataTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadataTest.java
@@ -51,6 +51,12 @@ public class CompressionStatsMetadataTest {
assertNull(
properties.get(getKeyFor(COLUMN,
FORWARD_INDEX_DICTIONARY_ENCODED_UNCOMPRESSED_VALUE_SIZE_IN_BYTES)));
+ CompressionStatsMetadata.forRawForwardIndex(64, null).applyTo(properties,
COLUMN);
+ assertEquals(properties.get(getKeyFor(COLUMN,
FORWARD_INDEX_RAW_UNCOMPRESSED_VALUE_SIZE_IN_BYTES)), "64");
+ assertNull(properties.get(getKeyFor(COLUMN,
FORWARD_INDEX_RAW_CHUNK_COMPRESSION_TYPE)));
+ assertNull(
+ properties.get(getKeyFor(COLUMN,
FORWARD_INDEX_DICTIONARY_ENCODED_UNCOMPRESSED_VALUE_SIZE_IN_BYTES)));
+
CompressionStatsMetadata.unavailable().applyTo(properties, COLUMN);
assertNull(properties.get(getKeyFor(COLUMN,
FORWARD_INDEX_RAW_UNCOMPRESSED_VALUE_SIZE_IN_BYTES)));
assertNull(properties.get(getKeyFor(COLUMN,
FORWARD_INDEX_RAW_CHUNK_COMPRESSION_TYPE)));
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 85f3d963f62..8da8be2c939 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
@@ -138,7 +138,8 @@ public class ForwardIndexCreatorFactoryTest {
long[] values = storedType == DataType.INT
? new long[]{11, 13, 21}
: new long[]{Long.MIN_VALUE, (long) Integer.MAX_VALUE + 1,
Long.MAX_VALUE};
- // Legacy compression stats are opted in here on purpose: a codecSpec
column never reports them.
+ // Compression stats are opted in here on purpose: a V7 codecSpec column
reports its uncompressed
+ // value size (values written * storedType.size()) even though it has no
legacy ChunkCompressionType.
File indexFile = roundTripCodecSpecIndex(indexDir, codecSpec,
storedType, 2, values, true, 0);
try (PinotDataBuffer buffer =
PinotDataBuffer.mapReadOnlyBigEndianFile(indexFile)) {
assertEquals(buffer.getInt(0),
FixedByteChunkSVForwardIndexReaderV7.VERSION);
@@ -188,9 +189,10 @@ public class ForwardIndexCreatorFactoryTest {
/// Writes `values` through a `codecSpec` forward-index creator and reads
every one of them back, then
/// returns the index file so the caller can assert on its header. Covers
the parts both codecSpec round
/// trips share: the creator contract that holds for any codecSpec column
(not dictionary encoded, no
- /// legacy chunk compression type, and no legacy uncompressed value size
even when the table opts into
- /// compression stats), that the pipeline routed the file to the V7 reader,
and a sequential read of every
- /// value. `randomReads` further reads land on random doc ids through a
second, random-access context.
+ /// legacy chunk compression type, and an uncompressed value size of
`values.length * storedType.size()` when
+ /// the table opts into compression stats and UNAVAILABLE when it does not),
that the pipeline routed the
+ /// file to the V7 reader, and a sequential read of every value.
`randomReads` further reads land on
+ /// random doc ids through a second, random-access context.
private static File roundTripCodecSpecIndex(File indexDir, String codecSpec,
DataType storedType,
int targetDocsPerChunk, long[] values, boolean compressionStatsEnabled,
int randomReads)
throws Exception {
@@ -210,7 +212,8 @@ public class ForwardIndexCreatorFactoryTest {
}
}
creator.seal();
- assertEquals(creator.getRawForwardIndexUncompressedValueSizeInBytes(),
-1L);
+ assertEquals(creator.getRawForwardIndexUncompressedValueSizeInBytes(),
+ compressionStatsEnabled ? (long) values.length * storedType.size() :
-1L);
}
File indexFile = new File(indexDir, COLUMN_NAME +
V1Constants.Indexes.RAW_SV_FORWARD_INDEX_FILE_EXTENSION);
assertTrue(indexFile.exists());
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexTypeTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexTypeTest.java
index 43fde236a76..765f8ef7d76 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexTypeTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexTypeTest.java
@@ -25,17 +25,25 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Map;
import java.util.stream.Collectors;
+import org.apache.pinot.segment.local.io.writer.impl.DirectMemoryManager;
+import
org.apache.pinot.segment.local.realtime.impl.forward.FixedByteSVMutableForwardIndex;
import org.apache.pinot.segment.local.segment.index.AbstractSerdeIndexContract;
import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
import org.apache.pinot.segment.spi.compression.DictIdCompressionType;
import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
import org.apache.pinot.segment.spi.index.StandardIndexes;
+import org.apache.pinot.segment.spi.index.mutable.MutableIndex;
+import org.apache.pinot.segment.spi.index.mutable.provider.MutableIndexContext;
import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.utils.JsonUtils;
import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertSame;
@@ -495,4 +503,31 @@ public class ForwardIndexTypeTest {
assertSame(StandardIndexes.forward(), StandardIndexes.forward(), "Standard
index should use the same as "
+ "the ForwardIndexType static instance");
}
+
+ /// codecSpec applies only at immutable segment creation/conversion time.
The mutable (consuming)
+ /// forward index must build the standard in-memory format, ignoring the
configured codecSpec, so
+ /// realtime tables with a codecSpec keep consuming normally.
+ @Test
+ public void testCodecSpecBuildsStandardMutableIndexForRealtime()
+ throws Exception {
+ MutableIndexContext context = mock(MutableIndexContext.class);
+ when(context.getFieldSpec()).thenReturn(new DimensionFieldSpec("dimInt",
DataType.INT, true));
+ when(context.getSegmentName()).thenReturn("testSegment");
+ when(context.getCapacity()).thenReturn(16);
+ ForwardIndexConfig config = new
ForwardIndexConfig.Builder(FieldConfig.EncodingType.RAW)
+ .withCodecSpec("DELTA,LZ4")
+ .build();
+
+ try (DirectMemoryManager memoryManager = new
DirectMemoryManager("testSegment")) {
+ when(context.getMemoryManager()).thenReturn(memoryManager);
+ MutableIndex mutableIndex =
StandardIndexes.forward().createMutableIndex(context, config);
+ assertNotNull(mutableIndex);
+ try {
+ assertTrue(mutableIndex instanceof FixedByteSVMutableForwardIndex,
+ "Expected the standard SV mutable forward index, got: " +
mutableIndex.getClass());
+ } finally {
+ mutableIndex.close();
+ }
+ }
+ }
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerCompressionStatsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerCompressionStatsTest.java
index 369bb5400db..9ac7e03fd5b 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerCompressionStatsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerCompressionStatsTest.java
@@ -27,15 +27,19 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import
org.apache.pinot.segment.local.segment.index.readers.forward.FixedByteChunkSVForwardIndexReaderV7;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
import org.apache.pinot.segment.local.segment.store.SegmentLocalFSDirectory;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.V1Constants;
import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.segment.spi.store.SegmentDirectory;
import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils;
import org.apache.pinot.spi.config.table.FieldConfig;
@@ -57,8 +61,8 @@ import static org.testng.Assert.*;
/// Tests compression metadata persistence across [ForwardIndexHandler] reload
operations.
///
-/// Coverage includes chunk-compression changes, dictionary/raw conversions,
forward-index disable/re-enable, and
-/// stale metadata cleanup.
+/// Coverage includes chunk-compression changes, codec-pipeline (V7) rewrites
and rollback, dictionary/raw
+/// conversions, forward-index disable/re-enable, and stale metadata cleanup.
public class ForwardIndexHandlerCompressionStatsTest {
private static final String RAW_TABLE_NAME = "compressionStatsReloadTest";
private static final String SEGMENT_NAME = "compressionStatsReloadSegment";
@@ -604,4 +608,141 @@ public class ForwardIndexHandlerCompressionStatsTest {
assertTrue(dictMeta.getDictionaryEncodedUncompressedValueSizeInBytes() > 0,
"Dictionary uncompressed value size should be collected during
raw-to-dictionary conversion");
}
+
+ @Test
+ public void testLegacyToCodecSpecRewritePersistsRawSizeWithoutChunkType()
+ throws Exception {
+ ColumnMetadata legacyMeta = new
SegmentMetadataImpl(INDEX_DIR).getColumnMetadataFor(RAW_INT_COL);
+ assertEquals(legacyMeta.getRawForwardIndexChunkCompressionType(),
ChunkCompressionType.SNAPPY);
+ assertEquals(legacyMeta.getRawForwardIndexUncompressedValueSizeInBytes(),
(long) NUM_ROWS * Integer.BYTES);
+
+ _fieldConfigMap.put(RAW_INT_COL, codecSpecFieldConfig(RAW_INT_COL,
"DELTA,ZSTD(3)", false));
+ updateForwardIndex(createIndexLoadingConfig());
+
+ // A V7 index has no single legacy chunk-compression type, but its
uncompressed value size is still reported.
+ assertEquals(readCodecSpec(RAW_INT_COL), "DELTA,ZSTD(3)");
+ ColumnMetadata v7Meta = new
SegmentMetadataImpl(INDEX_DIR).getColumnMetadataFor(RAW_INT_COL);
+ assertNull(v7Meta.getRawForwardIndexChunkCompressionType());
+ assertEquals(v7Meta.getRawForwardIndexUncompressedValueSizeInBytes(),
(long) NUM_ROWS * Integer.BYTES);
+ }
+
+ @Test
+ public void testCodecSpecChangeAndLegacyRollbackPersistRawStats()
+ throws Exception {
+ _fieldConfigMap.put(RAW_INT_COL, codecSpecFieldConfig(RAW_INT_COL,
"DELTA,LZ4", false));
+ updateForwardIndex(createIndexLoadingConfig());
+ _fieldConfigMap.put(RAW_INT_COL, codecSpecFieldConfig(RAW_INT_COL,
"T64,LZ4", false));
+ updateForwardIndex(createIndexLoadingConfig());
+
+ assertEquals(readCodecSpec(RAW_INT_COL), "T64,LZ4");
+ ColumnMetadata v7Meta = new
SegmentMetadataImpl(INDEX_DIR).getColumnMetadataFor(RAW_INT_COL);
+ assertNull(v7Meta.getRawForwardIndexChunkCompressionType());
+ assertEquals(v7Meta.getRawForwardIndexUncompressedValueSizeInBytes(),
(long) NUM_ROWS * Integer.BYTES);
+
+ // Rolling back to a legacy codec restores the legacy chunk-compression
type alongside the size.
+ _fieldConfigMap.put(RAW_INT_COL,
+ new FieldConfig(RAW_INT_COL, FieldConfig.EncodingType.RAW, List.of(),
CompressionCodec.LZ4, null));
+ updateForwardIndex(createIndexLoadingConfig());
+
+ assertNull(readCodecSpec(RAW_INT_COL));
+ ColumnMetadata legacyMeta = new
SegmentMetadataImpl(INDEX_DIR).getColumnMetadataFor(RAW_INT_COL);
+ assertEquals(legacyMeta.getRawForwardIndexChunkCompressionType(),
ChunkCompressionType.LZ4);
+ assertEquals(legacyMeta.getRawForwardIndexUncompressedValueSizeInBytes(),
(long) NUM_ROWS * Integer.BYTES);
+ }
+
+ @Test
+ public void testDictToCodecSpecPersistsRawStats()
+ throws Exception {
+ _noDictionaryColumns.add(DICT_INT_COL);
+ _fieldConfigMap.put(DICT_INT_COL, codecSpecFieldConfig(DICT_INT_COL,
"DELTA,LZ4", false));
+ updateForwardIndex(createIndexLoadingConfig());
+
+ assertEquals(readCodecSpec(DICT_INT_COL), "DELTA,LZ4");
+ ColumnMetadata colMeta = new
SegmentMetadataImpl(INDEX_DIR).getColumnMetadataFor(DICT_INT_COL);
+ assertFalse(colMeta.hasDictionary());
+ assertNull(colMeta.getRawForwardIndexChunkCompressionType());
+ assertEquals(colMeta.getRawForwardIndexUncompressedValueSizeInBytes(),
(long) NUM_ROWS * Integer.BYTES);
+ assertEquals(colMeta.getDictionaryEncodedUncompressedValueSizeInBytes(),
ColumnMetadata.UNAVAILABLE);
+ }
+
+ @Test
+ public void testEnableDictionaryOnCodecSpecColumnKeepsRawStats()
+ throws Exception {
+ _fieldConfigMap.put(RAW_INT_COL, codecSpecFieldConfig(RAW_INT_COL,
"DELTA,LZ4", false));
+ updateForwardIndex(createIndexLoadingConfig());
+
+ // Adding a dictionary keeps the V7 raw forward index, so its raw stats
are carried over unchanged.
+ _noDictionaryColumns.remove(RAW_INT_COL);
+ _fieldConfigMap.put(RAW_INT_COL, codecSpecFieldConfig(RAW_INT_COL,
"DELTA,LZ4", true));
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+ SegmentDirectory.Writer writer = segmentDirectory.createWriter()) {
+ ForwardIndexHandler handler = new ForwardIndexHandler(segmentDirectory,
createIndexLoadingConfig());
+ assertEquals(handler.computeOperations(writer),
+ Map.of(RAW_INT_COL,
List.of(ForwardIndexHandler.Operation.ENABLE_DICTIONARY)));
+ handler.updateIndices(writer);
+ handler.postUpdateIndicesCleanup(writer);
+ }
+
+ assertEquals(readCodecSpec(RAW_INT_COL), "DELTA,LZ4");
+ ColumnMetadata colMeta = new
SegmentMetadataImpl(INDEX_DIR).getColumnMetadataFor(RAW_INT_COL);
+ assertTrue(colMeta.hasDictionary());
+ assertEquals(colMeta.getForwardIndexEncoding(),
FieldConfig.EncodingType.RAW);
+ assertNull(colMeta.getRawForwardIndexChunkCompressionType());
+ assertEquals(colMeta.getRawForwardIndexUncompressedValueSizeInBytes(),
(long) NUM_ROWS * Integer.BYTES);
+ }
+
+ @Test
+ public void testCodecSpecRewriteClearsStatsWhenDisabled()
+ throws Exception {
+ _fieldConfigMap.put(RAW_INT_COL, codecSpecFieldConfig(RAW_INT_COL,
"DELTA,LZ4", false));
+ TableConfig configWithStatsDisabled = new
TableConfigBuilder(TableType.OFFLINE)
+ .setTableName(RAW_TABLE_NAME)
+ .setNoDictionaryColumns(new ArrayList<>(_noDictionaryColumns))
+ .setFieldConfigList(new ArrayList<>(_fieldConfigMap.values()))
+ .build();
+ updateForwardIndex(new IndexLoadingConfig(configWithStatsDisabled,
SCHEMA));
+
+ assertEquals(readCodecSpec(RAW_INT_COL), "DELTA,LZ4");
+ ColumnMetadata colMeta = new
SegmentMetadataImpl(INDEX_DIR).getColumnMetadataFor(RAW_INT_COL);
+ assertNull(colMeta.getRawForwardIndexChunkCompressionType());
+ assertEquals(colMeta.getRawForwardIndexUncompressedValueSizeInBytes(),
ColumnMetadata.UNAVAILABLE);
+ }
+
+ /// Builds a RAW [FieldConfig] whose forward index uses `codecSpec`,
optionally keeping a dictionary.
+ private static FieldConfig codecSpecFieldConfig(String column, String
codecSpec, boolean withDictionary) {
+ ObjectNode forward = JsonUtils.newObjectNode();
+ forward.put("codecSpec", codecSpec);
+ ObjectNode indexes = JsonUtils.newObjectNode();
+ indexes.set("forward", forward);
+ if (withDictionary) {
+ indexes.set("dictionary", JsonUtils.newObjectNode());
+ }
+ return new FieldConfig.Builder(column)
+ .withEncodingType(FieldConfig.EncodingType.RAW)
+ .withIndexes(indexes)
+ .build();
+ }
+
+ private static void updateForwardIndex(IndexLoadingConfig indexLoadingConfig)
+ throws Exception {
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+ SegmentDirectory.Writer writer = segmentDirectory.createWriter()) {
+ ForwardIndexHandler handler = new ForwardIndexHandler(segmentDirectory,
indexLoadingConfig);
+ assertTrue(handler.needUpdateIndices(writer), "Handler should detect the
forward-index change");
+ handler.updateIndices(writer);
+ handler.postUpdateIndicesCleanup(writer);
+ }
+ }
+
+ /// Returns the canonical codec spec from the column's V7 forward-index
header, or null for a legacy raw index.
+ @Nullable
+ private static String readCodecSpec(String column)
+ throws Exception {
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+ SegmentDirectory.Reader reader = segmentDirectory.createReader()) {
+ PinotDataBuffer buffer = reader.getIndexFor(column,
StandardIndexes.forward());
+ return
FixedByteChunkSVForwardIndexReaderV7.hasCodecPipelineHeader(buffer)
+ ? FixedByteChunkSVForwardIndexReaderV7.readCodecSpec(buffer) : null;
+ }
+ }
}
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 1c91e348e9d..8baa99d66c5 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
@@ -19,27 +19,44 @@
package org.apache.pinot.segment.local.segment.index.openstruct;
import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.pinot.segment.local.utils.TableConfigUtils;
import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.OpenStructIndexConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.ComplexFieldSpec;
import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.expectThrows;
public class OpenStructIndexTypeTest {
+ @DataProvider(name = "codecSpecOpenStructConfigs")
+ public Object[][] codecSpecOpenStructConfigs() {
+ OpenStructIndexConfig valueConfig = new OpenStructIndexConfig(false, null,
-1, null, 0.5,
+ List.of(rawCodecSpecFieldConfig("clicks")), null, null, null);
+ OpenStructIndexConfig defaultConfig = new OpenStructIndexConfig(false,
rawCodecSpecFieldConfig("default"), -1,
+ null, 0.5, null, null, null, null);
+ return new Object[][]{{valueConfig, "key 'clicks'"}, {defaultConfig,
"defaultValueFieldConfig"}};
+ }
+
@Test
public void testServiceLoaderResolves() {
assertNotNull(StandardIndexes.openStruct(),
@@ -184,4 +201,37 @@ public class OpenStructIndexTypeTest {
// Must not throw - validation is skipped entirely when the index is
disabled.
StandardIndexes.openStruct().validate(fieldIndexConfigs, openStructSpec,
null);
}
+
+ @Test(dataProvider = "codecSpecOpenStructConfigs")
+ public void
testTableValidationRejectsCodecSpecForMaterializedChildren(OpenStructIndexConfig
openStructConfig,
+ String expectedTarget) {
+ ObjectNode indexes = JsonUtils.newObjectNode();
+ indexes.set(StandardIndexes.openStruct().getPrettyName(),
JsonUtils.objectToJsonNode(openStructConfig));
+ FieldConfig parentFieldConfig = new
FieldConfig.Builder("payload").withIndexes(indexes).build();
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName("openStructCodecSpecTest")
+ .setFieldConfigList(List.of(parentFieldConfig))
+ .build();
+ Schema schema = new Schema.SchemaBuilder()
+ .setSchemaName("openStructCodecSpecTest")
+ .addOpenStruct("payload", Map.of())
+ .build();
+
+ IllegalStateException exception = expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(tableConfig, schema));
+ 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");
+ }
+
+ private static FieldConfig rawCodecSpecFieldConfig(String name) {
+ ObjectNode forward = JsonUtils.newObjectNode();
+ forward.put("encodingType", FieldConfig.EncodingType.RAW.name());
+ forward.put("codecSpec", "LZ4");
+ ObjectNode indexes = JsonUtils.newObjectNode();
+ indexes.set(StandardIndexes.forward().getPrettyName(), forward);
+ return new FieldConfig.Builder(name)
+ .withEncodingType(FieldConfig.EncodingType.RAW)
+ .withIndexes(indexes)
+ .build();
+ }
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 42b2639974b..97a0263b5a2 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -1812,12 +1812,24 @@ public class TableConfigUtilsTest {
fieldConfigWithCodecSpec("intCol", FieldConfig.EncodingType.RAW,
"DELTA,LZ4"),
fieldConfigWithCodecSpec("longCol", FieldConfig.EncodingType.RAW,
"ZSTD(3)")));
TableConfigUtils.validate(tableConfig, schema);
+ // Chained value transforms, a value transform feeding a packing
transform, and transform chains on LONG.
+ assertCodecSpecValidationPasses(schema, "intCol", "DELTA,ZSTD(3)");
+ assertCodecSpecValidationPasses(schema, "intCol", "DELTA,DELTADELTA,LZ4");
+ assertCodecSpecValidationPasses(schema, "intCol", "DELTA,T64,LZ4");
+ assertCodecSpecValidationPasses(schema, "longCol", "DELTADELTA,LZ4");
assertCodecSpecValidationFails(schema, "intCol",
FieldConfig.EncodingType.RAW, "LZ4,UNKNOWN", "Unknown codec");
assertCodecSpecValidationFails(schema, "intCol",
FieldConfig.EncodingType.RAW, "LZ4,DELTA",
"all transforms must precede any compression stage");
+ // T64 output is not a typed value array, so a value transform cannot
follow it.
+ assertCodecSpecValidationFails(schema, "intCol",
FieldConfig.EncodingType.RAW, "T64,DELTA,LZ4",
+ "must operate on column values");
+ // Every codecSpec, compression-only or transform, uses the V7 writer,
which only supports single-value
+ // INT/LONG columns.
assertCodecSpecValidationFails(schema, "mvIntCol",
FieldConfig.EncodingType.RAW, "LZ4",
"only supports single-value columns");
+ assertCodecSpecValidationFails(schema, "mvIntCol",
FieldConfig.EncodingType.RAW, "DELTA,LZ4",
+ "only supports single-value columns");
assertCodecSpecValidationFails(schema, "stringCol",
FieldConfig.EncodingType.RAW, "SNAPPY",
"only supports INT and LONG columns");
assertCodecSpecChunkSizeValidationFails(schema, -1, "numDocsPerChunk must
be positive");
@@ -1862,6 +1874,12 @@ public class TableConfigUtilsTest {
exception.getMessage());
}
+ private static void assertCodecSpecValidationPasses(Schema schema, String
column, String codecSpec) {
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
+ tableConfig.setFieldConfigList(List.of(fieldConfigWithCodecSpec(column,
FieldConfig.EncodingType.RAW, codecSpec)));
+ TableConfigUtils.validate(tableConfig, schema);
+ }
+
private static void assertCodecSpecValidationFails(Schema schema, String
column,
FieldConfig.EncodingType encodingType, String codecSpec, String
expectedMessage) {
TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/SegmentCompressionStatsReader.java
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/SegmentCompressionStatsReader.java
index 7084843bc4c..d9f3ef5d68a 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/SegmentCompressionStatsReader.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/SegmentCompressionStatsReader.java
@@ -105,7 +105,7 @@ final class SegmentCompressionStatsReader {
chunkCompressionType = null;
} else {
long uncompressedValueSize =
columnMetadata.getRawForwardIndexUncompressedValueSizeInBytes();
- if (uncompressedValueSize < 0 ||
columnMetadata.getRawForwardIndexChunkCompressionType() == null) {
+ if (uncompressedValueSize < 0) {
if (!includeColumnCompressionStats) {
return incomplete(segmentMetadata);
}
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/api/resources/SegmentCompressionStatsReaderTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/SegmentCompressionStatsReaderTest.java
index 91720990279..14059aa83eb 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/api/resources/SegmentCompressionStatsReaderTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/SegmentCompressionStatsReaderTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.server.api.resources;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.File;
import java.nio.file.Files;
import java.util.ArrayList;
@@ -26,6 +27,7 @@ import java.util.Map;
import java.util.TreeMap;
import javax.ws.rs.WebApplicationException;
import org.apache.commons.io.FileUtils;
+import
org.apache.pinot.common.restlet.resources.ColumnCompressionStatsContribution;
import
org.apache.pinot.common.restlet.resources.SegmentCompressionStatsContribution;
import
org.apache.pinot.common.restlet.resources.ServerCompressionStatsResponse;
import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
@@ -44,6 +46,7 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -120,6 +123,80 @@ public class SegmentCompressionStatsReaderTest {
}
}
+ @Test
+ public void testRawCodecPipelineWithoutLegacyCompressionTypeIsComplete() {
+ ColumnMetadata columnMetadata = mock(ColumnMetadata.class);
+ when(columnMetadata.getColumnName()).thenReturn("value");
+
when(columnMetadata.getIndexSizeFor(StandardIndexes.forward())).thenReturn(20L);
+
when(columnMetadata.getForwardIndexEncoding()).thenReturn(FieldConfig.EncodingType.RAW);
+
when(columnMetadata.getRawForwardIndexUncompressedValueSizeInBytes()).thenReturn(80L);
+
when(columnMetadata.getRawForwardIndexChunkCompressionType()).thenReturn(null);
+
+ SegmentMetadata segmentMetadata = mock(SegmentMetadata.class);
+ when(segmentMetadata.getName()).thenReturn("v7");
+ when(segmentMetadata.getColumnMetadataMap()).thenReturn(new
TreeMap<>(Map.of("value", columnMetadata)));
+
+ SegmentCompressionStatsContribution contribution =
SegmentCompressionStatsReader.read(segmentMetadata, true);
+ assertTrue(contribution.isComplete());
+ assertEquals(contribution.getUncompressedValueSizeInBytes(), 80L);
+
assertEquals(contribution.getForwardIndexAndDictionaryStorageSizeInBytes(),
20L);
+ ColumnCompressionStatsContribution.EncodingContribution encoding =
+
contribution.getColumnCompressionStats().get("value").getEncodingBreakdown().get(0);
+ assertEquals(encoding.getEncoding(), FieldConfig.EncodingType.RAW);
+ assertNull(encoding.getChunkCompressionType());
+ }
+
+ @Test
+ public void testV7SegmentCreationPersistsAndReportsCompressionStats()
+ throws Exception {
+ File outputDir =
Files.createTempDirectory("SegmentCompressionStatsReaderV7Test").toFile();
+ String segmentName = "trackedV7";
+ try {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(segmentName)
+ .addSingleValueDimension("value", DataType.INT)
+ .build();
+ ObjectNode forward = JsonUtils.newObjectNode();
+ forward.put("codecSpec", "DELTA,LZ4");
+ ObjectNode indexes = JsonUtils.newObjectNode();
+ indexes.set("forward", forward);
+ FieldConfig fieldConfig = new FieldConfig.Builder("value")
+ .withEncodingType(FieldConfig.EncodingType.RAW)
+ .withIndexes(indexes)
+ .build();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(segmentName)
+ .setFieldConfigList(List.of(fieldConfig))
+ .setCompressionStatsEnabled(true)
+ .build();
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig,
schema);
+ config.setOutDir(outputDir.getAbsolutePath());
+ config.setSegmentName(segmentName);
+ List<GenericRow> rows = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ GenericRow row = new GenericRow();
+ row.putValue("value", i);
+ rows.add(row);
+ }
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(rows));
+ driver.build();
+
+ SegmentMetadata metadata = new SegmentMetadataImpl(new File(outputDir,
segmentName));
+ ColumnMetadata columnMetadata = metadata.getColumnMetadataFor("value");
+
assertEquals(columnMetadata.getRawForwardIndexUncompressedValueSizeInBytes(),
10L * Integer.BYTES);
+ assertNull(columnMetadata.getRawForwardIndexChunkCompressionType());
+
+ SegmentCompressionStatsContribution contribution =
SegmentCompressionStatsReader.read(metadata, true);
+ assertTrue(contribution.isComplete());
+ assertEquals(contribution.getUncompressedValueSizeInBytes(), 10L *
Integer.BYTES);
+ ColumnCompressionStatsContribution.EncodingContribution encoding =
+
contribution.getColumnCompressionStats().get("value").getEncodingBreakdown().get(0);
+ assertEquals(encoding.getEncoding(), FieldConfig.EncodingType.RAW);
+ assertNull(encoding.getChunkCompressionType());
+ } finally {
+ FileUtils.deleteQuietly(outputDir);
+ }
+ }
+
@Test
public void testServerRejectsOversizedColumnContributionResponse() {
SegmentMetadata segmentMetadata = mock(SegmentMetadata.class);
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/FieldConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/FieldConfig.java
index bbeec54847b..3e459eacf9f 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/FieldConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/FieldConfig.java
@@ -148,23 +148,49 @@ public class FieldConfig extends BaseJsonConfig {
public enum CompressionCodec {
//@formatter:off
+ /// No compression. This is the default for `METRIC` columns and has no
`codecSpec` equivalent:
+ /// the DSL has no identity codec and rejects a blank spec, so this
remains the only way to state
+ /// "uncompressed" explicitly.
PASS_THROUGH(true, false),
+ /// Snappy compression for raw forward indexes. For single-value INT/LONG
raw columns on a cluster
+ /// where every server reads the V7 format, `codecSpec="SNAPPY"` is the
codec-pipeline equivalent;
+ /// other column shapes keep using this value.
SNAPPY(true, false),
+ /// Zstandard compression for raw forward indexes. For single-value
INT/LONG raw columns on a cluster
+ /// where every server reads the V7 format, `codecSpec="ZSTD(3)"` is the
codec-pipeline equivalent;
+ /// other column shapes keep using this value.
ZSTANDARD(true, false),
+ /// LZ4 compression for raw forward indexes. For single-value INT/LONG raw
columns on a cluster
+ /// where every server reads the V7 format, `codecSpec="LZ4"` is the
codec-pipeline equivalent;
+ /// other column shapes keep using this value.
LZ4(true, false),
+ /// GZIP (DEFLATE) compression for raw forward indexes. For single-value
INT/LONG raw columns on a
+ /// cluster where every server reads the V7 format, `codecSpec="GZIP"` is
the codec-pipeline
+ /// equivalent; other column shapes keep using this value.
GZIP(true, false),
- // For MV dictionary encoded forward index, add a second level dictionary
encoding for the multi-value entries
+ /// Second-level dictionary encoding of the multi-value entries of a
dictionary-encoded MV forward
+ /// index. No `codecSpec` equivalent: `codecSpec` applies only to RAW
forward indexes, so this
+ /// remains the only way to express it.
MV_ENTRY_DICT(false, true),
- // CLP is a special type of compression codec that isn't generally
applicable to all RAW columns and has a special
- // handling for log lines (see {@link CLPForwardIndexCreatorV1} and {@link
CLPForwardIndexCreatorV2)
+ /// CLP is a special type of compression codec that isn't generally
applicable to all RAW columns and has
+ /// special handling for log lines (see `CLPForwardIndexCreatorV1` and
`CLPForwardIndexCreatorV2`).
+ /// The CLP family has no `codecSpec` equivalent and remains the only way
to express it: these are
+ /// whole-index formats for STRING columns rather than chunk codecs, and
they are validated against the
+ /// column's stored type instead of the raw/dictionary applicability flags
below.
CLP(false, false),
CLPV2(false, false),
CLPV2_ZSTD(false, false),
CLPV2_LZ4(false, false),
+ /// Delta encoding for raw forward indexes. Rejected by table-config
validation on every column with a
+ /// forward index: it is applicable to neither raw nor dictionary-encoded
forward indexes. Use
+ /// `codecSpec="DELTA,LZ4"` on SV INT/LONG raw columns instead, which
writes the codec-pipeline format
+ /// rather than the legacy chunk format.
DELTA(false, false),
+ /// Second-order delta encoding for raw forward indexes. Rejected by
table-config validation for the
+ /// same reason as [#DELTA]; use `codecSpec="DELTADELTA,LZ4"` on SV
INT/LONG raw columns instead.
DELTADELTA(false, false);
//@formatter:on
@@ -212,6 +238,8 @@ public class FieldConfig extends BaseJsonConfig {
return _tierOverwrites;
}
+ /// Returns the raw forward-index compression codec, or `null` when using
`codecSpec` or the
+ /// default. Still the only way to express `MV_ENTRY_DICT` and the CLP
family — not deprecated.
@Nullable
public CompressionCodec getCompressionCodec() {
return _compressionCodec;
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/OpenStructIndexConfig.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/OpenStructIndexConfig.java
index 7feb8923a4d..46dcd1a749e 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/OpenStructIndexConfig.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/OpenStructIndexConfig.java
@@ -56,7 +56,9 @@ import org.apache.pinot.spi.utils.JsonUtils;
/// **Per-key index settings** are specified via `valueFieldConfigs` — each
entry is a standard
/// [FieldConfig] (modern `indexes` format) for one materialized OPEN_STRUCT
key. Keys without an
/// entry fall back to `defaultValueFieldConfig`. When neither is set, the
built-in default is
-/// DICTIONARY encoding with an inverted index.
+/// DICTIONARY encoding with an inverted index. A per-key config selects the
dictionary vs raw
+/// encoding and the dictionary, inverted, range and bloom filter indexes;
forward-index tuning is not
+/// configurable per key (a raw key always uses LZ4 compression), and a
per-key `codecSpec` is rejected.
public class OpenStructIndexConfig extends IndexConfig {
public static final OpenStructIndexConfig DISABLED = new
OpenStructIndexConfig(false);
public static final OpenStructIndexConfig DEFAULT = new
OpenStructIndexConfig(true);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
