[feature] Add pluggable codec pipeline for RAW forward indexes

Replace the single compressionCodec with a codecPipeline field that supports
an ordered list of transform and compression stages. Transforms (DELTA,
DOUBLE_DELTA) are applied left-to-right on write and right-to-left on read,
with at most one terminal compressor. Uses writer version 7 with a new header
format that stores the full pipeline. Includes 53 tests covering unit,
pipeline round-trip, and full end-to-end writer/reader validation including
triple-delta stacking scenarios.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: xiangfu0

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

@@ -18,6 +18,7 @@
  */
 package org.apache.pinot.segment.local.io.compression;
 
+import org.apache.pinot.segment.spi.compression.ChunkCodecPipeline;
 import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
 import org.apache.pinot.segment.spi.compression.ChunkCompressor;
 import org.apache.pinot.segment.spi.compression.ChunkDecompressor;
@@ -118,4 +119,38 @@ public static ChunkDecompressor getDecompressor(ChunkCompressionType compression
         throw new IllegalArgumentException("Illegal decompressor name " + compressionType);
     }
   }
+
+  /**
+   * Returns a compressor for a codec pipeline. If the pipeline has transforms, returns a
+   * {@link PipelineChunkCompressor}; otherwise returns the plain compressor for the terminal codec.
+   * Legacy compound codecs (DELTA_LZ4, DOUBLE_DELTA_LZ4) are handled by their existing compressor
+   * implementations.
+   *
+   * @param pipeline the codec pipeline
+   * @param valueSizeInBytes size of each typed value (4 for INT, 8 for LONG); used by transforms
+   * @return compressor for the pipeline
+   */
+  public static ChunkCompressor getCompressor(ChunkCodecPipeline pipeline, int valueSizeInBytes) {
+    if (pipeline.hasTransforms()) {
+      return new PipelineChunkCompressor(pipeline, valueSizeInBytes);
+    }
+    return getCompressor(pipeline.getChunkCompressionType());
+  }
+
+  /**
+   * Returns a decompressor for a codec pipeline. If the pipeline has transforms, returns a
+   * {@link PipelineChunkDecompressor}; otherwise returns the plain decompressor for the terminal codec.
+   * Legacy compound codecs (DELTA_LZ4, DOUBLE_DELTA_LZ4) are handled by their existing decompressor
+   * implementations.
+   *
+   * @param pipeline the codec pipeline
+   * @param valueSizeInBytes size of each typed value (4 for INT, 8 for LONG); used by transforms
+   * @return decompressor for the pipeline
+   */
+  public static ChunkDecompressor getDecompressor(ChunkCodecPipeline pipeline, int valueSizeInBytes) {
+    if (pipeline.hasTransforms()) {
+      return new PipelineChunkDecompressor(pipeline, valueSizeInBytes);
+    }
+    return getDecompressor(pipeline.getChunkCompressionType());
+  }
 }

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

@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.io.compression;
+
+import org.apache.pinot.segment.spi.compression.ChunkCodec;
+import org.apache.pinot.segment.spi.compression.ChunkTransform;
+
+
+/**
+ * Factory for obtaining {@link ChunkTransform} instances by their {@link ChunkCodec} identifier.
+ */
+public class ChunkTransformFactory {
+
+  private ChunkTransformFactory() {
+  }
+
+  /**
+   * Returns the singleton {@link ChunkTransform} for the given transform codec.
+   *
+   * @param codec the transform codec (must be a {@link ChunkCodec.CodecKind#TRANSFORM})
+   * @return the corresponding transform instance
+   * @throws IllegalArgumentException if the codec is not a known transform
+   */
+  public static ChunkTransform getTransform(ChunkCodec codec) {
+    switch (codec) {
+      case DELTA:
+        return DeltaTransform.INSTANCE;
+      case DOUBLE_DELTA:
+        return DoubleDeltaTransform.INSTANCE;
+      case XOR:
+        return XorTransform.INSTANCE;
+      default:
+        throw new IllegalArgumentException("Unknown transform codec: " + codec);
+    }
+  }
+}

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

