Jackie-Jiang commented on code in PR #19397:
URL: https://github.com/apache/pinot/pull/19397#discussion_r3897400312
##########
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:
[BUG-PERF/C4.3] This replaces one bulk block copy followed by heap-array
reads with roughly 4–8 absolute direct ByteBuffer.get(int) calls per INT/LONG
value. The published DELTA+ZSTD ingestion benchmark covers neither T64 nor
decoding. Add a matched T64 decode JMH benchmark, or retain reusable block
scratch/use word-at-a-time decoding, before assuming this query-path change is
safe.
##########
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:
[BUG-TEST/C6.11] The new process-wide access-ordered plan and alias caches
are shared concurrent state, but eviction tests are single-threaded and the
existing concurrency test creates the executor before starting workers. Add a
test that concurrently calls create() with canonical and alias spellings while
exceeding the 256-plan limit, then verifies identity, type separation, and
stale-alias recovery.
##########
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:
[BUG-PERF/C4.3] The reusable buffers are permanently sized from generic
maxEncodedSize bounds, but T64's bound assumes LONG-sized blocks even though
the executor knows the stored type. For a 1 MiB INT chunk, T64 reports
2,134,021 bytes versus a type-specific worst case of 1,069,061; a T64,LZ4
workspace therefore retains about 4.28 MiB instead of the previous roughly 2.14
MiB peak. Make the bound type-aware before using it to size retained buffers.
--
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]
