Support codecSpec on segment reload and enable codecSpec end to end

Author: xiangfu0

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java

@@ -309,9 +309,11 @@ 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.hasCodecSpec(),
-        "codecSpec is not supported yet for OPEN_STRUCT key: %s", key);
+        "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

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexCreatorFactory.java

@@ -92,18 +92,10 @@ public static ForwardIndexCreator createIndexCreator(IndexCreationContext contex
         chunkCompressionType = CodecSpecUtils.toLegacyChunkCompressionType(codecSpec);
         if (chunkCompressionType == null) {
           // The spec is either a transform or a compression-only spec with arguments that can't be
-          // represented by a legacy ChunkCompressionType (e.g. ZSTD(5)). Both routes need V7. The
-          // closed feature gate in ForwardIndexType.validate/shouldCreateIndex still rejects
-          // codecSpec table configs, so this branch is only reachable through direct factory calls;
-          // these checks are defense-in-depth for such paths.
-          Preconditions.checkArgument(fieldSpec.isSingleValueField(),
-              "codecSpec '%s' requires the V7 codec-pipeline writer (transform, chain, or non-default options), "
-                  + "which only supports single-value columns. Column '%s' is multi-value.",
-              codecSpec, columnName);
-          Preconditions.checkArgument(storedType == DataType.INT || storedType == DataType.LONG,
-              "codecSpec '%s' requires the V7 codec-pipeline writer (transform, chain, or non-default options), "
-                  + "which only supports INT and LONG columns. Column '%s' has type: %s.",
-              codecSpec, columnName, storedType);
+          // represented by a legacy ChunkCompressionType (e.g. ZSTD(5)). Both routes need V7.
+          // ForwardIndexType.validate enforces the same shape constraints at table-config time;
+          // this check is defense-in-depth for direct factory calls that bypass validation.
+          ForwardIndexType.validateV7WriterShape(codecSpec, fieldSpec);
           CodecPipelineExecutor executor = CodecPipelineExecutor.create(codecSpec, storedType);
           creator = new SingleValueFixedByteRawIndexCreator(indexDir, columnName, numTotalDocs, storedType,
               indexConfig.getTargetDocsPerChunk(), executor);

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/forward/ForwardIndexType.java

@@ -31,6 +31,8 @@
 import java.util.Map;
 import java.util.Set;
 import javax.annotation.Nullable;
+import org.apache.pinot.segment.local.io.codec.CodecPipelineExecutor;
+import org.apache.pinot.segment.local.io.codec.CodecSpecUtils;
 import org.apache.pinot.segment.local.realtime.impl.forward.CLPMutableForwardIndexV2;
 import org.apache.pinot.segment.local.realtime.impl.forward.FixedByteMVMutableForwardIndex;
 import org.apache.pinot.segment.local.realtime.impl.forward.FixedByteSVMutableForwardIndex;
@@ -104,24 +106,21 @@ public ForwardIndexConfig getDefaultConfig() {
   @Override
   public void validate(FieldIndexConfigs indexConfigs, FieldSpec fieldSpec, TableConfig tableConfig) {
     ForwardIndexConfig forwardIndexConfig = indexConfigs.getConfig(StandardIndexes.forward());
-    rejectUnsupportedCodecSpec(forwardIndexConfig, fieldSpec.getName());
     if (forwardIndexConfig.isEnabled()) {
       validateForwardIndexEnabled(forwardIndexConfig, indexConfigs, fieldSpec);
     } else {
       validateForwardIndexDisabled(indexConfigs, fieldSpec, tableConfig);
     }
   }
 
-  static void rejectUnsupportedCodecSpec(ForwardIndexConfig config, String column) {
-    Preconditions.checkState(!config.hasCodecSpec(),
-        "codecSpec is not supported yet for column: %s", column);
-  }
-
   private void validateForwardIndexEnabled(ForwardIndexConfig forwardIndexConfig, FieldIndexConfigs indexConfigs,
       FieldSpec fieldSpec) {
     String column = fieldSpec.getName();
     CompressionCodec compressionCodec = forwardIndexConfig.getCompressionCodec();
     DictionaryIndexConfig dictionaryConfig = indexConfigs.getConfig(StandardIndexes.dictionary());
+    if (forwardIndexConfig.hasCodecSpec()) {
+      validateCodecSpec(forwardIndexConfig, fieldSpec);
+    }
     // Dictionary-encoded forward index requires a dictionary to translate dict ids back to values.
     if (forwardIndexConfig.getEncodingType() == FieldConfig.EncodingType.DICTIONARY) {
       Preconditions.checkState(dictionaryConfig.isEnabled(),
@@ -143,6 +142,51 @@ private void validateForwardIndexEnabled(ForwardIndexConfig forwardIndexConfig,
     }
   }
 
+  /// Semantic validation for `codecSpec` at table-config time: parses and validates the pipeline
+  /// (unknown codecs, stage ordering, per-codec type compatibility) against the column's stored
+  /// type, then enforces the V7 codec-pipeline writer's shape constraints — single-value INT/LONG —
+  /// for specs that cannot be served by the legacy raw forward-index formats.
+  private void validateCodecSpec(ForwardIndexConfig forwardIndexConfig, FieldSpec fieldSpec) {
+    String column = fieldSpec.getName();
+    Preconditions.checkState(forwardIndexConfig.getEncodingType() == FieldConfig.EncodingType.RAW,
+        "codecSpec requires RAW forward-index encoding for column: %s", column);
+    FieldSpec.DataType storedType = fieldSpec.getDataType().getStoredType();
+
+    String codecSpec = forwardIndexConfig.getCodecSpec();
+    try {
+      // CodecPipelineExecutor.create parses the spec and runs full pipeline validation.
+      CodecPipelineExecutor.create(codecSpec, storedType);
+    } catch (IllegalArgumentException e) {
+      throw new IllegalStateException(
+          "Codec pipeline validation failed for column '" + column + "' (codecSpec='" + codecSpec + "'): "
+              + e.getMessage(), e);
+    }
+    // Transforms, chains, and non-default options all fail the legacy ChunkCompressionType mapping, so a
+    // null mapping is exactly "needs the V7 writer".
+    if (CodecSpecUtils.toLegacyChunkCompressionType(codecSpec) == null) {
+      validateV7WriterShape(codecSpec, fieldSpec);
+    }
+  }
+
+  /// Shared V7 codec-pipeline writer shape constraints: specs that cannot be represented by a legacy
+  /// [org.apache.pinot.segment.spi.compression.ChunkCompressionType] (transform, chain, or non-default
+  /// options) require the V7 writer, which only supports single-value INT/LONG columns. Called at
+  /// table-config validation time, and again from [ForwardIndexCreatorFactory] as defense-in-depth for
+  /// direct factory calls that bypass validation.
+  static void validateV7WriterShape(String codecSpec, FieldSpec fieldSpec) {
+    String column = fieldSpec.getName();
+    Preconditions.checkArgument(fieldSpec.isSingleValueField(),
+        "codecSpec '%s' requires the V7 codec-pipeline writer (transform, chain, or non-default options), "
+            + "which only supports single-value columns. Column '%s' is multi-value; use a compression-only "
+            + "spec representable by ChunkCompressionType (LZ4, SNAPPY, GZIP, ZSTD/ZSTD(3)).",
+        codecSpec, column);
+    FieldSpec.DataType storedType = fieldSpec.getDataType().getStoredType();
+    Preconditions.checkArgument(storedType == FieldSpec.DataType.INT || storedType == FieldSpec.DataType.LONG,
+        "codecSpec '%s' requires the V7 codec-pipeline writer (transform, chain, or non-default options), "
+            + "which only supports INT and LONG columns. Column '%s' has type: %s.",
+        codecSpec, column, storedType);
+  }
+
   private void validateForwardIndexDisabled(FieldIndexConfigs indexConfigs, FieldSpec fieldSpec,
       TableConfig tableConfig) {
     String column = fieldSpec.getName();
@@ -314,7 +358,6 @@ public static ChunkCompressionType getDefaultCompressionType(FieldSpec.FieldType
 
   @Override
   public boolean shouldCreateIndex(IndexCreationContext context, ForwardIndexConfig indexConfig) {
-    rejectUnsupportedCodecSpec(indexConfig, context.getFieldSpec().getName());
     return context.getFieldSpec().getDataType() != FieldSpec.DataType.OPEN_STRUCT;
   }
 
@@ -377,7 +420,9 @@ public List<String> getFileExtensions(@Nullable ColumnMetadata columnMetadata) {
   @Nullable
   @Override
   public MutableIndex createMutableIndex(MutableIndexContext context, ForwardIndexConfig config) {
-    rejectUnsupportedCodecSpec(config, context.getFieldSpec().getName());
+    // Note: codecSpec applies only when an immutable segment is created (initial creation or
+    // conversion/commit of a consuming segment). The mutable (consuming) forward index always uses
+    // the standard in-memory formats below, so a configured codecSpec is intentionally ignored here.
     if (config.isDisabled()) {
       return null;
     }