Address PR review comments

- Allow transformCodec=NONE on dict-encoded columns (treat NONE same
  as null in validation)
- Use explicit value-based lookup in TransformCodec.valueOf(int)
  instead of ordinal to prevent silent corruption if enum is reordered
- Fix empty-array tests to actually exercise encode/decode with
  numBytes=0 instead of returning early
- Fix misleading comment in DoubleDeltaTransform.encodeInts() that
  said "reverse order" when the code encodes forward

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: xiangfu0

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/compression/DoubleDeltaTransform.java

@@ -68,9 +68,8 @@ private void encodeInts(ByteBuffer buffer, int numBytes) {
     }
     int pos = buffer.position();
 
-    // Encode in reverse to avoid overwriting values we still need.
-    // First pass: compute all original values we need (they're in the buffer).
-    // We encode from the end to index 2, then handle index 1.
+    // Encode in forward order, keeping the previous original value and delta in local variables
+    // so we can safely overwrite each position in-place as we advance from index 2 onward.
     int prevPrevVal = buffer.getInt(pos);
     int prevVal = buffer.getInt(pos + Integer.BYTES);
     int prevDelta = prevVal - prevPrevVal;

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

@@ -113,7 +113,11 @@ private void validateForwardIndexEnabled(ForwardIndexConfig forwardIndexConfig,
     if (dictionaryConfig.isEnabled()) {
       Preconditions.checkState(compressionCodec == null || compressionCodec.isApplicableToDictEncodedIndex(),
           "Compression codec: %s is not applicable to dictionary encoded column: %s", compressionCodec, column);
-      Preconditions.checkState(forwardIndexConfig.getTransformCodec() == null,
+      org.apache.pinot.segment.spi.compression.TransformCodec dictTransformCodec =
+          forwardIndexConfig.getTransformCodec();
+      Preconditions.checkState(
+          dictTransformCodec == null
+              || dictTransformCodec == org.apache.pinot.segment.spi.compression.TransformCodec.NONE,
           "Transform codec is not supported for dictionary-encoded column: %s", column);
     } else {
       boolean isCLPCodec = compressionCodec == CompressionCodec.CLP || compressionCodec == CompressionCodec.CLPV2

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/io/compression/ChunkTransformTest.java

@@ -191,16 +191,14 @@ public void testDoubleDeltaRandomLongs() {
   // --- Helpers ---
 
   private void assertRoundTripInts(ChunkTransform transform, int[] original) {
-    if (original.length == 0) {
-      return;
-    }
-    ByteBuffer buffer = ByteBuffer.allocate(original.length * Integer.BYTES);
+    int numBytes = original.length * Integer.BYTES;
+    ByteBuffer buffer = ByteBuffer.allocate(Math.max(numBytes, 1));
     for (int v : original) {
       buffer.putInt(v);
     }
     buffer.flip();
 
-    int numBytes = original.length * Integer.BYTES;
+    // Exercise encode/decode even on empty buffers to verify no-throw behavior
     transform.encode(buffer, numBytes, Integer.BYTES);
     transform.decode(buffer, numBytes, Integer.BYTES);
 
@@ -211,16 +209,14 @@ private void assertRoundTripInts(ChunkTransform transform, int[] original) {
   }
 
   private void assertRoundTripLongs(ChunkTransform transform, long[] original) {
-    if (original.length == 0) {
-      return;
-    }
-    ByteBuffer buffer = ByteBuffer.allocate(original.length * Long.BYTES);
+    int numBytes = original.length * Long.BYTES;
+    ByteBuffer buffer = ByteBuffer.allocate(Math.max(numBytes, 1));
     for (long v : original) {
       buffer.putLong(v);
     }
     buffer.flip();
 
-    int numBytes = original.length * Long.BYTES;
+    // Exercise encode/decode even on empty buffers to verify no-throw behavior
     transform.encode(buffer, numBytes, Long.BYTES);
     transform.decode(buffer, numBytes, Long.BYTES);
 

pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/compression/TransformCodec.java

@@ -46,7 +46,19 @@ public enum TransformCodec {
    */
   DOUBLE_DELTA(2);
 
-  private static final TransformCodec[] VALUES = values();
+  private static final TransformCodec[] BY_VALUE;
+
+  static {
+    TransformCodec[] all = values();
+    int maxValue = 0;
+    for (TransformCodec tc : all) {
+      maxValue = Math.max(maxValue, tc._value);
+    }
+    BY_VALUE = new TransformCodec[maxValue + 1];
+    for (TransformCodec tc : all) {
+      BY_VALUE[tc._value] = tc;
+    }
+  }
 
   private final int _value;
 
@@ -58,10 +70,13 @@ public int getValue() {
     return _value;
   }
 
-  public static TransformCodec valueOf(int ordinal) {
-    if (ordinal < 0 || ordinal >= VALUES.length) {
-      throw new IllegalArgumentException("Invalid TransformCodec ordinal: " + ordinal);
+  /**
+   * Look up a TransformCodec by its on-disk int value (not ordinal).
+   */
+  public static TransformCodec valueOf(int value) {
+    if (value < 0 || value >= BY_VALUE.length || BY_VALUE[value] == null) {
+      throw new IllegalArgumentException("Invalid TransformCodec value: " + value);
     }
-    return VALUES[ordinal];
+    return BY_VALUE[value];
   }
 }