rohityadav1993 opened a new issue, #19924:
URL: https://github.com/apache/pinot/issues/19924
Part of #18667. Follow-up to #19120.
## Context
In #19120, `StreamingSelectionOrderByCombineOperator` was added. Rows from
all segment leaves are merged with a
k-way heap on one consumer thread. No worker threads are used.
```mermaid
flowchart LR
L1["Leaf, segment 1"] --> H
L2["Leaf, segment 2"] --> H
LN["Leaf, segment N"] --> H
subgraph C["StreamingSelectionOrderByCombineOperator (consumer thread)"]
H["k-way heap<br/>(one cursor per activated segment)"] --> O["Output
block<br/>(sortedSelectionMergeBlockSize)"]
O --> T["Terminal MetadataResultsBlock<br/>(stats, errors)"]
end
O --> S["StreamingInstanceResponseOperator"]
T --> S
A["Planner: AUTO resolution"] --> C
```
| Item | Where | Source |
|---|---|---|
| B1 Lifecycle without workers | class hierarchy | review body note 1 |
| B2 Bulk run copy | merge loop | in-code TODO, r3871714056 |
| B3 Activated-segment count stat | terminal block | in-code TODO,
r3871714063 |
| B4 Schema of a zero-match query | terminal block | in-code TODO (test),
r3871714002 |
| B5 Bounded live heap | cursors | review thread r3871714047 |
| B6 AUTO heuristic and benchmarks | planner | review body, thread [15] part
2 |
## B1. Separate the streaming lifecycle from worker execution
Source: review body note 1,
https://github.com/apache/pinot/pull/19120#pullrequestreview-5494854939
```
BaseCombineOperator phaser, futures, worker CPU counters,
getNumWorkerThreads()
| + used helpers: attachExecutionStats(),
termination checks
+-- BaseStreamingCombineOperator results queue (100 slots),
query-satisfied tracker,
| worker loop, start() / stop()
+-- StreamingSelectionOrderByCombineOperator
null merger; start() is a no-op; processSegments() and
isQuerySatisfied() throw
```
- The unused allocations are small. The issue is architectural: the class is
coupled to a worker model that it
does not use.
- A concrete result of this coupling was fixed in #19120 (P2e): the
consumer-thread CPU and allocations were
under-reported over streamed blocks.
- Only two places depend on the type `BaseStreamingCombineOperator` for
`start()` / `stop()`:
`StreamingInstanceResponseOperator` and `CombineTerminationTest`.
- Expected change: `BaseCombineOperator` is split into (a) stats and
termination helpers and (b) the worker
lifecycle. A streaming combine without workers is then built on (a) only.
Every combine extends
`BaseCombineOperator`, so all of them are touched; a separate PR is
recommended.
- A smaller step (an interface for `start()` / `stop()`, with this combine
on `BaseCombineOperator`) was
considered and not taken: the coupling is only moved one level down.
## B2. Copy whole runs in the merge loop
Source: in-code TODO in the merge loop (cites
https://github.com/apache/pinot/pull/19120#discussion_r3871714056)
- Today: rows are moved one at a time from the leader cursor to the output
block.
- Proposed: the cut point is found with a binary search, and the run is
moved with one `addAll(subList(...))`.
- Blocker: `_numRowsEmitted` would then step over the `0x1FFF` mask of the
termination check, so cancellation,
pause and the deadline would not be observed. An unconditional check per
run, or a run-length clamp, is needed.
- A benchmark is needed to show a gain.
## B3. Report the activated-segment count
Source: in-code TODO in the class javadoc (cites
https://github.com/apache/pinot/pull/19120#discussion_r3871714063)
- Today: `numSegmentsProcessed` counts every segment. A never-activated
segment scans no docs, so it is not in
`numSegmentsMatched`. The difference also includes activated segments that
matched no rows, so the activated
count cannot be derived.
- Expected change: a new `DataTable.MetadataKey` is added. Mixed-version
brokers and servers must be considered.
## B4. Schema of a zero-match streaming query
Source: in-code TODO in `testEmptyResultOnStreamingPath` (cites
https://github.com/apache/pinot/pull/19120#discussion_r3871714002)
- Today: when no segment matches, only the terminal `MetadataResultsBlock`
is sent. It carries no `DataSchema`.
- This convention is shared by all streaming combines.
`StreamingReduceService` uses the null schema as the
discriminator.
- Since #19120, every leaf of this combine sends at least one
schema-carrying block, so the combine knows the
schema. A change is a cross-combine convention change; discussion is
requested before any change is made.
## B5. Bound the live heap of overlapping segments
Source: review thread
https://github.com/apache/pinot/pull/19120#discussion_r3871714047
- Today: one block is held per activated cursor. When all segments overlap,
all cursors stay open.
Live memory is about (number of activated segments) x (block per cursor).
- Capping the number of active cursors is not correct and is not proposed.
- Candidate levers:
- a smaller `maxDocsPerCall` for streaming children;
- late materialization for unsorted children;
- an overlap-depth fallback in AUTO;
- parallel merges (batch activation, or stride-partitioned merges with a
shared boundary).
- Reference workload: full scan or `LIMIT 10000` over 100 overlapping
segments.
## B6. AUTO heuristic and representative benchmarks
Source: review body (AUTO paragraph); review thread [15] part 2
- Today: AUTO resolves ON when the fraction of sorted segments is at least
`sortedSelectionMergeAutoMinSortedRatio` (default 0.8). Segment size,
overlap and run cardinality are not used.
- Benchmarks that are requested:
- skewed segment sizes;
- consuming segments;
- short runs with compressed payload columns (see #19923 A1);
- more than one sorted-prefix column;
- a sweep of `sortedSelectionMergeBlockSize` (the harness pins 10000
today). The trade is per-block overhead
against downstream buffering, so it is best measured with a sort-aware
mailbox receiver.
- Expected change: inputs for the heuristic are selected from the benchmark
results.
## Out of scope
Leaf-side follow-ups are tracked in #19923.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
