Report V7 compression stats and reject per-key OPEN_STRUCT codecSpec

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/io/writer/impl/FixedByteChunkForwardIndexWriterV7.java

@@ -255,6 +255,12 @@ public void putDouble(double value) {
     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.

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,7 +41,11 @@
 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 void close()
 
   @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();
     }

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;
@@ -408,6 +409,13 @@ 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,
+        "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

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/openstruct/OpenStructIndexType.java

@@ -20,7 +20,6 @@
 
 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 void validate(FieldIndexConfigs indexConfigs, FieldSpec fieldSpec, TableC
       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 @@ private static boolean isCoercible(FieldSpec.DataType storedType) {
     }
   }
 
-  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);
     }
   }
 

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;
@@ -51,6 +52,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;
@@ -65,6 +67,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 {
@@ -97,6 +100,16 @@ private OpenStructIndexConfig config(double minFillRate, int maxDenseKeys, Set<S
         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 void testRawStringForwardIndexViaStandardCreator()
         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 {

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/ForwardIndexCreatorFactoryTest.java

@@ -138,7 +138,8 @@ public void testCodecSpecRoundTripUsesV7Format(String codecSpec, DataType stored
       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 void testCodecSpecMultiChunkRoundTrip(String codecSpec, DataType storedTy
   /// 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 @@ private static File roundTripCodecSpecIndex(File indexDir, String codecSpec, Dat
         }
       }
       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());

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

@@ -25,17 +25,25 @@
 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 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 = 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();
+      }
+    }
+  }
 }