Select the stats store by name and purge orphaned statistics on request - #19411
Draft
gortiz wants to merge 5 commits into
Draft
Select the stats store by name and purge orphaned statistics on request#19411gortiz wants to merge 5 commits into
gortiz wants to merge 5 commits into
Conversation
Defines what a Pinot statistic is, before anything produces or consumes one: TableStatistics and ColumnStatistics as the values, StatConfidence as the trust tier attached to each one, StatsStore as broker-local persistence over per-segment rows, and ColumnStatsSource as where per-column statistics are fetched from. Confidence is per statistic rather than per table, so a source that can only estimate some values does not have to devalue the rest. Consumers are expected to treat a low-confidence statistic as absent instead of trusting it, which is what keeps table types with biased raw counts (upsert, dedup, consuming segments) on today's behavior rather than producing confidently wrong plans. StatsAggregations carries the rollup semantics every store must share, so two implementations cannot disagree about what the same stored rows mean: time overlap interpolation, document-weighted averages that exclude the "unknown" sentinel rather than averaging it in as a measurement, and min/max folded under the ordering the column actually has. That ordering is recorded per row as a ColumnValueType rather than guessed from the text, because guessing is wrong in both directions: a string column holding "9" and "10" orders lexically in Pinot but would compare numerically, and a long beyond 2^53 loses digits as a double -- in the direction that narrows the range, which would exclude rows that exist. When the ordering is unknown or segments disagree about it, both bounds are reported as absent: a bound folded under two different orderings is neither a true minimum nor a true maximum, and there is no honest way to describe it as merely untrusted. The store contract includes getTables(), so a caller can find tables it no longer serves. Per-table cleanup is otherwise driven by an event -- a routing entry being removed -- which a broker cannot observe for a table dropped while it was down.
Statistics for hundreds of thousands of segments must not compete with query execution for JVM heap, so the default store is an embedded SQLite database (WAL, a small read-connection pool, and drop-and-rebuild recovery if the file is ever unreadable). Surviving a restart is a side benefit rather than the goal: the file also lets the broker skip re-collecting segments whose crc has not changed. There is no migration framework. SQLite's own user_version pragma carries the schema version, and a store whose version does not match is discarded and rebuilt rather than migrated. Every row here is derived from ZooKeeper metadata the broker re-reads at startup, so a rebuild costs nothing that was not already being read -- and it puts schema change, corruption and an unreadable file on one recovery path. This adds exactly one third-party dependency, sqlite-jdbc. InMemoryStatsStore is the alternative for brokers that cannot or should not write a file. It keeps the same semantics -- consuming segments excluded from aggregates but present for crc reconciliation, no-data reported as absent rather than zero -- by folding through the same StatsAggregations helpers. Both serve table-level reads from a rollup recomputed only after a write, since query planning asks for them on every compile while writes arrive at segment push cadence. The version stamp is what makes that safe without holding the write lock: a rollup computed from rows that changed underneath it is used for that call but never published. Tests are written as a contract both stores must satisfy -- the optimizer must not change with the configured store -- with each implementation adding only what is specific to it: durability and corruption recovery for SQLite, starting empty for in-memory.
The broker already subscribes to segment ZK metadata to build routing, so collection rides that existing stream rather than adding cluster traffic: row counts, sizes and time boundaries cost nothing extra. Collection attaches through the routing manager's generic listener-provider seam, so the routing manager itself needs no knowledge of statistics. Reconciliation is by crc, so a restart re-collects only what actually changed, and segments that leave the online set are dropped. Statistics for a table are released when its routing entry goes away, via a new default onRoutingRemoved() on the listener interface -- deliberately named for the event it is, since routing is also removed when a broker stops serving a table that still exists. One listener failing there cannot abort the rest of the teardown. LogicalTableStatsResolver turns stored per-segment rows into the logical view a planner would ask for: hybrid tables merge at the live time boundary rather than double counting, upsert and dedup tables are marked low confidence because physical doc counts over-count logical rows, and a table with consuming segments is marked estimated because committed counts undercount fresh data. Splitting a hybrid table at its boundary means one side can contribute rows to a range while the other structurally cannot. A side whose sub-range is empty contributes a certain zero, since no rows fall in an empty interval; a side that must be answered from statistics and has none is unknown, and reporting that as zero would hand the planner a confident "this table is empty" -- the worst possible input -- instead of no estimate at all. The boundary itself comes from TimeBoundaryManager, which already derives it in epoch milliseconds and owns the DateTimeFormatSpec needed to interpret TimeBoundaryInfo's formatted value. Re-deriving it at the call site would be free to drift, and parsing the formatted value directly is wrong for any column stored in days, seconds or a numeric date pattern: it parses cleanly as a number that is not an instant. Off by default (pinot.broker.stats.enabled=false), broker-local, and every store error degrades to no-statistics rather than failing a query.
pinot.broker.stats.store resolves against StatsStoreProvider implementations discovered through ServiceLoader, each declaring its own name -- following the existing TimeBoundaryStrategyService pattern. The value is deliberately a name and not a class name. A class name in configuration is a rename trap: moving or renaming an implementation would silently invalidate an operator's config. Duplicate names fail loudly, and an unknown name fails startup listing the names that do exist, rather than quietly disabling statistics an operator explicitly enabled. Discovery walks the plugin realms as well as the context classloader, since a provider contributed by a plugin is invisible to the context loader alone -- which would make the documented extension seam unusable. The same provider class reached through two classloaders is deduplicated by class name, so a fat jar plus a plugin realm is not mistaken for a name collision. A provider's own failure is kept distinct from an operator typo: the typo fails broker startup, while a provider that cannot build a store from the given configuration degrades to no-statistics. Getting those two the wrong way round inverts the operator-visible outcome.
Tables dropped while a broker is down leave rows that nothing would ever revisit, because no routing entry -- and so no listener -- is created for them again. Reclaiming those is exposed as DELETE /statistics/orphaned rather than run automatically. A startup sweep is wrong in the worst way: BaseBrokerRoutingManager.init() only wires up ZooKeeper accessors, and routing entries are created later by Helix state transitions. routingExists() would answer false for every table and the sweep would delete the whole store on every boot, logging it as tables the broker no longer serves -- making the SQLite store, its schema versioning and the crc fast path all pointless. Moving it later does not fix it either. Routing is built asynchronously, so before it settles "no routing for T" also matches a table that has merely not loaded yet; purging then discards the column statistics that are the reason to persist anything at all. There is a readiness signal -- service status is GOOD only once CurrentState matches IdealState, which by default means every assigned table -- but consuming it needs a background waiter, and an operator can lower minResourcePercentForStart and quietly break the guarantee. An operator calling this on a running broker removes the ambiguity entirely. The liveness test and the purge are not atomic with respect to a routing build, which publishes its routing entry only after its listener has written that table's rows. A table that becomes served in that window has just had those rows deleted underneath a listener that believes they are still there, and since a listener only removes segments its mirror knows about, the loss would persist until restart. The purge re-tests afterwards and hands such a listener a clean slate instead.
This was referenced Aug 31, 2026
Closed
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19411 +/- ##
============================================
- Coverage 67.56% 57.59% -9.98%
============================================
Files 3486 2692 -794
Lines 224173 164194 -59979
Branches 35381 26651 -8730
============================================
- Hits 151462 94567 -56895
- Misses 60684 61620 +936
+ Partials 12027 8007 -4020
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Contributes to #18740. Stacked on #19409 (contracts and stores) and #19410 (collection) — review those first. Part of the split that replaces #18741.
This PR makes the feature configurable and operable: it selects a store by name, wires collection
into broker startup behind
pinot.broker.stats.enabled, and adds an endpoint to reclaim statisticsfor tables the broker no longer serves. Still no query-behavior change.
Selecting a store
pinot.broker.stats.storeresolves againstStatsStoreProviderimplementations discovered throughServiceLoader, each declaring its own name — the same pattern asTimeBoundaryStrategyService.The value is a name, not a class name. A class name in configuration is a rename trap: moving or
renaming an implementation would silently invalidate an operator's config. Duplicate names fail
loudly, and an unknown name fails startup listing the names that do exist, rather than quietly
disabling statistics an operator explicitly asked for.
Discovery walks the plugin realms as well as the context classloader, since a provider contributed
by a plugin is invisible to the context loader alone — which would make the documented extension
seam unusable. The same provider class reached through two classloaders is deduplicated by class
name, so a fat jar plus a plugin realm is not mistaken for a collision.
A provider's own failure is kept distinct from an operator typo: the typo fails broker startup,
while a provider that cannot build a store from the given configuration degrades to no-statistics.
Getting those two the wrong way round inverts the operator-visible outcome.
Reclaiming orphaned statistics
A broker drops a table's statistics as soon as its routing entry goes away (PR 2). That cannot cover
a table dropped while the broker was down: on restart no routing entry — and so no listener — is
ever created for it, leaving rows nothing would revisit.
This is exposed as
DELETE /statistics/orphanedrather than run automatically, because a startupsweep is wrong in the worst way.
BaseBrokerRoutingManager.init()only wires up ZooKeeper accessors;routing entries are created later by Helix state transitions.
routingExists()would answer falsefor every table and the sweep would delete the whole store on every boot, logging it as tables the
broker no longer serves — making the store, its schema versioning and the crc fast path pointless.
Moving it later does not fix it either. Routing is built asynchronously, so before it settles "no
routing for T" also matches a table that has merely not loaded yet, and purging then discards exactly
the column statistics that are expensive to re-fetch. There is a readiness signal — service status
is GOOD only once CurrentState matches IdealState — but consuming it needs a background waiter, and
an operator can lower
minResourcePercentForStartand quietly break the guarantee. An operatorcalling this on a running broker removes the ambiguity entirely.
The liveness test and the purge are not atomic with respect to a routing build, which publishes its
routing entry only after its listener has written that table's rows. A table that becomes served in
that window has just had those rows deleted underneath a listener that believes they are still
there; since a listener only removes segments its mirror knows about, the loss would persist until
restart. The purge re-tests afterwards and hands such a listener a clean slate instead.
Reviewer notes
pinot.broker.stats.enabled=false) and broker-local: no wire format, nomixed-version concern, nothing to coordinate on a rolling upgrade.
[EXPERIMENTAL]:pinot.broker.stats.enabled,pinot.broker.stats.store,pinot.broker.stats.dir(the last applies only to thesqlitestore). The default directory is
<java.io.tmpdir>/<instanceId>/broker-stats— per-instance,because two brokers on one host sharing a single SQLite file would corrupt it.
DELETE /statistics/orphaned, behind a newActions.Cluster.DELETE_STATISTICSauthorization action. Returns 404 when collection is disabled, so it is inert on an unconfigured
broker.
BrokerAdminApiApplicationgains a nullable constructor parameter for the stats manager; it isbound only when statistics are enabled, and the resource injects it as
@Optional.Testing
StatsStoreFactoryTestcovers every configured value, the fail-fast path for a typo, duplicatenames, and the deliberate split between a provider failure (degrade) and an operator typo (fail
startup).
PinotBrokerStatisticsTestcovers the endpoint: the 404 when collection is disabled, and that theliveness predicate is actually wired to the routing table — an inverted one would purge tables the
broker still serves.
BrokerStatsCollectionIntegrationTeststarts a real broker withpinot.broker.stats.enabled=trueagainst a real controller and ZooKeeper. Everything else in thisstack is unit-tested against hand-built ZooKeeper records, which cannot see the wiring: that the
listener provider is registered before the routing manager initialises, that the
pinot.broker.stats.*subset reaches the provider, that the directory default resolves, and thatsqlite-jdbc can load its native library in an assembled broker. Each of those fails by silently
collecting nothing, so no other assertion would notice. It asserts that segments announced through
Helix arrive in the store with the counts ZooKeeper carries, that the time boundary of a
DAYS-typed column resolves to an instant rather than the day number, and that
DELETE /statistics/orphanedis reachable and leaves a served table alone. It justifies its owncluster because the feature is off by default and cannot ride an existing broker fixture.