@@ -0,0 +1,120 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.io.compression;
+
+import java.nio.ByteBuffer;
+import org.apache.pinot.segment.spi.compression.ChunkTransform;
+
+
+/**
+ * Delta encoding transform for numeric values. Stores the first value as-is, then each
+ * subsequent value as the difference from the previous value. Operates in-place on the
+ * ByteBuffer.
+ *
+ * <p>Effective for monotonically increasing data (e.g., timestamps, counters).
+ * Java two's complement arithmetic guarantees correct wrap-around for overflow.</p>
+ */
+public class DeltaTransform implements ChunkTransform {
+
+  public static final DeltaTransform INSTANCE = new DeltaTransform();
+
+  private DeltaTransform() {
+  }
+
+  @Override
+  public void encode(ByteBuffer buffer, int numBytes, int valueSizeInBytes) {
+    if (valueSizeInBytes == Integer.BYTES) {
+      encodeInts(buffer, numBytes);
+    } else {
+      encodeLongs(buffer, numBytes);
+    }
+  }
+
+  @Override
+  public void decode(ByteBuffer buffer, int numBytes, int valueSizeInBytes) {
+    if (valueSizeInBytes == Integer.BYTES) {
+      decodeInts(buffer, numBytes);
+    } else {
+      decodeLongs(buffer, numBytes);
+    }
+  }
+
+  private void encodeInts(ByteBuffer buffer, int numBytes) {
+    int numValues = numBytes / Integer.BYTES;
+    if (numValues <= 1) {
+      return;
+    }
+    int pos = buffer.position();
+    // Encode backwards so each value can be read before being overwritten
+    int prev = buffer.getInt(pos + (numValues - 2) * Integer.BYTES);
+    for (int i = numValues - 1; i >= 1; i--) {
+      int offset = pos + i * Integer.BYTES;
+      int curr = buffer.getInt(offset);
+      buffer.putInt(offset, curr - prev);
+      if (i > 1) {
+        prev = buffer.getInt(pos + (i - 2) * Integer.BYTES);
+      }
+    }
+  }
+
+  private void encodeLongs(ByteBuffer buffer, int numBytes) {
+    int numValues = numBytes / Long.BYTES;
+    if (numValues <= 1) {
+      return;
+    }
+    int pos = buffer.position();
+    long prev = buffer.getLong(pos + (numValues - 2) * Long.BYTES);
+    for (int i = numValues - 1; i >= 1; i--) {
+      int offset = pos + i * Long.BYTES;
+      long curr = buffer.getLong(offset);
+      buffer.putLong(offset, curr - prev);
+      if (i > 1) {
+        prev = buffer.getLong(pos + (i - 2) * Long.BYTES);
+      }
+    }
+  }
+
+  private void decodeInts(ByteBuffer buffer, int numBytes) {
+    int numValues = numBytes / Integer.BYTES;
+    if (numValues <= 1) {
+      return;
+    }
+    int pos = buffer.position();
+    for (int i = 1; i < numValues; i++) {
+      int offset = pos + i * Integer.BYTES;
+      int prevValue = buffer.getInt(pos + (i - 1) * Integer.BYTES);
+      int delta = buffer.getInt(offset);
+      buffer.putInt(offset, prevValue + delta);
+    }
+  }
+
+  private void decodeLongs(ByteBuffer buffer, int numBytes) {
+    int numValues = numBytes / Long.BYTES;
+    if (numValues <= 1) {
+      return;
+    }
+    int pos = buffer.position();
+    for (int i = 1; i < numValues; i++) {
+      int offset = pos + i * Long.BYTES;
+      long prevValue = buffer.getLong(pos + (i - 1) * Long.BYTES);
+      long delta = buffer.getLong(offset);
+      buffer.putLong(offset, prevValue + delta);
+    }
+  }
+}