xiangfu0 commented on code in PR #19397:
URL: https://github.com/apache/pinot/pull/19397#discussion_r3900280062
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/T64CodecDefinition.java:
##########
@@ -409,19 +390,18 @@ private static void writeBits(byte[] buf, long bitOffset,
long value, int bitWid
}
}
- /// Read `bitWidth` bits from `buf` starting at `bitOffset`.
- private static long readBits(byte[] buf, long bitOffset, int bitWidth) {
- long byteIndex = bitOffset >>> 3;
+ /// Read `bitWidth` bits from `buf` starting at `baseOffset + bitOffset`
without copying the
+ /// packed block into a temporary heap array.
+ private static long readBits(ByteBuffer buf, int baseOffset, long bitOffset,
int bitWidth) {
+ int idx = baseOffset + (int) (bitOffset >>> 3);
int bitInByte = (int) (bitOffset & 7);
long out = 0L;
int shift = 0;
int bitsRemaining = bitWidth;
- int idx = (int) byteIndex;
while (bitsRemaining > 0) {
- long b = ((long) buf[idx]) & 0xFFL;
+ long b = ((long) buf.get(idx)) & 0xFFL;
Review Comment:
Updated in 11b7ce6cfc: T64 decode now uses a sequential block-local 64-bit
reservoir. Each packed word is loaded and reversed once, and crossing values
reuse the next word. Matched JMH across widths 1/2/8/32/64 and heap/direct
sources improved 31.7–93.8%; all 149 T64 tests pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/CodecPipelineExecutor.java:
##########
@@ -245,115 +464,59 @@ public int maxEncodedSize(int decodedSize, int
maxStageSize, long maxCumulativeS
"Codec pipeline cumulative stage-output bound " + cumulativeSize +
" exceeds " + maxCumulativeSize
+ " at stage " + i + " for pipeline " + _canonicalSpec);
}
- }
- return size;
- }
-
- /// Encodes `src` through the pipeline and returns the encoded bytes ready
for read.
- ///
- /// The readable range is interpreted in the persisted big-endian
typed-value order, independent
- /// of the caller view's byte order. The caller's position, limit, and order
are not modified.
- ///
- /// @param src decoded chunk data, ready for read (position=0,
limit=dataSize)
- /// @return encoded buffer ready for read; the caller owns this buffer
- public ByteBuffer encode(ByteBuffer src) throws IOException {
- ByteBuffer input = src.duplicate().order(ByteOrder.BIG_ENDIAN);
- ByteBuffer current = input;
- try {
- for (BoundStage<?> stage : _stages) {
- ByteBuffer previous = current;
- current = stage.encode(previous);
- if (previous != input && previous != current) {
- CleanerUtil.cleanQuietly(previous);
- }
+ if (stageBounds != null) {
+ stageBounds[i] = size;
}
- return current;
- } catch (IOException | RuntimeException | Error e) {
- if (current != input) {
- CleanerUtil.cleanQuietly(current);
- }
- throw e;
- }
- }
-
- /// Decodes `src` through the reversed pipeline and returns the original
bytes.
- ///
- /// @param src encoded chunk data, ready for read
- /// @return decoded buffer ready for read; the caller owns this buffer
- ByteBuffer decode(ByteBuffer src) throws IOException {
- ByteBuffer current = src;
- try {
- for (int i = _stages.size() - 1; i >= 0; i--) {
- ByteBuffer previous = current;
- current = _stages.get(i).decode(previous);
- if (previous != src && previous != current) {
- CleanerUtil.cleanQuietly(previous);
- }
- }
- return current;
- } catch (IOException | RuntimeException | Error e) {
- if (current != src) {
- CleanerUtil.cleanQuietly(current);
- }
- throw e;
}
+ return size;
}
- /// Decodes `src` through the reversed pipeline, writing the result directly
into
- /// `dst`. On return `dst` is flipped and ready for read (position=0,
- /// limit=decoded size).
+ /// Encodes through capacity-bounded caller-owned scratch buffers.
///
- /// For single-stage pipelines (transform-only or compression-only), the
decoded bytes are
- /// written directly into `dst`, avoiding an intermediate allocation. For
multi-stage
- /// pipelines an intermediate buffer is allocated for any stage that is not
the last in the
- /// reversed pipeline; the final stage writes directly into `dst`.
+ /// Every stage output is written directly into one of two alternating
direct buffers. The
+ /// returned readable view is owned by `scratch`: callers must consume or
copy it before the next
+ /// call using that workspace, and must not use it after the workspace is
closed. The caller's
+ /// source position, limit, and byte order are not modified. Typed input is
interpreted in
+ /// persisted big-endian order, independent of the caller view's byte order.
The source must
+ /// not alias this workspace's buffers (including a view returned by a
previous call).
///
- /// @param src encoded chunk data, ready for read
- /// @param dst caller-supplied output buffer; must be a *direct*
[ByteBuffer] when
- /// the pipeline requires it (see
[ChunkCodecHandler#requiresDirectDstBuffer]);
- /// must have sufficient [ByteBuffer#capacity()] for the decoded
data; its
- /// position and limit are overwritten before returning
- /// @throws IOException if decoding fails
- /// @throws IllegalArgumentException if `dst` is not direct when required,
or does not have
- /// enough capacity
- void decode(ByteBuffer src, ByteBuffer dst) throws IOException {
- decode(src, dst, dst.capacity());
- }
-
- /// Decodes into `dst` while bounding every intermediate allocation from the
expected final
- /// decoded size. This is the segment-reader entry point: codec frames are
untrusted and may
- /// advertise arbitrary decoded lengths, so an intermediate stage must never
allocate directly
- /// from its frame header.
- ///
- /// @param src encoded chunk data, ready for read
- /// @param dst caller-owned final output buffer
- /// @param expectedDecodedSize exact decoded bytes declared and validated by
the outer format
- void decode(ByteBuffer src, ByteBuffer dst, int expectedDecodedSize) throws
IOException {
- decode(src, dst, expectedDecodedSize, Integer.MAX_VALUE, Long.MAX_VALUE);
- }
-
- /// Variant of [#decode(ByteBuffer, ByteBuffer, int)] that caps every
intermediate scratch
- /// buffer. Callers reading untrusted persisted data should pass the
format's validated limit.
- void decode(ByteBuffer src, ByteBuffer dst, int expectedDecodedSize, int
maxIntermediateSize)
- throws IOException {
- decode(src, dst, expectedDecodedSize, maxIntermediateSize, Long.MAX_VALUE);
- }
-
- /// Variant that also caps the sum of stage-output bounds, limiting total
work and allocation
- /// churn for an otherwise-valid long pipeline. This convenience overload
owns a temporary
- /// workspace; production chunk readers should use the caller-owned
[DecodeScratch] overload.
- void decode(ByteBuffer src, ByteBuffer dst, int expectedDecodedSize, int
maxIntermediateSize,
- long maxCumulativeSize)
+ /// @param src decoded chunk data, ready for read
+ /// @param maxStageSize maximum permitted output bound for any stage
+ /// @param maxCumulativeSize maximum permitted sum of all stage-output bounds
+ /// @param scratch caller-owned workspace, not shared across
concurrent calls
+ /// @return scratch-owned encoded bytes ready for read
+ public ByteBuffer encode(ByteBuffer src, int maxStageSize, long
maxCumulativeSize, EncodeScratch scratch)
throws IOException {
- try (DecodeScratch scratch = new DecodeScratch()) {
- decode(src, dst, expectedDecodedSize, maxIntermediateSize,
maxCumulativeSize, scratch);
+ scratch.ensureOpen();
+ ByteBuffer current = src.duplicate().order(ByteOrder.BIG_ENDIAN);
+ int[] stageBounds = scratch.stageBounds(_stages.size());
+ encodedStageBounds(current.remaining(), maxStageSize, maxCumulativeSize,
stageBounds);
Review Comment:
Updated in 11b7ce6cfc: maxEncodedSize is now context-aware and T64
calculates its bound from the stored INT/LONG element width. For a 1 MiB input
the exact bounds are 1,069,061 bytes for INT and 1,067,013 bytes for LONG,
eliminating the generic LONG over-retention; focused tests cover both.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/codec/CodecPipelineExecutor.java:
##########
@@ -166,7 +342,43 @@ public void close() {
/// @param spec the codec DSL string (e.g. `"LZ4,ZSTD(3)"`)
/// @param storedType stored data type used for validation
public static CodecPipelineExecutor create(String spec, DataType storedType)
{
- return create(spec, new CodecContext(storedType), CodecRegistry.DEFAULT);
+ // Preserve parser rejection before any normalization-based cache lookup.
In particular, outer
+ // whitespace must not let an overlong spec collapse to a cached canonical
spelling.
+ if (spec == null || spec.length() > CodecSpecParser.MAX_SPEC_LENGTH ||
spec.isBlank()) {
+ return create(spec, new CodecContext(storedType), CodecRegistry.DEFAULT);
+ }
+ PlanKey requestedKey = new PlanKey(normalizeRequestedSpec(spec),
storedType);
+ synchronized (PLAN_CACHE_LOCK) {
Review Comment:
Updated in 11b7ce6cfc: added a 20-second-bounded, no-mock concurrency test
where four workers race canonical and alias create() calls while churning 289
plans past the 256-entry limit. It verifies shared identity, eviction recovery,
stale-alias recovery, and stored-type separation.
--
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]
