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 cce98a1ab5f4..240df22bc5c1 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 EncodingType getEncoding() { 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 5ea44700c359..943876cbae7d 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 EncodingType getEncoding() { 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/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 6550f6553e65..8fc541dfb44a 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 15594826c18c..8e088eaa6a01 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 @@ 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,6 +41,7 @@ public class SingleValueFixedByteRawIndexCreator implements CompressionStatsTrackingForwardIndexCreator { private final FixedByteValueWriter _indexWriter; private final DataType _valueType; + @Nullable private final ChunkCompressionType _chunkCompressionType; /// Constructor for the class @@ -142,6 +144,7 @@ public long getRawForwardIndexUncompressedValueSizeInBytes() { } @Override + @Nullable public ChunkCompressionType getRawForwardIndexChunkCompressionType() { return _chunkCompressionType; } 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 01565e066052..d0a9c97f95b1 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 java.io.File; import java.io.IOException; import java.nio.charset.StandardCharsets; @@ -405,6 +406,12 @@ private void writeDenseKeyColumn(String key) 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, + "codecSpec is not supported for OPEN_STRUCT key: %s", 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 42329efa4249..95fe70bb1af4 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 @@ private CompressionStatsMetadata(@Nullable Long forwardIndexUncompressedValueSiz _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/loader/ForwardIndexHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java index 2c1403de6bd3..9863eacc3026 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 @@ -299,7 +299,9 @@ Map> computeOperations(SegmentDirectory.Reader segmentRe /// `desiredDict = newIsDict || any-enabled-index-requires-dict`. The "force on if required" rule is the /// only place this method consults other indexes — once `desiredDict` is computed, the rest of the logic /// treats it as the source of truth. - /// 3. **Compression-type change** — only when no encoding change happened (forward + dict both unchanged). + /// 3. **Compression-type change** — whenever an existing RAW forward index remains RAW after the other + /// operations. Adding/removing a standalone dictionary does not recreate that raw index, so a codec change + /// must be queued alongside the dictionary operation. /// 4. **Cross-cutting guards** — sorted columns can't toggle forward; range index format is incompatible /// with disabling the dictionary; enabling forward needs dict + inverted on disk; enabling dict needs /// forward to be on so the dict can be bootstrapped. 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 1b0e5c296aa1..cc31cb6d30d7 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 @@ -104,6 +104,11 @@ private void validatePerKeyIndexes(OpenStructIndexConfig config) { if (indexes == null) { continue; } + 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"), + "codecSpec is not supported for OPEN_STRUCT key: %s", fieldConfig.getName()); Iterator indexNames = indexes.fieldNames(); while (indexNames.hasNext()) { String indexName = indexNames.next(); diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java index 8ef3c48727e0..54e16a6a8944 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java @@ -1781,6 +1781,12 @@ private static void validateIndexingConfigAndFieldConfigList(TableConfig tableCo // Validate DELTA / DELTADELTA compression codecs compatibility validateGorillaCompressionCodecIfPresent(fieldConfig, schema.getFieldSpecFor(column)); + + // Note: codecSpec is validated below via `indexType.validate(...)` once FieldIndexConfigsUtil + // has resolved the effective ForwardIndexConfig (merging the legacy `noDictionaryColumns` / + // `noDictionaryConfig` signals into the resolved encoding type). Validating the raw FieldConfig + // here would incorrectly reject legacy tables that express RAW via `noDictionaryColumns` while + // configuring codecSpec under `indexes.forward`. } validateIndexingConfigAndFieldConfigListCompatibility(indexingConfig, fieldConfigs); } 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 b8795f02c53e..56a1bb89e9cb 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 @@ 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; @@ -50,6 +51,7 @@ 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; @@ -64,6 +66,7 @@ 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 { @@ -96,6 +99,16 @@ private OpenStructIndexConfig config(double minFillRate, int maxDenseKeys, Set TableConfigUtils.validate(tableConfig, schema)); + assertEquals(exception.getMessage(), + "codecSpec is not supported for OPEN_STRUCT key: " + configuredKey); + } + + 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 317fcf322fc1..3e7dad0dca95 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 @@ -1798,6 +1798,145 @@ public void testValidateFieldConfig() { } } + /// Semantic table-config-time validation of `codecSpec` (via `ForwardIndexType.validate`): + /// well-formed specs on supported column shapes are accepted, and malformed or type-incompatible + /// specs fail with precise errors. + @Test + public void testCodecSpecValidation() { + Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME) + .addSingleValueDimension("intCol", DataType.INT) + .addSingleValueDimension("longCol", DataType.LONG) + .addSingleValueDimension("stringCol", DataType.STRING) + .addMultiValueDimension("mvIntCol", DataType.INT) + .build(); + + // An unknown codec name inside indexes.forward.codecSpec fails with a precise error. + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("intCol", "LZ4,UNKNOWN"))); + TableConfig unknownCodecTableConfig = tableConfig; + Exception exception = expectThrows(Exception.class, + () -> TableConfigUtils.validate(unknownCodecTableConfig, schema)); + assertTrue(exception.getMessage().contains("Unknown codec"), "Unexpected error: " + exception.getMessage()); + + // A multi-stage codecSpec requires the V7 codec-pipeline writer, which only supports + // single-value columns, so it is rejected on a multi-value column. + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("mvIntCol", "LZ4,SNAPPY"))); + TableConfig mvChainTableConfig = tableConfig; + exception = expectThrows(Exception.class, () -> TableConfigUtils.validate(mvChainTableConfig, schema)); + assertTrue(exception.getMessage().contains("only supports single-value columns") + && exception.getMessage().contains("mvIntCol"), "Unexpected error: " + exception.getMessage()); + + // A transform codecSpec likewise requires the V7 writer, so DELTA on a multi-value column is + // rejected even though the column's stored type (INT) is supported. + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("mvIntCol", "DELTA,LZ4"))); + TableConfig mvTransformTableConfig = tableConfig; + exception = expectThrows(Exception.class, () -> TableConfigUtils.validate(mvTransformTableConfig, schema)); + assertTrue(exception.getMessage().contains("only supports single-value columns") + && exception.getMessage().contains("mvIntCol"), "Unexpected error: " + exception.getMessage()); + + // A V7-requiring spec on a single-value column of an unsupported stored type (STRING) is + // rejected with the INT/LONG-only error. ZSTD(5) is compression-only but its non-default level + // cannot be represented by a legacy ChunkCompressionType, so it needs the V7 writer. + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("stringCol", "ZSTD(5)"))); + TableConfig stringV7TableConfig = tableConfig; + exception = expectThrows(Exception.class, () -> TableConfigUtils.validate(stringV7TableConfig, schema)); + assertTrue(exception.getMessage().contains("only supports INT and LONG columns") + && exception.getMessage().contains("stringCol"), "Unexpected error: " + exception.getMessage()); + + // A transform after a packing transform must be rejected (T64 output is not a typed value + // array, so DELTA cannot consume it). + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("intCol", "T64,DELTA,LZ4"))); + TableConfig misorderedTableConfig = tableConfig; + exception = expectThrows(Exception.class, () -> TableConfigUtils.validate(misorderedTableConfig, schema)); + assertTrue(exception.getMessage().contains("must operate on column values"), + "Unexpected error: " + exception.getMessage()); + + // A disabled modern forward-index config cannot retain an ignored codecSpec. This must be rejected even + // without the legacy FieldConfig.forwardIndexDisabled property. + ObjectNode disabledForward = JsonUtils.newObjectNode(); + disabledForward.put("disabled", true); + disabledForward.put("codecSpec", "DELTA,LZ4"); + ObjectNode disabledIndexes = JsonUtils.newObjectNode(); + disabledIndexes.set("forward", disabledForward); + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(new FieldConfig.Builder("intCol") + .withEncodingType(FieldConfig.EncodingType.RAW) + .withIndexes(disabledIndexes) + .build())); + TableConfig disabledCodecSpecTableConfig = tableConfig; + exception = expectThrows(Exception.class, + () -> TableConfigUtils.validate(disabledCodecSpecTableConfig, schema)); + assertTrue(exception.getMessage().contains("codecSpec cannot be configured when the forward index is disabled") + && exception.getMessage().contains("intCol"), "Unexpected error: " + exception.getMessage()); + + // A well-formed compression-only RAW codecSpec passes table-config validation. + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("intCol", "ZSTD(3)"))); + TableConfigUtils.validate(tableConfig, schema); + + // A transform + compression chain on a RAW single-value INT column passes validation. + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("intCol", "DELTA,ZSTD(3)"))); + TableConfigUtils.validate(tableConfig, schema); + + // A transform pipeline on a LONG column passes validation. + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("longCol", "DELTADELTA,LZ4"))); + TableConfigUtils.validate(tableConfig, schema); + + // Every non-null codecSpec routes through the V7 writer, including a single compression stage. Reject + // unsupported stored types and multi-value columns consistently with transform pipelines. + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("stringCol", "SNAPPY"))); + TableConfig stringCompressionTableConfig = tableConfig; + exception = expectThrows(Exception.class, () -> TableConfigUtils.validate(stringCompressionTableConfig, schema)); + assertTrue(exception.getMessage().contains("only supports INT and LONG columns") + && exception.getMessage().contains("stringCol"), "Unexpected error: " + exception.getMessage()); + + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("mvIntCol", "LZ4"))); + TableConfig mvCompressionTableConfig = tableConfig; + exception = expectThrows(Exception.class, () -> TableConfigUtils.validate(mvCompressionTableConfig, schema)); + assertTrue(exception.getMessage().contains("only supports single-value columns") + && exception.getMessage().contains("mvIntCol"), "Unexpected error: " + exception.getMessage()); + + // Regression: codecSpec validation runs via `IndexType.validate(...)` after + // `FieldIndexConfigsUtil` resolves overrides — not by an early raw-FieldConfig pre-pass. + // A column whose RAW encoding is resolved from `noDictionaryColumns`, with codecSpec set under + // `indexes.forward`, must pass validation. + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.getIndexingConfig().setNoDictionaryColumns(List.of("intCol")); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("intCol", "DELTA,LZ4"))); + TableConfigUtils.validate(tableConfig, schema); + + // Chained value-transforms + compression are valid ("DELTA,DELTADELTA,LZ4"). + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("intCol", "DELTA,DELTADELTA,LZ4"))); + TableConfigUtils.validate(tableConfig, schema); + + // A value-transform → packing transform → compression chain is valid ("DELTA,T64,LZ4"). + tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + tableConfig.setFieldConfigList(List.of(rawFieldConfigWithCodecSpec("intCol", "DELTA,T64,LZ4"))); + TableConfigUtils.validate(tableConfig, schema); + } + + /// Builds a RAW FieldConfig whose codecSpec is configured via the modern `indexes.forward` block + /// (the only supported path; there is no top-level FieldConfig.codecSpec field). + private static FieldConfig rawFieldConfigWithCodecSpec(String column, String codecSpec) { + ObjectNode forward = JsonUtils.newObjectNode(); + forward.put("codecSpec", codecSpec); + ObjectNode indexes = JsonUtils.newObjectNode(); + indexes.set("forward", forward); + return new FieldConfig.Builder(column) + .withEncodingType(FieldConfig.EncodingType.RAW) + .withIndexes(indexes) + .build(); + } + @Test public void testCodecSpecTableConfigValidation() { Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME) 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 7084843bc4cd..d9f3ef5d68a1 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 @@ private static SegmentCompressionStatsContribution readUncached(SegmentMetadata 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 91720990279d..14059aa83ebf 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.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.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 void testV1SegmentUsesSidecarIndexFileSizes() } } + @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 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 bbeec54847b2..65ed2a45ff74 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,45 @@ public enum IndexType { 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. Prefer `codecSpec="SNAPPY"` in new configs; + /// existing `compressionCodec` uses remain supported. SNAPPY(true, false), + /// Zstandard compression for raw forward indexes. Prefer `codecSpec="ZSTD(3)"` in new configs; + /// existing `compressionCodec` uses remain supported. ZSTANDARD(true, false), + /// LZ4 compression for raw forward indexes. Prefer `codecSpec="LZ4"` in new configs; + /// existing `compressionCodec` uses remain supported. LZ4(true, false), + /// GZIP (DEFLATE) compression for raw forward indexes. Prefer `codecSpec="GZIP"` in new configs; + /// existing `compressionCodec` uses remain supported. 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: it is applicable to + /// neither raw nor dictionary-encoded columns, so a config that sets it fails validation for every + /// column shape. 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 +234,8 @@ public JsonNode getTierOverwrites() { 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;