xiangfu0 commented on code in PR #19307: URL: https://github.com/apache/pinot/pull/19307#discussion_r3888191301
########## 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: Added caller-owned `EncodeScratch` retained for the V7 writer lifecycle plus direct `encodeInto` implementations for the built-in stages. Scratch memory is bounded and closed with the writer. A matched V7 `DELTA,ZSTD(3)` ingestion JMH improved from 14.511 to 12.296 ms/op (15.3%) with zero GC in both runs. ########## 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: The consolidated real reload test now removes `codecSpec` without setting a replacement compression codec, asserts rewrite to the legacy field-type default `LZ4`, verifies legacy reader dispatch, and checks every value remains unchanged. The same test also covers an equivalent alias no-op and a changed V7 spec rewrite. ########## pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/ForwardIndexConfig.java: ########## @@ -79,6 +79,15 @@ public static ForwardIndexConfig getDisabled() { return new Builder(EncodingType.DICTIONARY).withDisabled(true).build(); } + /// Writer version used for codec-pipeline forward indexes. Legacy fixed-byte writers + /// accept arbitrary versions greater than or equal to 4, so version 7 alone does not identify + /// this format; readers also require the explicit codec-pipeline header magic. + /// + /// **The version and structural header marker together are the frozen on-disk identifier.** + /// Version 7 must never be used alone to dispatch this format because legacy fixed-byte writers + /// can also emit that value. A new format must define an unambiguous discriminator. + public static final int CODEC_PIPELINE_WRITER_VERSION = 7; Review Comment: Resolved by making the format version writer-owned as `FixedByteChunkForwardIndexWriterV7.VERSION`. Factory dispatch now requires both the historical version field and the V7 magic discriminator, so legacy version 7 remains on the legacy reader path. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/writer/impl/FixedByteChunkForwardIndexWriter.java: ########## @@ -27,7 +27,8 @@ /// Chunk-based raw (non-dictionary-encoded) forward index writer where each chunk contains fixed number of docs, and /// each entry has fixed number of bytes. @NotThreadSafe -public class FixedByteChunkForwardIndexWriter extends BaseChunkForwardIndexWriter { +public class FixedByteChunkForwardIndexWriter extends BaseChunkForwardIndexWriter Review Comment: Added `@Override` to the fixed-byte writer methods implemented by the legacy writer. ########## 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: Moved codec scratch ownership out of the shared `ChunkReaderContext` and into the V7 reader-specific `Context`. Its lazy mutating accessor is now `getOrCreateDecodeScratch()`, while the shared legacy context remains codec-agnostic. ########## 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: Renamed the operation to `REWRITE_FORWARD_INDEX` and the helper to `rewriteForwardIndex`, covering compression, codec-spec, and legacy/V7 format transitions. ########## 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: Added a bounded immutable plan cache keyed by canonical spec and stored type, with a bounded alias cache for equivalent spellings. `ForwardIndexHandler` also resolves and retains the configured plan once per column for its lifecycle; the real reload test covers alias no-op and changed-spec rewrite paths. ########## 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 now uses the bounded header-only `readCodecSpec(PinotDataBuffer)` inspector for V7. It does not construct a reader or scan chunk frames for unchanged indexes. ########## 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: Removed this standalone public test while consolidating the oversized suite. Scratch ownership now lives in the documented V7-specific context and its lifecycle is exercised by the real creator/reader round trip. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/writer/impl/FixedByteValueWriter.java: ########## @@ -0,0 +1,53 @@ +/** + * 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.Closeable; +import javax.annotation.concurrent.NotThreadSafe; + + +/// Common write API for single-value fixed-width raw forward-index chunk writers, letting callers +/// hold a single writer reference regardless of the on-disk format that backs it. +/// +/// Two implementations exist, selected by the configured codec spec: +/// - [FixedByteChunkForwardIndexWriter] — the legacy chunk format (versions 2, 3, and arbitrary +/// fixed-writer tags greater than or equal to 4); used for one plain `ChunkCompressionType`. +/// Supports INT/LONG/FLOAT/DOUBLE. +/// - [FixedByteChunkForwardIndexWriterV7] — the explicitly marked V7 codec-pipeline format; used +/// for every configured `codecSpec`. Supports INT/LONG only; +/// `putFloat`/`putDouble` throw [UnsupportedOperationException]. +/// +/// Instances require serialized access and are not safe for concurrent use. +@NotThreadSafe +public interface FixedByteValueWriter extends Closeable { Review Comment: Renamed the interface and file to `FixedByteChunkWriter`; both the legacy and V7 writers now implement that name. -- 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]
