You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2026/08/30 20:57:04 UTC
[PR] Use caller-owned buffers for codec pipelines [pinot]
xiangfu0 opened a new pull request, #19397:
URL: https://github.com/apache/pinot/pull/19397
## Summary
Extracts the codec-runtime changes requested in #19307 into an independently reviewable prerequisite.
- Replaces the unreleased allocating handler API with required destination-based `encode` and `decode` methods; no compatibility adapters.
- Uses bounded caller-owned encode/decode workspaces, with reusable intermediate decode views and immutable cached plans.
- Migrates existing real-codec fixtures and adds focused destination, scratch lifecycle, corruption-bound, cache, and concurrency tests. No mocks.
V7 reader/writer, routing, table-config validation, and reload integration remain in #19307.
## Validation
- Fresh prerequisite-only checkout: 307 codec/parser tests pass, including 39 destination-contract cases and 31 executor cases.
- Spotless, Checkstyle, license formatting, and license checks pass.
- Optional warning-enabled compilation fails on both base `80dbb7d02d` and this change because dependency metadata references missing JetBrains `NotNull`; the normal build and tests pass.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Use caller-owned buffers for codec pipelines [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19397:
URL: https://github.com/apache/pinot/pull/19397#issuecomment-5471431509
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19397?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `0%` with `255 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 57.70%. Comparing base ([`80dbb7d`](https://app.codecov.io/gh/apache/pinot/commit/80dbb7d02dfd92ccc20ae2606ac712ea4f73d3cf?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`9fc595c`](https://app.codecov.io/gh/apache/pinot/commit/9fc595c09f53a2ae2e65d204595d2897471c3a9c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/19397?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../segment/local/io/codec/CodecPipelineExecutor.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FCodecPipelineExecutor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9Db2RlY1BpcGVsaW5lRXhlY3V0b3IuamF2YQ==) | 0.00% | [153 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...not/segment/local/io/codec/T64CodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FT64CodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9UNjRDb2RlY0RlZmluaXRpb24uamF2YQ==) | 0.00% | [22 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...segment/local/io/codec/GorillaCodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FGorillaCodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9Hb3JpbGxhQ29kZWNEZWZpbml0aW9uLmphdmE=) | 0.00% | [16 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ot/segment/local/io/codec/GzipCodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FGzipCodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9HemlwQ29kZWNEZWZpbml0aW9uLmphdmE=) | 0.00% | [13 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...gment/local/io/codec/BaseDeltaCodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FBaseDeltaCodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9CYXNlRGVsdGFDb2RlY0RlZmluaXRpb24uamF2YQ==) | 0.00% | [10 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../segment/local/io/codec/SnappyCodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FSnappyCodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9TbmFwcHlDb2RlY0RlZmluaXRpb24uamF2YQ==) | 0.00% | [10 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ot/segment/local/io/codec/ZstdCodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FZstdCodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9ac3RkQ29kZWNEZWZpbml0aW9uLmphdmE=) | 0.00% | [10 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...not/segment/local/io/codec/Lz4CodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FLz4CodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9MejRDb2RlY0RlZmluaXRpb24uamF2YQ==) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ment/local/io/codec/DeltaDeltaCodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FDeltaDeltaCodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9EZWx0YURlbHRhQ29kZWNEZWZpbml0aW9uLmphdmE=) | 0.00% | [8 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...t/segment/local/io/codec/DeltaCodecDefinition.java](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fio%2Fcodec%2FDeltaCodecDefinition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9pby9jb2RlYy9EZWx0YUNvZGVjRGVmaW5pdGlvbi5qYXZh) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (80dbb7d) and HEAD (9fc595c). Click for more details.
>
> <details><summary>HEAD has 24 uploads less than BASE</summary>
>
>| Flag | BASE (80dbb7d) | HEAD (9fc595c) |
>|------|------|------|
>|unittests|2|1|
>|java-25|6|1|
>|temurin|6|1|
>|lane-a|2|0|
>|integration|4|0|
>|integration1|2|0|
>|lane-b|2|0|
>|integration2|2|0|
>|unittests2|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19397 +/- ##
============================================
- Coverage 67.56% 57.70% -9.87%
+ Complexity 1430 1 -1429
============================================
Files 3486 2683 -803
Lines 224173 163948 -60225
Branches 35381 26624 -8757
============================================
- Hits 151462 94604 -56858
- Misses 60684 61341 +657
+ Partials 12027 8003 -4024
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.70% <0.00%> (-9.87%)` | :arrow_down: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.70% <0.00%> (-9.87%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.70% <0.00%> (-9.86%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.70% <0.00%> (+0.03%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19397/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19397?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Use caller-owned buffers for codec pipelines [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 merged PR #19397:
URL: https://github.com/apache/pinot/pull/19397
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Use caller-owned buffers for codec pipelines [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Use caller-owned buffers for codec pipelines [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #19397:
URL: https://github.com/apache/pinot/pull/19397#issuecomment-5488052626
Rebased and updated at `11b7ce6cfc`; all review threads are addressed and resolved.
Key follow-ups:
- T64 now uses sequential word-at-a-time decode and stored-type-aware retained-buffer bounds.
- Plan-cache creation/eviction now has bounded concurrent canonical/alias coverage.
- Cached scratch views preserve caller state, moving windows, byte order, and separate-scratch concurrency.
- DELTA/DELTADELTA indexed access removes the warmed regression: matched three-fork JMH is 28.1% faster transform-only and 3.8% faster with LZ4, with allocations reduced to profiler noise.
Validation: 312 codec/parser tests pass; Spotless, Checkstyle, license formatting, and license checks pass. Please re-review when convenient.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Use caller-owned buffers for codec pipelines [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Use caller-owned buffers for codec pipelines [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #19397:
URL: https://github.com/apache/pinot/pull/19397#issuecomment-5490731339
Rebased onto current `master` and force-updated at `1ce44658f54`. The PR remains one commit, and `range-diff` plus the binary patch hash confirm the rebase did not change the reviewed patch.
Fresh validation:
- 312 focused codec/parser tests pass.
- Spotless, Checkstyle, license formatting, and license checks pass.
- An independent eight-domain review found no actionable findings.
- All existing review threads are addressed and resolved.
The PR description now reflects the current base and the inherited warning-enabled compile issue reproduced on exact `master`. Fresh CI is expected to rerun on this head. Please re-review when convenient.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Use caller-owned buffers for codec pipelines [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #19397:
URL: https://github.com/apache/pinot/pull/19397#issuecomment-5519222007
Rebased and updated at `64edf3219378`. All four current review requests are addressed:
1. Removed the process-wide plan/alias cache and its three cache-only tests; `create()` is again pure parse/validate and lifecycle owners retain immutable executors.
2. Decode now calls the shared `encodedStageBounds()` helper and uses its returned composed bound.
3. Renamed the decode-only destination contract and related state to `requiresDirectDecodeDstBuffer()`.
4. Renamed `maxIntermediateSize` to `maxStageSize` consistently in the public decode API, messages, callers, and tests.
The three older inline threads remain resolved. Fresh validation is 309 focused tests, all four module hygiene gates, and an independent eight-domain review with no remaining findings. Please re-review when convenient.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org