Repository: phoenix Updated Branches: refs/heads/calcite be86248f5 -> f00a3c981
Implement SUM aggregate function Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f00a3c98 Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f00a3c98 Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f00a3c98 Branch: refs/heads/calcite Commit: f00a3c981aed3bef91baf99d4836a17194d7fa38 Parents: be86248 Author: maryannxue <wei....@intel.com> Authored: Wed Nov 4 20:37:41 2015 -0500 Committer: maryannxue <wei....@intel.com> Committed: Wed Nov 4 20:37:41 2015 -0500 ---------------------------------------------------------------------- .../org/apache/phoenix/calcite/CalciteIT.java | 21 +++++++++++++++ .../apache/phoenix/calcite/CalciteUtils.java | 23 ++++++++++------ .../calcite/jdbc/PhoenixCalciteDriver.java | 14 +++++++--- .../calcite/jdbc/PhoenixPrepareImpl.java | 1 - .../calcite/type/PhoenixRelDataTypeSystem.java | 28 ++++++++++++++++++++ 5 files changed, 74 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/phoenix/blob/f00a3c98/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java b/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java index 5b32383..6623b37 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/calcite/CalciteIT.java @@ -748,6 +748,27 @@ public class CalciteIT extends BaseClientManagedTimeIT { {"0000000005", 1L}, {"0000000006", 1L}}) .close(); + + start(false).sql("select a_string, sum(a_integer) from aTable group by a_string") + .explainIs("PhoenixToEnumerableConverter\n" + + " PhoenixServerAggregate(group=[{2}], EXPR$1=[SUM($4)], isOrdered=[false])\n" + + " PhoenixTableScan(table=[[phoenix, ATABLE]])\n") + .resultIs(new Object[][] { + {"a", 10L}, + {"b", 26L}, + {"c", 9L}}) + .close(); + + start(false).sql("select mypk0, avg(mypk1) from " + SALTED_TABLE_NAME + " group by mypk0") + .explainIs("PhoenixToEnumerableConverter\n" + + " PhoenixClientProject(MYPK0=[$0], EXPR$1=[CAST(/($1, $2)):INTEGER NOT NULL])\n" + + " PhoenixServerAggregate(group=[{0}], agg#0=[$SUM0($1)], agg#1=[COUNT()], isOrdered=[false])\n" + + " PhoenixTableScan(table=[[phoenix, SALTED_TEST_TABLE]])\n") + .resultIs(new Object[][] { + {1, 2}, + {2, 3}, + {3, 4}}) + .close(); } @Test public void testDistinct() { http://git-wip-us.apache.org/repos/asf/phoenix/blob/f00a3c98/phoenix-core/src/main/java/org/apache/phoenix/calcite/CalciteUtils.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/calcite/CalciteUtils.java b/phoenix-core/src/main/java/org/apache/phoenix/calcite/CalciteUtils.java index 27062cf..e5c70c0 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/calcite/CalciteUtils.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/CalciteUtils.java @@ -73,6 +73,7 @@ import org.apache.phoenix.expression.function.PowerFunction; import org.apache.phoenix.expression.function.RoundDecimalExpression; import org.apache.phoenix.expression.function.RoundTimestampExpression; import org.apache.phoenix.expression.function.SqrtFunction; +import org.apache.phoenix.expression.function.SumAggregateFunction; import org.apache.phoenix.expression.function.TrimFunction; import org.apache.phoenix.expression.function.UpperFunction; import org.apache.phoenix.parse.JoinTableNode.JoinType; @@ -760,14 +761,20 @@ public class CalciteUtils { return new CountAggregateFunction(args); } }); - // TODO Buggy. Disable for now. - //FUNCTION_MAP.put("$SUM0", new FunctionFactory() { - // @Override - // public FunctionExpression newFunction(SqlFunction sqlFunc, - // List<Expression> args) { - // return new SumAggregateFunction(args); - // } - //}); + FUNCTION_MAP.put("$SUM0", new FunctionFactory() { + @Override + public FunctionExpression newFunction(SqlFunction sqlFunc, + List<Expression> args) { + return new SumAggregateFunction(args); + } + }); + FUNCTION_MAP.put("SUM", new FunctionFactory() { + @Override + public FunctionExpression newFunction(SqlFunction sqlFunc, + List<Expression> args) { + return new SumAggregateFunction(args); + } + }); FUNCTION_MAP.put("MAX", new FunctionFactory() { @Override public FunctionExpression newFunction(SqlFunction sqlFunc, http://git-wip-us.apache.org/repos/asf/phoenix/blob/f00a3c98/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixCalciteDriver.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixCalciteDriver.java b/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixCalciteDriver.java index b4f6598..e219639 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixCalciteDriver.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixCalciteDriver.java @@ -6,6 +6,7 @@ import java.util.Map; import java.util.Properties; import java.util.Map.Entry; +import org.apache.calcite.config.CalciteConnectionProperty; import org.apache.calcite.jdbc.CalciteConnection; import org.apache.calcite.jdbc.CalcitePrepare; import org.apache.calcite.jdbc.Driver; @@ -13,6 +14,7 @@ import org.apache.calcite.linq4j.function.Function0; import org.apache.calcite.schema.SchemaPlus; import org.apache.phoenix.calcite.PhoenixSchema; import org.apache.phoenix.calcite.rules.PhoenixConverterRules; +import org.apache.phoenix.calcite.type.PhoenixRelDataTypeSystem; import org.apache.phoenix.util.PhoenixRuntime; import com.google.common.collect.Maps; @@ -39,20 +41,24 @@ public class PhoenixCalciteDriver extends Driver { @Override protected String getConnectStringPrefix() { return CONNECT_STRING_PREFIX; - } - + } public Connection connect(String url, Properties info) throws SQLException { if (!acceptsURL(url)) { return null; } + + Properties info2 = new Properties(info); + info2.setProperty(CalciteConnectionProperty.TYPE_SYSTEM.camelName(), + PhoenixRelDataTypeSystem.class.getName()); + final String prefix = getConnectStringPrefix(); assert url.startsWith(prefix); final String urlSuffix = url.substring(prefix.length()); final int delimiter = urlSuffix.indexOf(';'); final int eq = urlSuffix.indexOf('='); if ((delimiter < 0 && eq > 0) || eq < delimiter) { - return super.connect(url, info); + return super.connect(url, info2); } // URLs that start with a non-property-pair string will be treated as Phoenix @@ -60,7 +66,7 @@ public class PhoenixCalciteDriver extends Driver { // of this URL can be the connection string prefix itself. final String phoenixUrl = delimiter < 0 ? urlSuffix : urlSuffix.substring(0, delimiter); url = delimiter < 0 ? prefix : (prefix + urlSuffix.substring(delimiter + 1)); - final CalciteConnection connection = (CalciteConnection) super.connect(url, info); + final CalciteConnection connection = (CalciteConnection) super.connect(url, info2); Map<String, Object> operand = Maps.newHashMap(); for (Entry<Object, Object> entry : info.entrySet()) { operand.put((String) entry.getKey(), entry.getValue()); http://git-wip-us.apache.org/repos/asf/phoenix/blob/f00a3c98/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixPrepareImpl.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixPrepareImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixPrepareImpl.java index b0102fe..dd3c9bf 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixPrepareImpl.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/jdbc/PhoenixPrepareImpl.java @@ -13,7 +13,6 @@ import org.apache.calcite.prepare.CalcitePrepareImpl; import org.apache.calcite.prepare.Prepare.Materialization; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.rules.JoinCommuteRule; -import org.apache.calcite.rel.rules.SortJoinTransposeRule; import org.apache.calcite.rel.rules.SortUnionTransposeRule; import org.apache.calcite.runtime.Hook; import org.apache.calcite.sql.SqlNode; http://git-wip-us.apache.org/repos/asf/phoenix/blob/f00a3c98/phoenix-core/src/main/java/org/apache/phoenix/calcite/type/PhoenixRelDataTypeSystem.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/calcite/type/PhoenixRelDataTypeSystem.java b/phoenix-core/src/main/java/org/apache/phoenix/calcite/type/PhoenixRelDataTypeSystem.java new file mode 100644 index 0000000..8f7cb67 --- /dev/null +++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/type/PhoenixRelDataTypeSystem.java @@ -0,0 +1,28 @@ +package org.apache.phoenix.calcite.type; + +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelDataTypeSystemImpl; +import org.apache.calcite.sql.type.SqlTypeName; + +public class PhoenixRelDataTypeSystem extends RelDataTypeSystemImpl { + + public PhoenixRelDataTypeSystem() { + super(); + } + + @Override + public RelDataType deriveSumType( + RelDataTypeFactory typeFactory, RelDataType argumentType) { + RelDataType type; + if (argumentType.getSqlTypeName() == SqlTypeName.DECIMAL) { + type = typeFactory.createSqlType(SqlTypeName.DECIMAL); + } else if (argumentType.getSqlTypeName() == SqlTypeName.FLOAT + || argumentType.getSqlTypeName() == SqlTypeName.DOUBLE) { + type = typeFactory.createSqlType(SqlTypeName.DOUBLE); + } else { + type = typeFactory.createSqlType(SqlTypeName.BIGINT); + } + return typeFactory.createTypeWithNullability(type, argumentType.isNullable()); + } +}
