You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "praveenc7 (via GitHub)" <gi...@apache.org> on 2026/04/20 08:56:28 UTC
[PR] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
praveenc7 opened a new pull request, #18262:
URL: https://github.com/apache/pinot/pull/18262
## Description
Adds the foundation types needed to represent MSE data as Apache Arrow VectorSchemaRoots, behind a feature flag that is off by default.
Added foundation classes
- ArrowDataBlock — DataBlock implementation over Arrow VectorSchemaRoot
- ArrowBlock — MseBlock.Data wrapper with reference-counted lifecycle
- ArrowBlockConverter — converts RowHeapDataBlock/SerializedDataBlock → ArrowBlock column-by-column, with a fast path for ColumnarDataBlock that copies directly from the backing buffer
- ArrowBuffers — manages the Arrow RootAllocator lifecycle; not a singleton, created and injected explicitly
Currently with this PR
- No operator creates ArrowBlocks
- No operator consumes ArrowBlocks differently
- MailboxReceiveOperator does not convert incoming blocks
- HashJoinOperator is unchanged
- Wire format is unchanged — if an ArrowBlock does reach GrpcSendingMailbox (e.g., via a test), it falls back through asRowHeap() → legacy protobuf, same as before
## Testing done
Unit test added
## Future PRs
- Arrow-native HashJoinOperator with Arrow-aware KeySelector hashers/comparators
- Arrow IPC serialization in GrpcSendingMailbox (metadata field selects format; same gRPC transport)
--
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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "praveenc7 (via GitHub)" <gi...@apache.org>.
praveenc7 commented on PR #18262:
URL: https://github.com/apache/pinot/pull/18262#issuecomment-4840628107
@siddharthteotia I have the doc attached to the description. Also mentioned the implementation plan there on using pushing the changes in phases with a feature flag(false by default)
--
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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on PR #18262:
URL: https://github.com/apache/pinot/pull/18262#issuecomment-4386432835
# Review: PR #18262 — [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data
**Repo:** apache/pinot
**Branch:** pchaganl/arrow-block-representation → master
**Author:** praveenc7 (Praveen)
**Date:** 2026-04-20
**Repo context loaded:**
- [x] CLAUDE.md read
- [x] CODEOWNERS read (not present in repo; noted as observation)
- [x] Module-level docs for touched paths read
---
## Stage 1: Triage & Evidence Collection
### PR description
- [x] Description present
- [x] States what problem is being solved
- [x] States why this approach
- [x] States risky areas
**Findings:** The description is thorough — it includes an ASCII diagram showing the four-piece stack, one-line summaries for each layer, and an explicit list of what is NOT wired yet (no operator produces ArrowBlocks, wire format unchanged). Risky areas (off-heap lifetime, allocator ownership) are documented. No issues.
### Change summary
- [x] Full diff read
- [x] Summary produced
- [x] Files listed by module
**Summary:** Introduces four new classes and wires them behind a feature flag (`pinot.multistage.engine.use.arrow`, default false). No operator consumes or produces `ArrowBlock`s; the PR establishes infrastructure only. All three existing `MseBlock.Data.Visitor` implementations (`GrpcSendingMailbox`, `BlockSplitter`) add a `visit(ArrowBlock)` stub that throws `UnsupportedOperationException`. Tests exercise round-tripping and null handling for every supported scalar type.
**Files:**
| Module | Files |
|---|---|
| pinot-common | `ArrowDataBlock.java` (new), `DataBlock.java` (ARROW enum type), `ColumnarDataBlock.java` (comment fix), `ArrowDataBlockTest.java` (new) |
| pinot-query-runtime | `ArrowBlock.java` (new), `ArrowBlockConverter.java` (new), `ArrowBuffers.java` (new), `MseBlock.java` (asArrow/isArrow), `BlockSplitter.java` (stub), `GrpcSendingMailbox.java` (stub), `ArrowBlockConverterTest.java` (new), `pom.xml` (arrow-vector dep) |
| pinot-spi | `CommonConstants.java` (7 new constants) |
### Risk classification
- [x] Classified as medium
- [x] Reason stated
**Classification:** Medium
**Reason:** The feature is off by default; no operator path is activated. However, changes touch `DataBlock.java` (an SPI used across the codebase), `MseBlock.Data.Visitor` (every mailbox and exchange path), and `CommonConstants` (configuration contracts). The off-heap allocator adds a new failure mode (OOM, leak) invisible to existing monitoring. The `DataBlock.Type.ARROW(3)` enum value changes the binary serialization header format.
### Test gaps
- [x] Changed behaviors/contracts checked for corresponding tests
- [x] Test coverage of new/changed behavior verified
**Gaps:**
1. **Concern — medium confidence.** `ArrowBuffers.create(PinotConfiguration)` is not tested anywhere. All tests construct `ArrowBuffers` directly via the public constructor. The factory method references `CommonConstants.Helix.CONFIG_OF_ARROW_ALLOCATOR_MAX_SIZE` and related constants but is never exercised end-to-end. Evidence: `ArrowBlockConverterTest:412` uses `new ArrowBuffers(true, new RootAllocator(Long.MAX_VALUE), 0, Long.MAX_VALUE)` directly.
2. **Observation — high confidence.** No test verifies that a disabled `ArrowBuffers` (the default) correctly refuses `newQueryAllocator` / `newAllocator` calls. This is minor since the guard code is simple, but the path is not covered.
3. **Observation — high confidence.** No test covers `ArrowBlock.asSerialized()` in isolation (only through `testRoundTripViaAsRowHeap`). This is adequate given the method is a one-liner.
---
## Stage 2: Deep Review
### Does the implementation address the stated problem?
- [x] Problem statement identified from the PR description
- [x] Implementation traced through the code to check whether it appears to address the stated problem
- [x] No obvious gaps between what the PR claims to fix and what the code actually does
**Findings:** The PR claims to add foundation types only, with no operator wiring. This is verified: no existing operator or mailbox receives or produces `ArrowBlock`s. The feature flag gates everything. The implementation matches the stated goals.
### High-risk files (medium, high)
- [x] Files ranked by logical risk
- [x] Risk reasons stated per file
| Rank | File | Risk reason |
|---|---|---|
| 1 | `DataBlock.java` | SPI interface change; adds `ARROW(3)` enum and two new methods. Downstream serialization and deserialization switch on this enum. |
| 2 | `MseBlock.java` | Adds `visit(ArrowBlock)` to the `Data.Visitor` interface; breaks all existing implementations that don't implement the new method — except that GrpcSendingMailbox and BlockSplitter are updated in the same PR. |
| 3 | `ArrowBuffers.java` | Manages off-heap RootAllocator; incorrect config constant reference or missing close() call causes off-heap leak or silent misconfiguration. |
| 4 | `ArrowBlockConverter.java` | Data integrity on conversion; null handling, type coercion, and resource leak on failure. |
| 5 | `ArrowBlock.java` | asRowHeap() / asSerialized() fallbacks; incorrect type dispatch causes ClassCastException or silent data corruption. |
**Findings:** No critical structural issues found in the high-risk files individually. Specific issues are documented below.
### Center of gravity (medium, high)
- [x] 1-3 key files or decisions identified
**Files:** `ArrowDataBlock.java`, `ArrowBlockConverter.java`, `MseBlock.java`
**Why these are central:** `ArrowDataBlock` is the foundational data-layer type — all reads and null semantics flow through it. `ArrowBlockConverter` is the only bridge from legacy format to Arrow and determines type correctness and null fidelity for every conversion. `MseBlock.java` defines the `Visitor` contract that will require updates from every consumer as Arrow adoption grows.
### Invariants — targeted (medium, high)
- [x] Authorization — N/A, no trust boundary touched
- [x] Idempotency — N/A, no mutations
- [x] Backward compatibility — APIs: existing endpoints unchanged; `MseBlock.Data.Visitor` gains a new `visit(ArrowBlock)` method. All existing non-test implementations in the PR add the method, so compilation is preserved.
- [x] Backward compatibility — Schemas: `DataBlock.Type.ARROW(3)` added. `ZeroCopyDataBlockSerde.deserialize` hits its `default` case for type 3, throwing `UnsupportedOperationException`. This is safe because `ArrowDataBlock.serialize()` already throws `UnsupportedOperationException`, so no ARROW-typed frame can appear on the wire.
- [x] Backward compatibility — SPI: `DataBlock` interface gains `getBigDecimalArray()` and `getBytesArray()` (see blocker below).
- [x] Backward compatibility — Configuration: 7 new constants added, all with safe defaults. Feature off by default.
- [x] Ordering guarantees — N/A
- [x] Null/empty handling — Verified: `ArrowDataBlock.getNullRowIds` uses `vector.isNull(row)`, correct. `ArrowBlockConverter` passes `nullBitmap` to all column writers. All writers skip null rows and call `setNull(row)` on the vector. Empty-block case (`numRows == 0`) correctly sets `valueCount(0)` and returns.
- [x] Retry safety — N/A, stateless conversion
- [x] Observability — No metrics added. Off-heap allocation is invisible to existing JVM heap monitoring. Noted as concern.
- [x] Resource cleanup — `ArrowDataBlock.close()` closes `_root`. `ArrowBlock.close()` delegates to `ArrowDataBlock.close()`. `ArrowBlockConverter.fromDataBlock()` wraps population in try/catch and closes `root` on any exception. `ArrowBuffers` closes its `RootAllocator` on `close()`. No obvious leaks in happy path.
- [x] Migration reversibility — N/A, no schema migration
- [x] Performance in hot paths — Feature is off by default; zero cost on default path. When enabled, `ArrowBlockConverter.toArrowBlock` calls `block.asSerialized()` first (an expensive serialization step), then re-reads from the serialized form into Arrow vectors. This double-materialization is a performance concern for large blocks when the feature is eventually enabled.
- [x] API design — `ArrowBlock.isSerialized()` is explicitly overridden to return `false`. Without the override, the `MseBlock.Data` default (`return !isRowHeap()`) would return `true` for `ArrowBlock`, misrouting `QueryDispatcher` to `reduceSerialized`. The override is present and correct, but the default is semantically broken for any future third block type that doesn't remember to override. Noted as concern.
**Findings:**
- **Concern — high confidence.** `ArrowBlock.close()` is not reference-counted and the Javadoc warns that calling it twice is undefined. If a caller holds both an `ArrowBlock` and the underlying `ArrowDataBlock` and closes both, `VectorSchemaRoot.close()` is called twice. Arrow's `VectorSchemaRoot` does have a safety check on double-close, so this would throw an exception rather than corrupt memory, but the contract is brittle. Evidence: `ArrowBlock.java:168`, `ArrowDataBlock.java:252`.
- **Observation — medium confidence.** The `MseBlock.Data.isSerialized()` default (`return !isRowHeap()`) is now semantically incorrect for `ArrowBlock` without explicit override. The override is present, but any future `MseBlock.Data` implementor that forgets to override it would silently misroute in `QueryDispatcher`. Consider changing the default to `return false` or documenting the invariant explicitly. Evidence: `MseBlock.java:102-104`.
- **Observation — medium confidence.** No metrics or log events are emitted when Arrow allocation fails or when the Arrow path is entered. For debugging a production incident with OOM errors in off-heap memory, there's no signal to distinguish Arrow allocation failures from regular heap OOM. This should be addressed before Arrow is enabled in production.
### Invariants — full sweep (high only)
- [x] N/A — skipped (medium risk)
### Test adequacy (medium, high)
- [x] Tests prove intended behavior (not just that code runs)
- [x] Failure modes covered
- [x] Boundary conditions covered
- [x] Not over-coupled to implementation details
**Findings:** Test quality is high. `ArrowDataBlockTest.testAllScalarTypesHappyAndNullPath` covers every supported scalar type with a non-null/null/non-null row pattern and verifies the null bitmap for each column independently. `ArrowBlockConverterTest.testConverterRoundTripAllTypes` does the same through the full conversion path and adds a regression guard that `DataSchema` column types survive the Arrow round-trip (preventing TIMESTAMP from collapsing to LONG, etc.). `testCloseReleasesOffHeapMemory` verifies the allocator reports zero after close, which is a strong invariant for off-heap tracking.
**Gap:** No test verifies what happens when the `RootAllocator` runs out of budget mid-conversion (off-heap OOM). This is acceptable for a foundation PR but should be noted for future hardening.
### Collateral damage (medium, high)
- [x] No copy-paste duplication introduced
- [x] No partial renames (all references updated)
- [x] No dead code left behind
- [x] Docs and config match code changes
- [x] No one-off exceptions added to shared abstractions
**Findings:**
- **Observation — high confidence.** The `ColumnarDataBlock.java` change (moving a misindented comment) is a drive-by fix with no functional effect. Fine.
- **Observation — medium confidence.** `ArrowBuffers.create()` reads `CONFIG_OF_ARROW_ALLOCATOR_MAX_SIZE` from `CommonConstants.Helix`, while `CONFIG_OF_MULTI_STAGE_ENGINE_ENABLED` and the new `CONFIG_OF_MULTI_STAGE_ENGINE_USE_ARROW` are also in `CommonConstants.Helix` (confirmed from the diff hunk at `@@ -266` which is after the `Instance` inner class closes). The placement is internally consistent. No issue.
### Path-triggered review (medium, high; also low if boundary paths touched)
- [x] Touched paths listed
- [x] Failure mode inferred per path
- [x] Reviewed for inferred risks
- [x] Boundary areas checked first
**Touched paths and inferred risks:**
| Path | Inferred failure mode | What was checked | Findings |
|---|---|---|---|
| `pinot-common/src/main/java/org/apache/pinot/common/datablock/DataBlock.java` | SPI change breaks implementors | Checked all `DataBlock` implementors for the new interface methods | **BLOCKER** — `ArrowDataBlock` is missing `getBigDecimalArray()` and `getBytesArray()` (see blocker section) |
| `pinot-query-runtime/src/main/java/.../blocks/MseBlock.java` | Visitor extension breaks all implementors | Checked `GrpcSendingMailbox` and `BlockSplitter` for new `visit(ArrowBlock)` | Both updated in this PR; no other visitor implementations found in production code |
| `pinot-query-runtime/src/main/java/.../memory/ArrowBuffers.java` | Off-heap OOM, leak on shutdown | Checked `close()`, `checkEnabled()`, disabled-mode RootAllocator creation | Disabled mode correctly creates a zero-sized allocator; close() unconditional and correct |
| `pinot-spi/src/main/java/.../spi/utils/CommonConstants.java` | Config namespace collision | Checked constant placement and ArrowBuffers references | No collision; all constants in `Helix` class, consistent with `ArrowBuffers` usage |
| `pinot-query-runtime/pom.xml` | New transitive dependency breaks dependency graph | Checked `arrow-vector` and `arrow-memory-netty` scope | `arrow-vector` compile scope (correct), `arrow-memory-netty` runtime scope (correct) |
**Findings:** See blocker below.
---
## Verdict
**Risk:** Medium
**Verdict:** changes-requested
**Verdict confidence:** High
### Confidence reasoning
The diff is clean and well-documented. All CI checks pass against the PR's base commit. The blocker is a post-PR interface addition (PR #18326 was merged 5 days after this PR was opened, adding `getBigDecimalArray()` and `getBytesArray()` to the `DataBlock` interface). This is a factual, traceable finding with a specific fix. The other findings are concerns and observations that do not block correctness of the infrastructure at its current (feature-off) state, but should be addressed before enabling Arrow in production.
### Blockers
1. **BLOCKER — high confidence.** `ArrowDataBlock` does not implement `getBigDecimalArray(int rowId, int colId)` or `getBytesArray(int rowId, int colId)`, both of which are required by the `DataBlock` interface. These methods were added to `DataBlock` by PR #18326 (commit `aad170afaa`, merged 2026-04-25), which post-dates this PR's creation (2026-04-20). The PR must be rebased or the methods must be added (both should throw `UnsupportedOperationException`, consistent with the other unsupported array methods in `ArrowDataBlock`). Evidence: `DataBlock.java:86,90`; `ArrowDataBlock.java` (absent implementation). Severity: blocker. Confidence: high.
### Concerns
1. **Concern — medium confidence.** `ArrowBuffers.create(PinotConfiguration)` is the intended production entry point for initializing the allocator, but no test calls it. The only tests use the direct constructor. If a config key name drifts or the `PinotConfiguration` lookup semantics change, `create()` will silently misconfigure without any test failing. Evidence: `ArrowBuffers.java:257-270`, `ArrowBlockConverterTest.java:412`. Severity: concern.
2. **Concern — medium confidence.** Off-heap allocations by `ArrowBuffers` are invisible to existing JVM heap monitoring and metrics. When Arrow is eventually enabled, off-heap OOM will produce `OutOfMemoryError` with no Pinot-level metric to distinguish it from heap OOM. Before enabling in production, emit a metric (e.g., `arrowAllocatedBytes`) via `getAllocatedMemory()`. Evidence: `ArrowBuffers.java:313-316`. Severity: concern.
3. **Concern — medium confidence.** `ArrowBlock.close()` is not idempotent and not reference-counted. The Javadoc states "calling it twice will attempt to close the underlying `ArrowDataBlock` twice." Arrow's `VectorSchemaRoot` closes gracefully on a second call (it is a no-op after the first close), but the contract is brittle for future consumers who hold both references. Consider adding a `_closed` guard. Evidence: `ArrowBlock.java:165-168`. Severity: concern.
4. **Concern — low confidence.** `ArrowBlockConverter.toArrowBlock()` always routes through `block.asSerialized()` for non-Arrow inputs, including `RowHeapDataBlock`. This means every conversion suffers a full heap→serialized→Arrow materialization. For the foundation PR this is acceptable, but the path should be revisited before Arrow is used in a hot query path. Evidence: `ArrowBlockConverter.java:101-108`. Severity: concern.
### Observations
1. The `MseBlock.Data.isSerialized()` default (`return !isRowHeap()`) is a semantic time bomb for future block types. Any new `MseBlock.Data` implementor that forgets to override `isSerialized()` will incorrectly report itself as serialized, misrouting in `QueryDispatcher`. Consider changing the default to `return false`. Evidence: `MseBlock.java:102-104`.
2. `ArrowBlock.asRowHeap()` and `asSerialized()` are marked with `TODO: remove this method`. These fallbacks exist for operator compatibility but add code that will need cleanup. The TODO is appropriate as-is given the staged rollout plan.
3. `ColumnarDataBlock.java` has a drive-by fix for a misindented comment. Harmless.
4. CODEOWNERS file was not found in this repository. Reviews are unassigned; human auditors should route this to the MSE/query-runtime team.
### Areas recommended for human focus
1. **The rebase gap**: Confirm that `ArrowDataBlock` is updated to add `getBigDecimalArray()` and `getBytesArray()` throwing `UnsupportedOperationException`, and re-run the unit tests after rebasing on master.
2. **ArrowBuffers.create() lifecycle**: Verify that production component startup code (ServerQueryExecutorRunnerFactory or equivalent) will call `ArrowBuffers.create()` and inject the result correctly before this feature is enabled. This path is completely untested.
3. **Double-close semantics**: Validate that `ArrowDataBlock.close()` being called twice is safe in Arrow 19.0.0 (the version in use). If it is not a no-op, add a `_closed` guard.
### Status
- [x] All applicable Stage 1 boxes checked
- [x] All applicable Stage 2 boxes checked (or N/A for skipped tiers)
- [x] Verdict committed
- [x] Verdict confidence assigned with reasoning
--
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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "praveenc7 (via GitHub)" <gi...@apache.org>.
praveenc7 commented on PR #18262:
URL: https://github.com/apache/pinot/pull/18262#issuecomment-4436671906
> Thanks for the contribution Praveen — the four-class layering is clean and the test coverage for null/non-null round-trips is thorough. Before this merges I'd like to discuss two things: the overall merge strategy, and the ownership model for off-heap memory.
>
> **Merge strategy**
>
> Small, reviewable PRs are the right way to develop this incrementally, and we're happy to review them that way. However, we'd prefer not to merge partial Arrow infrastructure into `apache/pinot` master until the design is complete and proven end-to-end. Dead code in master (classes that exist but nothing uses) creates maintenance burden and makes it harder to reason about the codebase.
>
> A better workflow would be:
>
> * Create a long-lived integration branch — either in `apache/pinot` or your own fork
> * Stack your incremental PRs against that branch so each one is reviewable in isolation
> * Open a single draft PR from that branch into `apache/pinot` master, so we have full visibility into the end state
>
> This way we get the review ergonomics of small PRs without shipping partial infrastructure to master prematurely.
>
> **Ownership model: off-heap memory requires explicit release**
>
> With heap-backed blocks (`RowHeapDataBlock`, `SerializedDataBlock`) the GC handles reclamation automatically. Arrow blocks are different — the off-heap memory is invisible to the GC and must be explicitly freed. This means the design needs a clear answer to: who releases each block, and when?
>
> **Why a simple single-owner model is insufficient**
>
> Most of the operator pipeline is fine with single-owner linear transfer — a block is consumed, rows extracted, and then released. The case that breaks this is the **local broadcast / singleton exchange**: the same block is enqueued into multiple local mailbox queues without copying. The first consumer to release the block would free memory that other consumers are still reading.
>
> Note: for hashing distribution this is not a problem — new disjoint blocks are created per bucket, each with a single owner.
>
> **Proposed ownership model: retain / release**
>
> The right primitive here is reference counting:
>
> * Initial refcount = 1 (the producer holds the first reference)
> * `retain()` — increments refcount; throws if already 0 (use-after-free guard)
> * `release()` — decrements; frees the underlying `VectorSchemaRoot` only when it reaches 0
>
> A few notes on API design:
>
> * The primary API should be `retain()` / `release()`, not `close()`. `close()` implies "destroy unconditionally", which conflicts with ref counting semantics.
> * I'd recommend **not** implementing `AutoCloseable` on `ArrowBlock`. `try-with-resources` calls `close()` unconditionally regardless of refcount, and callers that share a block would silently release it early. Making the release explicit forces the caller to reason about ownership rather than relying on scope.
> * `VectorSchemaRoot` in Arrow Java doesn't have block-level ref counting built in; you'd implement it with an `AtomicInteger` on `ArrowBlock`, similar to Netty's `ReferenceCounted` pattern.
>
> **Mailbox implications**
>
> For the local mailbox send path, the sender would call `block.retain()` once per local recipient before enqueuing, then `release()` its own producer reference after all enqueues complete. The remote path serializes to bytes (a copy), so no retain is needed — the block can be released after serialization. Each consumer calls `release()` when done.
>
> This is worth designing explicitly before the operator and transport PRs are written, so the contract is clear from the start.
>
> Happy to discuss — wanted to raise this while the API surface is still small.
Thanks for looking into this. I will get back to you by this week on the 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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "praveenc7 (via GitHub)" <gi...@apache.org>.
praveenc7 commented on PR #18262:
URL: https://github.com/apache/pinot/pull/18262#issuecomment-4465646357
@gortiz The approaches sounds good to me
I drafted a short design doc on the off-heap ownership contract: covers the retain/release rules, the broadcast pattern, why we're not implementing AutoCloseable, and a few open questions (leak detection, empty blocks, exception-path cleanup): https://docs.google.com/document/d/1koNPiW8L0IbijQi1JarFscH-4Ct7IF8Jv5HrP3fZw3k/edit would like our review on that
On Integration branch on apache/pinot — I don't have push rights to the repo. Could you help create an integration branch off master having on a branch here is better for visibility than on my fork. Something like arrow-integration? Once it exists I'll retarget this PR to it and stack future PRs against the same branch, keeping #18207 on master as the end-state draft.
--
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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "praveenc7 (via GitHub)" <gi...@apache.org>.
praveenc7 commented on PR #18262:
URL: https://github.com/apache/pinot/pull/18262#issuecomment-4311117727
@gortiz / @yashmayya Would appreciate your review on this
--
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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18262:
URL: https://github.com/apache/pinot/pull/18262#discussion_r3130641396
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/ArrowBlock.java:
##########
@@ -0,0 +1,157 @@
+/**
+ * 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.query.runtime.blocks;
+
+import java.util.Arrays;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.NullVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.pinot.common.datablock.ArrowDataBlock;
+import org.apache.pinot.common.utils.DataSchema;
+
+
+/**
+ * An {@link MseBlock.Data} backed by an Apache Arrow {@link VectorSchemaRoot}.
+ *
+ * <p>This is the columnar block type for the Multi-Stage Query Engine when Arrow is enabled. The
+ * {@link #asRowHeap()} and {@link #asSerialized()} methods provide a fallback path for operators that
+ * don't yet consume Arrow blocks directly — they materialize the off-heap columnar data into the legacy
+ * row-heap format. These fallback conversions are expensive (every primitive gets boxed, every string
+ * allocated on heap); they exist as a compatibility bridge, not as the intended hot path.
+ *
+ * <p><b>Lifetime:</b> the block's off-heap buffers are owned by the {@code BufferAllocator} used to
+ * construct it (typically a per-query or per-stage child of the root allocator). When that allocator
+ * closes, every block it produced is freed atomically. Individual blocks <em>do not</em> need to be
+ * released by operator code — the allocator is the unit of ownership.
+ *
+ * <p>{@link #close()} is provided for explicit early disposal (e.g. in tests, or when a caller wants to
+ * free a block before its allocator closes). It is <b>not</b> reference-counted; calling it twice will
+ * attempt to close the underlying {@link ArrowDataBlock} twice.
+ *
+ * <p>To move a block across allocator scopes (e.g. across stage boundaries or onto the wire), use
+ * Arrow's {@code TransferPair} API — this is zero-copy and preserves the allocator-ownership invariant.
+ */
+public class ArrowBlock implements MseBlock.Data, AutoCloseable {
+ private final ArrowDataBlock _dataBlock;
+
+ public ArrowBlock(ArrowDataBlock dataBlock) {
+ _dataBlock = dataBlock;
+ }
+
+ public ArrowDataBlock getDataBlock() {
+ return _dataBlock;
+ }
+
+ // ----- MseBlock.Data -----
+
+ @Override
+ public int getNumRows() {
+ return _dataBlock.getNumberOfRows();
+ }
+
+ @Override
+ public DataSchema getDataSchema() {
+ return _dataBlock.getDataSchema();
+ }
+
+ @Override
+ public boolean isRowHeap() {
+ return false;
+ }
+
+ @Override
+ public boolean isSerialized() {
+ return false;
+ }
+
+ @Override
+ public boolean isArrow() {
+ return true;
+ }
+
+ @Override
+ public ArrowBlock asArrow() {
+ return this;
+ }
+
+ /**
+ * Materializes the Arrow columnar data into a row-heap block. This is the fallback path for operators
+ * that don't yet consume Arrow blocks directly; it is expensive (every primitive gets boxed, every string
+ * allocated on heap).
+ *
+ * <p>The returned {@link RowHeapDataBlock} is independent of this block's off-heap buffers — the caller
+ * may close this block (or let its allocator close) without affecting the row-heap copy. This block is
+ * <em>not</em> closed as a side effect.
+ *
+ * <p>TODO: remove this method once all operators consume {@link ArrowBlock} directly.
+ */
+ @Override
+ public RowHeapDataBlock asRowHeap() {
+ int numRows = getNumRows();
+ int numCols = _dataBlock.getNumberOfColumns();
+ VectorSchemaRoot root = _dataBlock.getRoot();
+ Object[][] rows = new Object[numRows][numCols];
+ for (int colIdx = 0; colIdx < numCols; colIdx++) {
+ FieldVector vector = root.getVector(colIdx);
+ if (vector instanceof NullVector) {
+ continue;
+ }
+ if (vector instanceof VarCharVector) {
+ for (int row = 0; row < numRows; row++) {
+ Object value = vector.getObject(row);
+ if (value != null) {
+ rows[row][colIdx] = value.toString();
+ }
+ }
+ } else {
+ for (int row = 0; row < numRows; row++) {
+ rows[row][colIdx] = vector.getObject(row);
Review Comment:
Arrow getObject() returns Arrow/native objects here, such as byte[] for VarBinaryVector and Boolean for BitVector, but RowHeapDataBlock.asSerialized() expects Pinot internal values like ByteArray for BYTES and 0/1 Integer for BOOLEAN. This fallback bridge will throw or serialize the wrong type for supported scalar columns. Convert via the DataSchema/ArrowDataBlock typed getters before building the row heap, and add BYTES/BOOLEAN coverage through asSerialized().
--
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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on PR #18262:
URL: https://github.com/apache/pinot/pull/18262#issuecomment-4399445123
I started to review the pr and I have some comments to add, but I'm at a conference this week, so I'll add my feedback next week
--
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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on PR #18262:
URL: https://github.com/apache/pinot/pull/18262#issuecomment-4418591435
Thanks for the contribution Praveen — the four-class layering is clean and the test coverage for null/non-null round-trips is thorough. Before this merges I'd like to discuss two things: the overall merge strategy, and the ownership model for off-heap memory.
**Merge strategy**
Small, reviewable PRs are the right way to develop this incrementally, and we're happy to review them that way. However, we'd prefer not to merge partial Arrow infrastructure into `apache/pinot` master until the design is complete and proven end-to-end. Dead code in master (classes that exist but nothing uses) creates maintenance burden and makes it harder to reason about the codebase.
A better workflow would be:
- Create a long-lived integration branch — either in `apache/pinot` or your own fork
- Stack your incremental PRs against that branch so each one is reviewable in isolation
- Open a single draft PR from that branch into `apache/pinot` master, so we have full visibility into the end state
This way we get the review ergonomics of small PRs without shipping partial infrastructure to master prematurely.
**Ownership model: off-heap memory requires explicit release**
With heap-backed blocks (`RowHeapDataBlock`, `SerializedDataBlock`) the GC handles reclamation automatically. Arrow blocks are different — the off-heap memory is invisible to the GC and must be explicitly freed. This means the design needs a clear answer to: who releases each block, and when?
**Why a simple single-owner model is insufficient**
Most of the operator pipeline is fine with single-owner linear transfer — a block is consumed, rows extracted, and then released. The case that breaks this is the **local broadcast / singleton exchange**: the same block is enqueued into multiple local mailbox queues without copying. The first consumer to release the block would free memory that other consumers are still reading.
Note: for hashing distribution this is not a problem — new disjoint blocks are created per bucket, each with a single owner.
**Proposed ownership model: retain / release**
The right primitive here is reference counting:
- Initial refcount = 1 (the producer holds the first reference)
- `retain()` — increments refcount; throws if already 0 (use-after-free guard)
- `release()` — decrements; frees the underlying `VectorSchemaRoot` only when it reaches 0
A few notes on API design:
- The primary API should be `retain()` / `release()`, not `close()`. `close()` implies "destroy unconditionally", which conflicts with ref counting semantics.
- I'd recommend **not** implementing `AutoCloseable` on `ArrowBlock`. `try-with-resources` calls `close()` unconditionally regardless of refcount, and callers that share a block would silently release it early. Making the release explicit forces the caller to reason about ownership rather than relying on scope.
- `VectorSchemaRoot` in Arrow Java doesn't have block-level ref counting built in; you'd implement it with an `AtomicInteger` on `ArrowBlock`, similar to Netty's `ReferenceCounted` pattern.
**Mailbox implications**
For the local mailbox send path, the sender would call `block.retain()` once per local recipient before enqueuing, then `release()` its own producer reference after all enqueues complete. The remote path serializes to bytes (a copy), so no retain is needed — the block can be released after serialization. Each consumer calls `release()` when done.
This is worth designing explicitly before the operator and transport PRs are written, so the contract is clear from the start.
Happy to discuss — wanted to raise this while the API surface is still small.
--
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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #18262:
URL: https://github.com/apache/pinot/pull/18262#issuecomment-4279592470
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18262?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 `405 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 34.90%. Comparing base ([`9ff14da`](https://app.codecov.io/gh/apache/pinot/commit/9ff14da6dbcb8dddbdcb7edc04b2db7f93fb70de?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`323ee7f`](https://app.codecov.io/gh/apache/pinot/commit/323ee7f2a20dadd374331cc1284c66b0b5511b75?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 32 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/18262?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../apache/pinot/common/datablock/ArrowDataBlock.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcommon%2Fdatablock%2FArrowDataBlock.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vZGF0YWJsb2NrL0Fycm93RGF0YUJsb2NrLmphdmE=) | 0.00% | [175 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...inot/query/runtime/blocks/ArrowBlockConverter.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Fblocks%2FArrowBlockConverter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9ibG9ja3MvQXJyb3dCbG9ja0NvbnZlcnRlci5qYXZh) | 0.00% | [135 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../apache/pinot/query/runtime/blocks/ArrowBlock.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Fblocks%2FArrowBlock.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9ibG9ja3MvQXJyb3dCbG9jay5qYXZh) | 0.00% | [54 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/pinot/query/runtime/memory/ArrowBuffers.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Fmemory%2FArrowBuffers.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9tZW1vcnkvQXJyb3dCdWZmZXJzLmphdmE=) | 0.00% | [32 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...a/org/apache/pinot/common/datablock/DataBlock.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcommon%2Fdatablock%2FDataBlock.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vZGF0YWJsb2NrL0RhdGFCbG9jay5qYXZh) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ache/pinot/common/datablock/ColumnarDataBlock.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcommon%2Fdatablock%2FColumnarDataBlock.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vZGF0YWJsb2NrL0NvbHVtbmFyRGF0YUJsb2NrLmphdmE=) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rg/apache/pinot/query/runtime/blocks/MseBlock.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Fblocks%2FMseBlock.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9ibG9ja3MvTXNlQmxvY2suamF2YQ==) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...apache/pinot/query/mailbox/GrpcSendingMailbox.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fmailbox%2FGrpcSendingMailbox.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvbWFpbGJveC9HcnBjU2VuZGluZ01haWxib3guamF2YQ==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ache/pinot/query/runtime/blocks/BlockSplitter.java](https://app.codecov.io/gh/apache/pinot/pull/18262?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Fblocks%2FBlockSplitter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9ibG9ja3MvQmxvY2tTcGxpdHRlci5qYXZh) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18262?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 (9ff14da) and HEAD (323ee7f). Click for more details.
>
> <details><summary>HEAD has 32 uploads less than BASE</summary>
>
>| Flag | BASE (9ff14da) | HEAD (323ee7f) |
>|------|------|------|
>|java-21|5|1|
>|unittests1|2|0|
>|unittests|3|1|
>|temurin|9|1|
>|java-11|4|0|
>|integration|6|0|
>|integration2|2|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18262 +/- ##
=============================================
- Coverage 63.34% 34.90% -28.45%
+ Complexity 1627 783 -844
=============================================
Files 3238 3248 +10
Lines 197003 197767 +764
Branches 30464 30609 +145
=============================================
- Hits 124801 69032 -55769
- Misses 62202 122661 +60459
+ Partials 10000 6074 -3926
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/18262/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/18262/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/18262/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/18262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/18262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/18262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `34.90% <0.00%> (-28.42%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/18262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `34.90% <0.00%> (-28.45%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/18262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `34.90% <0.00%> (-28.45%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `34.90% <0.00%> (-0.06%)` | :arrow_down: |
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 Sentry](https://app.codecov.io/gh/apache/pinot/pull/18262?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] [MSE] Introduce Apache Arrow as a columnar block format for MseBlock.Data [pinot]
Posted by "praveenc7 (via GitHub)" <gi...@apache.org>.
praveenc7 commented on code in PR #18262:
URL: https://github.com/apache/pinot/pull/18262#discussion_r3135824310
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/ArrowBlock.java:
##########
@@ -0,0 +1,157 @@
+/**
+ * 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.query.runtime.blocks;
+
+import java.util.Arrays;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.NullVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.pinot.common.datablock.ArrowDataBlock;
+import org.apache.pinot.common.utils.DataSchema;
+
+
+/**
+ * An {@link MseBlock.Data} backed by an Apache Arrow {@link VectorSchemaRoot}.
+ *
+ * <p>This is the columnar block type for the Multi-Stage Query Engine when Arrow is enabled. The
+ * {@link #asRowHeap()} and {@link #asSerialized()} methods provide a fallback path for operators that
+ * don't yet consume Arrow blocks directly — they materialize the off-heap columnar data into the legacy
+ * row-heap format. These fallback conversions are expensive (every primitive gets boxed, every string
+ * allocated on heap); they exist as a compatibility bridge, not as the intended hot path.
+ *
+ * <p><b>Lifetime:</b> the block's off-heap buffers are owned by the {@code BufferAllocator} used to
+ * construct it (typically a per-query or per-stage child of the root allocator). When that allocator
+ * closes, every block it produced is freed atomically. Individual blocks <em>do not</em> need to be
+ * released by operator code — the allocator is the unit of ownership.
+ *
+ * <p>{@link #close()} is provided for explicit early disposal (e.g. in tests, or when a caller wants to
+ * free a block before its allocator closes). It is <b>not</b> reference-counted; calling it twice will
+ * attempt to close the underlying {@link ArrowDataBlock} twice.
+ *
+ * <p>To move a block across allocator scopes (e.g. across stage boundaries or onto the wire), use
+ * Arrow's {@code TransferPair} API — this is zero-copy and preserves the allocator-ownership invariant.
+ */
+public class ArrowBlock implements MseBlock.Data, AutoCloseable {
+ private final ArrowDataBlock _dataBlock;
+
+ public ArrowBlock(ArrowDataBlock dataBlock) {
+ _dataBlock = dataBlock;
+ }
+
+ public ArrowDataBlock getDataBlock() {
+ return _dataBlock;
+ }
+
+ // ----- MseBlock.Data -----
+
+ @Override
+ public int getNumRows() {
+ return _dataBlock.getNumberOfRows();
+ }
+
+ @Override
+ public DataSchema getDataSchema() {
+ return _dataBlock.getDataSchema();
+ }
+
+ @Override
+ public boolean isRowHeap() {
+ return false;
+ }
+
+ @Override
+ public boolean isSerialized() {
+ return false;
+ }
+
+ @Override
+ public boolean isArrow() {
+ return true;
+ }
+
+ @Override
+ public ArrowBlock asArrow() {
+ return this;
+ }
+
+ /**
+ * Materializes the Arrow columnar data into a row-heap block. This is the fallback path for operators
+ * that don't yet consume Arrow blocks directly; it is expensive (every primitive gets boxed, every string
+ * allocated on heap).
+ *
+ * <p>The returned {@link RowHeapDataBlock} is independent of this block's off-heap buffers — the caller
+ * may close this block (or let its allocator close) without affecting the row-heap copy. This block is
+ * <em>not</em> closed as a side effect.
+ *
+ * <p>TODO: remove this method once all operators consume {@link ArrowBlock} directly.
+ */
+ @Override
+ public RowHeapDataBlock asRowHeap() {
+ int numRows = getNumRows();
+ int numCols = _dataBlock.getNumberOfColumns();
+ VectorSchemaRoot root = _dataBlock.getRoot();
+ Object[][] rows = new Object[numRows][numCols];
+ for (int colIdx = 0; colIdx < numCols; colIdx++) {
+ FieldVector vector = root.getVector(colIdx);
+ if (vector instanceof NullVector) {
+ continue;
+ }
+ if (vector instanceof VarCharVector) {
+ for (int row = 0; row < numRows; row++) {
+ Object value = vector.getObject(row);
+ if (value != null) {
+ rows[row][colIdx] = value.toString();
+ }
+ }
+ } else {
+ for (int row = 0; row < numRows; row++) {
+ rows[row][colIdx] = vector.getObject(row);
Review Comment:
Thanks @xiangfu0 for the review. That's a valid point. Changed the implementation to account for this and updated the test
--
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