xiangfu0 commented on code in PR #19974:
URL: https://github.com/apache/pinot/pull/19974#discussion_r4329667620
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java:
##########
@@ -135,6 +141,28 @@ public void testClusterConfigAppliesWithoutRestart(boolean
useMultiStageQueryEng
}
}
+ /// The multi-stage engine turns `COUNT(DISTINCT CASE WHEN c THEN x END)`
into `COUNT(DISTINCT x) FILTER (WHERE c)`,
+ /// and the rewrite must not stop that, or the NULL of every row that fails
`c` counts as one more value. The
+ /// single-stage engine has no such conversion, but its answer must not
change either.
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testConditionalDistinctCountMatchesExact(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ JsonNode exact = postQuery(CONDITIONAL_DISTINCT_COUNT_QUERY);
+ assertApplied(exact, false);
+ if (useMultiStageQueryEngine) {
+ for (JsonNode row : rows(exact)) {
+ assertEquals(row.get(3).asLong(), 0L, exact.toString());
+ }
+ }
+
+ setApproximateFunctionOverride("threshold=1000000");
+ waitForOverride(true);
+ JsonNode approximate = postQuery(CONDITIONAL_DISTINCT_COUNT_QUERY);
Review Comment:
Neither the exact nor the approximate run checks the response's `exceptions`
array or that rows are non-empty, so an error response carrying an empty
`resultTable` passes every assertion vacuously: the `row.get(3)==0` loop never
executes and `assertSameResult` compares two equal empty arrays. A regression
that makes the filtered `DISTINCTCOUNTSMARTHLL` plan fail on servers would go
green here. The sibling test guards exactly this with
`assertEquals(response.get("exceptions").size(), 0)` at line 138 — this test
needs the same, plus a rows-non-empty assertion.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java:
##########
@@ -135,6 +141,28 @@ public void testClusterConfigAppliesWithoutRestart(boolean
useMultiStageQueryEng
}
}
+ /// The multi-stage engine turns `COUNT(DISTINCT CASE WHEN c THEN x END)`
into `COUNT(DISTINCT x) FILTER (WHERE c)`,
+ /// and the rewrite must not stop that, or the NULL of every row that fails
`c` counts as one more value. The
+ /// single-stage engine has no such conversion, but its answer must not
change either.
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testConditionalDistinctCountMatchesExact(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ JsonNode exact = postQuery(CONDITIONAL_DISTINCT_COUNT_QUERY);
+ assertApplied(exact, false);
+ if (useMultiStageQueryEngine) {
+ for (JsonNode row : rows(exact)) {
+ assertEquals(row.get(3).asLong(), 0L, exact.toString());
+ }
+ }
+
+ setApproximateFunctionOverride("threshold=1000000");
Review Comment:
`setApproximateFunctionOverride` writes three separate ZK cluster-config
keys, but `waitForOverride(true)` only polls the enabled flag. If the broker
processes the `USE_APPROXIMATE_FUNCTION` callback before the
`DISTINCT_COUNT_PARAMS` write lands, the planner sees enabled=true with empty
params, skips the param project, and plans single-arg `DISTINCTCOUNTSMARTHLL` —
the shape that was never broken — so the test passes without exercising the
two-argument path this PR fixes. (`DefaultClusterConfigChangeHandler` re-reads
the whole record per callback and `ApproximateFunctionOverrideProvider` falls
back to empty params.) Writing all three keys in one `ConfigAccessor.set(scope,
Map)` call closes the race; it never causes a spurious failure, only a silent
coverage hole.
##########
pinot-query-planner/src/test/java/org/apache/pinot/query/ApproximateAggregateRewriteTest.java:
##########
@@ -108,6 +110,35 @@ public void testAppendsParamsUnderAJoinAndASubquery() {
assertTrue(subquery.contains("threshold=23"), subquery);
}
+ /// `AggregateCaseToFilterRule` turns `COUNT(DISTINCT CASE WHEN c THEN x
END)` into `COUNT(DISTINCT x) FILTER (WHERE
+ /// c)`, so the rows that fail `c` add nothing to the count. It only matches
a single-argument call, so the rewrite
+ /// has to run after it: with the parameters appended first, the CASE stays
and its NULL counts as one more value.
+ @Test
+ public void testKeepsCaseToFilterConversion() {
+ QueryEnvironment exactEnv = newQueryEnvironment(false, "", "");
+ QueryEnvironment approximateEnv = newQueryEnvironment(true, "", "");
+ QueryEnvironment approximateWithParamsEnv = newQueryEnvironment(true,
"threshold=17", "");
+ for (String query : List.of(
Review Comment:
No test (planner or integration) covers an explicit `FILTER` clause under
the rewrite with params — neither `COUNT(DISTINCT x) FILTER (WHERE d)` nor the
`AggregateCaseToFilterRule` `filterArg >= 0` branch that ANDs a pre-existing
FILTER with the CASE condition — although the class doc now advertises that the
rewrite keeps the call's filter. Every query here derives its filter solely
from the CASE conversion, so a future change to the params-project construction
in `PinotApproximateAggregateRewriteRule.onMatch` (e.g. projecting only
referenced fields, or inserting the literal before input refs) could invalidate
`filterArg` for user-written FILTER clauses undetected. One query with an
explicit FILTER would pin it.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java:
##########
@@ -135,6 +141,28 @@ public void testClusterConfigAppliesWithoutRestart(boolean
useMultiStageQueryEng
}
}
+ /// The multi-stage engine turns `COUNT(DISTINCT CASE WHEN c THEN x END)`
into `COUNT(DISTINCT x) FILTER (WHERE c)`,
+ /// and the rewrite must not stop that, or the NULL of every row that fails
`c` counts as one more value. The
+ /// single-stage engine has no such conversion, but its answer must not
change either.
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testConditionalDistinctCountMatchesExact(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ JsonNode exact = postQuery(CONDITIONAL_DISTINCT_COUNT_QUERY);
+ assertApplied(exact, false);
+ if (useMultiStageQueryEngine) {
Review Comment:
This direct `row.get(3)==0` "counted NULL shows up as 1" assertion runs only
on the pre-override exact response, where the rewrite never fires regardless of
rule order — on un-fixed code it passes either way, since
`AggregateCaseToFilterRule` converts fine without the rewrite. Detection of the
actual regression rests solely on `assertSameResult`, which (combined with the
params-propagation race above) leaves no direct `==0` check on a response that
exercises the rewrite. Duplicating this per-row zero check on the `approximate`
response would assert the fix itself.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java:
##########
@@ -56,6 +56,11 @@
/// It must run before [PinotAggregateExchangeNodeInsertRule], which derives
the leaf-to-final intermediate result
/// format from the function name, hence `Phase.BASIC` rather than that rule's
`POST_LOGICAL`. That is also why the
/// runtime is not a valid place to do this rewrite.
+///
+/// It must also run after `AggregateCaseToFilterRule`, which only matches a
single-argument call. With the parameter
Review Comment:
The "only matches a single-argument call" rationale is incomplete for the
`DISTINCTCOUNT`/`DISTINCTCOUNTMV` spellings: their conversion goes through
Calcite's Case-A1 branch, which is additionally gated on the aggregation's
`allowsFilter()` and `skipsNullInputs()` defaults (both true on
`PinotSqlAggFunction` by inheritance) — and `skipsNullInputs()==true`
mis-describes Pinot's actual DISTINCTCOUNT null behavior and is pinned nowhere
as a contract. If `PinotSqlAggFunction` ever overrides `skipsNullInputs()` to
reflect runtime semantics (or Calcite tightens the Case-A1 gate),
`DISTINCTCOUNT(CASE ...)` silently stops converting while `COUNT(DISTINCT CASE
...)` still converts via the distinct-COUNT branch, and the two spellings
diverge by the NULL count under the override. Worth mentioning the real gate
here so a maintainer isn't steered toward arity alone.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java:
##########
@@ -46,6 +46,12 @@ public class ApproximateFunctionOverrideIntegrationTest
extends BaseClusterInteg
"SELECT Carrier, COUNT(DISTINCT AirlineID) 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";
+ // The last predicate matches no row, so the exact multi-stage count is 0
and a counted NULL would show up as 1.
+ private static final String CONDITIONAL_DISTINCT_COUNT_QUERY = "SELECT
Carrier, "
+ + "COUNT(DISTINCT CASE WHEN ArrDelay > 0 THEN Origin ELSE NULL END), "
+ + "DISTINCTCOUNT(CASE WHEN ArrDelay > 0 THEN Origin ELSE NULL END), "
Review Comment:
Nit: this query carries both spellings (`COUNT(DISTINCT CASE ...)` and
`DISTINCTCOUNT(CASE ...)`) of the same aggregate over the same expression, and
the planner test's four-query loop re-runs the same rule interaction across
GROUP BY variants that exercise no additional code path in either rule. If
trimming, note the dual spelling does pin the two distinct Calcite conversion
branches (distinct-COUNT vs Case-A1), so keep one query per spelling rather
than dropping a spelling entirely.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java:
##########
@@ -135,6 +141,28 @@ public void testClusterConfigAppliesWithoutRestart(boolean
useMultiStageQueryEng
}
}
+ /// The multi-stage engine turns `COUNT(DISTINCT CASE WHEN c THEN x END)`
into `COUNT(DISTINCT x) FILTER (WHERE c)`,
+ /// and the rewrite must not stop that, or the NULL of every row that fails
`c` counts as one more value. The
+ /// single-stage engine has no such conversion, but its answer must not
change either.
+ @Test(dataProvider = "useBothQueryEngines")
Review Comment:
Nit (CI wall-clock only): this method adds four full cluster-config override
flip/poll cycles (2 engines × set+clear, each polling every 100ms up to 60s) on
a cluster whose existing `testClusterConfigAppliesWithoutRestart` already
performs the identical `threshold=1000000` override lifecycle per engine.
Appending `CONDITIONAL_DISTINCT_COUNT_QUERY` to that test's exact/approximate
pair would add zero flips.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotQueryRuleSets.java:
##########
@@ -175,14 +175,17 @@ private PinotQueryRuleSets() {
PinotAggregateFunctionRewriteRule
.instanceWithDescription(PlannerRuleNames.AGGREGATE_FUNCTION_REWRITE),
- // Must stay ahead of PinotAggregateExchangeNodeInsertRule
(POST_LOGICAL), which fixes the leaf-to-final
- // intermediate result format from the function name.
- PinotApproximateAggregateRewriteRule.INSTANCE,
-
// convert CASE-style filtered aggregates into true filtered aggregates
// put it after AGGREGATE_REDUCE_FUNCTIONS where SUM is converted to SUM0
+ // put it before APPROXIMATE_AGGREGATE_REWRITE, which appends a
parameter argument this rule cannot match
AggregateCaseToFilterRule.Config.DEFAULT
- .withDescription(PlannerRuleNames.AGGREGATE_CASE_TO_FILTER).toRule()
+ .withDescription(PlannerRuleNames.AGGREGATE_CASE_TO_FILTER).toRule(),
+
+ // Must run after AGGREGATE_CASE_TO_FILTER: once the parameter argument
is appended, COUNT(DISTINCT CASE WHEN c
Review Comment:
Nit: the same ordering mechanism (param appended → single-argument match
fails → NULL counted) is restated near-verbatim in five places — the rule's
class doc, the comments on both rule entries here, the planner test doc, and
the integration test doc/query comment. When the mechanism changes (e.g. a
Calcite upgrade lets `AggregateCaseToFilterRule` match multi-argument calls),
all five must move in lockstep. One canonical statement (the rule's class doc)
with one-line pointers elsewhere would do the same job.
##########
pinot-query-planner/src/test/java/org/apache/pinot/query/ApproximateAggregateRewriteTest.java:
##########
@@ -130,6 +161,11 @@ private static void assertNotRewritten(String plan) {
assertFalse(plan.toLowerCase().contains("smart"), "Unexpected rewrite
in:\n" + plan);
}
+ private static void assertFilteredWithoutCase(String plan) {
+ assertTrue(plan.contains("FILTER"), "Expected a filtered aggregate in:\n"
+ plan);
Review Comment:
Keying on raw `plan.contains("FILTER")` / `!plan.contains("CASE")` over the
whole explain string is correct for today's four queries but a trap for
extension: a correctly converted `SUM(CASE ...)` plan legitimately contains
`CASE` in its null-handling final project (see
`QueryCompilationTest.testAggregateCaseToFilter`'s golden plan), and a
WHERE-bearing query would satisfy `contains("FILTER")` without the aggregate
being filtered. Matching `) FILTER $` or the aggregate-call digest would make
the helper self-contained.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotQueryRuleSets.java:
##########
@@ -175,14 +175,17 @@ private PinotQueryRuleSets() {
PinotAggregateFunctionRewriteRule
.instanceWithDescription(PlannerRuleNames.AGGREGATE_FUNCTION_REWRITE),
- // Must stay ahead of PinotAggregateExchangeNodeInsertRule
(POST_LOGICAL), which fixes the leaf-to-final
- // intermediate result format from the function name.
- PinotApproximateAggregateRewriteRule.INSTANCE,
-
// convert CASE-style filtered aggregates into true filtered aggregates
// put it after AGGREGATE_REDUCE_FUNCTIONS where SUM is converted to SUM0
+ // put it before APPROXIMATE_AGGREGATE_REWRITE, which appends a
parameter argument this rule cannot match
AggregateCaseToFilterRule.Config.DEFAULT
- .withDescription(PlannerRuleNames.AGGREGATE_CASE_TO_FILTER).toRule()
+ .withDescription(PlannerRuleNames.AGGREGATE_CASE_TO_FILTER).toRule(),
+
+ // Must run after AGGREGATE_CASE_TO_FILTER: once the parameter argument
is appended, COUNT(DISTINCT CASE WHEN c
+ // THEN x END) is no longer converted to COUNT(DISTINCT x) FILTER (WHERE
c), and the CASE's NULL counts as a
+ // value. Must stay ahead of PinotAggregateExchangeNodeInsertRule
(POST_LOGICAL), which fixes the leaf-to-final
+ // intermediate result format from the function name.
+ PinotApproximateAggregateRewriteRule.INSTANCE
Review Comment:
(Speculative, surfacing for the record.) The CASE-to-FILTER-before-rewrite
constraint is enforced only by list position in `BASIC_RULES` plus behavior
tests that run the default rule set. The `RuleSetCustomizer` SPI (ServiceLoader
plugins, and this file's own TODO to consolidate lists into customizers) can
re-seed or append `Phase.BASIC` rules with no structural check that the rewrite
stays after CASE-affecting rules — a customizer that reintroduces an
Aggregate-over-CASE shape after this rule would over-count silently with
`distinctCountParams` configured, since `AggregateCaseToFilterRule` never
reruns.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotQueryRuleSets.java:
##########
@@ -175,14 +175,17 @@ private PinotQueryRuleSets() {
PinotAggregateFunctionRewriteRule
.instanceWithDescription(PlannerRuleNames.AGGREGATE_FUNCTION_REWRITE),
- // Must stay ahead of PinotAggregateExchangeNodeInsertRule
(POST_LOGICAL), which fixes the leaf-to-final
- // intermediate result format from the function name.
- PinotApproximateAggregateRewriteRule.INSTANCE,
-
// convert CASE-style filtered aggregates into true filtered aggregates
// put it after AGGREGATE_REDUCE_FUNCTIONS where SUM is converted to SUM0
+ // put it before APPROXIMATE_AGGREGATE_REWRITE, which appends a
parameter argument this rule cannot match
AggregateCaseToFilterRule.Config.DEFAULT
- .withDescription(PlannerRuleNames.AGGREGATE_CASE_TO_FILTER).toRule()
+ .withDescription(PlannerRuleNames.AGGREGATE_CASE_TO_FILTER).toRule(),
Review Comment:
The correctness guarantee this ordering buys is silently voided by
`plannerRule_skipAggregateCaseToFilter=true` (query option) or by disabling
`AggregateCaseToFilter` in the broker's default-disabled rules: the plan
reverts to exactly the pre-PR shape where the CASE's NULL counts as one extra
value per group. Verified the bypassed plan executes without crashing — it just
over-counts with no indication the skip reintroduced the bug. A sentence in the
ordering comment documenting the bypass would be the minimal fix.
##########
pinot-query-planner/src/test/java/org/apache/pinot/query/ApproximateAggregateRewriteTest.java:
##########
@@ -108,6 +110,35 @@ public void testAppendsParamsUnderAJoinAndASubquery() {
assertTrue(subquery.contains("threshold=23"), subquery);
}
+ /// `AggregateCaseToFilterRule` turns `COUNT(DISTINCT CASE WHEN c THEN x
END)` into `COUNT(DISTINCT x) FILTER (WHERE
+ /// c)`, so the rows that fail `c` add nothing to the count. It only matches
a single-argument call, so the rewrite
+ /// has to run after it: with the parameters appended first, the CASE stays
and its NULL counts as one more value.
+ @Test
+ public void testKeepsCaseToFilterConversion() {
+ QueryEnvironment exactEnv = newQueryEnvironment(false, "", "");
+ QueryEnvironment approximateEnv = newQueryEnvironment(true, "", "");
+ QueryEnvironment approximateWithParamsEnv = newQueryEnvironment(true,
"threshold=17", "");
+ for (String query : List.of(
+ "SELECT COUNT(DISTINCT CASE WHEN col3 > 0 THEN col1 ELSE NULL END)
FROM a",
+ "SELECT col2, COUNT(DISTINCT CASE WHEN col3 > 0 THEN col1 ELSE NULL
END) FROM a GROUP BY col2",
+ "SELECT DISTINCTCOUNT(CASE WHEN col3 > 0 THEN col1 ELSE NULL END) FROM
a",
+ "SELECT col2, DISTINCTCOUNT(CASE WHEN col3 > 0 THEN col1 END) FROM a
GROUP BY col2")) {
+ String exact = explain(exactEnv, query);
+ assertFilteredWithoutCase(exact);
+
+ // Without parameters the only difference from the exact plan is the
function name.
+ String approximate = explain(approximateEnv, query);
+ assertRewritten(approximate, "distinctcountsmarthll");
+ assertEquals(approximate.replace("DISTINCTCOUNTSMARTHLL",
"DISTINCTCOUNT"), exact);
+
+ // This is the case that kept the CASE, because the parameter makes it a
two-argument call.
+ String approximateWithParams = explain(approximateWithParamsEnv, query);
+ assertRewritten(approximateWithParams, "distinctcountsmarthll");
+ assertTrue(approximateWithParams.contains("threshold=17"),
approximateWithParams);
+ assertFilteredWithoutCase(approximateWithParams);
+ }
+ }
+
/// The physical optimizer splits aggregates in its own rule, which also
keys the leaf-to-final intermediate format
/// off the function name, so the rewrite has to hold on that path too.
Review Comment:
The new filterArg + trailing-literal plan shape is never exercised under the
v2 physical optimizer: `testKeepsCaseToFilterConversion` never passes
`usePhysicalOptimizer=true` and `testRewritesUnderThePhysicalOptimizer` has no
CASE query, so `AggregatePushdownRule`'s leaf/final split of
`DISTINCTCOUNTSMARTHLL(x, 'params') FILTER (c)` is untested. Structural reading
shows it mirrors the default rule (filterArg kept at leaf, literal inlined via
`findImmediateProjects`), so this is a coverage gap rather than a known break —
one physical-optimizer variant in the new test's loop would close it.
--
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]
