You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "lowka (via GitHub)" <gi...@apache.org> on 2024/03/14 07:32:47 UTC
[PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
lowka opened a new pull request, #3413:
URL: https://github.com/apache/ignite-3/pull/3413
---
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533835397
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
We need an additional projection because COUNT returns a BIGINT, but SUM0 that returns a DECIMAL requires a DECIMAL type . Otherwise we get an error:
> java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.math.BigDecimal (java.lang.Long and java.math.BigDecimal are in module java.base of loader 'bootstrap')
at org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators$DecimalSumEmptyIsZero.add(Accumulators.java:538) ~[main/:?]
Plan w/o projection
> Query: SELECT AVG(dec4_2_col) FROM numbers
> Plan :
Project (EXPR$0=[CASE(=($0, 0:DECIMAL(32767, 0)), null:DECIMAL(4, 2), DECIMAL_DIVIDE($0, $1, 4, 2))]) row=(DECIMAL(4, 2) EXPR$0)
ReduceHashAggregate (group=[{}], AVG_SUM0=[SUM($0)], AVG_SUM00=[$SUM0($1)]) row=(DECIMAL(32767, 2) f0_0, DECIMAL(32767, 0) f1_1)
Exchange row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)]
MapHashAggregate (group=[{}], AVG_SUM0=[SUM($0)], AVG_COUNT0=[COUNT($0)]) row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
TableScan traits=IGNITE.random.[], row=(DECIMAL(4, 2) DEC4_2_COL)
Alternatively, we can introduce an accumulator operator that that accepts BIGINT and returns DECIMAL.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1535072370
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/HashAggregateConverterRule.java:
##########
@@ -121,13 +124,26 @@ public IgniteRel makeMapAgg(RelOptCluster cluster, RelNode input, ImmutableBitSe
}
@Override
- public IgniteRel makeReduceAgg(RelOptCluster cluster, RelNode map, ImmutableBitSet groupSet,
+ public IgniteRel makeProject(RelOptCluster cluster, RelNode input, List<RexNode> reduceInputExprs,
+ RelDataType projectRowType) {
+
+ return new IgniteProject(
+ agg.getCluster(),
+ outTrait.replace(IgniteDistributions.single()),
+ convert(input, inTrait.replace(IgniteDistributions.single())),
Review Comment:
Sure, Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533258895
##########
modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AggregatePlannerTest.java:
##########
@@ -536,7 +537,21 @@ public void countDistinctGroupSetHash() throws Exception {
checkCountDistinctHash(TestCase.CASE_24_1B);
checkCountDistinctHash(TestCase.CASE_24_1D);
- Predicate<RelNode> colocated = nodeOrAnyChild(isInstanceOf(IgniteReduceSortAggregate.class)
+ Predicate<RelNode> hash = nodeOrAnyChild(isInstanceOf(IgniteReduceHashAggregate.class)
+ .and(hasNoGroupSets(IgniteReduceHashAggregate::getGroupSets))
+ .and(input(isInstanceOf(IgniteExchange.class)
+ .and(hasDistribution(single())
+ .and(input(isInstanceOf(IgniteMapHashAggregate.class)
+ .and(hasNoGroupSets(IgniteMapHashAggregate::getGroupSets))
+ .and(input(isInstanceOf(IgniteColocatedHashAggregate.class)
+ .and(hasGroupSets(IgniteColocatedHashAggregate::getGroupSets, 1))
+ ))
+ ))
+ )
+ ))
+ );
+
+ Predicate<RelNode> sort = nodeOrAnyChild(isInstanceOf(IgniteReduceSortAggregate.class)
Review Comment:
Both sort and hash map reduce plan have the same cost.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1535067252
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/PlanUtils.java:
##########
@@ -127,9 +127,17 @@ private static void addAccumulatorFields(IgniteTypeFactory typeFactory, List<Agg
for (int i = 0; i < aggregateCalls.size(); i++) {
AggregateCall call = aggregateCalls.get(i);
-
Accumulator acc = accumulators.accumulatorFactory(call).get();
- RelDataType fieldType = acc.returnType(typeFactory);
+ RelDataType fieldType;
+ // For a decimal type Accumulator::returnType returns a type with default precision and scale,
+ // that can cause precision loss when a tuple is sent over the wire by an exchanger/outbox.
+ // Outbox uses its input type as wire format, so if a scale is 0, then the scale is lost
+ // (see Outbox::sendBatch -> RowHandler::toBinaryTuple -> BinaryTupleBuilder::appendDecimalNotNull).
Review Comment:
Because implementation of `Accumulator` is precision/scale independent and it relies on post-computation type conversion to adjust its result. This keeps accumulator logic simple and allows to reuse existing type conversion functions defined by runtime (so type conversion work the same everywhere).
The only way to figure out a type of an aggregate function that runtime is going to use is to call `Accumulator::returnType`, but since it returns a type without default precision and scale, and default scale is 0, the scale can be lost during serialisation.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1530463054
##########
modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/MapReduceHashAggregatePlannerTest.java:
##########
@@ -476,42 +473,42 @@ public void twoPhaseCountAgg() throws Exception {
assertPlan(TestCase.CASE_22, nonColocated, disableRules);
assertPlan(TestCase.CASE_22A, nonColocated, disableRules);
-
- Predicate<RelNode> colocated = hasChildThat(isInstanceOf(IgniteReduceHashAggregate.class)
Review Comment:
Identical to `Predicate<RelNode> nonColocated = ...`
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533835397
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
We need an additional projection because COUNT returns a BIGINT, but SUM0 requires a DECIMAL time. Otherwise we get an error:
> java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.math.BigDecimal (java.lang.Long and java.math.BigDecimal are in module java.base of loader 'bootstrap')
at org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators$DecimalSumEmptyIsZero.add(Accumulators.java:538) ~[main/:?]
Plan w/o projection
> Query: SELECT AVG(dec4_2_col) FROM numbers
> Plan :
Project (EXPR$0=[CASE(=($0, 0:DECIMAL(32767, 0)), null:DECIMAL(4, 2), DECIMAL_DIVIDE($0, $1, 4, 2))]) traits=IGNITE.single.[], row=(DECIMAL(4, 2) EXPR$0)
ReduceHashAggregate (group=[{}], AVG_SUM0=[SUM($0)], AVG_SUM00=[$SUM0($1)]) traits=IGNITE.single.[], row=(DECIMAL(32767, 2) f0_0, DECIMAL(32767, 0) f1_1)
Exchange traits=IGNITE.single.[], row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)]
MapHashAggregate traits=IGNITE.random.[], row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
TableScan traits=IGNITE.random.[], row=(DECIMAL(4, 2) DEC4_2_COL)
Alternatively, we can introduce an accumulator operator that that accepts BIGINT and returns DECIMAL.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1530463054
##########
modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/MapReduceHashAggregatePlannerTest.java:
##########
@@ -476,42 +473,42 @@ public void twoPhaseCountAgg() throws Exception {
assertPlan(TestCase.CASE_22, nonColocated, disableRules);
assertPlan(TestCase.CASE_22A, nonColocated, disableRules);
-
- Predicate<RelNode> colocated = hasChildThat(isInstanceOf(IgniteReduceHashAggregate.class)
Review Comment:
Identical to `Predicate<RelNode> nonColocated = ...`, so it can be removed and `Predicate<RelNode> colocated` can be used instead.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1535067252
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/PlanUtils.java:
##########
@@ -127,9 +127,17 @@ private static void addAccumulatorFields(IgniteTypeFactory typeFactory, List<Agg
for (int i = 0; i < aggregateCalls.size(); i++) {
AggregateCall call = aggregateCalls.get(i);
-
Accumulator acc = accumulators.accumulatorFactory(call).get();
- RelDataType fieldType = acc.returnType(typeFactory);
+ RelDataType fieldType;
+ // For a decimal type Accumulator::returnType returns a type with default precision and scale,
+ // that can cause precision loss when a tuple is sent over the wire by an exchanger/outbox.
+ // Outbox uses its input type as wire format, so if a scale is 0, then the scale is lost
+ // (see Outbox::sendBatch -> RowHandler::toBinaryTuple -> BinaryTupleBuilder::appendDecimalNotNull).
Review Comment:
Because implementation of `Accumulator` is precision/scale independent and it relies on post-computation type conversion to adjust its result. This keeps accumulator logic simple and allows to reuse existing type conversion functions defined by runtime (so type conversion work the same everywhere).
The only way to figure out a type of an aggregate function that runtime is going to use is to call `Accumulator::returnType`, but since it returns a type without default precision and scale, and default scale is 0, the scale can be lost during serialisation.
There is no output conversion on MAP phase, and there is no input conversion on REDUCE phase, so one needs to preserve correct scale.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533835397
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
We need an additional projection because COUNT returns a BIGINT, but SUM0 requires a DECIMAL time. Otherwise we get an error:
> java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.math.BigDecimal (java.lang.Long and java.math.BigDecimal are in module java.base of loader 'bootstrap')
at org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators$DecimalSumEmptyIsZero.add(Accumulators.java:538) ~[main/:?]
Plan w/o projection
> Query: SELECT AVG(dec4_2_col) FROM numbers
> Plan :
Project (EXPR$0=[CASE(=($0, 0:DECIMAL(32767, 0)), null:DECIMAL(4, 2), DECIMAL_DIVIDE($0, $1, 4, 2))]) traits=IGNITE.single.[], row=(DECIMAL(4, 2) EXPR$0)
ReduceHashAggregate (group=[{}], AVG_SUM0=[SUM($0)], AVG_SUM00=[$SUM0($1)]) traits=IGNITE.single.[], row=(DECIMAL(32767, 2) f0_0, DECIMAL(32767, 0) f1_1)
Exchange traits=IGNITE.single.[], row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)]
MapHashAggregate traits=IGNITE.random.[], row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
TableScan traits=IGNITE.random.[], row=(DECIMAL(4, 2) DEC4_2_COL)
Alternatively, we can introduce an accumulator operator that that accepts BIGINT and returns DECIMAL.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533258895
##########
modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AggregatePlannerTest.java:
##########
@@ -536,7 +537,21 @@ public void countDistinctGroupSetHash() throws Exception {
checkCountDistinctHash(TestCase.CASE_24_1B);
checkCountDistinctHash(TestCase.CASE_24_1D);
- Predicate<RelNode> colocated = nodeOrAnyChild(isInstanceOf(IgniteReduceSortAggregate.class)
+ Predicate<RelNode> hash = nodeOrAnyChild(isInstanceOf(IgniteReduceHashAggregate.class)
+ .and(hasNoGroupSets(IgniteReduceHashAggregate::getGroupSets))
+ .and(input(isInstanceOf(IgniteExchange.class)
+ .and(hasDistribution(single())
+ .and(input(isInstanceOf(IgniteMapHashAggregate.class)
+ .and(hasNoGroupSets(IgniteMapHashAggregate::getGroupSets))
+ .and(input(isInstanceOf(IgniteColocatedHashAggregate.class)
+ .and(hasGroupSets(IgniteColocatedHashAggregate::getGroupSets, 1))
+ ))
+ ))
+ )
+ ))
+ );
+
+ Predicate<RelNode> sort = nodeOrAnyChild(isInstanceOf(IgniteReduceSortAggregate.class)
Review Comment:
Both sort and hash map reduce plan have the same cost.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533835397
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
We need an additional projection because COUNT returns a BIGINT, but SUM0 requires a DECIMAL time. Otherwise we get an error:
> java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.math.BigDecimal (java.lang.Long and java.math.BigDecimal are in module java.base of loader 'bootstrap')
at org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators$DecimalSumEmptyIsZero.add(Accumulators.java:538) ~[main/:?]
Plan w/o projection
> Query: SELECT AVG(dec4_2_col) FROM numbers
> Plan :
Project (EXPR$0=[CASE(=($0, 0:DECIMAL(32767, 0)), null:DECIMAL(4, 2), DECIMAL_DIVIDE($0, $1, 4, 2))]) row=(DECIMAL(4, 2) EXPR$0)
ReduceHashAggregate (group=[{}], AVG_SUM0=[SUM($0)], AVG_SUM00=[$SUM0($1)]) row=(DECIMAL(32767, 2) f0_0, DECIMAL(32767, 0) f1_1)
Exchange row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)]
MapHashAggregate (group=[{}], AVG_SUM0=[SUM($0)], AVG_COUNT0=[COUNT($0)]) row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
TableScan traits=IGNITE.random.[], row=(DECIMAL(4, 2) DEC4_2_COL)
Alternatively, we can introduce an accumulator operator that that accepts BIGINT and returns DECIMAL.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533835397
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
We need an additional projection because COUNT returns a BIGINT, but SUM0 requires a DECIMAL time. Otherwise we get an error:
> java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.math.BigDecimal (java.lang.Long and java.math.BigDecimal are in module java.base of loader 'bootstrap')
at org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators$DecimalSumEmptyIsZero.add(Accumulators.java:538) ~[main/:?]
Plan w/o projection
> Query: SELECT AVG(dec4_2_col) FROM numbers
> Plan :
Project (EXPR$0=[CASE(=($0, 0:DECIMAL(32767, 0)), null:DECIMAL(4, 2), DECIMAL_DIVIDE($0, $1, 4, 2))]) traits=IGNITE.single.[], row=(DECIMAL(4, 2) EXPR$0)
ReduceHashAggregate traits=IGNITE.single.[], row=(DECIMAL(32767, 2) f0_0, DECIMAL(32767, 0) f1_1)
Exchange traits=IGNITE.single.[], row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)]
MapHashAggregate traits=IGNITE.random.[], row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
TableScan traits=IGNITE.random.[], row=(DECIMAL(4, 2) DEC4_2_COL)
Alternatively, we can introduce an accumulator operator that that accepts BIGINT and returns DECIMAL.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533835397
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
We need an additional projection because COUNT returns a BIGINT, but SUM0 requires a DECIMAL time. Otherwise we get an error:
> java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.math.BigDecimal (java.lang.Long and java.math.BigDecimal are in module java.base of loader 'bootstrap')
at org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators$DecimalSumEmptyIsZero.add(Accumulators.java:538) ~[main/:?]
Plan w/o projection
> Query: SELECT AVG(dec4_2_col) FROM numbers
> Plan :
Project traits=IGNITE.single.[], row=(DECIMAL(4, 2) EXPR$0)
ReduceHashAggregate traits=IGNITE.single.[], row=(DECIMAL(32767, 2) f0_0, DECIMAL(32767, 0) f1_1)
Exchange traits=IGNITE.single.[], row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)]
MapHashAggregate traits=IGNITE.random.[], row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
TableScan traits=IGNITE.random.[], row=(DECIMAL(4, 2) DEC4_2_COL)
Alternatively, we can introduce an accumulator operator that that accepts BIGINT and returns DECIMAL.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1530463054
##########
modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/MapReduceHashAggregatePlannerTest.java:
##########
@@ -476,42 +473,42 @@ public void twoPhaseCountAgg() throws Exception {
assertPlan(TestCase.CASE_22, nonColocated, disableRules);
assertPlan(TestCase.CASE_22A, nonColocated, disableRules);
-
- Predicate<RelNode> colocated = hasChildThat(isInstanceOf(IgniteReduceHashAggregate.class)
Review Comment:
Identical to `Predicate<RelNode> nonColocated = ...`, so it can be removed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "korlov42 (via GitHub)" <gi...@apache.org>.
korlov42 commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533745061
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
why do we need an additional projection in between map and reduce nodes?
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/PlanUtils.java:
##########
@@ -127,9 +127,17 @@ private static void addAccumulatorFields(IgniteTypeFactory typeFactory, List<Agg
for (int i = 0; i < aggregateCalls.size(); i++) {
AggregateCall call = aggregateCalls.get(i);
-
Accumulator acc = accumulators.accumulatorFactory(call).get();
- RelDataType fieldType = acc.returnType(typeFactory);
+ RelDataType fieldType;
+ // For a decimal type Accumulator::returnType returns a type with default precision and scale,
+ // that can cause precision loss when a tuple is sent over the wire by an exchanger/outbox.
+ // Outbox uses its input type as wire format, so if a scale is 0, then the scale is lost
+ // (see Outbox::sendBatch -> RowHandler::toBinaryTuple -> BinaryTupleBuilder::appendDecimalNotNull).
Review Comment:
not sure I fully understand the reasoning... If scale is 0, why there is something after period then?
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -356,10 +486,190 @@ private static MapReduceAgg createSimpleAgg(AggregateCall call, int reduceArgume
return new MapReduceAgg(argList, call, reduceCall, USE_INPUT_FIELD);
}
+ /**
+ * TODO: Rename to MakeExpr.
+ * Produces intermediate expressions that modify results of MAP/REDUCE aggregate.
+ * For example: after splitting a function into a MAP aggregate and REDUCE aggregate it is necessary to add casts to
+ * output of a REDUCE phase aggregate.
+ *
+ * <p>In order to avoid creating unnecessary projections, use {@link MapReduceAggregates#USE_INPUT_FIELD}.
+ */
@FunctionalInterface
private interface MakeReduceExpr {
- /** Creates an expression that produces result of REDUCE phase of an aggregate. */
+ /**
+ * Creates an expression that applies a performs computation (e.g. applies some function, adds a cast)
+ * on {@code args} fields of input relation.
+ *
+ * @param rexBuilder Expression builder.
+ * @param input Input relation.
+ * @param args Arguments.
+ * @param typeFactory Type factory.
+ *
+ * @return Expression.
+ */
RexNode makeExpr(RexBuilder rexBuilder, RelNode input, List<Integer> args, IgniteTypeFactory typeFactory);
}
+
+ private static MapReduceAgg createAvgAgg(
+ RelOptCluster cluster,
+ AggregateCall call,
+ int reduceArgumentOffset,
+ RelDataType inputType,
+ boolean canBeNull
+ ) {
+ RelDataTypeFactory tf = cluster.getTypeFactory();
+ RelDataTypeSystem typeSystem = tf.getTypeSystem();
+
+ RelDataType fieldType = inputType.getFieldList().get(call.getArgList().get(0)).getType();
+
+ // In case of AVG(NULL) return a simple version of an aggregate, because result is always NULL.
+ if (fieldType.getSqlTypeName() == SqlTypeName.NULL) {
+ return createSimpleAgg(call, reduceArgumentOffset);
+ }
+
+ // AVG(x) : SUM(x)/COUNT0(x)
+ // MAP : SUM(x) / COUNT(x)
+
+ // SUM(x) as s
+ RelDataType mapSumType = typeSystem.deriveSumType(tf, fieldType);
+ if (canBeNull) {
+ mapSumType = tf.createTypeWithNullability(mapSumType, true);
+ }
+
+ AggregateCall mapSum0 = AggregateCall.create(
+ SqlStdOperatorTable.SUM,
+ call.isDistinct(),
+ call.isApproximate(),
+ call.ignoreNulls(),
+ ImmutableList.of(),
+ call.getArgList(),
+ call.filterArg,
+ null,
+ call.collation,
+ mapSumType,
+ "AVG" + reduceArgumentOffset + "_MAP_SUM");
+
+ // COUNT(x) as c
+ RelDataType mapCountType = tf.createSqlType(SqlTypeName.BIGINT);
+
+ AggregateCall mapCount0 = AggregateCall.create(
+ SqlStdOperatorTable.COUNT,
+ call.isDistinct(),
+ call.isApproximate(),
+ call.ignoreNulls(),
+ ImmutableList.of(),
+ call.getArgList(),
+ call.filterArg,
+ null,
+ call.collation,
+ mapCountType,
+ "AVG" + reduceArgumentOffset + "_MAP_COUNT");
+
+ // REDUCE : SUM(s) as reduce_sum, SUM0(c) as reduce_count
+ List<Integer> reduceSumArgs = List.of(reduceArgumentOffset);
+
+ // SUM0(s)
+ RelDataType reduceSumType = typeSystem.deriveSumType(tf, mapSumType);
+ if (canBeNull) {
+ reduceSumType = tf.createTypeWithNullability(reduceSumType, true);
+ }
+
+ AggregateCall reduceSum0 = AggregateCall.create(
+ SqlStdOperatorTable.SUM,
+ call.isDistinct(),
+ call.isApproximate(),
+ call.ignoreNulls(),
+ ImmutableList.of(),
+ reduceSumArgs,
+ // there is no filtering on REDUCE phase
+ -1,
+ null,
+ call.collation,
+ reduceSumType,
+ "AVG" + reduceArgumentOffset + "_RED_VAL_SUM");
Review Comment:
such a name pattern seems too verbose. Let's have a look at output of EXPLAIN:
```
ReduceHashAggregate(group=[{0, 1}], SUM_QTY=[SUM($2)], SUM_BASE_PRICE=[SUM($3)], SUM_DISC_PRICE=[SUM($4)], SUM_CHARGE=[SUM($5)], AVG6_RED_VAL_SUM=[SUM($6)], AVG6_RED_COUNT_SUM=[$SUM0($7)], AVG8_RED_VAL_SUM=[SUM($8)], AVG8_RED_COUNT_SUM=[$SUM0($9)], AVG10_RED_VAL_SUM=[SUM($10)], AVG10_RED_COUNT_SUM=[$SUM0($11)], COUNT_12_MAP_SUM=[$SUM0($12)]), id = 211
```
* AVG -- this is an aggregate node, so everything is agg call.
* RED -- again, I can derive this from node itself
* VAL -- have no idea why do we need this
* SUM -- seems like this relates to the aggregating function
Do we really need all this parts, or we may short this to something like `"_SUM" + reduceArgumentOffset`? WDYT?
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/exp/IgniteSqlFunctions.java:
##########
@@ -526,6 +536,8 @@ private static BigDecimal convertToBigDecimal(Number value) {
return dec;
}
+
Review Comment:
unnecessary change
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/HashAggregateConverterRule.java:
##########
@@ -121,13 +124,26 @@ public IgniteRel makeMapAgg(RelOptCluster cluster, RelNode input, ImmutableBitSe
}
@Override
- public IgniteRel makeReduceAgg(RelOptCluster cluster, RelNode map, ImmutableBitSet groupSet,
+ public IgniteRel makeProject(RelOptCluster cluster, RelNode input, List<RexNode> reduceInputExprs,
+ RelDataType projectRowType) {
+
+ return new IgniteProject(
+ agg.getCluster(),
+ outTrait.replace(IgniteDistributions.single()),
+ convert(input, inTrait.replace(IgniteDistributions.single())),
Review Comment:
does it make sense to reuse traits of reduce node?
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533835397
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
We need an additional projection because COUNT returns a BIGINT, but SUM0 that returns a DECIMAL requires a DECIMAL type as an argument. Otherwise we get an error:
> java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.math.BigDecimal (java.lang.Long and java.math.BigDecimal are in module java.base of loader 'bootstrap')
at org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators$DecimalSumEmptyIsZero.add(Accumulators.java:538) ~[main/:?]
Plan w/o projection
> Query: SELECT AVG(dec4_2_col) FROM numbers
> Plan :
Project (EXPR$0=[CASE(=($0, 0:DECIMAL(32767, 0)), null:DECIMAL(4, 2), DECIMAL_DIVIDE($0, $1, 4, 2))]) row=(DECIMAL(4, 2) EXPR$0)
ReduceHashAggregate (group=[{}], AVG_SUM0=[SUM($0)], AVG_SUM00=[$SUM0($1)]) row=(DECIMAL(32767, 2) f0_0, DECIMAL(32767, 0) f1_1)
Exchange row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)]
MapHashAggregate (group=[{}], AVG_SUM0=[SUM($0)], AVG_COUNT0=[COUNT($0)]) row=(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
TableScan traits=IGNITE.random.[], row=(DECIMAL(4, 2) DEC4_2_COL)
Alternatively, we can introduce an accumulator operator that that accepts BIGINT and returns DECIMAL.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#issuecomment-2016769527
> [IGNITE-21827: Sql. Union returns incorrect result for DECIMAL and INT…
Wrong message, caused by switching between branches.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "korlov42 (via GitHub)" <gi...@apache.org>.
korlov42 merged PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1530450989
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java:
##########
@@ -744,7 +744,10 @@ public Node<RowT> visit(IgniteReduceHashAggregate rel) {
RelDataType rowType = rel.getRowType();
Supplier<List<AccumulatorWrapper<RowT>>> accFactory = expressionFactory.accumulatorsFactory(
- type, rel.getAggregateCalls(), null);
Review Comment:
`IgniteReduceHashAggregate / IgniteReduceSortAggregate` should have the same accFactory setup.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1535067252
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/PlanUtils.java:
##########
@@ -127,9 +127,17 @@ private static void addAccumulatorFields(IgniteTypeFactory typeFactory, List<Agg
for (int i = 0; i < aggregateCalls.size(); i++) {
AggregateCall call = aggregateCalls.get(i);
-
Accumulator acc = accumulators.accumulatorFactory(call).get();
- RelDataType fieldType = acc.returnType(typeFactory);
+ RelDataType fieldType;
+ // For a decimal type Accumulator::returnType returns a type with default precision and scale,
+ // that can cause precision loss when a tuple is sent over the wire by an exchanger/outbox.
+ // Outbox uses its input type as wire format, so if a scale is 0, then the scale is lost
+ // (see Outbox::sendBatch -> RowHandler::toBinaryTuple -> BinaryTupleBuilder::appendDecimalNotNull).
Review Comment:
Because implementation of `Accumulator` is precision/scale independent and it relies on post-computation type conversion to adjust its result. This keeps accumulator logic simple and allows to reuse existing type conversion functions defined by runtime (so type conversion work the same everywhere).
The only way to figure out a type of an aggregate function that runtime is going to use (so we can provide matching MAP / REDUCE with matching types) is to call `Accumulator::returnType`, but since it returns a type without default precision and scale, and default scale is 0, the scale can be lost during serialisation.
There is no output conversion on MAP phase, and there is no input conversion on REDUCE phase, so one needs to preserve correct scale.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1535067252
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/PlanUtils.java:
##########
@@ -127,9 +127,17 @@ private static void addAccumulatorFields(IgniteTypeFactory typeFactory, List<Agg
for (int i = 0; i < aggregateCalls.size(); i++) {
AggregateCall call = aggregateCalls.get(i);
-
Accumulator acc = accumulators.accumulatorFactory(call).get();
- RelDataType fieldType = acc.returnType(typeFactory);
+ RelDataType fieldType;
+ // For a decimal type Accumulator::returnType returns a type with default precision and scale,
+ // that can cause precision loss when a tuple is sent over the wire by an exchanger/outbox.
+ // Outbox uses its input type as wire format, so if a scale is 0, then the scale is lost
+ // (see Outbox::sendBatch -> RowHandler::toBinaryTuple -> BinaryTupleBuilder::appendDecimalNotNull).
Review Comment:
Because implementation of `Accumulator` is precision/scale independent and it relies on post-computation type conversion to adjust its result. This keeps accumulator logic simple and allows to reuse existing type conversion functions defined by runtime (so type conversion work the same everywhere).
c = convertInput(a)
result = compute(c)
convertOutput(result)
The only way to figure out a type of an aggregate function that runtime is going to use is to call `Accumulator::returnType`, but since it returns a type without scale, the scale can be lost during serialisation.
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/PlanUtils.java:
##########
@@ -127,9 +127,17 @@ private static void addAccumulatorFields(IgniteTypeFactory typeFactory, List<Agg
for (int i = 0; i < aggregateCalls.size(); i++) {
AggregateCall call = aggregateCalls.get(i);
-
Accumulator acc = accumulators.accumulatorFactory(call).get();
- RelDataType fieldType = acc.returnType(typeFactory);
+ RelDataType fieldType;
+ // For a decimal type Accumulator::returnType returns a type with default precision and scale,
+ // that can cause precision loss when a tuple is sent over the wire by an exchanger/outbox.
+ // Outbox uses its input type as wire format, so if a scale is 0, then the scale is lost
+ // (see Outbox::sendBatch -> RowHandler::toBinaryTuple -> BinaryTupleBuilder::appendDecimalNotNull).
Review Comment:
Because implementation of `Accumulator` is precision/scale independent and it relies on post-computation type conversion to adjust its result. This keeps accumulator logic simple and allows to reuse existing type conversion functions defined by runtime (so type conversion work the same everywhere).
The only way to figure out a type of an aggregate function that runtime is going to use is to call `Accumulator::returnType`, but since it returns a type without scale, the scale can be lost during serialisation.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "AMashenkov (via GitHub)" <gi...@apache.org>.
AMashenkov commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1532074928
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -356,10 +486,190 @@ private static MapReduceAgg createSimpleAgg(AggregateCall call, int reduceArgume
return new MapReduceAgg(argList, call, reduceCall, USE_INPUT_FIELD);
}
+ /**
+ * TODO: Rename to MakeExpr.
+ * Produces intermediate expressions that modify results of MAP/REDUCE aggregate.
+ * For example: after splitting a function into a MAP aggregate and REDUCE aggregate it is necessary to add casts to
+ * output of a REDUCE phase aggregate.
+ *
+ * <p>In order to avoid creating unnecessary projections, use {@link MapReduceAggregates#USE_INPUT_FIELD}.
+ */
@FunctionalInterface
private interface MakeReduceExpr {
- /** Creates an expression that produces result of REDUCE phase of an aggregate. */
+ /**
+ * Creates an expression that applies a performs computation (e.g. applies some function, adds a cast)
+ * on {@code args} fields of input relation.
+ *
+ * @param rexBuilder Expression builder.
+ * @param input Input relation.
+ * @param args Arguments.
+ * @param typeFactory Type factory.
+ *
+ * @return Expression.
+ */
RexNode makeExpr(RexBuilder rexBuilder, RelNode input, List<Integer> args, IgniteTypeFactory typeFactory);
}
+
+ private static MapReduceAgg createAvgAgg(
+ RelOptCluster cluster,
+ AggregateCall call,
+ int reduceArgumentOffset,
+ RelDataType inputType,
+ boolean canBeNull
+ ) {
+ RelDataTypeFactory tf = cluster.getTypeFactory();
+ RelDataTypeSystem typeSystem = tf.getTypeSystem();
+
+ RelDataType fieldType = inputType.getFieldList().get(call.getArgList().get(0)).getType();
+
+ // In case of AVG(NULL) return a simple version of an aggregate, because result is always NULL.
+ if (fieldType.getSqlTypeName() == SqlTypeName.NULL) {
+ return createSimpleAgg(call, reduceArgumentOffset);
+ }
+
+ // AVG(x) : SUM(x)/COUNT0(x)
+ // MAP : SUM(x) / COUNT(x)
+
+ // SUM(x) as s
+ RelDataType mapSumType = typeSystem.deriveSumType(tf, fieldType);
+ if (canBeNull) {
+ mapSumType = tf.createTypeWithNullability(mapSumType, true);
+ }
+
+ AggregateCall mapSum0 = AggregateCall.create(
+ SqlStdOperatorTable.SUM,
+ call.isDistinct(),
+ call.isApproximate(),
+ call.ignoreNulls(),
+ ImmutableList.of(),
+ call.getArgList(),
+ call.filterArg,
+ null,
+ call.collation,
+ mapSumType,
+ "AVG" + reduceArgumentOffset + "_MAP_SUM");
+
+ // COUNT(x) as c
+ RelDataType mapCountType = tf.createSqlType(SqlTypeName.BIGINT);
+
+ AggregateCall mapCount0 = AggregateCall.create(
+ SqlStdOperatorTable.COUNT,
+ call.isDistinct(),
+ call.isApproximate(),
+ call.ignoreNulls(),
+ ImmutableList.of(),
+ call.getArgList(),
+ call.filterArg,
+ null,
+ call.collation,
+ mapCountType,
+ "AVG" + reduceArgumentOffset + "_MAP_COUNT");
+
+ // REDUCE : SUM0(s) as reduce_sum, SUM0(c) as reduce_count
Review Comment:
```suggestion
// REDUCE : SUM(s) as reduce_sum, SUM0(c) as reduce_count
```
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533780816
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -356,10 +486,190 @@ private static MapReduceAgg createSimpleAgg(AggregateCall call, int reduceArgume
return new MapReduceAgg(argList, call, reduceCall, USE_INPUT_FIELD);
}
+ /**
+ * TODO: Rename to MakeExpr.
+ * Produces intermediate expressions that modify results of MAP/REDUCE aggregate.
+ * For example: after splitting a function into a MAP aggregate and REDUCE aggregate it is necessary to add casts to
+ * output of a REDUCE phase aggregate.
+ *
+ * <p>In order to avoid creating unnecessary projections, use {@link MapReduceAggregates#USE_INPUT_FIELD}.
+ */
@FunctionalInterface
private interface MakeReduceExpr {
- /** Creates an expression that produces result of REDUCE phase of an aggregate. */
+ /**
+ * Creates an expression that applies a performs computation (e.g. applies some function, adds a cast)
+ * on {@code args} fields of input relation.
+ *
+ * @param rexBuilder Expression builder.
+ * @param input Input relation.
+ * @param args Arguments.
+ * @param typeFactory Type factory.
+ *
+ * @return Expression.
+ */
RexNode makeExpr(RexBuilder rexBuilder, RelNode input, List<Integer> args, IgniteTypeFactory typeFactory);
}
+
+ private static MapReduceAgg createAvgAgg(
+ RelOptCluster cluster,
+ AggregateCall call,
+ int reduceArgumentOffset,
+ RelDataType inputType,
+ boolean canBeNull
+ ) {
+ RelDataTypeFactory tf = cluster.getTypeFactory();
+ RelDataTypeSystem typeSystem = tf.getTypeSystem();
+
+ RelDataType fieldType = inputType.getFieldList().get(call.getArgList().get(0)).getType();
+
+ // In case of AVG(NULL) return a simple version of an aggregate, because result is always NULL.
+ if (fieldType.getSqlTypeName() == SqlTypeName.NULL) {
+ return createSimpleAgg(call, reduceArgumentOffset);
+ }
+
+ // AVG(x) : SUM(x)/COUNT0(x)
+ // MAP : SUM(x) / COUNT(x)
+
+ // SUM(x) as s
+ RelDataType mapSumType = typeSystem.deriveSumType(tf, fieldType);
+ if (canBeNull) {
+ mapSumType = tf.createTypeWithNullability(mapSumType, true);
+ }
+
+ AggregateCall mapSum0 = AggregateCall.create(
+ SqlStdOperatorTable.SUM,
+ call.isDistinct(),
+ call.isApproximate(),
+ call.ignoreNulls(),
+ ImmutableList.of(),
+ call.getArgList(),
+ call.filterArg,
+ null,
+ call.collation,
+ mapSumType,
+ "AVG" + reduceArgumentOffset + "_MAP_SUM");
+
+ // COUNT(x) as c
+ RelDataType mapCountType = tf.createSqlType(SqlTypeName.BIGINT);
+
+ AggregateCall mapCount0 = AggregateCall.create(
+ SqlStdOperatorTable.COUNT,
+ call.isDistinct(),
+ call.isApproximate(),
+ call.ignoreNulls(),
+ ImmutableList.of(),
+ call.getArgList(),
+ call.filterArg,
+ null,
+ call.collation,
+ mapCountType,
+ "AVG" + reduceArgumentOffset + "_MAP_COUNT");
+
+ // REDUCE : SUM(s) as reduce_sum, SUM0(c) as reduce_count
+ List<Integer> reduceSumArgs = List.of(reduceArgumentOffset);
+
+ // SUM0(s)
+ RelDataType reduceSumType = typeSystem.deriveSumType(tf, mapSumType);
+ if (canBeNull) {
+ reduceSumType = tf.createTypeWithNullability(reduceSumType, true);
+ }
+
+ AggregateCall reduceSum0 = AggregateCall.create(
+ SqlStdOperatorTable.SUM,
+ call.isDistinct(),
+ call.isApproximate(),
+ call.ignoreNulls(),
+ ImmutableList.of(),
+ reduceSumArgs,
+ // there is no filtering on REDUCE phase
+ -1,
+ null,
+ call.collation,
+ reduceSumType,
+ "AVG" + reduceArgumentOffset + "_RED_VAL_SUM");
Review Comment:
Sure, I was thinking about removing this as well.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20009: Sql. Rework 2-phase aggregates part 2. AVG as SUM / COUNT. [ignite-3]
Posted by "lowka (via GitHub)" <gi...@apache.org>.
lowka commented on code in PR #3413:
URL: https://github.com/apache/ignite-3/pull/3413#discussion_r1533835397
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/agg/MapReduceAggregates.java:
##########
@@ -158,26 +183,84 @@ public static IgniteRel buildAggregates(LogicalAggregate agg, AggregateRelBuilde
// It consists of columns from agg.groupSet and aggregate expressions.
for (int i = 0; i < groupByColumns; i++) {
+ List<RelDataTypeField> outputRowFields = agg.getRowType().getFieldList();
RelDataType type = outputRowFields.get(i).getType();
reduceType.add("f" + reduceType.getFieldCount(), type);
}
+ RexBuilder rexBuilder = agg.getCluster().getRexBuilder();
+ IgniteTypeFactory typeFactory = (IgniteTypeFactory) agg.getCluster().getTypeFactory();
+
+ List<RexNode> reduceInputExprs = new ArrayList<>();
+
+ for (int i = 0; i < map.getRowType().getFieldList().size(); i++) {
+ RelDataType type = map.getRowType().getFieldList().get(i).getType();
+ RexInputRef ref = new RexInputRef(i, type);
+ reduceInputExprs.add(ref);
+ }
+
+ // Build a list of projections for reduce operator,
+ // if all projections are identity, it is not necessary
+ // to create a projection between MAP and REDUCE operators.
+
+ boolean additionalProjectionsForReduce = false;
+
+ for (int i = 0, argOffset = 0; i < mapReduceAggs.size(); i++) {
+ MapReduceAgg mapReduceAgg = mapReduceAggs.get(i);
+ int argIdx = groupByColumns + argOffset;
+
+ for (int j = 0; j < mapReduceAgg.reduceCalls.size(); j++) {
+ RexNode projExpr = mapReduceAgg.makeReduceInputExpr.makeExpr(rexBuilder, map, List.of(argIdx), typeFactory);
+ reduceInputExprs.set(argIdx, projExpr);
+
+ if (mapReduceAgg.makeReduceInputExpr != USE_INPUT_FIELD) {
+ additionalProjectionsForReduce = true;
+ }
+
+ argIdx += 1;
+ }
+
+ argOffset += mapReduceAgg.reduceCalls.size();
+ }
+
+ RelNode reduceInputNode;
+ if (additionalProjectionsForReduce) {
+ RelDataTypeFactory.Builder projectRow = new Builder(agg.getCluster().getTypeFactory());
+
+ for (int i = 0; i < reduceInputExprs.size(); i++) {
+ RexNode rexNode = reduceInputExprs.get(i);
+ projectRow.add(String.valueOf(i), rexNode.getType());
+ }
+
+ RelDataType projectRowType = projectRow.build();
+
+ reduceInputNode = builder.makeProject(agg.getCluster(), map, reduceInputExprs, projectRowType);
Review Comment:
We need an additional projection because COUNT returns a BIGINT, but SUM0 requires a DECIMAL time. Otherwise we get an error:
> java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.math.BigDecimal (java.lang.Long and java.math.BigDecimal are in module java.base of loader 'bootstrap')
at org.apache.ignite.internal.sql.engine.exec.exp.agg.Accumulators$DecimalSumEmptyIsZero.add(Accumulators.java:538) ~[main/:?]
Plan w/o projection
> Query: SELECT AVG(dec4_2_col) FROM numbers
> Plan :
Project traits=IGNITE.single.[], row=RecordType(DECIMAL(4, 2) EXPR$0)
ReduceHashAggregate traits=IGNITE.single.[], row=RecordType(DECIMAL(32767, 2) f0_0, DECIMAL(32767, 0) f1_1)
Exchange traits=IGNITE.single.[], row=RecordType(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
MapHashAggregate traits=IGNITE.random.[], row=RecordType(DECIMAL(32767, 2) _ACC0, BIGINT _ACC1, TINYINT _GROUP_ID)
TableScan traits=IGNITE.random.[], row=RecordType(DECIMAL(4, 2) DEC4_2_COL)
Alternatively, we can introduce an accumulator operator that that accepts BIGINT and returns DECIMAL.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org