This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch release-1.6.0-rc
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/release-1.6.0-rc by this push:
new 49d9994479f [1.6.0] Backport Replace approximateFunctionApplied with
approximatedResult, set when a SMART function turns approximate (#19972)
49d9994479f is described below
commit 49d9994479fe5fbe385e203c6a850b88ab7f3898
Author: Yash Mayya <[email protected]>
AuthorDate: Wed Oct 21 10:08:32 2026 -0700
[1.6.0] Backport Replace approximateFunctionApplied with
approximatedResult, set when a SMART function turns approximate (#19972)
Keeps the wire ids of master: DataTable metadata key 49, and the
LeafOperator stat ordinals, with the four remote access stats of master added
as unused placeholders.
(cherry picked from commit be94fc67e474828fa5e605bfe6df70084afbc00f)
---
.../apache/pinot/broker/querylog/QueryLogger.java | 4 +-
.../BaseSingleStageBrokerRequestHandler.java | 14 ++-
.../MultiStageBrokerRequestHandler.java | 12 +--
.../pinot/broker/querylog/QueryLoggerTest.java | 2 +-
.../apache/pinot/common/datatable/DataTable.java | 9 +-
.../apache/pinot/common/metrics/BrokerMeter.java | 9 +-
.../pinot/common/response/BrokerResponse.java | 17 ++--
.../response/broker/BrokerResponseNative.java | 17 ++--
.../response/broker/BrokerResponseNativeV2.java | 29 +++---
.../response/broker/CursorResponseNative.java | 3 +-
.../response/broker/BrokerResponseNativeTest.java | 14 +++
.../blocks/results/AggregationResultsBlock.java | 44 +++++++--
.../blocks/results/GroupByResultsBlock.java | 6 ++
.../aggregation/function/AggregationFunction.java | 12 +++
.../function/AggregationFunctionUtils.java | 14 +++
...istinctCountSmartSketchAggregationFunction.java | 15 +++
.../DistinctCountSmartHLLAggregationFunction.java | 1 +
...stinctCountSmartHLLPlusAggregationFunction.java | 1 +
.../DistinctCountSmartULLAggregationFunction.java | 1 +
.../PercentileSmartTDigestAggregationFunction.java | 11 +++
.../core/query/reduce/BrokerReduceService.java | 6 ++
.../query/reduce/ExecutionStatsAggregator.java | 6 ++
.../results/AggregationResultsBlockTest.java | 85 ++++++++++++++++
.../SmartFunctionApproximatedResultTest.java | 83 ++++++++++++++++
.../core/query/reduce/BrokerReduceServiceTest.java | 52 ++++++++++
.../query/reduce/ExecutionStatsAggregatorTest.java | 17 ++++
...ApproximateFunctionOverrideIntegrationTest.java | 107 +++++++++++++++------
.../PinotApproximateAggregateRewriteRule.java | 5 -
.../apache/pinot/query/context/PlannerContext.java | 14 ---
.../query/runtime/operator/AggregateOperator.java | 19 +++-
.../pinot/query/runtime/operator/LeafOperator.java | 14 ++-
.../query/runtime/operator/MultiStageOperator.java | 1 +
.../runtime/operator/AggregateOperatorTest.java | 37 +++++++
.../query/runtime/operator/LeafOperatorTest.java | 24 +++++
34 files changed, 588 insertions(+), 117 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/querylog/QueryLogger.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/querylog/QueryLogger.java
index 1d79a6f9916..82e6fdc7396 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/querylog/QueryLogger.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/querylog/QueryLogger.java
@@ -450,10 +450,10 @@ public class QueryLogger {
builder.append(params._response.getRLSFiltersApplied());
}
},
- APPROXIMATE_FUNCTION_APPLIED("approximateFunctionApplied") {
+ APPROXIMATED_RESULT("approximatedResult") {
@Override
void doFormat(StringBuilder builder, QueryLogger logger, QueryLogParams
params) {
- builder.append(params._response.isApproximateFunctionApplied());
+ builder.append(params._response.isApproximatedResult());
}
},
WORKLOAD_NAME("workloadName") {
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
index d93ffc1ff70..6ca06199301 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
@@ -674,13 +674,9 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
HandlerContext handlerContext =
getHandlerContext(offlineTableQueryConfig, realtimeTableQueryConfig,
pinotQuery.getQueryOptions());
validateGroovyScript(serverPinotQuery, handlerContext._disableGroovy);
- boolean approximateFunctionApplied = false;
if (handlerContext._useApproximateFunction) {
- approximateFunctionApplied =
handleApproximateFunctionOverride(serverPinotQuery,
- handlerContext._distinctCountParams,
handlerContext._percentileParams);
- if (approximateFunctionApplied) {
- _brokerMetrics.addMeteredTableValue(rawTableName,
BrokerMeter.APPROXIMATE_FUNCTION_OVERRIDES, 1);
- }
+ handleApproximateFunctionOverride(serverPinotQuery,
handlerContext._distinctCountParams,
+ handlerContext._percentileParams);
}
// Validate the request
@@ -1049,6 +1045,9 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
if (brokerResponse.isGroupsTrimmed()) {
_brokerMetrics.addMeteredTableValue(rawTableName,
BrokerMeter.BROKER_RESPONSES_WITH_GROUPS_TRIMMED, 1);
}
+ if (brokerResponse.isApproximatedResult()) {
+ _brokerMetrics.addMeteredTableValue(rawTableName,
BrokerMeter.BROKER_RESPONSES_WITH_APPROXIMATED_RESULT, 1);
+ }
// server returns STRING as default dataType for all columns in (some)
scenarios where no rows are returned
// this is an attempt to return more faithful information based on other
sources
@@ -1086,7 +1085,6 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
}
brokerResponse.setRLSFiltersApplied(rlsFiltersApplied.get());
- brokerResponse.setApproximateFunctionApplied(approximateFunctionApplied);
// Record per-server stats on the SSE BrokerResponse so downstream
consumers can read it.
brokerResponse.setServerStats(serverStats.getServerStats());
@@ -1996,7 +1994,7 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
/// The rewritten functions stay exact until an accumulator crosses their
conversion threshold, which the
/// parameters carry. An empty parameter string appends nothing, so the
aggregation function defaults apply.
///
- /// @return true if at least one function was rewritten, which makes the
result of the query approximate
+ /// @return true if at least one function was rewritten
@VisibleForTesting
static boolean handleApproximateFunctionOverride(PinotQuery pinotQuery,
String distinctCountParams,
String percentileParams) {
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
index 06991be29da..85a92f467e9 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
@@ -898,6 +898,9 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
if (brokerResponse.isGroupsTrimmed()) {
_brokerMetrics.addMeteredTableValue(table,
BrokerMeter.BROKER_RESPONSES_WITH_GROUPS_TRIMMED, 1);
}
+ if (brokerResponse.isApproximatedResult()) {
+ _brokerMetrics.addMeteredTableValue(table,
BrokerMeter.BROKER_RESPONSES_WITH_APPROXIMATED_RESULT, 1);
+ }
if (brokerResponse.isMseLiteLeafStageLimitReached()) {
_brokerMetrics.addMeteredTableValue(table,
BrokerMeter.BROKER_RESPONSES_WITH_MSE_LITE_LEAF_STAGE_LIMIT_REACHED, 1);
@@ -921,15 +924,6 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
// set if rls (row level security) filters have been applied on the query
brokerResponse.setRLSFiltersApplied(rlsFiltersApplied);
- // set if an exact aggregation was rewritten into its approximate
counterpart, which makes the result
- // approximate rather than exact
- if (query.getPlannerContext().isApproximateFunctionApplied()) {
- brokerResponse.setApproximateFunctionApplied(true);
- for (String tableName : tableNames) {
- _brokerMetrics.addMeteredTableValue(tableName,
BrokerMeter.APPROXIMATE_FUNCTION_OVERRIDES, 1);
- }
- }
-
// Log query and stats
_queryLogger.logQueryCompleted(
new QueryLogger.QueryLogParams(requestContext,
tableNames.toString(), brokerResponse,
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/querylog/QueryLoggerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/querylog/QueryLoggerTest.java
index 15a9bf7d382..796a0df53e2 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/querylog/QueryLoggerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/querylog/QueryLoggerTest.java
@@ -149,7 +149,7 @@ public class QueryLoggerTest {
+ "realtimeMemAllocatedBytes(total/thread/resSer):0/0/0,"
+ "pools=[],"
+ "rlsFiltersApplied=true,"
- + "approximateFunctionApplied=false,"
+ + "approximatedResult=false,"
+ "workloadName=workloadName,"
+ "query=SELECT * FROM foo");
//@formatter:on
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java
b/pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java
index d3b3a79dbaa..5fa30eafe68 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java
@@ -160,11 +160,16 @@ public interface DataTable {
// inputs. How a downstream consumer reacts (skip, retry, accept with
annotation) is the
// consumer's policy.
INCOMPLETE_MERGE(43, "incompleteMerge", MetadataValueType.STRING),
- LITE_MODE_LEAF_STAGE_LIMIT_REACHED(44, "liteModeLeafStageLimitReached",
MetadataValueType.STRING);
+ LITE_MODE_LEAF_STAGE_LIMIT_REACHED(44, "liteModeLeafStageLimitReached",
MetadataValueType.STRING),
+ // Ids 45 to 48 belong to the remote access stats on master, which this
release line does not have. The next key
+ // keeps its master id, so that both lines read it the same way.
+ // 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.
// Increase it when adding new keys, but NEVER DECREASE IT!!!
- private static final int MAX_ID =
LITE_MODE_LEAF_STAGE_LIMIT_REACHED.getId();
+ private static final int MAX_ID = APPROXIMATED_RESULT.getId();
private static final MetadataKey[] ID_TO_ENUM_KEY_MAP = new
MetadataKey[MAX_ID + 1];
private static final Map<String, MetadataKey> NAME_TO_ENUM_KEY_MAP = new
HashMap<>();
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
index bc2a8b354e9..e9d5d8c34f9 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
@@ -43,10 +43,6 @@ public class BrokerMeter implements AbstractMetrics.Meter {
/// At this moment this counter does not include queries executed in
multi-stage mode.
public static final BrokerMeter QUERIES = create("QUERIES", "queries",
false);
- /// Queries in which the broker rewrote at least one exact aggregation into
its approximate counterpart, because
- /// `pinot.broker.use.approximate.function` was on. The results of those
queries are approximate.
- public static final BrokerMeter APPROXIMATE_FUNCTION_OVERRIDES =
- create("APPROXIMATE_FUNCTION_OVERRIDES", "queries", false);
/// Number of single-stage queries that have been started.
///
/// Unlike [#QUERIES], this metric is global and not attached to a
particular table.
@@ -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,
+ // such as a SMART function past its conversion threshold, returned an
approximate result.
+ public static final BrokerMeter BROKER_RESPONSES_WITH_APPROXIMATED_RESULT =
create(
+ "BROKER_RESPONSES_WITH_APPROXIMATED_RESULT", "badResponses", false);
+
public static final BrokerMeter
BROKER_RESPONSES_WITH_MSE_LITE_LEAF_STAGE_LIMIT_REACHED = create(
"BROKER_RESPONSES_WITH_MSE_LITE_LEAF_STAGE_LIMIT_REACHED",
"badResponses", false);
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
index 98362b1b4e2..ce1f9cbd973 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
@@ -311,16 +311,13 @@ public interface BrokerResponse {
/// @return true if RLS filters were applied, false otherwise
boolean getRLSFiltersApplied();
- /// Set whether the broker rewrote an exact aggregation into its approximate
counterpart, for example
- /// `DISTINCT_COUNT` into `DISTINCT_COUNT_SMART_HLL`. The result is then
approximate, not exact.
- /// The default is a no-op so that implementations which do not track this
need no change.
- /// @param approximateFunctionApplied true if at least one function was
rewritten
- default void setApproximateFunctionApplied(boolean
approximateFunctionApplied) {
- }
-
- /// Get whether the broker rewrote an exact aggregation into its approximate
counterpart.
- /// @return true if at least one function was rewritten, false otherwise
- default boolean isApproximateFunctionApplied() {
+ /// Get whether the result is approximated: a function that can return an
exact result returned an approximate one.
+ /// For example, a SMART function such as `DISTINCT_COUNT_SMART_HLL` crossed
its conversion threshold and switched to
+ /// a sketch, including the SMART functions that the broker rewrites exact
`DISTINCT_COUNT` and `PERCENTILE` into.
+ /// Functions that are always approximate, such as `DISTINCT_COUNT_HLL`, do
not set it.
+ /// The default returns `false` so that implementations which do not track
this need no change.
+ /// @return true if at least one function returned an approximated result,
false otherwise
+ default boolean isApproximatedResult() {
return false;
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
index b38bae56605..f1fa7a5d5a1 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
@@ -57,7 +57,7 @@ import org.apache.pinot.spi.utils.JsonUtils;
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tablesQueried",
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
"realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes",
- "pools", "rlsFiltersApplied", "approximateFunctionApplied",
"groupsTrimmed", "materializedViewQueried",
+ "pools", "rlsFiltersApplied", "approximatedResult", "groupsTrimmed",
"materializedViewQueried",
"serverStats"
})
@JsonIgnoreProperties(ignoreUnknown = true)
@@ -122,7 +122,7 @@ public class BrokerResponseNative implements BrokerResponse
{
private Set<Integer> _pools = Set.of();
private boolean _rlsFiltersApplied = false;
- private boolean _approximateFunctionApplied = false;
+ private boolean _approximatedResult = false;
@Nullable
private String _materializedViewQueried;
@@ -639,16 +639,15 @@ public class BrokerResponseNative implements
BrokerResponse {
return _rlsFiltersApplied;
}
- @JsonProperty("approximateFunctionApplied")
- @Override
- public void setApproximateFunctionApplied(boolean
approximateFunctionApplied) {
- _approximateFunctionApplied = approximateFunctionApplied;
+ @JsonProperty("approximatedResult")
+ public void setApproximatedResult(boolean approximatedResult) {
+ _approximatedResult = approximatedResult;
}
- @JsonProperty("approximateFunctionApplied")
+ @JsonProperty("approximatedResult")
@Override
- public boolean isApproximateFunctionApplied() {
- return _approximateFunctionApplied;
+ public boolean isApproximatedResult() {
+ return _approximatedResult;
}
@JsonProperty("materializedViewQueried")
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
index 9919057b565..77f5b45a6fa 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
@@ -54,7 +54,7 @@ import org.apache.pinot.common.response.ProcessingException;
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tablesQueried",
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
"realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes",
- "pools", "rlsFiltersApplied", "approximateFunctionApplied",
"groupsTrimmed",
+ "pools", "rlsFiltersApplied", "approximatedResult", "groupsTrimmed",
"mseLiteLeafStageLimitReached", "mseLiteLeafStageEffectiveLimit",
"mseLiteFanOutAdjustedLimitApplied",
"responseMetadata"
})
@@ -92,7 +92,6 @@ public class BrokerResponseNativeV2 implements BrokerResponse
{
private Set<Integer> _pools = Set.of();
private boolean _rlsFiltersApplied = false;
- private boolean _approximateFunctionApplied = false;
@Nullable
private Integer _mseLiteLeafStageEffectiveLimit;
@Nullable
@@ -184,6 +183,17 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
_brokerStats.merge(StatKey.LITE_MODE_LEAF_STAGE_LIMIT_REACHED,
mseLiteLeafStageLimitReached);
}
+ /// Merged from the stats that the operators of every stage report.
+ @JsonProperty("approximatedResult")
+ @Override
+ public boolean isApproximatedResult() {
+ return _brokerStats.getBoolean(StatKey.APPROXIMATED_RESULT);
+ }
+
+ public void mergeApproximatedResult(boolean approximatedResult) {
+ _brokerStats.merge(StatKey.APPROXIMATED_RESULT, approximatedResult);
+ }
+
@JsonInclude(JsonInclude.Include.NON_NULL)
@Nullable
public Integer getMseLiteLeafStageEffectiveLimit() {
@@ -522,18 +532,6 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
return _rlsFiltersApplied;
}
- @JsonProperty("approximateFunctionApplied")
- @Override
- public void setApproximateFunctionApplied(boolean
approximateFunctionApplied) {
- _approximateFunctionApplied = approximateFunctionApplied;
- }
-
- @JsonProperty("approximateFunctionApplied")
- @Override
- public boolean isApproximateFunctionApplied() {
- return _approximateFunctionApplied;
- }
-
public void addBrokerStats(StatMap<StatKey> brokerStats) {
_brokerStats.merge(brokerStats);
}
@@ -571,7 +569,8 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
}
},
EARLY_TERMINATION_REASONS(StatMap.Type.STRING_SET),
- LITE_MODE_LEAF_STAGE_LIMIT_REACHED(StatMap.Type.BOOLEAN);
+ LITE_MODE_LEAF_STAGE_LIMIT_REACHED(StatMap.Type.BOOLEAN),
+ APPROXIMATED_RESULT(StatMap.Type.BOOLEAN);
private final StatMap.Type _type;
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
index 96565b68cad..97e10fab1f1 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
@@ -35,7 +35,7 @@ import org.apache.pinot.common.response.CursorResponse;
"offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs",
"realtimeTotalCpuTimeNs",
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tableQueries",
- "groupsTrimmed", "approximateFunctionApplied",
+ "groupsTrimmed", "approximatedResult",
// Fields specific to CursorResponse
"offset", "numRows", "cursorResultWriteTimeMs", "cursorFetchTimeMs",
"submissionTimeMs", "expirationTimeMs",
"brokerHost", "brokerPort", "bytesWritten"
@@ -60,6 +60,7 @@ public class CursorResponseNative extends
BrokerResponseNative implements Cursor
setNumRowsResultSet(response.getNumRowsResultSet());
setExceptions(response.getExceptions());
setGroupsTrimmed(response.isGroupsTrimmed());
+ setApproximatedResult(response.isApproximatedResult());
setNumGroupsLimitReached(response.isNumGroupsLimitReached());
setNumGroupsWarningLimitReached(response.isNumGroupsWarningLimitReached());
setTimeUsedMs(response.getTimeUsedMs());
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
index 7ede9c2932b..41288e492ee 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -38,6 +39,19 @@ public class BrokerResponseNativeTest {
Assert.assertEquals(actual.getResultTable(), expected.getResultTable());
}
+ /// The flag is serialized as `approximatedResult`, survives a round trip,
and is carried into a cursor response.
+ @Test
+ public void testApproximatedResult()
+ throws IOException {
+ BrokerResponseNative response = new BrokerResponseNative();
+
Assert.assertFalse(JsonUtils.stringToJsonNode(response.toJsonString()).get("approximatedResult").asBoolean());
+
+ response.setApproximatedResult(true);
+
Assert.assertTrue(JsonUtils.stringToJsonNode(response.toJsonString()).get("approximatedResult").asBoolean());
+
Assert.assertTrue(BrokerResponseNative.fromJsonString(response.toJsonString()).isApproximatedResult());
+ Assert.assertTrue(new
CursorResponseNative(response).isApproximatedResult());
+ }
+
@Test
public void testNullResponse()
throws IOException {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java
index 141367beb3e..5b711429716 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java
@@ -26,8 +26,11 @@ import it.unimi.dsi.fastutil.objects.ObjectArrayList;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.common.datatable.DataTable;
+import org.apache.pinot.common.datatable.DataTable.MetadataKey;
import org.apache.pinot.common.request.context.FilterContext;
import org.apache.pinot.common.utils.ArrayListUtils;
import org.apache.pinot.common.utils.DataSchema;
@@ -46,6 +49,9 @@ public class AggregationResultsBlock extends BaseResultsBlock
{
private final AggregationFunction[] _aggregationFunctions;
private final List<Object> _results;
private final QueryContext _queryContext;
+ /// The final results, extracted once when the server returns final results.
See [#getFinalResults()].
+ @Nullable
+ private Object[] _finalResults;
public AggregationResultsBlock(AggregationFunction[] aggregationFunctions,
List<Object> results,
QueryContext queryContext) {
@@ -101,12 +107,23 @@ public class AggregationResultsBlock extends
BaseResultsBlock {
// that the rows are consistent with the schema; otherwise an intermediate
object (e.g. a HyperLogLogPlus) would be
// left in a column typed as its final type (e.g. LONG) and fail when the
block is serialized. This mirrors the
// finalization done in getDataTable() and in GroupByCombineOperator for
the group-by case.
- int numColumns = _results.size();
- Object[] row = new Object[numColumns];
- for (int i = 0; i < numColumns; i++) {
- row[i] = _aggregationFunctions[i].extractFinalResult(_results.get(i));
+ return List.<Object[]>of(getFinalResults().clone());
+ }
+
+ /// Extracts the final results once, for [#getResultsMetadata()],
[#getRows()] and [#getDataTable()]. Synchronized
+ /// because two threads can ask at once: a streamed block reaches the
multi-stage leaf, which reads its rows, while
+ /// the server thread is still reading its metadata. Extraction can mutate
the intermediate results in place, for
+ /// example `PERCENTILE` sorts its values, so it must not run twice
concurrently.
+ private synchronized Object[] getFinalResults() {
+ if (_finalResults == null) {
+ int numColumns = _results.size();
+ Object[] finalResults = new Object[numColumns];
+ for (int i = 0; i < numColumns; i++) {
+ finalResults[i] =
_aggregationFunctions[i].extractFinalResult(_results.get(i));
+ }
+ _finalResults = finalResults;
}
- return List.<Object[]>of(row);
+ return _finalResults;
}
@Override
@@ -128,8 +145,9 @@ public class AggregationResultsBlock extends
BaseResultsBlock {
// return a null intermediate result in both modes, and their result
column type is not OBJECT.
dataTableBuilder.startRow();
if (_queryContext.isServerReturnFinalResult()) {
+ Object[] finalResults = getFinalResults();
for (int i = 0; i < numColumns; i++) {
- Object result =
_aggregationFunctions[i].extractFinalResult(_results.get(i));
+ Object result = finalResults[i];
if (result == null) {
dataTableBuilder.setNull(i);
} else {
@@ -152,6 +170,20 @@ public class AggregationResultsBlock extends
BaseResultsBlock {
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() {
+ Map<String, String> metadata = super.getResultsMetadata();
+ if (_queryContext.isServerReturnFinalResult()) {
+ getFinalResults();
+ if
(AggregationFunctionUtils.hasApproximatedResult(_aggregationFunctions)) {
+ metadata.put(MetadataKey.APPROXIMATED_RESULT.getName(), "true");
+ }
+ }
+ return metadata;
+ }
+
private void setFinalResult(DataTableBuilder dataTableBuilder,
ColumnDataType[] columnDataTypes, int index,
Object result)
throws IOException {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java
index 1ffc5b8ee93..2a4d056596b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java
@@ -35,6 +35,7 @@ import org.apache.pinot.core.data.table.IntermediateRecord;
import org.apache.pinot.core.data.table.Record;
import org.apache.pinot.core.data.table.Table;
import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
+import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
import
org.apache.pinot.core.query.aggregation.groupby.AggregationGroupByResult;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.spi.query.QueryThreadContext;
@@ -237,6 +238,11 @@ public class GroupByResultsBlock extends BaseResultsBlock {
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. Unlike the aggregation
+ // block, this is not limited to that mode: ORDER BY trimming extracts
too, also for groups it then trims away.
+ if
(AggregationFunctionUtils.hasApproximatedResult(_queryContext.getAggregationFunctions()))
{
+ metadata.put(MetadataKey.APPROXIMATED_RESULT.getName(), "true");
+ }
metadata.put(MetadataKey.NUM_RESIZES.getName(),
Integer.toString(_numResizes));
metadata.put(MetadataKey.RESIZE_TIME_MS.getName(),
Long.toString(_resizeTimeMs));
return metadata;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
index 79ca2736e8a..d0c08009466 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
@@ -263,6 +263,18 @@ public interface AggregationFunction<IntermediateResult,
FinalResult extends Com
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() {
+ return false;
+ }
+
/// Returns whether a star-tree index with the specified properties can be
used for this aggregation function.
default boolean canUseStarTree(Map<String, Object> functionParameters) {
// Implementations can override this method to perform additional checks
on the function parameters
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
index f8c240a3f5c..4c8aac474e4 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
@@ -150,6 +150,20 @@ public class AggregationFunctionUtils {
return aggregationFunction.mergeFinalResult(finalResult1, finalResult2);
}
+ /// Returns whether any of the functions has returned an approximate final
result, see
+ /// [AggregationFunction#hasApproximatedResult()].
+ public static boolean hasApproximatedResult(@Nullable AggregationFunction[]
aggregationFunctions) {
+ if (aggregationFunctions == null) {
+ return false;
+ }
+ for (AggregationFunction aggregationFunction : aggregationFunctions) {
+ if (aggregationFunction.hasApproximatedResult()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/// Creates a map from expression required by the [AggregationFunction] to
[BlockValSet] fetched from the
/// [ValueBlock].
public static Map<ExpressionContext, BlockValSet>
getBlockValSetMap(AggregationFunction aggregationFunction,
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java
index 181fc11297b..f53aff83a7e 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java
@@ -52,11 +52,26 @@ 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;
+
protected BaseDistinctCountSmartSketchAggregationFunction(ExpressionContext
expression,
boolean nullHandlingEnabled) {
super(expression, nullHandlingEnabled);
}
+ @Override
+ public boolean hasApproximatedResult() {
+ return _approximatedResult;
+ }
+
+ /// Reads before it writes, so that the threads that extract many groups
share the field instead of contending on it.
+ protected void markApproximatedResult() {
+ if (!_approximatedResult) {
+ _approximatedResult = true;
+ }
+ }
+
protected abstract int getThreshold();
protected abstract Object convertSetToSketch(Set valueSet, DataType
storedType);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunction.java
index d40686e8d2f..667084aa174 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunction.java
@@ -400,6 +400,7 @@ public class DistinctCountSmartHLLAggregationFunction
extends BaseDistinctCountS
return 0;
}
if (intermediateResult instanceof HyperLogLog) {
+ markApproximatedResult();
return (int) ((HyperLogLog) intermediateResult).cardinality();
} else {
return ((Set) intermediateResult).size();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
index a2d11c9351e..cf30ff5f30b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
@@ -349,6 +349,7 @@ public class DistinctCountSmartHLLPlusAggregationFunction
extends BaseDistinctCo
return 0;
}
if (intermediateResult instanceof HyperLogLogPlus) {
+ markApproximatedResult();
return (int) ((HyperLogLogPlus) intermediateResult).cardinality();
} else {
return ((Set) intermediateResult).size();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartULLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartULLAggregationFunction.java
index a3634cf4006..bc9ffec6e41 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartULLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartULLAggregationFunction.java
@@ -381,6 +381,7 @@ public class DistinctCountSmartULLAggregationFunction
extends BaseDistinctCountS
return 0;
}
if (intermediateResult instanceof UltraLogLog) {
+ markApproximatedResult();
return (int) Math.round(((UltraLogLog)
intermediateResult).getDistinctCountEstimate());
} else {
return ((Set) intermediateResult).size();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java
index 062c1ca9252..0d87bb22910 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java
@@ -58,6 +58,8 @@ public class PercentileSmartTDigestAggregationFunction
extends BaseSingleInputAg
private final double _percentile;
private final int _threshold;
private final int _compression;
+ /// Set by [#extractFinalResult] when it reads a digest, see
[AggregationFunction#hasApproximatedResult()].
+ private volatile boolean _approximatedResult;
public PercentileSmartTDigestAggregationFunction(List<ExpressionContext>
arguments, boolean nullHandlingEnabled) {
super(arguments.get(0), nullHandlingEnabled);
@@ -90,6 +92,11 @@ public class PercentileSmartTDigestAggregationFunction
extends BaseSingleInputAg
return _compression;
}
+ @Override
+ public boolean hasApproximatedResult() {
+ return _approximatedResult;
+ }
+
@Override
public AggregationFunctionType getType() {
return AggregationFunctionType.PERCENTILESMARTTDIGEST;
@@ -368,6 +375,10 @@ public class PercentileSmartTDigestAggregationFunction
extends BaseSingleInputAg
if (_nullHandlingEnabled && tDigest.size() == 0L) {
return null;
}
+ // Read first, so that the threads that extract many groups do not
contend on the field
+ if (!_approximatedResult) {
+ _approximatedResult = true;
+ }
return tDigest.quantile(_percentile / 100.0);
} else {
DoubleArrayList valueList = (DoubleArrayList) intermediateResult;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/BrokerReduceService.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/BrokerReduceService.java
index 05021e4bf0e..d59baf6a8d1 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/BrokerReduceService.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/BrokerReduceService.java
@@ -33,6 +33,7 @@ import
org.apache.pinot.common.response.broker.BrokerResponseNative;
import org.apache.pinot.common.response.broker.QueryProcessingException;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
import org.apache.pinot.core.query.request.context.QueryContext;
import
org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils;
import org.apache.pinot.core.transport.ServerRoutingInstance;
@@ -132,6 +133,11 @@ public class BrokerReduceService extends BaseReduceService
{
"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()))
{
+ brokerResponseNative.setApproximatedResult(true);
+ }
QueryContext queryContext;
if (brokerRequest == serverBrokerRequest) {
queryContext = serverQueryContext;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java
index 643b9fe606c..cbe7d68cc52 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java
@@ -76,6 +76,7 @@ public class ExecutionStatsAggregator {
private boolean _groupsTrimmed = false;
private boolean _numGroupsLimitReached = false;
private boolean _numGroupsWarningLimitReached = false;
+ private boolean _approximatedResult = false;
private boolean _maxRowsInDistinctReached = false;
private boolean _maxRowsWithoutChangeInDistinctReached = false;
private boolean _maxExecutionTimeInDistinctReached = false;
@@ -240,6 +241,7 @@ public class ExecutionStatsAggregator {
Boolean.parseBoolean(metadata.get(DataTable.MetadataKey.NUM_GROUPS_LIMIT_REACHED.getName()));
_numGroupsWarningLimitReached |=
Boolean.parseBoolean(metadata.get(DataTable.MetadataKey.NUM_GROUPS_WARNING_LIMIT_REACHED.getName()));
+ _approximatedResult |=
Boolean.parseBoolean(metadata.get(DataTable.MetadataKey.APPROXIMATED_RESULT.getName()));
String distinctEarlyTermination =
metadata.get(DataTable.MetadataKey.EARLY_TERMINATION_REASON.getName());
if (distinctEarlyTermination != null) {
@@ -286,6 +288,7 @@ public class ExecutionStatsAggregator {
brokerResponseNative.setGroupsTrimmed(_groupsTrimmed);
brokerResponseNative.setNumGroupsLimitReached(_numGroupsLimitReached);
brokerResponseNative.setNumGroupsWarningLimitReached(_numGroupsWarningLimitReached);
+ brokerResponseNative.setApproximatedResult(_approximatedResult);
brokerResponseNative.setMaxRowsInDistinctReached(_maxRowsInDistinctReached);
brokerResponseNative.setMaxRowsWithoutChangeInDistinctReached(_maxRowsWithoutChangeInDistinctReached);
brokerResponseNative.setMaxExecutionTimeInDistinctReached(_maxExecutionTimeInDistinctReached);
@@ -434,6 +437,9 @@ public class ExecutionStatsAggregator {
if (_numGroupsWarningLimitReached) {
metadata.put(DataTable.MetadataKey.NUM_GROUPS_WARNING_LIMIT_REACHED.getName(),
"true");
}
+ if (_approximatedResult) {
+ metadata.put(DataTable.MetadataKey.APPROXIMATED_RESULT.getName(),
"true");
+ }
// EARLY_TERMINATION_REASON: 1-string wire format ↔ 3-boolean accumulator.
aggregate() OR-reduces
// multiple per-server reasons into the three DISTINCT booleans (each
per-server DataTable carries
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlockTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlockTest.java
new file mode 100644
index 00000000000..4efeabbc9e7
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlockTest.java
@@ -0,0 +1,85 @@
+/**
+ * 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.doubles.DoubleArrayList;
+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.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
+
+
+/// Covers how [AggregationResultsBlock] extracts final results when the
server returns them.
+public class AggregationResultsBlockTest {
+
+ /// The response metadata is read before the data table is built, so a
server that returns final results has to
+ /// extract them early to report a SMART function that switched to a sketch.
+ @Test
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ public void testServerFinalResultReportsApproximatedResult()
+ throws IOException {
+ for (int threshold : new int[]{1, 10}) {
+ QueryContext queryContext = QueryContextConverterUtils.getQueryContext(
+ "SELECT DISTINCTCOUNTSMARTHLL(col, 'threshold=" + threshold + "')
FROM testTable");
+ queryContext.setServerReturnFinalResult(true);
+ AggregationFunction[] aggregationFunctions =
queryContext.getAggregationFunctions();
+ Object merged =
+ aggregationFunctions[0].merge(new IntOpenHashSet(new int[]{1}), new
IntOpenHashSet(new int[]{2}));
+ AggregationResultsBlock resultsBlock =
+ new AggregationResultsBlock(aggregationFunctions, new
ArrayList<>(List.of(merged)), queryContext);
+
+ Map<String, String> metadata = resultsBlock.getResultsMetadata();
+
assertEquals(Boolean.parseBoolean(metadata.get(MetadataKey.APPROXIMATED_RESULT.getName())),
threshold == 1);
+ DataTable dataTable = resultsBlock.getDataTable();
+ assertEquals(dataTable.getNumberOfRows(), 1);
+ assertEquals(resultsBlock.getRows().get(0)[0], dataTable.getInt(0, 0));
+ }
+ }
+
+ /// The metadata, the rows and the data table share one extraction.
Extraction can sort the values in place, and the
+ /// rows of a streamed block are read on another thread than its metadata.
+ @Test
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ public void testFinalResultsAreExtractedOnce()
+ throws IOException {
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext("SELECT PERCENTILE(col, 50) FROM
testTable");
+ queryContext.setServerReturnFinalResult(true);
+ AggregationFunction function =
spy(queryContext.getAggregationFunctions()[0]);
+ AggregationResultsBlock resultsBlock = new AggregationResultsBlock(new
AggregationFunction[]{function},
+ new ArrayList<>(List.of(new DoubleArrayList(new double[]{3.0, 1.0,
2.0}))), queryContext);
+
+ resultsBlock.getResultsMetadata();
+ assertEquals(resultsBlock.getRows().get(0)[0], 2.0);
+ assertEquals(resultsBlock.getDataTable().getDouble(0, 0), 2.0);
+ verify(function, times(1)).extractFinalResult(any());
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SmartFunctionApproximatedResultTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SmartFunctionApproximatedResultTest.java
new file mode 100644
index 00000000000..90e07643f40
--- /dev/null
+++
b/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);
+ assertEquals(function.extractFinalResult(null), Double.NEGATIVE_INFINITY);
+ assertEquals(function.extractFinalResult(new DoubleArrayList(new
double[]{1.0})), 1.0);
+ assertFalse(function.hasApproximatedResult());
+
+ Object merged = function.merge(new DoubleArrayList(new double[]{1.0}), new
DoubleArrayList(new double[]{2.0}));
+ function.extractFinalResult(merged);
+ assertTrue(function.hasApproximatedResult());
+ }
+
+ @Test
+ public void testExactFunctionKeepsTheDefault() {
+ DistinctCountAggregationFunction function = new
DistinctCountAggregationFunction(List.of(COLUMN), false);
+ function.extractFinalResult(function.merge(new IntOpenHashSet(new
int[]{1}), new IntOpenHashSet(new int[]{2})));
+ assertFalse(function.hasApproximatedResult());
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/BrokerReduceServiceTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/BrokerReduceServiceTest.java
index 9ef6a9fdcdc..a32537450da 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/BrokerReduceServiceTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/BrokerReduceServiceTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.query.reduce;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
@@ -25,12 +26,15 @@ import java.util.Map;
import org.apache.pinot.common.datatable.DataTable;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.response.broker.BrokerResponseNative;
import org.apache.pinot.common.response.broker.QueryProcessingException;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.core.common.datatable.DataTableBuilder;
import org.apache.pinot.core.common.datatable.DataTableBuilderFactory;
+import
org.apache.pinot.core.query.aggregation.function.DistinctCountSmartHLLAggregationFunction;
import org.apache.pinot.core.transport.ServerRoutingInstance;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.env.PinotConfiguration;
@@ -201,6 +205,54 @@ public class BrokerReduceServiceTest {
}
}
+ /// Each server holds one distinct value, which is exact on its own. Merging
them on the broker crosses a threshold of
+ /// one, and only that threshold, so the response reports an approximate
result for it alone.
+ @Test
+ public void testApproximatedResultFromBrokerMerge()
+ throws IOException {
+ BrokerReduceService brokerReduceService =
+ new BrokerReduceService(new
PinotConfiguration(Map.of(Broker.CONFIG_OF_MAX_REDUCE_THREADS_PER_QUERY, 2)));
+ for (int threshold : new int[]{1, 10}) {
+ String params = "threshold=" + threshold;
+ BrokerRequest brokerRequest = CalciteSqlCompiler.compileToBrokerRequest(
+ "SELECT DISTINCTCOUNTSMARTHLL(col1, '" + params + "') FROM
testTable");
+ DistinctCountSmartHLLAggregationFunction function = new
DistinctCountSmartHLLAggregationFunction(
+ List.of(ExpressionContext.forIdentifier("col1"),
ExpressionContext.forLiteral(Literal.stringValue(params))),
+ false);
+ Map<ServerRoutingInstance, DataTable> dataTableMap = new HashMap<>();
+ for (int i = 0; i < 2; i++) {
+ DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(
+ new DataSchema(new String[]{"distinctcountsmarthll(col1)"}, new
ColumnDataType[]{ColumnDataType.OBJECT}));
+ dataTableBuilder.startRow();
+ dataTableBuilder.setColumn(0, function.serializeIntermediateResult(new
IntOpenHashSet(new int[]{i})));
+ dataTableBuilder.finishRow();
+ dataTableMap.put(new ServerRoutingInstance("localhost", i,
TableType.OFFLINE), dataTableBuilder.build());
+ }
+
+ BrokerResponseNative response = reduce(brokerReduceService,
brokerRequest, dataTableMap, 10_000L);
+ assertEquals(response.getExceptions().size(), 0,
response.getExceptions().toString());
+ assertEquals(response.isApproximatedResult(), threshold == 1, params);
+ }
+ brokerReduceService.shutDown();
+ }
+
+ /// A server that extracted final results itself reports an approximate
result in its metadata.
+ @Test
+ public void testApproximatedResultReportedByServer() {
+ BrokerReduceService brokerReduceService =
+ new BrokerReduceService(new
PinotConfiguration(Map.of(Broker.CONFIG_OF_MAX_REDUCE_THREADS_PER_QUERY, 2)));
+ BrokerRequest brokerRequest =
CalciteSqlCompiler.compileToBrokerRequest("SELECT COUNT(*) FROM testTable");
+ DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(
+ new DataSchema(new String[]{"count(*)"}, new
ColumnDataType[]{ColumnDataType.LONG}));
+ DataTable dataTable = dataTableBuilder.build().toMetadataOnlyDataTable();
+
dataTable.getMetadata().put(DataTable.MetadataKey.APPROXIMATED_RESULT.getName(),
"true");
+ Map<ServerRoutingInstance, DataTable> dataTableMap = new HashMap<>();
+ dataTableMap.put(new ServerRoutingInstance("localhost", 12345,
TableType.OFFLINE), dataTable);
+
+ assertTrue(reduce(brokerReduceService, brokerRequest, dataTableMap,
10_000L).isApproximatedResult());
+ brokerReduceService.shutDown();
+ }
+
private BrokerResponseNative reduce(BrokerReduceService brokerReduceService,
BrokerRequest brokerRequest,
Map<ServerRoutingInstance, DataTable> dataTableMap, long
reduceTimeoutMs) {
try (QueryThreadContext ignore = QueryThreadContext.openForSseTest()) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregatorTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregatorTest.java
index 0759fc13d75..a577f90c495 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregatorTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregatorTest.java
@@ -29,6 +29,7 @@ import org.apache.pinot.spi.config.table.TableType;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
@@ -165,4 +166,20 @@ public class ExecutionStatsAggregatorTest {
assertFalse(flags[1]);
assertFalse(flags[2]);
}
+
+ @Test
+ public void testApproximatedResultIsOrReducedAndRoundTrips() {
+ DataTable table = emptyServerTable();
+ table.getMetadata().put(MetadataKey.APPROXIMATED_RESULT.getName(), "true");
+ ExecutionStatsAggregator aggregator = new ExecutionStatsAggregator(false);
+ aggregator.aggregate(SERVER, emptyServerTable());
+ aggregator.aggregate(SERVER, table);
+ BrokerResponseNative response = new BrokerResponseNative();
+ aggregator.setStats("testTable", response, mock(BrokerMetrics.class));
+ assertTrue(response.isApproximatedResult());
+
+ DataTable merged = emptyServerTable();
+ aggregator.setStatsOnMergedDataTable(merged);
+
assertEquals(merged.getMetadata().get(MetadataKey.APPROXIMATED_RESULT.getName()),
"true");
+ }
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java
index 67596a1072f..73bef16c83c 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java
@@ -21,6 +21,7 @@ package org.apache.pinot.integration.tests;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.File;
import java.util.List;
+import java.util.Locale;
import org.apache.commons.io.FileUtils;
import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
@@ -37,13 +38,14 @@ import static org.testng.Assert.assertEquals;
/// End-to-end test of the cluster config that rewrites exact aggregations
into their approximate counterparts. It
-/// covers the two properties unit tests cannot show: the config takes effect
on a running broker without a restart,
-/// and the rewrite is invisible to the caller apart from the flag on the
response.
+/// covers the properties unit tests cannot show: the config takes effect on a
running broker without a restart, the
+/// rewrite is invisible to the caller while the rewritten functions stay
exact, and the `approximatedResult` flag on
+/// the response reports a result that turned approximate.
public class ApproximateFunctionOverrideIntegrationTest extends
BaseClusterIntegrationTestSet {
private static final String DISTINCT_COUNT_QUERY =
- "SELECT Carrier, DISTINCTCOUNT(AirlineID) FROM mytable GROUP BY Carrier
ORDER BY Carrier LIMIT 100";
+ "SELECT Carrier, DISTINCTCOUNT(Origin) FROM mytable GROUP BY Carrier
ORDER BY Carrier LIMIT 100";
private static final String COUNT_DISTINCT_QUERY =
- "SELECT Carrier, COUNT(DISTINCT AirlineID) FROM mytable GROUP BY Carrier
ORDER BY Carrier LIMIT 100";
+ "SELECT Carrier, COUNT(DISTINCT Origin) FROM mytable GROUP BY Carrier
ORDER BY Carrier LIMIT 100";
private static final String PERCENTILE_QUERY =
"SELECT Carrier, PERCENTILE(ArrDelay, 90) FROM mytable GROUP BY Carrier
ORDER BY Carrier LIMIT 100";
@@ -98,58 +100,73 @@ public class ApproximateFunctionOverrideIntegrationTest
extends BaseClusterInteg
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");
+ waitForApproximatedResult();
+
// COUNT(DISTINCT x) is the standard SQL spelling and must be rewritten
too.
- assertApplied(postQuery(COUNT_DISTINCT_QUERY), true);
+ assertApproximated(postQuery(COUNT_DISTINCT_QUERY), true);
// The query option is the escape hatch back to exact results.
JsonNode forcedExact = postQuery("SET useApproximateFunction = false; " +
DISTINCT_COUNT_QUERY);
- assertApplied(forcedExact, false);
+ assertApproximated(forcedExact, false);
assertSameResult(exactDistinctCount, forcedExact);
- // A threshold below the cardinality has to make the answer approximate
rather than fail, keeping the same shape.
- // This is also what proves the parameters reach the servers and parse
there.
- setApproximateFunctionOverride("threshold=1");
- waitForOverride(true);
for (JsonNode exact : List.of(exactDistinctCount, exactPercentile)) {
String query = exact == exactDistinctCount ? DISTINCT_COUNT_QUERY :
PERCENTILE_QUERY;
JsonNode approximate = postQuery(query);
- assertApplied(approximate, true);
+ assertApproximated(approximate, true);
assertEquals(approximate.get("exceptions").size(), 0,
approximate.toString());
assertEquals(columnDataTypes(approximate), columnDataTypes(exact));
assertEquals(rows(approximate).size(), rows(exact).size());
}
}
+ /// The flag reports the result, however the SMART function got into the
query: a query that asks for one itself
+ /// sets it too, once the function switches to a sketch.
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testExplicitSmartFunction(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ for (int threshold : new int[]{1, 1000000}) {
+ JsonNode response = postQuery("SELECT Carrier,
DISTINCTCOUNTSMARTHLL(Origin, 'threshold=" + threshold
+ + "') FROM mytable GROUP BY Carrier ORDER BY Carrier LIMIT 100");
+ assertEquals(response.get("exceptions").size(), 0, response.toString());
+ assertApproximated(response, threshold == 1);
+ }
+ }
+
/// Pins the one visible difference the rewrite makes to a single-stage
response, so that changing it is a
/// deliberate decision. The multi-stage engine keeps the Calcite-derived
column name.
@Test
public void testSingleStageRenamesTheColumn()
throws Exception {
setUseMultiStageQueryEngine(false);
- assertEquals(columnName(postQuery(DISTINCT_COUNT_QUERY)),
"distinctcount(AirlineID)");
+ assertEquals(columnName(postQuery(DISTINCT_COUNT_QUERY)),
"distinctcount(Origin)");
setApproximateFunctionOverride("threshold=1000000");
waitForOverride(true);
- assertEquals(columnName(postQuery(DISTINCT_COUNT_QUERY)),
"distinctcountsmarthll(AirlineID)");
+ assertEquals(columnName(postQuery(DISTINCT_COUNT_QUERY)),
"distinctcountsmarthll(Origin)");
// An explicit alias is the way to keep the old name.
assertEquals(columnName(postQuery(
- "SELECT Carrier, DISTINCTCOUNT(AirlineID) AS dc FROM mytable GROUP BY
Carrier ORDER BY Carrier LIMIT 100")),
+ "SELECT Carrier, DISTINCTCOUNT(Origin) AS dc FROM mytable GROUP BY
Carrier ORDER BY Carrier LIMIT 100")),
"dc");
}
@@ -160,8 +177,8 @@ public class ApproximateFunctionOverrideIntegrationTest
extends BaseClusterInteg
assertEquals(rows(actual), rows(expected));
}
- private static void assertApplied(JsonNode response, boolean expected) {
- assertEquals(response.get("approximateFunctionApplied").asBoolean(),
expected, response.toString());
+ private static void assertApproximated(JsonNode response, boolean expected) {
+ assertEquals(response.get("approximatedResult").asBoolean(), expected,
response.toString());
}
private static JsonNode columnDataTypes(JsonNode response) {
@@ -176,6 +193,28 @@ public class ApproximateFunctionOverrideIntegrationTest
extends BaseClusterInteg
return
response.get("resultTable").get("dataSchema").get("columnNames").get(1).asText();
}
+ /// The rewrite is already on, so [#waitForOverride] would return before the
new threshold arrives. Waits until both
+ /// engines report an approximated percentile instead. The percentile params
are the last key that
+ /// [#setApproximateFunctionOverride] writes, so the distinct count params
have arrived too by then.
+ private void waitForApproximatedResult() {
+ boolean useMultiStageQueryEngine = useMultiStageQueryEngine();
+ try {
+ TestUtils.waitForCondition(aVoid -> {
+ try {
+ setUseMultiStageQueryEngine(false);
+ boolean singleStage =
postQuery(PERCENTILE_QUERY).get("approximatedResult").asBoolean();
+ setUseMultiStageQueryEngine(true);
+ boolean multiStage =
postQuery(PERCENTILE_QUERY).get("approximatedResult").asBoolean();
+ return singleStage && multiStage;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 100L, 60_000L, "Broker did not pick up the new conversion threshold");
+ } finally {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ }
+ }
+
private void setApproximateFunctionOverride(String params) {
_helixManager.getConfigAccessor().set(_clusterScope,
Broker.USE_APPROXIMATE_FUNCTION, "true");
_helixManager.getConfigAccessor().set(_clusterScope,
Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS, params);
@@ -190,13 +229,25 @@ public class ApproximateFunctionOverrideIntegrationTest
extends BaseClusterInteg
}
/// The broker picks the change up from a ZooKeeper watch, so the test waits
for it instead of restarting anything.
+ /// The response flag only reports a result that turned approximate, so the
wait checks the rewrite itself, in both
+ /// engines: the single-stage column name, and the multi-stage plan.
private void waitForOverride(boolean expected) {
- TestUtils.waitForCondition(aVoid -> {
- try {
- return
postQuery(DISTINCT_COUNT_QUERY).get("approximateFunctionApplied").asBoolean()
== expected;
- } catch (Exception e) {
- return false;
- }
- }, 100L, 60_000L, "Broker did not pick up the approximate function cluster
config");
+ boolean useMultiStageQueryEngine = useMultiStageQueryEngine();
+ try {
+ TestUtils.waitForCondition(aVoid -> {
+ try {
+ setUseMultiStageQueryEngine(false);
+ boolean singleStage =
columnName(postQuery(DISTINCT_COUNT_QUERY)).startsWith("distinctcountsmarthll");
+ setUseMultiStageQueryEngine(true);
+ boolean multiStage = postQuery("EXPLAIN PLAN FOR " +
DISTINCT_COUNT_QUERY).toString()
+ .toUpperCase(Locale.ROOT).contains("DISTINCTCOUNTSMARTHLL");
+ return singleStage == expected && multiStage == expected;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 100L, 60_000L, "Broker did not pick up the approximate function
cluster config");
+ } finally {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ }
}
}
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java
index 5f2bf560704..449bef9add7 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java
@@ -39,7 +39,6 @@ import org.apache.calcite.sql.type.ReturnTypes;
import org.apache.calcite.tools.RelBuilderFactory;
import org.apache.pinot.common.function.sql.PinotSqlAggFunction;
import org.apache.pinot.query.QueryEnvironment;
-import org.apache.pinot.query.context.PlannerContext;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.spi.utils.CommonConstants.Broker.PlannerRuleNames;
@@ -138,10 +137,6 @@ public class PinotApproximateAggregateRewriteRule extends
RelOptRule {
rewrittenAggCalls.add(rewrite(aggCall, target, argList, newInput,
aggRel.getGroupCount()));
}
- PlannerContext plannerContext =
call.getPlanner().getContext().unwrap(PlannerContext.class);
- if (plannerContext != null) {
- plannerContext.setApproximateFunctionApplied();
- }
call.transformTo(
aggRel.copy(aggRel.getTraitSet(), newInput, aggRel.getGroupSet(),
aggRel.getGroupSets(), rewrittenAggCalls));
}
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/context/PlannerContext.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/context/PlannerContext.java
index c5708d3cddf..33f95acc16f 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/context/PlannerContext.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/context/PlannerContext.java
@@ -64,10 +64,6 @@ public class PlannerContext implements AutoCloseable,
Context {
@Nullable
private final PhysicalPlannerContext _physicalPlannerContext;
private final SearchSealer _searchSealer;
- /// Set by the approximate aggregation rewrite rule when it rewrites at
least one aggregation, so that the broker
- /// can report on the response that the results are approximate. Written
during planning, which is single threaded
- /// for a query, and read after planning completes.
- private boolean _approximateFunctionApplied;
public PlannerContext(FrameworkConfig config, Prepare.CatalogReader
catalogReader, RelDataTypeFactory typeFactory,
HepProgram optProgram, HepProgram traitProgram, Map<String, String>
options, QueryEnvironment.Config envConfig,
@@ -137,16 +133,6 @@ public class PlannerContext implements AutoCloseable,
Context {
return _envConfig;
}
- /// Records that an exact aggregation was rewritten into its approximate
counterpart.
- public void setApproximateFunctionApplied() {
- _approximateFunctionApplied = true;
- }
-
- /// Returns whether any exact aggregation was rewritten into its approximate
counterpart while planning this query.
- public boolean isApproximateFunctionApplied() {
- return _approximateFunctionApplied;
- }
-
/// Unwraps this context. Returns `this` when asked for [PlannerContext] or
/// [Context], and delegates to [#_envConfig] when asked for
/// [QueryEnvironment.Config] so that existing rules remain compatible.
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java
index 4717d5a2b96..ca8e0182921 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java
@@ -41,6 +41,7 @@ import
org.apache.pinot.core.operator.docvalsets.FilteredRowBasedBlockValSet;
import org.apache.pinot.core.operator.docvalsets.RowBasedBlockValSet;
import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionFactory;
+import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
import
org.apache.pinot.core.query.aggregation.function.CountAggregationFunction;
import org.apache.pinot.core.util.DataBlockExtractUtils;
import org.apache.pinot.core.util.GroupByUtils;
@@ -240,7 +241,9 @@ public class AggregateOperator extends MultiStageOperator {
private MseBlock produceAggregatedBlock() {
if (!_isGroupBy) {
assert _aggregationExecutor != null;
- return new RowHeapDataBlock(_aggregationExecutor.getResult(),
_resultSchema, _aggFunctions);
+ List<Object[]> rows = _aggregationExecutor.getResult();
+ recordApproximatedResult();
+ return new RowHeapDataBlock(rows, _resultSchema, _aggFunctions);
} else {
assert _groupByExecutor != null;
List<Object[]> rows;
@@ -249,6 +252,7 @@ public class AggregateOperator extends MultiStageOperator {
} else {
rows = _groupByExecutor.getResult(_groupTrimSize);
}
+ recordApproximatedResult();
// Record stat before we check for limit so we can propagate to query
response
_statMap.merge(StatKey.NUM_GROUPS, _groupByExecutor.getNumGroups());
@@ -280,6 +284,14 @@ public class AggregateOperator extends MultiStageOperator {
}
}
+ /// Records a function that returned an approximate final result while the
executor extracted the results, see
+ /// [AggregationFunction#hasApproximatedResult()].
+ private void recordApproximatedResult() {
+ if (AggregationFunctionUtils.hasApproximatedResult(_aggFunctions)) {
+ _statMap.merge(StatKey.APPROXIMATED_RESULT, true);
+ }
+ }
+
@Override
public StatMap<StatKey> copyStatMaps() {
return new StatMap<>(_statMap);
@@ -502,7 +514,10 @@ public class AggregateOperator extends MultiStageOperator {
/// Allocated memory in bytes for this operator or its children in the
same stage.
ALLOCATED_MEMORY_BYTES(StatMap.Type.LONG),
/// Time spent on GC while this operator or its children in the same stage
were running.
- GC_TIME_MS(StatMap.Type.LONG);
+ GC_TIME_MS(StatMap.Type.LONG),
+ /// Whether a function that can answer exactly returned an approximate
final result, for example a SMART function
+ /// that switched to a sketch.
+ APPROXIMATED_RESULT(StatMap.Type.BOOLEAN);
private final StatMap.Type _type;
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java
index dff4f49b05d..477e119b54c 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java
@@ -394,6 +394,9 @@ public class LeafOperator extends MultiStageOperator
implements ExplainableOpera
case LITE_MODE_LEAF_STAGE_LIMIT_REACHED:
_statMap.merge(StatKey.LITE_MODE_LEAF_STAGE_LIMIT_REACHED,
Boolean.parseBoolean(entry.getValue()));
break;
+ case APPROXIMATED_RESULT:
+ _statMap.merge(StatKey.APPROXIMATED_RESULT,
Boolean.parseBoolean(entry.getValue()));
+ break;
case TIME_USED_MS:
_statMap.merge(StatKey.SSE_EXECUTION_TIME_MS,
Long.parseLong(entry.getValue()));
break;
@@ -798,7 +801,16 @@ public class LeafOperator extends MultiStageOperator
implements ExplainableOpera
/// given something to read, which is the common case. A worker with no
segment was given no work at all, which
/// is different from a worker that read segments and produced nothing:
compare this against the `parallelism`
/// the stats tree renders on this node, and against what the operators
above it report.
- NON_ACTIVE_WORKERS(StatMap.Type.INT, null);
+ NON_ACTIVE_WORKERS(StatMap.Type.INT, null),
+ /// The remote access stats of master, which this release line does not
report. They hold the same ordinals as
+ /// on master, so that the keys after them match it on the wire: StatMap
identifies keys by their ordinal.
+ NUM_REMOTE_ACCESSES(StatMap.Type.LONG, null),
+ REMOTE_ACCESS_BYTES(StatMap.Type.LONG, null),
+ NUM_DIRECT_REMOTE_ACCESSES(StatMap.Type.LONG, null),
+ DIRECT_REMOTE_ACCESS_BYTES(StatMap.Type.LONG, null),
+ /// Whether a function that can answer exactly returned an approximate
final result, when the leaf returns final
+ /// results.
+ APPROXIMATED_RESULT(StatMap.Type.BOOLEAN);
// IMPORTANT: When adding new StatKeys, make sure to either create the
same key in BrokerResponseNativeV2.StatKey or
// call the constructor that accepts a String as last argument and set it
to null.
// Otherwise the constructor will fail with an IllegalArgumentException
which will not be caught and will
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
index 88dafd2e012..39265dba560 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
@@ -368,6 +368,7 @@ public abstract class MultiStageOperator implements
Operator<MseBlock>, AutoClos
response.mergeNumGroupsLimitReached(stats.getBoolean(AggregateOperator.StatKey.NUM_GROUPS_LIMIT_REACHED));
response.mergeNumGroupsWarningLimitReached(
stats.getBoolean(AggregateOperator.StatKey.NUM_GROUPS_WARNING_LIMIT_REACHED));
+
response.mergeApproximatedResult(stats.getBoolean(AggregateOperator.StatKey.APPROXIMATED_RESULT));
response.mergeNumGroups(stats.getLong(AggregateOperator.StatKey.NUM_GROUPS));
response.mergeMaxRowsInOperator(stats.getLong(AggregateOperator.StatKey.EMITTED_ROWS));
}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/AggregateOperatorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/AggregateOperatorTest.java
index 60ab3e50356..f30304abc48 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/AggregateOperatorTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/AggregateOperatorTest.java
@@ -26,6 +26,7 @@ import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.sql.SqlKind;
import org.apache.pinot.calcite.rel.hint.PinotHintOptions;
import org.apache.pinot.common.datatable.StatMap;
+import org.apache.pinot.common.response.broker.BrokerResponseNativeV2;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.query.planner.logical.RexExpression;
@@ -44,6 +45,7 @@ import org.apache.pinot.spi.utils.CommonConstants.Server;
import org.mockito.Mock;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.apache.pinot.common.utils.DataSchema.ColumnDataType.BOOLEAN;
@@ -351,6 +353,41 @@ public class AggregateOperatorTest {
assertEquals(getAggregateOperator(context, nodeHint, 5, List.of(new
RelFieldCollation(1))).getGroupTrimSize(), 30);
}
+ @DataProvider
+ public Object[][] approximatedResultCases() {
+ return new Object[][]{{true, 1, true}, {true, 10, false}, {false, 1,
true}, {false, 10, false}};
+ }
+
+ /// A SMART function that switches to a sketch is reported in the operator
stats, which the broker reads.
+ @Test(dataProvider = "approximatedResultCases")
+ public void testApproximatedResultStat(boolean groupBy, int threshold,
boolean expected) {
+ OpChainExecutionContext context = OperatorTestUtil.getContext(Map.of());
+ List<RexExpression.FunctionCall> aggCalls = List.of(new
RexExpression.FunctionCall(INT, "DISTINCTCOUNTSMARTHLL",
+ List.of(new RexExpression.InputRef(1), new
RexExpression.Literal(STRING, "threshold=" + threshold))));
+ DataSchema inputSchema = new DataSchema(new String[]{"group", "value"},
new ColumnDataType[]{INT, INT});
+ BlockListMultiStageOperator.Builder input = new
BlockListMultiStageOperator.Builder(context, inputSchema);
+ for (int value = 0; value < 3; value++) {
+ input.addRow(1, value);
+ }
+ List<Integer> groupKeys = groupBy ? List.of(0) : List.of();
+ DataSchema resultSchema = groupBy
+ ? new DataSchema(new String[]{"group", "distinctCount"}, new
ColumnDataType[]{INT, INT})
+ : new DataSchema(new String[]{"distinctCount"}, new
ColumnDataType[]{INT});
+ AggregateOperator operator = new AggregateOperator(context,
input.buildWithEos(),
+ new AggregateNode(-1, resultSchema, PlanNode.NodeHint.EMPTY,
List.of(), aggCalls, List.of(-1), groupKeys,
+ AggType.DIRECT, false, null, 0));
+
+ assertEquals(((MseBlock.Data)
operator.nextBlock()).asRowHeap().getRows().size(), 1);
+ assertTrue(operator.nextBlock().isSuccess());
+ StatMap<AggregateOperator.StatKey> statMap =
+ OperatorTestUtil.getStatMap(AggregateOperator.StatKey.class,
operator.calculateStats());
+
assertEquals(statMap.getBoolean(AggregateOperator.StatKey.APPROXIMATED_RESULT),
expected);
+
+ BrokerResponseNativeV2 brokerResponse = new BrokerResponseNativeV2();
+ MultiStageOperator.Type.AGGREGATE.mergeInto(brokerResponse, statMap);
+ assertEquals(brokerResponse.isApproximatedResult(), expected);
+ }
+
private AggregateOperator getAggregateOperator(OpChainExecutionContext
context, PlanNode.NodeHint nodeHint, int limit,
@Nullable List<RelFieldCollation> collations) {
List<RexExpression.FunctionCall> aggCalls = List.of(getSum(new
RexExpression.InputRef(1)));
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafOperatorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafOperatorTest.java
index 237d7033ff0..64a081e6cef 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafOperatorTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafOperatorTest.java
@@ -546,6 +546,30 @@ public class LeafOperatorTest {
operator.close();
}
+ /// A leaf that returns final results reports a function that returned an
approximate result.
+ @Test
+ public void shouldPropagateApproximatedResult() {
+ DataSchema schema = new DataSchema(new String[]{"intCol"},
+ new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.INT});
+ InstanceResponseBlock metadataBlock = new InstanceResponseBlock(new
MetadataResultsBlock());
+
metadataBlock.getResponseMetadata().put(DataTable.MetadataKey.APPROXIMATED_RESULT.getName(),
"true");
+ QueryExecutor queryExecutor = mockQueryExecutor(List.of(), metadataBlock);
+ LeafOperator operator =
+ new LeafOperator(OperatorTestUtil.getTracingContext(),
mockQueryRequests(1), schema, queryExecutor,
+ _executorService);
+ _operatorRef.set(operator);
+
+ assertTrue(operator.nextBlock().isEos(), "Expected EOS after reading the
metadata block");
+ StatMap<LeafOperator.StatKey> leafStats = operator.copyStatMaps();
+ assertTrue(leafStats.getBoolean(LeafOperator.StatKey.APPROXIMATED_RESULT));
+
+ BrokerResponseNativeV2 brokerResponse = new BrokerResponseNativeV2();
+ MultiStageOperator.Type.LEAF.mergeInto(brokerResponse, leafStats);
+ assertTrue(brokerResponse.isApproximatedResult());
+
assertTrue(JsonUtils.objectToJsonNode(brokerResponse).path("approximatedResult").asBoolean());
+ operator.close();
+ }
+
@Test
public void shouldSkipNoneEarlyTerminationReason() {
assertSkippedEarlyTerminationReason(EarlyTerminationReason.NONE.name());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
