yashmayya opened a new pull request, #19510: URL: https://github.com/apache/pinot/pull/19510
Exact `DISTINCTCOUNT` and `PERCENTILE` hold unbounded per-group state on servers: a `Set` of every distinct value, or a `DoubleArrayList` of every raw value. Nothing caps them, because `numGroupsLimit` caps the number of groups rather than the size of each group's accumulator. This is a recurring source of server and multi-stage worker OOMs. `pinot.broker.use.approximate.function` already rewrites these into the SMART variants, which stay exact until an accumulator crosses a threshold. Three things stop it being usable as a fleet-wide guard rail: it is read once in the broker constructor, so a cluster-config change needs a restart; the multi-stage engine ignores it entirely; and the conversion threshold cannot be set outside the SQL call itself. ## What this adds **Cluster config with a live reload.** The existing key is now also read from the Helix cluster config, where it wins over the broker conf, and a change reaches a running broker through its ZooKeeper watch. Two new keys carry the conversion parameters, passed verbatim as the trailing argument of the rewritten call: ``` pinot.broker.use.approximate.function = true pinot.broker.approximate.function.distinct.count.params = threshold=10000;log2m=12;dictThreshold=10000 pinot.broker.approximate.function.percentile.params = threshold=1000;compression=100 ``` Worst-case memory for a group-by is `threshold` values per group, so the threshold wants sizing together with `numGroupsLimit` rather than in isolation. The parameters are validated on the broker by parsing them with the aggregation function itself, so a typo is rejected there instead of failing every query on the servers. **Off by default.** `pinot.broker.use.approximate.function` still defaults to `false`. **Multi-stage support.** `PinotApproximateAggregateRewriteRule` rewrites `DISTINCTCOUNT`, `DISTINCTCOUNTMV`, `COUNT(DISTINCT x)`, `PERCENTILE` and `PERCENTILEMV` on `LogicalAggregate`. It runs in `Phase.BASIC`, which it has to: `PinotAggregateExchangeNodeInsertRule` derives the leaf-to-final intermediate result format from the function name in `POST_LOGICAL`, so a later rewrite, or one in the runtime, would desync the planned schema from the bytes the leaf produces. The rule pins the original return type, so switching the config on does not change the result schema. **A per-query escape hatch.** A new `useApproximateFunction` query option overrides everything, in both directions. Precedence is query option > table `QueryConfig` > cluster config > broker conf. The table level applies to the single-stage engine only, because a multi-stage query can span tables and so resolves the setting before it knows the table set. **Reporting.** `approximateFunctionApplied` on the broker response, alongside `numGroupsLimitReached` and `rlsFiltersApplied`, plus an `APPROXIMATE_FUNCTION_OVERRIDES` meter and a query-log field. Trading exactness for memory should not be invisible. ## The group-by fix Both SMART functions only converted at merge time. `percentileSmartTDigest` had no per-group check at all, and `distinctCountSmartHLL` only checked dictionary-encoded columns; their class javadocs said "For aggregation-only queries". Per-group accumulators therefore grew for a whole segment, so the rewrite would not have bounded memory for group-by queries, which is where these OOMs happen. Both now apply the threshold per accumulator. Behaviour note, unrelated to the new config: this changes results for existing callers of `distinctCountSmartHLL`, `distinctCountSmartHLLPlus`, `distinctCountSmartULL` and `percentileSmartTDigest` with `GROUP BY`. Because a merged accumulator is a superset of each accumulator, a group that crosses the threshold in one segment would have crossed it at merge too, so the exact-or-approximate classification is unchanged wherever a merge happens; what changes is that the sketch is built per segment and then merged, rather than built once over the union. Where no merge happens at all, a single segment on a single server, an over-threshold group goes from exact to approximate, which is what the documented contract of these functions already said would happen. ## Two bugs found on the way, both needed here `AggregationFunctionType.getAggregationFunctionType` rejected every canonical `PERCENTILE*` name that is not one of the deprecated numeric-suffix spellings, so `percentileSmartTDigest` could not be planned in the multi-stage engine at all. It now tries the canonical name first. That change would have silently masked a second bug, so it is fixed explicitly: `RAWKLLMV` and `RAWKLL<n>MV` returned `PERCENTILEKLLMV`. **`percentileRawKLLMV(col)` therefore changes from returning a DOUBLE to returning the VARCHAR sketch it was always meant to return.** Separately, `DISTINCTCOUNTSMARTHLL`, `DISTINCTCOUNTSMARTHLLPLUS` and `DISTINCTCOUNTSMARTULL` declared no final return type, so Calcite inferred BIGINT while the runtime returns INT. They now declare `INTEGER`, matching `DISTINCTCOUNT`. ## Known caveat In the single-stage engine an unaliased aggregation column is named after its function, so a rewritten `distinctcount(x)` comes back as `distinctcountsmarthll(x)`. Values and column types are unchanged, the multi-stage engine keeps its Calcite-derived name, and an explicit `AS` alias keeps the old one. This is how every broker-side function override already behaves, including `distinctCountBitmap` and `segmentPartitionedDistinctCount`, and a test pins it. ## Testing `./mvnw test` over `pinot-spi`, `pinot-segment-spi`, `pinot-common`, `pinot-core`, `pinot-query-planner` and `pinot-broker` passes. New tests cover the per-group conversion for both function families, including multi-valued columns, the BYTES branch, and values arriving after a group has converted; the config precedence, live reload and parameter validation; the planner rule for every rewritten spelling, for the functions it must leave alone, for both parameter strings in one query, and under the physical optimizer. `ApproximateFunctionOverrideIntegrationTest` flips the cluster config on a running broker and checks both engines for the response flag, unchanged values and types, `COUNT(DISTINCT)`, the query option, and a threshold low enough to prove the parameters reach the servers and parse there. -- 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]
