Add a cluster config to rewrite exact DISTINCTCOUNT and PERCENTILE into approximations

[yashmayya]

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.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 79.87013% with 62 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.76%. Comparing base (a350f67) to head (0a29096).
⚠️ Report is 24 commits behind head on master.

Files with missing lines	Patch %	Lines
.../DistinctCountSmartHLLPlusAggregationFunction.java	0.00%	9 Missing ⚠️
...tion/DistinctCountSmartULLAggregationFunction.java	0.00%	9 Missing ⚠️
...seDistinctCountSmartSketchAggregationFunction.java	81.81%	4 Missing and 4 partials ⚠️
...el/rules/PinotApproximateAggregateRewriteRule.java	90.66%	2 Missing and 5 partials ⚠️
...e/pinot/broker/broker/helix/BaseBrokerStarter.java	14.28%	6 Missing ⚠️
...ion/PercentileSmartTDigestAggregationFunction.java	78.57%	5 Missing and 1 partial ⚠️
...sthandler/BaseSingleStageBrokerRequestHandler.java	87.17%	3 Missing and 2 partials ⚠️
...requesthandler/MultiStageBrokerRequestHandler.java	58.33%	5 Missing ⚠️
...che/pinot/segment/spi/AggregationFunctionType.java	33.33%	4 Missing ⚠️
...common/response/broker/BrokerResponseNativeV2.java	50.00%	2 Missing ⚠️
... and 1 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19510      +/-   ##
============================================
+ Coverage     67.63%   67.76%   +0.12%     
  Complexity     1430     1430              
============================================
  Files          3489     3491       +2     
  Lines        224427   224897     +470     
  Branches      35429    35525      +96     
============================================
+ Hits         151797   152397     +600     
+ Misses        60618    60457     -161     
- Partials      12012    12043      +31     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	67.76% <79.87%> (+0.12%)	⬆️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	67.76% <79.87%> (+0.12%)	⬆️
unittests	67.76% <79.87%> (+0.12%)	⬆️
unittests1	57.82% <75.75%> (+0.05%)	⬆️
unittests2	39.47% <36.36%> (+0.11%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.