Validate codecSpec and support codec-aware forward-index reloads

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() {

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() {

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();

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;
   }

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 @@
 
     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
@@ -624,7 +631,7 @@
         FieldSpec.FieldType.DIMENSION.name());
     props.setProperty(
         V1Constants.MetadataKeys.Column.getKeyFor(sparseCol, V1Constants.MetadataKeys.Column.IS_SINGLE_VALUED), true);
     props.setProperty(V1Constants.MetadataKeys.Column.getKeyFor(sparseCol, V1Constants.MetadataKeys.Column.TOTAL_DOCS),
         _numDocs);
     props.setProperty(V1Constants.MetadataKeys.Column.getKeyFor(sparseCol, V1Constants.MetadataKeys.Column.CARDINALITY),
         nonNullCount);
@@ -656,7 +663,7 @@
         V1Constants.MetadataKeys.Column.getKeyFor(_columnName, V1Constants.MetadataKeys.Column.IS_SINGLE_VALUED),
         true);
     props.setProperty(
         V1Constants.MetadataKeys.Column.getKeyFor(_columnName, V1Constants.MetadataKeys.Column.TOTAL_DOCS),
         _numDocs);
     props.setProperty(
         V1Constants.MetadataKeys.Column.getKeyFor(_columnName, V1Constants.MetadataKeys.Column.HAS_SPARSE_COLUMN),

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;
   }
 

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java

@@ -299,7 +299,9 @@ Map<String, List<Operation>> 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.

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<String> indexNames = indexes.fieldNames();
       while (indexNames.hasNext()) {
         String indexName = indexNames.next();

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);
     }

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 @@
         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)},
+        {new OpenStructIndexConfig(false, defaultValueFieldConfig, -1, null, 0.5, null, null)}
+    };
+  }
+
   @Test
   public void testClassifyByFillRate()
       throws Exception {
@@ -200,7 +213,7 @@
     assertEquals(p.getString(V1Constants.MetadataKeys.Column.getKeyFor(
         denseCol, V1Constants.MetadataKeys.Column.HAS_DICTIONARY)), "true");
     assertEquals(p.getInt(V1Constants.MetadataKeys.Column.getKeyFor(
         denseCol, V1Constants.MetadataKeys.Column.TOTAL_DOCS)), 10);
     assertEquals(p.getInt(V1Constants.MetadataKeys.Column.getKeyFor(
         denseCol, V1Constants.MetadataKeys.Column.CARDINALITY)), 10);
     assertEquals(p.getString(V1Constants.MetadataKeys.Column.getKeyFor(
@@ -397,6 +410,31 @@
         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(), "codecSpec is not supported for OPEN_STRUCT key: clicks");
+  }
+
+  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 {

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/CompressionStatsMetadataTest.java

@@ -51,6 +51,12 @@ public void testEveryStateUpdatesAllKeys() {
     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)));

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexTypeTest.java

@@ -25,13 +25,20 @@
 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;
 import org.apache.pinot.spi.utils.JsonUtils;
+import org.mockito.Mockito;
 import org.testng.Assert;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
@@ -495,4 +502,32 @@ public void testStandardIndex() {
     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 = Mockito.mock(MutableIndexContext.class);
+    Mockito.when(context.getFieldSpec()).thenReturn(
+        new DimensionFieldSpec("dimInt", FieldSpec.DataType.INT, true));
+    Mockito.when(context.getSegmentName()).thenReturn("testSegment");
+    Mockito.when(context.getCapacity()).thenReturn(16);
+    ForwardIndexConfig config = new ForwardIndexConfig.Builder(FieldConfig.EncodingType.RAW)
+        .withCodecSpec("DELTA,LZ4")
+        .build();
+
+    try (DirectMemoryManager memoryManager = new DirectMemoryManager("testSegment")) {
+      Mockito.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();
+      }
+    }
+  }
 }

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java

@@ -19,27 +19,45 @@
 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() {
+    FieldConfig valueFieldConfig = rawCodecSpecFieldConfig("clicks");
+    FieldConfig defaultValueFieldConfig = rawCodecSpecFieldConfig("default");
+    return new Object[][]{
+        {new OpenStructIndexConfig(false, null, -1, null, 0.5, List.of(valueFieldConfig), null), "clicks"},
+        {new OpenStructIndexConfig(false, defaultValueFieldConfig, -1, null, 0.5, null, null), "default"}
+    };
+  }
+
   @Test
   public void testServiceLoaderResolves() {
     assertNotNull(StandardIndexes.openStruct(),
@@ -154,4 +172,37 @@ public void testValidateSkipsIgnoredKeyChecksWhenIndexDisabled()
     // 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 configuredKey) {
+    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(),
+        "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();
+  }
 }