gortiz opened a new pull request, #19944: URL: https://github.com/apache/pinot/pull/19944
## Summary This PR adds an `auto` spool mode to the multi-stage engine and makes it the default. - `useSpools` (query option) and `pinot.broker.multistage.spools` (broker config) now take `off`, `auto` or `on`. `false`/`no` are aliases of `off`, and `true`/`yes` are aliases of `on`. - `auto` shares an equivalent stage only when a new static check, `SpoolDeadlockGuard`, proves that the sharing cannot block the query. - `on` keeps the behavior of the former `useSpools=true`: it shares every equivalent stage, without the check. - Two new broker meters, `SPOOLS_APPLIED` and `SPOOLS_REJECTED_BY_GUARD`, count the decisions. - With the new query option `spoolsFeedback=true`, the response carries a `spools` entry in `responseMetadata` that lists each decision and the reason for each rejection. Spools were added in [apache/pinot#14196](https://github.com/apache/pinot/issues/14196) as an experimental, opt-in feature. ## Background: why spools can block A spool is a stage that sends its output to more than one consumer stage. It sends each block to all its consumers, so it moves at the pace of the slowest one. When the mailbox of one consumer is full, the spool stops for all of them. Some operators read one input only after another input finishes: | Operator | Reads first | Reads later | |---|---|---| | Hash join (all `BaseJoinOperator`s) | right (build) input, in full | left (probe) input | | `INTERSECT`, `EXCEPT` | right input, in full | left input | | `UNION`, `UNION ALL` | input 0, until it finishes | input 1, then input 2, and so on | | Pipeline breaker | its receives, in round-robin with blocking reads | the main operator chain of the stage | A receive that waits is not read, so its sender stops. The stop moves up through every stage on a streaming path, and it stops each spool on that path. If a stopped stage is also necessary for the input that the operator reads first, that input never finishes: ```mermaid flowchart LR S["Spool stage"] -->|"mailbox full"| P["Probe receive<br/>(not read yet)"] S -->|"needs the spool"| B["Build receive"] P --> J["JOIN"] B --> J J -. "reads probe only after build finishes" .-> P ``` The query then hangs until it times out. The current spools work only while the data fits in the mailbox buffers. For example, `SpoolIntegrationTest.intermediateSpool` has this shape. ## How the guard works `SpoolDeadlockGuard` analyzes the stage DAG after fragmentation: 1. For each stage, it records each *deferral*: a pair of receives that an operator reads one after the other. 2. For each sender, it computes two sets: - The stages that **stop** when nobody reads the sender's output. These are reached through streaming paths only. - The stages that the sender **needs** in order to finish. These are reached through all paths. 3. Deferral `i` blocks deferral `j` when a stage that the waiting receive of `i` stops is needed by the earlier receive of `j`. 4. A cycle in this graph is a risk. The simplest cycle has one deferral. A longer one can span two joins that read the same two spools with the sides swapped. Some operators read their whole input before they emit their first row. A stop does not move through them, so they end a streaming path. These operators are: aggregates, windows, sorts with collations, the build input of a join, and the right input of `INTERSECT`/`EXCEPT`. A sort without collations (`LimitSortOperator`) streams. The streaming group-by and distinct operators exist only in the leaf stage, which receives data only through its pipeline breaker. The check is conservative. It does not use the size of the data, and it does not rely on an early end of the input (for example `LIMIT`) to free a spool. `EquivalentStagesReplacer` applies each substitution, runs the guard on the whole plan, and undoes the substitution when the guard finds a risk. It returns one `SpoolSubstitution` for each decision. That list feeds the meters and the feedback entry. ## Compatibility - **Wire format.** No change. Servers have read the multi-receiver send format since 1.4.0, so the new default is safe for upgrades from 1.6. A direct upgrade from a version before 1.4 needs `useSpools=off` until all servers run the new version. - **Default.** The default changes from off to `auto`. Queries that `auto` spools read a shared stage once. A CTE that is not deterministic (for example `LIMIT` without `ORDER BY`) can now return a different result, because it is evaluated once instead of once for each consumer. - **Parsing.** Values other than `off`/`auto`/`on`/`false`/`no`/`true`/`yes` now fail the query with `QUERY_VALIDATION`. Before, any value except `true` meant off. An invalid broker config fails the broker start. - **API.** `CommonConstants.Broker.DEFAULT_OF_SPOOLS` is replaced by `DEFAULT_SPOOLS_MODE`. `QueryEnvironment.Config.defaultUseSpools()` is replaced by `defaultSpoolMode()`, and `QueryEnvironment.useSpools(Map)` by `getSpoolMode(Map)`. - **Scope.** Spools apply only to the default planner. With `usePhysicalOptimizer`, the mode is ignored. ## Known limits - The spool meters also count EXPLAIN queries. `PRelNodeTreeValidator` has the same limitation. - `on` can still block. It exists for users who know that their data fits the mailbox buffers. ## Testing - New unit tests: - `SpoolDeadlockGuardTest`: 15 cases for joins, unions, intersections, draining operators, the pipeline breaker, and circular waits. - `SpoolModeTest`. - `PinotLogicalQueryPlannerSpoolFeedbackTest`. - New guarded cases in `EquivalentStagesReplacerTest`. - `PinotDispatchPlannerTest.testAutoSpoolsRejectCircularWaitBetweenJoins` plans the swapped-join query with `on` (two spools) and with `auto` (one spool). - `SpoolIntegrationTest` has two new end-to-end cases: - `auto` refuses the spool that can block. - `auto` applies a safe spool, and the result matches `useSpools=off`. - These suites pass with the new default: - The full `pinot-query-planner` suite: 2744 tests. - `ResourceBasedQueriesTest` and `QueryRunnerTest`: 4158 tests. - `SpoolIntegrationTest`. -- 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]
