Jackie-Jiang commented on code in PR #19307: URL: https://github.com/apache/pinot/pull/19307#discussion_r3887648184
########## pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/readers/forward/ChunkReaderContextTest.java: ########## @@ -0,0 +1,40 @@ +/** + * 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.segment.index.readers.forward; + +import org.apache.pinot.segment.local.io.codec.CodecPipelineExecutor; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertSame; +import static org.testng.Assert.expectThrows; + + +public class ChunkReaderContextTest { Review Comment: This new public test class is missing the required class-level Javadoc. Please document the context lifecycle and scratch-ownership behavior covered here. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/writer/impl/FixedByteChunkForwardIndexWriterV7.java: ########## @@ -0,0 +1,390 @@ +/** + * 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.writer.impl; + +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import javax.annotation.concurrent.NotThreadSafe; +import org.apache.pinot.segment.local.io.codec.CodecPipelineExecutor; +import org.apache.pinot.segment.spi.codec.CodecSpecParser; +import org.apache.pinot.segment.spi.index.ForwardIndexConfig; +import org.apache.pinot.segment.spi.memory.CleanerUtil; +import org.apache.pinot.spi.data.FieldSpec.DataType; + + +/// Chunk-based raw (non-dictionary-encoded) forward index writer for single-value fixed-width +/// columns (INT, LONG) that uses a [CodecPipelineExecutor] for encoding. +/// +/// This writer introduces **version 7** of the fixed-byte chunk raw forward index +/// format. The on-disk layout is: +/// +/// ``` +/// File header: +/// version (int, value = 7) +/// formatMagic (int, value = 0xC0DEC0DE) +/// numChunks (int) +/// numDocsPerChunk (int, normalised to power-of-2) +/// sizeOfEntry (int, bytes per logical value, e.g. 4 for INT) +/// totalDocs (int) +/// codecSpecLength (int, byte length of the UTF-8 encoded canonical codec spec) +/// dataHeaderStart (int, byte offset from file start where chunk-offset table begins) +/// codecSpec (byte[], UTF-8 encoded canonical spec, length = codecSpecLength) +/// chunkOffsets (long[numChunks], absolute byte offset of each chunk's per-chunk header) +/// Data (per chunk): +/// encodedSize (int, byte length of the encoded payload that follows) +/// decodedSize (int, byte length of the original decoded chunk data) +/// payload (byte[], encoded chunk data, length = encodedSize) +/// ``` +/// +/// Each chunk contains `numDocsPerChunk` values encoded by the pipeline. Chunk offsets +/// are 8-byte longs to support files larger than 2 GB. The per-chunk size header allows readers +/// to verify decoded output and to skip/read chunks without scanning adjacent offsets. +/// +/// This class is *not* thread-safe. +@NotThreadSafe +public class FixedByteChunkForwardIndexWriterV7 implements FixedByteValueWriter { + + public static final int VERSION = ForwardIndexConfig.CODEC_PIPELINE_WRITER_VERSION; + public static final int FORMAT_MAGIC = 0xC0DEC0DE; + + /// Upper bound for the canonical, ASCII-only codec spec embedded in the header. Keep the wire + /// limit aligned with the DSL parser so every accepted header is representable by public config. + public static final int MAX_CODEC_SPEC_LENGTH_BYTES = CodecSpecParser.MAX_SPEC_LENGTH; + + /// Maximum decoded bytes in one V7 chunk. The normal Pinot target is 1 MiB; this 64 MiB ceiling + /// bounds per-reader direct scratch and intermediate pipeline buffers for corrupt segments. + public static final int MAX_DECODED_CHUNK_SIZE_BYTES = 64 * 1024 * 1024; + + /// Maximum conservative encoded-size bound for every stage in a V7 pipeline. Writers reject a + /// pipeline/chunk-size combination whose composed bound exceeds this ceiling, so readers can + /// allocate bounded scratch without accepting a file that the same writer could not read back. + public static final int MAX_ENCODED_CHUNK_SIZE_BYTES = 128 * 1024 * 1024; + + /// Maximum sum of all stage-output bounds for one chunk. This prevents a long pipeline of + /// individually bounded transforms from causing unbounded allocation and CPU churn. + public static final long MAX_PIPELINE_WORK_SIZE_BYTES = 256L * 1024 * 1024; + + /// Bytes written before each chunk payload: encodedSize (int) + decodedSize (int). + public static final int CHUNK_HEADER_BYTES = 2 * Integer.BYTES; + + // Number of fixed int fields before the codec spec: version, formatMagic, numChunks, + // numDocsPerChunk, sizeOfEntry, totalDocs, codecSpecLength, dataHeaderStart + private static final int FIXED_HEADER_INT_COUNT = 8; + public static final int FIXED_HEADER_BYTES = FIXED_HEADER_INT_COUNT * Integer.BYTES; + + // Hold both the RAF and its FileChannel: closing the channel closes the underlying FD, but + // some JVM finalizers close the FD when the RAF becomes unreachable. Holding the RAF as a + // field anchors it to the writer's lifetime and removes any reliance on finalizer ordering. + private final RandomAccessFile _raf; + private final FileChannel _dataFile; + private final CodecPipelineExecutor _executor; + private final int _numDocsPerChunk; + private final int _sizeOfEntry; + private final int _chunkFullBytes; + private final int _maxFullChunkEncodedSize; + private final ByteBuffer _header; + private final ByteBuffer _chunkBuffer; + private final ByteBuffer _chunkHeaderBuffer = ByteBuffer.allocateDirect(CHUNK_HEADER_BYTES); + private final int _numChunks; + private final int _totalDocs; + + private long _dataOffset; + private int _docsWritten; + private int _chunksWritten; + private boolean _trackUncompressedValueSize; + + /// Creates a new writer. + /// + /// @param file output file + /// @param executor pre-validated pipeline executor + /// @param totalDocs total number of documents to write + /// @param numDocsPerChunk target documents per chunk (will be rounded up to power-of-2) + /// @param sizeOfEntry bytes per value (e.g. 4 for INT, 8 for LONG) + public FixedByteChunkForwardIndexWriterV7(File file, CodecPipelineExecutor executor, int totalDocs, + int numDocsPerChunk, int sizeOfEntry) + throws IOException { + if (totalDocs < 0) { + throw new IllegalArgumentException("totalDocs must be non-negative, got: " + totalDocs); + } + _executor = executor; + _numDocsPerChunk = validateChunkConfiguration(executor, sizeOfEntry, numDocsPerChunk); + _sizeOfEntry = sizeOfEntry; + _totalDocs = totalDocs; + long chunkSizeLong = (long) sizeOfEntry * _numDocsPerChunk; + _chunkFullBytes = (int) chunkSizeLong; + _maxFullChunkEncodedSize = executor.maxEncodedSize(_chunkFullBytes, MAX_ENCODED_CHUNK_SIZE_BYTES, + MAX_PIPELINE_WORK_SIZE_BYTES); + _numChunks = (int) (((long) totalDocs + _numDocsPerChunk - 1) / _numDocsPerChunk); + _docsWritten = 0; + _chunksWritten = 0; + + byte[] specBytes = executor.getCanonicalSpec().getBytes(StandardCharsets.UTF_8); + if (specBytes.length > MAX_CODEC_SPEC_LENGTH_BYTES) { + throw new IllegalArgumentException( + "Canonical codec spec is " + specBytes.length + " bytes; maximum is " + MAX_CODEC_SPEC_LENGTH_BYTES); + } + + // Header layout: + // 8 ints of fixed fields + // specBytes.length bytes of codec spec + // numChunks longs of chunk offsets + long fixedHeaderBytesLong = FIXED_HEADER_BYTES; + long dataHeaderStartLong = fixedHeaderBytesLong + specBytes.length; + long chunkOffsetTableBytesLong = (long) _numChunks * Long.BYTES; + long totalHeaderBytesLong = dataHeaderStartLong + chunkOffsetTableBytesLong; + if (totalHeaderBytesLong > Integer.MAX_VALUE) { + throw new IllegalArgumentException( + "Header size " + totalHeaderBytesLong + " bytes exceeds Integer.MAX_VALUE. Reduce totalDocs or" + + " increase numDocsPerChunk."); + } + int dataHeaderStart = (int) dataHeaderStartLong; + int totalHeaderBytes = (int) totalHeaderBytesLong; + + _header = ByteBuffer.allocateDirect(totalHeaderBytes); + _header.putInt(VERSION); + _header.putInt(FORMAT_MAGIC); + _header.putInt(_numChunks); + _header.putInt(_numDocsPerChunk); + _header.putInt(sizeOfEntry); + _header.putInt(totalDocs); + _header.putInt(specBytes.length); + _header.putInt(dataHeaderStart); + _header.put(specBytes); + // chunk offsets will be filled in during writeChunk() calls + + _dataOffset = totalHeaderBytes; + + // Open file first, then allocate the direct buffer under a try/catch so that an OOM during + // allocation closes the already-open file descriptor (the caller has no reference to a + // partially-constructed object and cannot invoke close() itself). + RandomAccessFile raf = new RandomAccessFile(file, "rw"); + FileChannel channel = raf.getChannel(); + try { + raf.setLength(0L); + _chunkBuffer = ByteBuffer.allocateDirect((int) chunkSizeLong); + } catch (Throwable t) { + try { + raf.close(); + } catch (IOException closeEx) { + t.addSuppressed(closeEx); + } + throw t; + } + _raf = raf; + _dataFile = channel; + } + + /// Writes a 4-byte integer value. + @Override + public void putInt(int value) { + if (_sizeOfEntry != Integer.BYTES) { + throw new IllegalStateException("putInt cannot write a LONG V7 forward index"); + } + checkRoomForOneMore(); + _chunkBuffer.putInt(value); + _docsWritten++; + flushIfNeeded(); + } + + /// Writes an 8-byte long value. + @Override + public void putLong(long value) { + if (_sizeOfEntry != Long.BYTES) { + throw new IllegalStateException("putLong cannot write an INT V7 forward index"); + } + checkRoomForOneMore(); + _chunkBuffer.putLong(value); + _docsWritten++; + flushIfNeeded(); + } + + /// The V7 codec-pipeline transforms (DELTA/DELTADELTA/T64/GORILLA) are defined for integral + /// INT/LONG values only, so FLOAT is not supported by this writer. + @Override + public void putFloat(float value) { + throw new UnsupportedOperationException("V7 codec-pipeline writer does not support FLOAT"); + } + + /// See [#putFloat] — DOUBLE is likewise unsupported by the V7 codec-pipeline writer. + @Override + public void putDouble(double value) { + throw new UnsupportedOperationException("V7 codec-pipeline writer does not support DOUBLE"); + } + + @Override + public long getRawForwardIndexUncompressedValueSizeInBytes() { + return _trackUncompressedValueSize ? (long) _docsWritten * _sizeOfEntry : -1; + } + + @Override + public void enableRawForwardIndexUncompressedValueSizeTracking() { + if (_docsWritten != 0) { + throw new IllegalStateException("Uncompressed-size tracking must be enabled before writing values"); + } + _trackUncompressedValueSize = true; + } + + /// 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. + private void checkRoomForOneMore() { + if (_docsWritten >= _totalDocs) { + throw new IllegalStateException( + "Cannot write past declared totalDocs=" + _totalDocs + " (already wrote " + _docsWritten + ")"); + } + } + + private void flushIfNeeded() { + if (_chunkBuffer.position() == _chunkFullBytes) { + writeChunk(); + } + } + + private void writeChunk() { + _chunkBuffer.flip(); + int decodedSize = _chunkBuffer.remaining(); + ByteBuffer encoded = null; + try { + encoded = _executor.encode(_chunkBuffer); Review Comment: Every chunk currently allocates and explicitly cleans roughly one direct buffer per codec stage. With the default 1,024-doc chunks, a three-stage pipeline causes about 2,931 native allocation/cleanup cycles per million rows per column, unlike the legacy writer which reuses its compression buffer. Please add reusable encode scratch buffers or an `encodeInto` API, and benchmark this ingestion path. ########## pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandlerTest.java: ########## @@ -2834,6 +2835,80 @@ public void testBackfillFromInvertedIndexRebuild() } } + /// Exercises the complete reload lifecycle with real segment files: legacy to V7 for even a + /// compression-only codecSpec, unchanged V7, V7 spec change, and rollback to a legacy format. + @Test + public void testCodecSpecReloadLifecycle() + throws Exception { + _fieldConfigMap.put(DIM_LZ4_INTEGER, rawFieldConfigWithCodecSpec(DIM_LZ4_INTEGER, "LZ4")); + applyCodecRewrite(DIM_LZ4_INTEGER); + assertRawForwardIndexState(DIM_LZ4_INTEGER, "LZ4", null); + + // The stored canonical spec matches, so a second reload is a no-op. + try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap); + SegmentDirectory.Writer writer = segmentDirectory.createWriter()) { + _segmentDirectory = segmentDirectory; + _writer = writer; + assertFalse(computeOperations().containsKey(DIM_LZ4_INTEGER)); + } + + _fieldConfigMap.put(DIM_LZ4_INTEGER, rawFieldConfigWithCodecSpec(DIM_LZ4_INTEGER, "DELTA,LZ4")); + applyCodecRewrite(DIM_LZ4_INTEGER); + assertRawForwardIndexState(DIM_LZ4_INTEGER, "DELTA,LZ4", null); + + _fieldConfigMap.put(DIM_LZ4_INTEGER, Review Comment: This only tests leaving V7 while explicitly selecting `SNAPPY`. It does not cover the documented remove-only rollback path where `codecSpec` is removed without a replacement `compressionCodec`; a regression to the old gate could therefore leave V7 files unreadable after rollback while this test still passes. Please add that transition and assert a legacy reader, the field-type default compression, and unchanged values. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java: ########## @@ -513,24 +514,37 @@ private boolean shouldDisableDictionary(String column, ColumnMetadata existingCo return true; } - private boolean shouldChangeRawCompressionType(String column, SegmentDirectory.Reader segmentReader) + private boolean shouldRewriteRawForwardIndex(String column, SegmentDirectory.Reader segmentReader) throws Exception { - // The compression type for an existing segment can only be determined by reading the forward index header. + // The persisted compression type / codec spec can only be determined from the forward-index header. ColumnMetadata existingColMetadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column); ChunkCompressionType existingCompressionType; + String existingCodecSpec; - // Get the forward index reader factory and create a reader IndexReaderFactory<ForwardIndexReader> readerFactory = StandardIndexes.forward().getReaderFactory(); try (ForwardIndexReader<?> fwdIndexReader = readerFactory.createIndexReader(segmentReader, Review Comment: Reload reconciliation constructs a complete reader only to obtain `codecSpec`. The V7 reader constructor then walks every chunk frame, potentially faulting pages throughout every unchanged index for every segment and V7 column. Please use a lightweight header-only format/spec inspector or persist a codec-spec fingerprint in metadata. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/FixedByteChunkSVForwardIndexReaderV7.java: ########## @@ -0,0 +1,398 @@ +/** + * 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.segment.index.readers.forward; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import org.apache.pinot.segment.local.io.codec.CodecPipelineExecutor; +import org.apache.pinot.segment.local.io.writer.impl.FixedByteChunkForwardIndexWriterV7; +import org.apache.pinot.segment.spi.index.ForwardIndexConfig; +import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader; +import org.apache.pinot.segment.spi.memory.PinotDataBuffer; +import org.apache.pinot.spi.data.FieldSpec.DataType; + + +/// Chunk-based single-value raw forward index reader for version-7 files written by +/// [FixedByteChunkForwardIndexWriterV7]. +/// +/// Reads the canonical codec spec from the file header, instantiates a +/// [CodecPipelineExecutor], and uses it to decode each chunk on demand. +/// +/// Supported data types: INT, LONG. +/// +/// **Threading:** the reader instance itself is immutable after construction and is safe to +/// share across threads, but each [ChunkReaderContext] is single-threaded. The +/// [ByteBuffer] returned by `getChunkBuffer` is the context's reusable scratch buffer — +/// its position/limit are mutated on every chunk transition and it must not be retained across +/// subsequent `getInt`/`getLong` calls. +public final class FixedByteChunkSVForwardIndexReaderV7 implements ForwardIndexReader<ChunkReaderContext> { + + public static final int VERSION = ForwardIndexConfig.CODEC_PIPELINE_WRITER_VERSION; + + private final PinotDataBuffer _dataBuffer; + private final DataType _storedType; + private final int _numChunks; + private final int _numDocsPerChunk; + private final int _shift; // log2(numDocsPerChunk) for fast chunk id calc + private final int _totalDocs; + private final int _dataHeaderStart; + private final int _chunkCapacityBytes; + private final CodecPipelineExecutor _executor; + private final String _canonicalSpec; + /// Composed pipeline encoded-size bound for a full chunk, computed once at construction; every + /// chunk except a partial final chunk reuses it instead of re-walking the pipeline stages. + private final int _maxFullChunkEncodedSize; + + /// Returns whether the buffer has the explicit header discriminator for the codec-pipeline V7 + /// format. Keep this predicate limited to the marker: once recognized, the constructor must see + /// and reject every corrupt structural field instead of letting the factory fall back to the + /// legacy version-7 reader. + public static boolean hasCodecPipelineHeader(PinotDataBuffer dataBuffer) { + return dataBuffer.size() >= 2L * Integer.BYTES && dataBuffer.getInt(0) == VERSION + && dataBuffer.getInt(Integer.BYTES) == FixedByteChunkForwardIndexWriterV7.FORMAT_MAGIC; + } + + public FixedByteChunkSVForwardIndexReaderV7(PinotDataBuffer dataBuffer, DataType storedType) { + this(dataBuffer, storedType, -1); + } + + /// Creates a V7 reader and, when `expectedTotalDocs` is non-negative, verifies that the index + /// belongs to segment metadata with the same document count. The two-argument constructor keeps + /// standalone fixture and StarTree helper reads available when no segment metadata is present. + public FixedByteChunkSVForwardIndexReaderV7(PinotDataBuffer dataBuffer, DataType storedType, + int expectedTotalDocs) { + _dataBuffer = dataBuffer; + _storedType = storedType; + + long bufferSize = dataBuffer.size(); + if (bufferSize < FixedByteChunkForwardIndexWriterV7.FIXED_HEADER_BYTES) { + throw new IllegalArgumentException( + "V7 forward index is truncated: " + bufferSize + " bytes; minimum header is " + + FixedByteChunkForwardIndexWriterV7.FIXED_HEADER_BYTES + " bytes"); + } + + if (storedType != DataType.INT && storedType != DataType.LONG) { + throw new IllegalArgumentException( + "FixedByteChunkSVForwardIndexReaderV7 only supports INT and LONG, got: " + storedType); + } + + int offset = 0; + int version = dataBuffer.getInt(offset); + if (version != VERSION) { + throw new IllegalArgumentException("Expected version " + VERSION + " but got " + version); + } + if (!hasCodecPipelineHeader(dataBuffer)) { + throw new IllegalArgumentException( + "Version " + VERSION + " buffer does not contain a valid codec-pipeline header discriminator"); + } + offset += Integer.BYTES; + + int formatMagic = dataBuffer.getInt(offset); + offset += Integer.BYTES; + if (formatMagic != FixedByteChunkForwardIndexWriterV7.FORMAT_MAGIC) { + throw new IllegalArgumentException("Invalid codec-pipeline format magic: " + formatMagic); + } + + _numChunks = dataBuffer.getInt(offset); + offset += Integer.BYTES; + if (_numChunks < 0) { + throw new IllegalArgumentException("Invalid numChunks in forward index header: " + _numChunks); + } + + _numDocsPerChunk = dataBuffer.getInt(offset); + offset += Integer.BYTES; + if (_numDocsPerChunk <= 0 || (_numDocsPerChunk & (_numDocsPerChunk - 1)) != 0) { + throw new IllegalArgumentException( + "Invalid numDocsPerChunk in forward index header: " + _numDocsPerChunk + + ". Expected a positive power of two."); + } + _shift = Integer.numberOfTrailingZeros(_numDocsPerChunk); + + int sizeOfEntry = dataBuffer.getInt(offset); + offset += Integer.BYTES; + if (sizeOfEntry != storedType.size()) { + throw new IllegalArgumentException( + "Header sizeOfEntry=" + sizeOfEntry + " does not match storedType=" + storedType + + " (expected " + storedType.size() + " bytes). Written for a different data type?"); + } + long chunkCapacity = (long) _numDocsPerChunk * sizeOfEntry; + if (chunkCapacity > FixedByteChunkForwardIndexWriterV7.MAX_DECODED_CHUNK_SIZE_BYTES) { + throw new IllegalArgumentException( + "Decoded chunk capacity " + chunkCapacity + " bytes exceeds V7 limit " + + FixedByteChunkForwardIndexWriterV7.MAX_DECODED_CHUNK_SIZE_BYTES + ". Segment may be corrupt."); + } + _chunkCapacityBytes = (int) chunkCapacity; + + _totalDocs = dataBuffer.getInt(offset); + offset += Integer.BYTES; + if (_totalDocs < 0) { + throw new IllegalArgumentException("Invalid totalDocs in forward index header: " + _totalDocs); + } + if (expectedTotalDocs >= 0 && _totalDocs != expectedTotalDocs) { + throw new IllegalArgumentException( + "V7 forward index totalDocs=" + _totalDocs + " does not match segment metadata totalDocs=" + + expectedTotalDocs); + } + + // Validate numChunks/totalDocs/numDocsPerChunk are mutually consistent. A corrupt header + // with mismatched values would otherwise let getChunkOffset() read past the chunk-offset table. + int expectedNumChunks = (int) (((long) _totalDocs + _numDocsPerChunk - 1) / _numDocsPerChunk); + if (_numChunks != expectedNumChunks) { + throw new IllegalArgumentException( + "Inconsistent header: numChunks=" + _numChunks + " but totalDocs=" + _totalDocs + " / numDocsPerChunk=" + + _numDocsPerChunk + " => expected " + expectedNumChunks + ". Segment may be corrupt."); + } + + int specLength = dataBuffer.getInt(offset); + offset += Integer.BYTES; + if (specLength <= 0 || specLength > FixedByteChunkForwardIndexWriterV7.MAX_CODEC_SPEC_LENGTH_BYTES) { + // V7 segments always embed a non-empty canonical codec spec; zero or negative is corruption. + throw new IllegalArgumentException( + "Invalid specLength in forward index header: " + specLength + "; expected [1, " + + FixedByteChunkForwardIndexWriterV7.MAX_CODEC_SPEC_LENGTH_BYTES + "]. Segment may be corrupt."); + } + + _dataHeaderStart = dataBuffer.getInt(offset); + offset += Integer.BYTES; + + // Validate dataHeaderStart, specLength, and the chunk-offset table bounds before using them. + long expectedSpecEnd = (long) offset + specLength; + long chunkOffsetTableEnd = _dataHeaderStart + (long) _numChunks * Long.BYTES; + if (specLength > bufferSize || _dataHeaderStart < 0 || _dataHeaderStart > bufferSize + || expectedSpecEnd != _dataHeaderStart || chunkOffsetTableEnd > bufferSize) { + throw new IllegalArgumentException( + "Forward index header is corrupt: specLength=" + specLength + ", dataHeaderStart=" + _dataHeaderStart + + ", numChunks=" + _numChunks + ", bufferSize=" + bufferSize); + } + + // Read codec spec bytes + byte[] specBytes = new byte[specLength]; + dataBuffer.copyTo(offset, specBytes, 0, specLength); + _canonicalSpec = new String(specBytes, StandardCharsets.UTF_8); + + try { + _executor = CodecPipelineExecutor.create(_canonicalSpec, storedType); Review Comment: Every segment reader reparses, validates, and rebuilds the same immutable, thread-safe codec execution plan; reload reconciliation constructs another executor from the table config per segment. Please cache plans by canonical specification and stored type, and resolve the configured plan once per handler/config lifecycle. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java: ########## @@ -405,17 +406,17 @@ private List<Operation> computeColumnOperations(String column, FieldSpec fieldSp } } - // 3. Compression-type change (only when no encoding change happened). - if (ops.isEmpty() && existingFwdEncoding != null && existingFwdEncoding == newFwdEncoding - && existingHasDict == desiredDict) { - if (existingFwdEncoding == EncodingType.RAW) { - // TODO: Also check if raw index version needs to be changed - if (shouldChangeRawCompressionType(column, segmentReader)) { - ops.add(Operation.CHANGE_INDEX_COMPRESSION_TYPE); - } - } else if (shouldChangeDictIdCompressionType(column, segmentReader)) { + // 3. Raw format/codec change. Adding or removing a standalone dictionary preserves a RAW + // forward index, so codec reconciliation must run independently of dictionary operations. + // Encoding conversions recreate the forward index with the new config and need no second rewrite. + if (existingFwdEncoding == EncodingType.RAW && newFwdEncoding == EncodingType.RAW) { + // TODO: Also check if raw index version needs to be changed + if (shouldRewriteRawForwardIndex(column, segmentReader)) { ops.add(Operation.CHANGE_INDEX_COMPRESSION_TYPE); Review Comment: This operation now covers codec-spec changes and legacy/V7 format transitions, but it is still named `CHANGE_INDEX_COMPRESSION_TYPE` and invokes `rewriteForwardIndexForCompressionChange`. Please rename the operation and helper around raw-forward-index format/codec rewrites so their names match the expanded behavior. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/ChunkReaderContext.java: ########## @@ -53,13 +55,27 @@ public void close() { return; } _closed = true; + if (_codecDecodeScratch != null) { + _codecDecodeScratch.close(); + _codecDecodeScratch = null; + } CleanerUtil.cleanQuietly(_chunkBuffer); } public ByteBuffer getChunkBuffer() { return _chunkBuffer; } + CodecPipelineExecutor.DecodeScratch getCodecDecodeScratch() { Review Comment: `getCodecDecodeScratch()` allocates and stores state on first access, so the getter name hides mutation. Please rename it to `getOrCreateCodecDecodeScratch()` or initialize the scratch object elsewhere. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
