apache / apache/pinot

[Umbrella] Cost-based optimization for the multi-stage query engine

Open
#18,740 4 comments 0 reactions 0 assignees View on GitHub
feature multi-stage
Dominant language
Java
Stars
6.1k
Forks
1.5k
Avg merge
2d 55m
Merged PRs (30d)
182

Description

# [Umbrella] Cost-based optimization for the multi-stage query engine

## Motivation

The multi-stage query engine plans queries with Apache Calcite, but optimization today is purely
rule-based (`HepPlanner`): join order is whatever the SQL says, join strategies are chosen via
explicit hints, and the planner has no cardinality information (`PinotTable` does not expose a
Calcite `Statistic`; there is a long-standing `TODO: add support for cost factory` in
`QueryEnvironment`). Calcite already ships the machinery for cost-based decisions — what Pinot is
missing is **statistics at the broker** and the integration to consume them.

This issue tracks the work to introduce cost-based optimization incrementally, gated and off by
default at every step.

## Why it matters (measured)

On TPC-H SF=1 (single-server quickstart, 150 samples per variant, medians with bootstrap 95% CIs),
for queries deliberately written in a poor syntactic join order (smallest tables first):

| query | literal SQL order | with cost-based reorder | hand-optimized SQL |
|---|---|---|---|
| `customer⋈orders⋈lineitem` (filtered) | 549 ms [516, 570] | 462 ms [438, 479] | 412 ms [390, 430] |
| `region⋈nation⋈supplier⋈lineitem` (filtered) | 253 ms [244, 277] | **80 ms [80, 81]** | 207 ms [204, 211] |

In the second query the optimizer found a bushy plan (reduce the dimension chain first, then a
single probe over the fact table) that is **2.6× faster than the hand-optimized left-deep SQL**
and 3.2× faster than the literal order. Results are identical in all variants, and the reorder
phase adds no measurable planning overhead when it decides to change nothing.

## Design overview

Three pillars (detailed design in the linked doc / PR descriptions):

1. **Broker statistics subsystem.** Per-segment stats collected from ZooKeeper metadata the broker
already watches (row count, size, time boundaries — effectively free), persisted off-heap in an
embedded SQLite store (bounded heap, warm restart, crc-based reconciliation). Per-column stats
(NDV, min/max, avg byte size) come later via a bounded server fan-out behind a swappable
`ColumnStatsSource`. Every stat carries a **confidence** tier so that table types where raw
doc counts are biased (upsert/dedup, consuming segments, hybrid time-boundary overlap) degrade
to today's behavior instead of producing silently-wrong plans.
2. **Cost model.** Rows-dominated for logical join ordering; extended later with columnar byte
sizes for exchange-aware decisions (shuffle/broadcast cost is the dominant term in MSE).
3. **Planner integration.** `PinotTable#getStatistic()` + a chained `RelMetadataProvider`
(row counts, selectivity incl. time-range estimation from segment time boundaries) feed
Calcite's `RelMetadataQuery`; a gated join-reorder phase (`useJoinReorder` query option,
default off) runs between the logical and physical planning phases with strict eligibility
gates (inner joins only, no hinted joins, all scans must have trusted row counts, join-count
cap, fall back to the original plan on any error).

## Phase 1 — statistics foundation + join reordering

#18741 carried this entire phase as one 7.2k-line PR and was too large to review in a single pass.
It is closed in favour of the stacked series below; each of the first three compiles and passes its
tests on its own, and none of them changes a query plan.

- [ ] #19409 — statistics contracts (`TableStatistics`, `ColumnStatistics`, `StatConfidence`,
`ColumnValueType`, and the `StatsStore` / `StatsStoreProvider` / `ColumnStatsSource` SPIs)
together with the two broker-local stores: SQLite (WAL, `user_version` schema versioning,
drop-and-rebuild recovery) and in-memory, behind one shared contract-test suite
- [ ] #19410 — broker stats collection from ZK segment metadata, and the table-type semantics that
turn raw doc counts into confidence-tiered statistics (hybrid merge at the time boundary,
upsert/dedup marked LOW, consuming segments marked ESTIMATED)
- [ ] #19411 — store selection by name (`pinot.broker.stats.store`), startup wiring
(`pinot.broker.stats.enabled`, default false), and `DELETE /statistics/orphaned`
- [ ] PR: planner wiring — `PinotTable#getStatistic()`, the statistics provider through
`QueryEnvironment.Config`, a `RelMetadataProvider` with stats-backed row counts and
selectivity (incl. time-range selectivity from segment time boundaries), and a rows-dominated
`RelOptCost`
- [ ] PR: gated cost-based join-reorder phase + guardrails (join cap, hint veto, error fallback) and
plan-level tests, together with the quickstart fixes its end-to-end run needs (`-configFile`
dropped by some quickstarts; multiple `-bootstrapTableDir` support)

## Phase 2 — column statistics and standalone payoffs

- [ ] Server endpoint audit/extension for lean per-segment column stats (NDV, min/max, avg bytes)
- [ ] `ColumnStatsSource` broker-pull implementation (rate-limited, jittered, debounced on
rebalance storms; crc-delta fetch)
- [ ] Selectivity refinement: NDV-based equality selectivity, min/max range selectivity,
null-sentinel handling (numeric null default pollutes min — track per-column trust)
- [ ] Broker-side min/max segment pruning for the single-stage engine (the per-column min/max in
the off-heap store removes the historical reason this lives only on servers — see the
existing `TODO` in `ColumnValueSegmentPruner`)
- [ ] Stats observability: broker metrics (stats age, store size, fallback reasons) and EXPLAIN
annotations showing the estimates used (the admin purge endpoint shipped in #19411)

## Phase 3 — beyond join ordering (not committed)

One concrete, self-contained item that falls out of the statistics work:

- [ ] **Build-side normalization** — use row counts to place the smaller input on the hash-build
side of a join. The benchmarks above show the logical reorderer minimizes intermediate
cardinality but cannot account for the engine's build-side convention (the remaining
462 vs 412 ms gap in query 1).

Anything beyond that — making colocation, broadcast-vs-hash, aggregate placement or parallelism
cost-driven — is **future work with no committed design, owner or timeline**. An earlier revision of
this issue sketched a specific two-phase search; that sketch is withdrawn rather than updated, since
it no longer reflects current thinking. If a proposal is made it will come as its own issue with its
own design discussion. Nothing in Phases 1–2 depends on it.

## Cross-cutting

- [ ] Vendor extensibility: statistics source, cost model and rule sets pluggable via the existing
`RuleSetCustomizer`-style SPI pattern
- [ ] Documentation (config reference, design notes, operations guide for the stats store)

## Compatibility and safety

- Everything is **off by default** (`pinot.broker.stats.enabled=false`; `useJoinReorder` query
option default false) and broker-local — no wire-format or mixed-version concerns.
- Stats-store failures never fail a query: all read paths degrade to the no-stats behavior.
- Note for reviewers: enabling stats collection changes the cardinality *estimates* visible to
all MSE planner rules (never correctness); the join-reorder phase is additionally gated by its
own option.
- New dependencies: `org.xerial:sqlite-jdbc`, `org.flywaydb:flyway-core` (broker only,
LICENSE-binary updated).

Contributor guide

Open the contributing guide

Research direction

This is an umbrella issue rather than a standalone change. Start by reading the separately scoped Phase 1 issues, then inspect QueryEnvironment, PinotTable, and the existing TODOs named in the issue. A contribution is done when one explicitly scoped statistics or planner milestone has its implementation and tests completed without changing default behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, sql, sqlite
Domain
backend, databases, distributed-systems, performance
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.