diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MaxSqlAggregator.java b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MaxSqlAggregator.java index 4f29b276a544..61afe7e115cf 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MaxSqlAggregator.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MaxSqlAggregator.java @@ -44,7 +44,8 @@ Aggregation getAggregation( final String name, final AggregateCall aggregateCall, final ExprMacroTable macroTable, - final String fieldName + final String fieldName, + final boolean filteredByElseZeroRewrite ) { final ColumnType valueType = Calcites.getColumnTypeForRelDataType(aggregateCall.getType()); diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MinSqlAggregator.java b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MinSqlAggregator.java index 93b87d376b52..70d3dd1425b0 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MinSqlAggregator.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/MinSqlAggregator.java @@ -44,7 +44,8 @@ Aggregation getAggregation( final String name, final AggregateCall aggregateCall, final ExprMacroTable macroTable, - final String fieldName + final String fieldName, + final boolean filteredByElseZeroRewrite ) { final ColumnType valueType = Calcites.getColumnTypeForRelDataType(aggregateCall.getType()); diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SimpleSqlAggregator.java b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SimpleSqlAggregator.java index 5da064c285d5..e5f0d5fa4240 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SimpleSqlAggregator.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SimpleSqlAggregator.java @@ -89,13 +89,27 @@ public Aggregation toDruidAggregation( fieldName = virtualColumnRegistry.getOrCreateVirtualColumnForExpression(arg, aggregateCall.getType()); } - return getAggregation(name, aggregateCall, macroTable, fieldName); + // 'finalizeAggregations' is also set by DruidAggregateCaseToFilterRule, to indicate that the filter of this + // aggregate call comes from the D1 rewrite of SUM(CASE WHEN COND THEN value ELSE 0 END), in which case it + // is semantically a SUM0 with 0 - rather than null - as its empty result. See that rule and + // SumZeroSqlAggregator for details. + return getAggregation(name, aggregateCall, macroTable, fieldName, finalizeAggregations); } + /** + * Returns the Druid {@link Aggregation} for this aggregate call. + * + * @param filteredByElseZeroRewrite true if the filter of this call was introduced by the + * {@code DruidAggregateCaseToFilterRule} rewrite of a + * {@code SUM(CASE WHEN COND THEN value ELSE 0 END)} expression, see + * {@link #toDruidAggregation}. Implementations that do not care about that + * distinction may ignore it. + */ abstract Aggregation getAggregation( String name, AggregateCall aggregateCall, ExprMacroTable macroTable, - String fieldName + String fieldName, + boolean filteredByElseZeroRewrite ); } diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SingleValueSqlAggregator.java b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SingleValueSqlAggregator.java index 088567e7be9e..27a917eb52c2 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SingleValueSqlAggregator.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SingleValueSqlAggregator.java @@ -50,7 +50,8 @@ Aggregation getAggregation( final String name, final AggregateCall aggregateCall, final ExprMacroTable macroTable, - final String fieldName + final String fieldName, + final boolean filteredByElseZeroRewrite ) { final ColumnType valueType = Calcites.getColumnTypeForRelDataType(aggregateCall.getType()); diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SumSqlAggregator.java b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SumSqlAggregator.java index 29790ab7154a..a8ed28a22a55 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SumSqlAggregator.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SumSqlAggregator.java @@ -55,7 +55,8 @@ Aggregation getAggregation( final String name, final AggregateCall aggregateCall, final ExprMacroTable macroTable, - final String fieldName + final String fieldName, + final boolean filteredByElseZeroRewrite ) { final ColumnType valueType = Calcites.getColumnTypeForRelDataType(aggregateCall.getType()); diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SumZeroSqlAggregator.java b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SumZeroSqlAggregator.java index aaa326645b6e..adbca8a46c18 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SumZeroSqlAggregator.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/aggregation/builtin/SumZeroSqlAggregator.java @@ -19,8 +19,18 @@ package org.apache.druid.sql.calcite.aggregation.builtin; +import com.google.common.collect.Iterables; +import org.apache.calcite.rel.core.AggregateCall; import org.apache.calcite.sql.SqlAggFunction; import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.druid.math.expr.ExprMacroTable; +import org.apache.druid.query.aggregation.AggregatorFactory; +import org.apache.druid.query.aggregation.post.ExpressionPostAggregator; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.sql.calcite.aggregation.Aggregation; +import org.apache.druid.sql.calcite.planner.Calcites; + +import javax.annotation.Nullable; public class SumZeroSqlAggregator extends SumSqlAggregator { @@ -29,4 +39,49 @@ public SqlAggFunction calciteFunction() { return SqlStdOperatorTable.SUM0; } + + @Override + @Nullable + Aggregation getAggregation( + final String name, + final AggregateCall aggregateCall, + final ExprMacroTable macroTable, + final String fieldName, + final boolean filteredByElseZeroRewrite + ) + { + final ColumnType valueType = Calcites.getColumnTypeForRelDataType(aggregateCall.getType()); + if (valueType == null) { + return null; + } + + if (filteredByElseZeroRewrite) { + // This SUM0 only exists because DruidAggregateCaseToFilterRule rewrote the D1 shape + // SUM(CASE WHEN COND THEN value ELSE 0 END) into SUM0(value) FILTER (WHERE COND), and the + // native filtered aggregator returns null when the filter matches no row, while the + // original CASE expression returned 0. Restore that by wrapping the filtered sum in an + // expression post-aggregator that replaces null with 0. + // + // The post-aggregator is created here, instead of leaving it to Aggregation.filter, on + // purpose: the Aggregation returned by this method carries no aggregator factory, so + // Aggregation.filter never pushes the filter down into a FilteredAggregatorFactory. The + // resulting post-aggregator-only Aggregation is what Windowing accepts for window + // aggregations, while the equivalent factory-based post-aggregator + // ([expressionPostAgg, filteredAgg]) is rejected by Windowing.fromCalciteStuff. + return Aggregation.create( + new ExpressionPostAggregator( + name, + "nvl(\"" + fieldName + "\", 0)", + null, + valueType, + macroTable + ) + ); + } + + // Plain SQL SUM0(x) and SUM0(x) FILTER (WHERE ...): keep the native nullable factory, which + // preserves the documented null-on-empty behavior for those calls. Only the D1 rewrite above + // is required to return 0. + return Aggregation.create(createSumAggregatorFactory(valueType, name, fieldName, macroTable)); + } } diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidAggregateCaseToFilterRule.java b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidAggregateCaseToFilterRule.java index 7950f62aab3b..7cba32e0cf32 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidAggregateCaseToFilterRule.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidAggregateCaseToFilterRule.java @@ -303,9 +303,13 @@ && isIntLiteral(arg2, BigDecimal.ZERO))) { newProjects.add(arg1); newProjects.add(filter); - RelDataType newType = typeFactory.createTypeWithNullability(call.getType(), true); + // Use SUM0 instead of SUM to return 0 instead of NULL when filter never matches. + // This fixes the known inconsistency documented in the class javadoc where + // SUM(CASE WHEN COND THEN COL1 ELSE 0 END) would return NULL instead of 0 + // when no rows match the condition. + RelDataType newType = typeFactory.createTypeWithNullability(call.getType(), false); return AggregateCall.create( - call.getAggregation(), + SqlStdOperatorTable.SUM0, false, false, false, diff --git a/sql/src/test/quidem/org.apache.druid.quidem.SqlQuidemTest/filtered_sum.iq b/sql/src/test/quidem/org.apache.druid.quidem.SqlQuidemTest/filtered_sum.iq index 23f91b897ead..bd17cecb25ab 100644 --- a/sql/src/test/quidem/org.apache.druid.quidem.SqlQuidemTest/filtered_sum.iq +++ b/sql/src/test/quidem/org.apache.druid.quidem.SqlQuidemTest/filtered_sum.iq @@ -7,7 +7,7 @@ SELECT COUNT(1)FILTER(WHERE l1=-1),COUNT(1)FILTER(WHERE l1!=-1),MIN(l2) is null, +--------+--------+--------+--------+--------+ | EXPR$0 | EXPR$1 | EXPR$2 | EXPR$3 | EXPR$4 | +--------+--------+--------+--------+--------+ -| 0 | 0 | true | | | +| 0 | 0 | true | 0 | | +--------+--------+--------+--------+--------+ (1 row) @@ -18,7 +18,7 @@ SELECT COUNT(1)FILTER(WHERE l1=-1),COUNT(1)FILTER(WHERE l1!=-1),MIN(l2) is null, +--------+--------+--------+--------+--------+ | EXPR$0 | EXPR$1 | EXPR$2 | EXPR$3 | EXPR$4 | +--------+--------+--------+--------+--------+ -| 0 | 1 | false | | | +| 0 | 1 | false | 0 | | +--------+--------+--------+--------+--------+ (1 row) @@ -37,21 +37,50 @@ SELECT COUNT(1)FILTER(WHERE l1=0),COUNT(1)FILTER(WHERE l1!=0),MIN(l2) is null, !ok +-- all rows match the condition, but all matching values are null: l1 IS NULL selects +-- the three rows that have neither l1 nor l2 set. The D1 rewrite turns +-- SUM(CASE WHEN l1 IS NULL THEN l2 ELSE 0 END) into SUM0(l2) FILTER (WHERE l1 IS NULL), +-- which matches rows but has no non-null value to sum, so it returns NULL and the +-- nvl post-aggregator of the D1 rewrite reports 0. The user-written filtered SUM stays +-- on the native path and keeps the pre-rewrite NULL. +SELECT SUM(CASE WHEN l1 IS NULL THEN l2 ELSE 0 END),SUM(l2) FILTER(WHERE l1 IS NULL) FROM numfoo WHERE l1 IS NULL; ++--------+--------+ +| EXPR$0 | EXPR$1 | ++--------+--------+ +| 0 | | ++--------+--------+ +(1 row) + +!ok + +-- window form of a user-written filtered aggregate: must not pick up the D1 rewrite +-- post-aggregator, otherwise Windowing.fromCalciteStuff rejects the translation. +-- The frame is the entire partition and holds the two non-null l2 values (325323 and 0). +SELECT SUM0(l2) FILTER(WHERE l2 IS NOT NULL) OVER () FROM numfoo ORDER BY l1 LIMIT 1; ++--------+ +| EXPR$0 | ++--------+ +| 325323 | ++--------+ +(1 row) + +!ok + -- 7=7,null SELECT COUNT(1)FILTER(WHERE l1=7),COUNT(1)FILTER(WHERE l1!=7),MIN(l2) is null, SUM(CASE WHEN l1 = 7 THEN l2 ELSE 0 END),SUM(l2) FILTER(WHERE l1=7) FROM numfoo where 0 < l1 and l1 < 10; +--------+--------+--------+--------+--------+ | EXPR$0 | EXPR$1 | EXPR$2 | EXPR$3 | EXPR$4 | +--------+--------+--------+--------+--------+ -| 1 | 0 | true | | | +| 1 | 0 | true | 0 | | +--------+--------+--------+--------+--------+ (1 row) !ok -LogicalProject(EXPR$0=[$0], EXPR$1=[$1], EXPR$2=[IS NULL($2)], EXPR$3=[$3], EXPR$4=[$4]) - LogicalAggregate(group=[{}], EXPR$0=[COUNT() FILTER $0], EXPR$1=[COUNT() FILTER $1], agg#2=[MIN($2)], EXPR$3=[SUM($3)], EXPR$4=[SUM($2) FILTER $0]) - LogicalProject($f1=[IS TRUE(=($0, 7))], $f2=[IS TRUE(<>($0, 7))], l2=[$1], $f4=[CASE(=($0, 7), $1, 0:BIGINT)]) +LogicalProject(EXPR$0=[$0], EXPR$1=[$1], EXPR$2=[IS NULL($2)], EXPR$3=[CAST($3):BIGINT], EXPR$4=[$4]) + LogicalAggregate(group=[{}], EXPR$0=[COUNT() FILTER $0], EXPR$1=[COUNT() FILTER $1], agg#2=[MIN($2)], EXPR$3=[$SUM0($3) FILTER $4], EXPR$4=[SUM($2) FILTER $0]) + LogicalProject($f0=[IS TRUE(=($0, 7))], $f1=[IS TRUE(<>($0, 7))], l2=[$1], l20=[$1], $f5=[IS TRUE(=($0, 7))]) LogicalFilter(condition=[SEARCH($0, Sarg[(0L:BIGINT..10L:BIGINT)]:BIGINT)]) LogicalProject(l1=[$11], l2=[$12]) LogicalTableScan(table=[[druid, numfoo]]) @@ -98,6 +127,18 @@ SELECT COUNT(1)FILTER(WHERE l1=0),COUNT(1)FILTER(WHERE l1!=0),MIN(l2) is null, !ok +-- all rows match the condition, but all matching values are null (control: rewrite off, +-- so the pre-rewrite expression is evaluated natively and both columns stay NULL) +SELECT SUM(CASE WHEN l1 IS NULL THEN l2 ELSE 0 END),SUM(l2) FILTER(WHERE l1 IS NULL) FROM numfoo WHERE l1 IS NULL; ++--------+--------+ +| EXPR$0 | EXPR$1 | ++--------+--------+ +| | | ++--------+--------+ +(1 row) + +!ok + -- 7=7,null SELECT COUNT(1)FILTER(WHERE l1=7),COUNT(1)FILTER(WHERE l1!=7),MIN(l2) is null, SUM(CASE WHEN l1 = 7 THEN l2 ELSE 0 END),SUM(l2) FILTER(WHERE l1=7) FROM numfoo where 0 < l1 and l1 < 10;