xiangfu0 commented on code in PR #19972:
URL: https://github.com/apache/pinot/pull/19972#discussion_r4329678003
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java:
##########
@@ -152,6 +166,20 @@ public DataTable getDataTable()
return dataTableBuilder.build();
}
+ /// Extracts the final results before the data table is built when the
server returns them, because the metadata
+ /// is read first, and has to report a function that returned an approximate
result.
+ @Override
+ public Map<String, String> getResultsMetadata() {
Review Comment:
**[CRITICAL] Cross-thread data race on `_finalResults` in the MSE leaf
DIRECT-aggregation path.**
When `ServerPlanRequestVisitor` sets `SERVER_RETURN_FINAL_RESULT=true`,
`StreamingInstanceResponseOperator.getNextBlock()` sends this block to
`LeafOperator`'s queue *and then* builds the `InstanceResponseBlock`, whose
constructor calls the new `getResultsMetadata()` → `getFinalResults()` on the
leaf thread. Concurrently the MSE operator thread dequeues the same block and
calls `getRows()` → `getFinalResults()`. The queue handoff happens before the
producer's extraction, so there is no happens-before edge: both threads can run
`extractFinalResult` on the same intermediate objects.
For `PERCENTILESMARTTDIGEST` below its threshold, the exact branch does an
in-place `Arrays.sort` on the shared `DoubleArrayList` backing array
(`PercentileSmartTDigestAggregationFunction.java:383-384`) — two interleaved
sorts can return a nondeterministically wrong percentile, and the consumer can
also observe a partially published `_finalResults` array (non-volatile field).
Pre-PR, only the consumer ever extracted (into a local array), so this race
is introduced here. Fix: extract before `_streamer.send()`, or make
`getFinalResults()` synchronized / publish via a volatile.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/BrokerReduceService.java:
##########
@@ -132,6 +133,11 @@ public BrokerResponseNative
reduceOnDataTable(BrokerRequest brokerRequest, Broke
"Caught exception while reducing data tables: " + e.getMessage()));
}
}
+ // The reducers extract the final results with the functions of this query
context, which record a result that
+ // turned approximate. Servers that extracted final results themselves
report it through the aggregator instead.
+ if
(AggregationFunctionUtils.hasApproximatedResult(serverQueryContext.getAggregationFunctions()))
{
Review Comment:
**[MAJOR] Gapfill aggregation never sets `approximatedResult`.**
For e.g. `SELECT time_col, DISTINCTCOUNTSMARTHLL(userId, 'threshold=10')
FROM (SELECT GAPFILL(...) ...) GROUP BY time_col`, `GapfillUtils.stripGapfill`
makes `serverBrokerRequest` the inner selection, so
`serverQueryContext.getAggregationFunctions()` is null here and this check is a
no-op. `GapfillProcessor` (lines ~289-303) then builds its *own* SMART function
instances; when a bucket crosses the threshold, `extractFinalResult` returns a
sketch estimate and sets `_approximatedResult` on an instance nobody reads. The
response returns approximate counts with `approximatedResult=false` — a false
negative on exactly the signal this PR introduces.
Fix: sample `hasApproximatedResult()` on the gapfill processor's function
instances after `process()` (or move this check after gapfill using those
instances).
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java:
##########
@@ -167,6 +163,11 @@ public class BrokerMeter implements AbstractMetrics.Meter {
public static final BrokerMeter
BROKER_RESPONSES_WITH_NUM_GROUPS_LIMIT_REACHED = create(
"BROKER_RESPONSES_WITH_NUM_GROUPS_LIMIT_REACHED", "badResponses", false);
+ // This metric tracks the number of broker responses with an approximated
result: a function that can answer exactly,
Review Comment:
**[MAJOR, release timing] The removed names exist in `release-1.6.0-rc0`.**
`APPROXIMATE_FUNCTION_OVERRIDES` and the `approximateFunctionApplied`
JSON/query-log fields are in no final release (verified absent in
`release-1.5.1`) but are present in `release-1.6.0-rc0`. If 1.6.0 ships from
that line before this rename reaches it, dashboards on
`pinot_broker_approximateFunctionOverrides` go silently flat and clients
reading `approximateFunctionApplied` silently read missing/false — a
released-API break in 1.7. Please make sure this rename is cherry-picked to the
1.6.0 release branch before the release finalizes, or keep a deprecated alias
for one release.
Also note the semantics narrowing: after flipping
`pinot.broker.use.approximate.function`, an operator can no longer verify from
metrics/response/logs that brokers picked up the config while data is below
threshold — this test itself had to fall back to column-name sniffing and
EXPLAIN. Consider keeping some rewrite-activity signal.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java:
##########
@@ -98,58 +100,73 @@ public void testClusterConfigAppliesWithoutRestart(boolean
useMultiStageQueryEng
JsonNode exactDistinctCount = postQuery(DISTINCT_COUNT_QUERY);
JsonNode exactPercentile = postQuery(PERCENTILE_QUERY);
- assertApplied(exactDistinctCount, false);
- assertApplied(exactPercentile, false);
+ assertApproximated(exactDistinctCount, false);
+ assertApproximated(exactPercentile, false);
// A threshold far above the cardinality of the test data keeps the
rewritten functions exact, which is what makes
- // the answers comparable before and after the flip.
+ // the answers comparable before and after the flip. The results stay
exact, so the flag stays off.
setApproximateFunctionOverride("threshold=1000000");
waitForOverride(true);
JsonNode distinctCount = postQuery(DISTINCT_COUNT_QUERY);
- assertApplied(distinctCount, true);
+ assertApproximated(distinctCount, false);
assertSameResult(exactDistinctCount, distinctCount);
JsonNode percentile = postQuery(PERCENTILE_QUERY);
- assertApplied(percentile, true);
+ assertApproximated(percentile, false);
assertSameResult(exactPercentile, percentile);
+ // A threshold below the cardinality has to make the answer approximate
rather than fail, keeping the same shape,
+ // and the flag reports it. This is also what proves the parameters reach
the servers and parse there.
+ setApproximateFunctionOverride("threshold=1");
+ waitForOverride(true);
Review Comment:
**[MAJOR, flaky test] `waitForOverride(true)` doesn't wait for the *new
threshold* to reach the broker.**
The wait's predicate (SSE column name starts with `distinctcountsmarthll` /
MSE EXPLAIN contains it) is already true under the previous
`threshold=1000000`, so the `threshold=1` switch returns immediately — before
the broker's async ZK watch delivers the new params. If the broker still
rewrites with the old threshold, results stay exact and the following
`assertApproximated(..., true)` assertions fail intermittently. (The pre-rename
test had no such race because `assertApplied(true)` held under either
threshold.)
Fix: after switching thresholds, poll until a query actually reports
`approximatedResult=true`.
##########
pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java:
##########
@@ -173,11 +173,14 @@ enum MetadataKey {
// The subset of the remote accesses made directly, i.e. outside the
buffers fetched for the query, and the number
// of bytes they read.
NUM_DIRECT_REMOTE_ACCESSES(47, "numDirectRemoteAccesses",
MetadataValueType.LONG),
- DIRECT_REMOTE_ACCESS_BYTES(48, "directRemoteAccessBytes",
MetadataValueType.LONG);
+ DIRECT_REMOTE_ACCESS_BYTES(48, "directRemoteAccessBytes",
MetadataValueType.LONG),
+ // Set when an aggregation function that can answer exactly returned an
approximate final result on the server,
+ // for example a SMART function that switched to a sketch.
+ APPROXIMATED_RESULT(49, "approximatedResult", MetadataValueType.STRING);
// We keep this constant to track the max id added so far for backward
compatibility.
Review Comment:
**[compat note, reverse direction]** During a rolling upgrade, an old server
that extracts final results (`serverReturnFinalResult=true` / MSE leaf final
results) never sends key 49, so a new broker reports `approximatedResult=false`
for a sketch-derived answer — `mergeFinalResult` on the broker never sets the
flag. Transient during rollout but a silent wrong exactness claim; worth a
release note.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java:
##########
@@ -316,6 +316,18 @@ default FinalResult mergeFinalResult(FinalResult
finalResult1, FinalResult final
throw new UnsupportedOperationException("Cannot merge final results for
function: " + getType());
}
+ /// Returns whether [#extractFinalResult] has returned an approximate result
for a function that can also answer
+ /// exactly. The SMART functions start exact and switch a group to a sketch
once it crosses their conversion
+ /// threshold. A merge with a sketch always returns a sketch, so the
extraction sees the switch, and records it here.
+ /// Read it after extracting, to report the result as approximate.
+ ///
+ /// A function instance belongs to one query, and so does the flag. Group
trimming also extracts final results, so a
+ /// group that is trimmed away can still set it. Functions that are always
exact or always approximate keep the
+ /// default.
+ default boolean hasApproximatedResult() {
Review Comment:
**[design/altitude]** The flag is a mutable per-instance side channel set
inside `extractFinalResult` and sampled at four uncoordinated points (two
results blocks, `BrokerReduceService`, `AggregateOperator`). Every result path
must independently remember to sample it — and the gapfill path already proves
the fragility: any extraction path without a paired sampler silently
under-reports, today and for future reducers/post-processors.
A single choke point would cover new paths for free — e.g.
`extractFinalResult` reporting through a collector, or recording into the
per-query `QueryThreadContext` that is already open on server combine threads,
broker reduce threads, and MSE operators. Fine as a follow-up, but worth
considering before more samplers accrete.
##########
pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java:
##########
@@ -173,11 +173,14 @@ enum MetadataKey {
// The subset of the remote accesses made directly, i.e. outside the
buffers fetched for the query, and the number
// of bytes they read.
NUM_DIRECT_REMOTE_ACCESSES(47, "numDirectRemoteAccesses",
MetadataValueType.LONG),
- DIRECT_REMOTE_ACCESS_BYTES(48, "directRemoteAccessBytes",
MetadataValueType.LONG);
+ DIRECT_REMOTE_ACCESS_BYTES(48, "directRemoteAccessBytes",
MetadataValueType.LONG),
+ // Set when an aggregation function that can answer exactly returned an
approximate final result on the server,
Review Comment:
**[compat note] New key 49 desynchronizes metadata deserialization on
pre-upgrade brokers.**
`DataTableImplV4.deserializeMetadata` skips unknown key ids with `continue`
*without consuming the value bytes*, so when a new server emits key 49 to an
old broker (canary servers, downgrade — i.e. outside the recommended
broker-before-server order), the 4-byte length of `"true"` is parsed as the
next keyId and every subsequent metadata entry is garbage (wrong stats or
`BufferUnderflowException`). The skip bug is pre-existing and the recommended
upgrade order avoids it, but this key fires on common queries — worth a
compatibility note in the PR description (or fixing the skip to consume the
value by encoding type/length).
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java:
##########
@@ -296,6 +298,9 @@ public void setStats(String rawTableName,
BrokerResponseNative brokerResponseNat
brokerResponseNative.setGroupsTrimmed(_groupsTrimmed);
brokerResponseNative.setNumGroupsLimitReached(_numGroupsLimitReached);
brokerResponseNative.setNumGroupsWarningLimitReached(_numGroupsWarningLimitReached);
+ if (_approximatedResult) {
Review Comment:
**[minor, consistency]** Every adjacent boolean in this method is set
unconditionally (`setGroupsTrimmed`, `setNumGroupsLimitReached`,
`setNumGroupsWarningLimitReached`), but this one is guarded. Both callers
invoke `setStats` before any other writer of the flag, so an unconditional
`brokerResponseNative.setApproximatedResult(_approximatedResult)` behaves
identically — the guard implies an ordering hazard that doesn't exist. Suggest
matching the neighbors' direct-set style.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java:
##########
@@ -52,11 +52,19 @@ abstract class
BaseDistinctCountSmartSketchAggregationFunction
// Use empty IntOpenHashSet as a placeholder for empty result
protected static final IntSet EMPTY_PLACEHOLDER = new IntOpenHashSet();
+ /// Set by `extractFinalResult` when it reads a sketch, see
[AggregationFunction#hasApproximatedResult()].
+ protected volatile boolean _approximatedResult;
Review Comment:
**[perf, minor]** Past the threshold, `extractFinalResult` does an
unconditional volatile store per group — once per group in
`IndexedTable.finish()` (multi-threaded when `numThreadsExtractFinalResult` is
set) and once per record per resize via `TableResizer` when ordering by the
aggregation. All threads repeatedly writing the same field of the shared
instance causes cache-line ping-pong, and the store also invalidates the line
holding fields read on every extraction (e.g. `_percentile` in the TDigest
function).
Cheaper and semantically identical: `if (!_approximatedResult) {
_approximatedResult = true; }` so the steady state is read-only.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java:
##########
@@ -237,6 +238,10 @@ public Map<String, String> getResultsMetadata() {
if (_numGroupsWarningLimitReached) {
metadata.put(MetadataKey.NUM_GROUPS_WARNING_LIMIT_REACHED.getName(),
"true");
}
+ // The combine step has already extracted the final results when the
server returns them.
Review Comment:
**[minor, comment accuracy]** "The combine step has already extracted the
final results when the server returns them" misstates when this unguarded
emission can fire: in intermediate mode, ORDER BY trim extraction
(`TableResizer` via `IndexedTable.finish`/resize) also sets the flag, including
for groups that are trimmed away — verified reachable with
`serverReturnFinalResult=false` plus ORDER BY on the SMART aggregation (yields
`approximatedResult=true` even when every returned row is exact). The
`hasApproximatedResult` javadoc documents this as intended, so this is a
comment fix: state that order-by/trim extraction in intermediate mode also
feeds this emission, so the asymmetry with `AggregationResultsBlock`'s
`isServerReturnFinalResult` guard doesn't get "fixed" into a false negative
later.
##########
pinot-core/src/test/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlockTest.java:
##########
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.operator.blocks.results;
+
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.common.datatable.DataTable;
+import org.apache.pinot.common.datatable.DataTable.MetadataKey;
+import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
+import org.apache.pinot.core.query.request.context.QueryContext;
+import
org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+public class AggregationResultsBlockTest {
Review Comment:
**[minor, convention]** New class is missing class-level Javadoc (CLAUDE.md:
"Add class-level Javadoc for new classes; describe behavior and thread-safety")
— the PR's other new class `SmartFunctionApproximatedResultTest` has one. A
one-line `///` class doc resolves it.
##########
pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SmartFunctionApproximatedResultTest.java:
##########
@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.query.aggregation.function;
+
+import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import java.util.List;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Covers [AggregationFunction#hasApproximatedResult()] for the SMART
functions: extracting an exact accumulator leaves
+/// it unset, and extracting one that a merge switched to a sketch sets it.
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class SmartFunctionApproximatedResultTest {
+ private static final ExpressionContext COLUMN =
ExpressionContext.forIdentifier("col");
+ private static final ExpressionContext THRESHOLD_ONE =
+ ExpressionContext.forLiteral(Literal.stringValue("threshold=1"));
+
+ @DataProvider
+ public static Object[][] distinctCountFunctions() {
+ List<ExpressionContext> arguments = List.of(COLUMN, THRESHOLD_ONE);
+ return new Object[][]{
+ {new DistinctCountSmartHLLAggregationFunction(arguments, false)},
+ {new DistinctCountSmartHLLPlusAggregationFunction(arguments, false)},
+ {new DistinctCountSmartULLAggregationFunction(arguments, false)}
+ };
+ }
+
+ @Test(dataProvider = "distinctCountFunctions")
+ public void testDistinctCount(AggregationFunction function) {
+ assertEquals(function.extractFinalResult(null), 0);
+ assertEquals(function.extractFinalResult(new IntOpenHashSet(new
int[]{1})), 1);
+ assertFalse(function.hasApproximatedResult());
+
+ // Two values cross the threshold of one, so the merge switches the
accumulator to a sketch.
+ Object merged = function.merge(new IntOpenHashSet(new int[]{1}), new
IntOpenHashSet(new int[]{2}));
+ function.extractFinalResult(merged);
+ assertTrue(function.hasApproximatedResult());
+ }
+
+ @Test
+ public void testPercentile() {
+ PercentileSmartTDigestAggregationFunction function = new
PercentileSmartTDigestAggregationFunction(
+ List.of(COLUMN,
ExpressionContext.forLiteral(Literal.doubleValue(50.0)), THRESHOLD_ONE), false);
+ function.extractFinalResult(null);
Review Comment:
**[minor, test gap]** `testPercentile` calls `extractFinalResult(null)`
without asserting the returned value, unlike `testDistinctCount` which asserts
`== 0`. The null-input contract (`DEFAULT_FINAL_RESULT` when null handling is
disabled) is exercised but never verified, so a regression in that branch would
pass. Suggest `assertEquals(function.extractFinalResult(null),
DEFAULT_FINAL_RESULT)`.
--
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]
