Fix review findings: move ChunkCodecHandler to pinot-segment-spi, fix buffer and type guards

- Move ChunkCodecHandler interface to pinot-segment-spi alongside CodecDefinition
- Add dst.clear() in ZstdCodecDefinition.decodeInto and Lz4CodecDefinition.decodeInto
- Guard empty-pipeline at construction time with a clear IllegalArgumentException
- Fix Delta/DeltaDelta encode to throw on unsupported stored types (restore old guard)
- Update Preconditions message to include canonical spec instead of hardcoded codec names
- Fix stale CodecDefinition Javadoc reference in CodecPipelineValidator

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>

Author: xiangfu0

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/CodecPipelineExecutor.java

@@ -23,6 +23,7 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecContext;
 import org.apache.pinot.segment.spi.codec.CodecInvocation;
 import org.apache.pinot.segment.spi.codec.CodecKind;
@@ -114,6 +115,9 @@ public static CodecPipelineExecutor create(String spec, CodecContext ctx, CodecR
   @SuppressWarnings({"unchecked", "rawtypes"})
   private CodecPipelineExecutor(CodecPipeline pipeline, CodecRegistry registry, CodecContext ctx) {
     List<CodecInvocation> invocations = pipeline.stages();
+    if (invocations.isEmpty()) {
+      throw new IllegalArgumentException("Codec pipeline must contain at least one stage");
+    }
     List<BoundStage<?>> stages = new ArrayList<>(invocations.size());
 
     for (CodecInvocation inv : invocations) {
@@ -193,7 +197,7 @@ public ByteBuffer decompress(ByteBuffer src) throws IOException {
    */
   public void decompress(ByteBuffer src, ByteBuffer dst) throws IOException {
     Preconditions.checkArgument(!_requiresDirectDstBuffer || dst.isDirect(),
-        "decompress(src, dst) requires a direct ByteBuffer when the pipeline includes Zstd or Snappy");
+        "decompress(src, dst) requires a direct ByteBuffer for pipeline: %s", _canonicalSpec);
 
     int stageCount = _stages.size();
     if (stageCount == 1) {

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/CodecPipelineValidator.java

@@ -19,6 +19,7 @@
 package org.apache.pinot.segment.local.io.codec;
 
 import java.util.List;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecContext;
 import org.apache.pinot.segment.spi.codec.CodecInvocation;
 import org.apache.pinot.segment.spi.codec.CodecKind;
@@ -35,7 +36,7 @@
  *   <li>At most one {@link CodecKind#COMPRESSION} stage is allowed.</li>
  *   <li>The compression stage, if present, must be last.</li>
  *   <li>All {@link CodecKind#TRANSFORM} stages must precede the compression stage.</li>
- *   <li>Each codec's {@link CodecDefinition#validateContext} must pass.</li>
+ *   <li>Each codec's {@link ChunkCodecHandler#validateContext} must pass.</li>
  * </ol>
  */
 public final class CodecPipelineValidator {

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/CodecRegistry.java

@@ -22,6 +22,7 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 import javax.annotation.Nullable;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecSpecParser;
 
 

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/DeltaCodecDefinition.java

@@ -20,6 +20,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.List;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecContext;
 import org.apache.pinot.segment.spi.codec.CodecKind;
 import org.apache.pinot.segment.spi.codec.CodecOptions;
@@ -101,8 +102,11 @@ public ByteBuffer encode(Options options, CodecContext ctx, ByteBuffer src) {
     int remaining = src.remaining();
     if (ctx.getDataType() == DataType.LONG) {
       return encodeLong(src, remaining);
+    } else if (ctx.getDataType() == DataType.INT) {
+      return encodeInt(src, remaining);
+    } else {
+      throw new IllegalArgumentException("DELTA does not support stored type: " + ctx.getDataType());
     }
-    return encodeInt(src, remaining);
   }
 
   @Override

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/DeltaDeltaCodecDefinition.java

@@ -20,6 +20,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.List;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecContext;
 import org.apache.pinot.segment.spi.codec.CodecKind;
 import org.apache.pinot.segment.spi.codec.CodecOptions;
@@ -103,8 +104,11 @@ public ByteBuffer encode(Options options, CodecContext ctx, ByteBuffer src) {
     int remaining = src.remaining();
     if (ctx.getDataType() == DataType.LONG) {
       return encodeLong(src, remaining);
+    } else if (ctx.getDataType() == DataType.INT) {
+      return encodeInt(src, remaining);
+    } else {
+      throw new IllegalArgumentException("DELTADELTA does not support stored type: " + ctx.getDataType());
     }
-    return encodeInt(src, remaining);
   }
 
   @Override

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/GzipCodecDefinition.java

@@ -24,6 +24,7 @@
 import java.util.zip.DataFormatException;
 import java.util.zip.Deflater;
 import java.util.zip.Inflater;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecContext;
 import org.apache.pinot.segment.spi.codec.CodecKind;
 import org.apache.pinot.segment.spi.codec.CodecOptions;

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/Lz4CodecDefinition.java

@@ -24,6 +24,7 @@
 import net.jpountz.lz4.LZ4CompressorWithLength;
 import net.jpountz.lz4.LZ4DecompressorWithLength;
 import net.jpountz.lz4.LZ4Factory;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecContext;
 import org.apache.pinot.segment.spi.codec.CodecKind;
 import org.apache.pinot.segment.spi.codec.CodecOptions;
@@ -115,6 +116,7 @@ public ByteBuffer decode(Options options, CodecContext ctx, ByteBuffer src) thro
 
   @Override
   public void decodeInto(Options options, CodecContext ctx, ByteBuffer src, ByteBuffer dst) throws IOException {
+    dst.clear();
     LZ4DecompressorWithLength decompressor = new LZ4DecompressorWithLength(LZ4_FACTORY.safeDecompressor());
     int decompressedLength = LZ4DecompressorWithLength.getDecompressedLength(src);
     if (decompressedLength > dst.capacity()) {

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/SnappyCodecDefinition.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.List;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecContext;
 import org.apache.pinot.segment.spi.codec.CodecKind;
 import org.apache.pinot.segment.spi.codec.CodecOptions;

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/ZstdCodecDefinition.java

@@ -23,6 +23,7 @@
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Objects;
+import org.apache.pinot.segment.spi.codec.ChunkCodecHandler;
 import org.apache.pinot.segment.spi.codec.CodecContext;
 import org.apache.pinot.segment.spi.codec.CodecKind;
 import org.apache.pinot.segment.spi.codec.CodecOptions;
@@ -166,6 +167,7 @@ public ByteBuffer decode(Options options, CodecContext ctx, ByteBuffer src) thro
 
   @Override
   public void decodeInto(Options options, CodecContext ctx, ByteBuffer src, ByteBuffer dst) throws IOException {
+    dst.clear();
     ByteBuffer directSrc = CodecBufferUtils.toDirectBuffer(src);
     long decompressedSize = Zstd.decompressedSize(directSrc);
     if (decompressedSize <= 0) {

…nt/local/io/codec/ChunkCodecHandler.java …segment/spi/codec/ChunkCodecHandler.javapinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/ChunkCodecHandler.java renamed to pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/codec/ChunkCodecHandler.java pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/ChunkCodecHandler.java renamed to pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/codec/ChunkCodecHandler.java

@@ -16,13 +16,10 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pinot.segment.local.io.codec;
+package org.apache.pinot.segment.spi.codec;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import org.apache.pinot.segment.spi.codec.CodecContext;
-import org.apache.pinot.segment.spi.codec.CodecDefinition;
-import org.apache.pinot.segment.spi.codec.CodecOptions;
 
 
 /**
@@ -35,8 +32,9 @@
  * <ul>
  *   <li>{@link #encode} and {@link #decode}: {@code src} is ready for read (position=0);
  *       the returned buffer is ready for read and owned by the caller.</li>
- *   <li>{@link #decodeInto}: {@code dst} must be pre-cleared (position=0); it is flipped
- *       (position=0, limit=decoded bytes) on return.</li>
+ *   <li>{@link #decodeInto}: implementations must treat {@code dst} as freshly cleared
+ *       (calling {@code dst.clear()} internally is recommended as a defensive first step);
+ *       {@code dst} is flipped (position=0, limit=decoded bytes) on return.</li>
  * </ul>
  *
  * @param <O> typed {@link CodecOptions} for this codec
@@ -65,15 +63,15 @@ public interface ChunkCodecHandler<O extends CodecOptions> extends CodecDefiniti
 
   /**
    * Decodes {@code src} directly into {@code dst}, avoiding an extra allocation.
-   * {@code dst} must be pre-cleared (position=0); it is flipped on return.
+   * Implementations must treat {@code dst} as freshly cleared and flip it before returning.
    *
    * <p>Callers must ensure {@code dst} is a direct {@link ByteBuffer} when
    * {@link #requiresDirectDstBuffer()} returns {@code true}.
    *
    * @param options parsed options for this codec invocation
    * @param ctx     column context
    * @param src     encoded data, ready for read
-   * @param dst     output buffer, pre-cleared; must be direct when required
+   * @param dst     output buffer; must be direct when required; must have sufficient capacity
    */
   void decodeInto(O options, CodecContext ctx, ByteBuffer src, ByteBuffer dst) throws IOException;