You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@calcite.apache.org by jh...@apache.org on 2015/01/22 01:33:40 UTC
[3/3] incubator-calcite git commit: [CALCITE-544] Implement Union in
Interpreter;
[CALCITE-562] Implement inner JOIN in interpreter and improve handling of
scalar expressions
[CALCITE-544] Implement Union in Interpreter;
[CALCITE-562] Implement inner JOIN in interpreter and improve handling of scalar expressions
Also, progress towards [CALCITE-558] Add BINDABLE convention and make it the default convention for query root, but currently ENUMERABLE is still the default. You can enable BINDABLE by setting CalcitePrepareImpl.ENABLE_BINDABLE; there are 184 test failures.
New interface ArrayBindable extends Bindable. Unlike Bindable the return is guaranteed to be an Object[]. Enumerable convention expects generated code to return a Bindable, which might return custom classes or arrays, and if there is a single column, the array is optimized to the raw column value. Bindable convention assumes that relational expressions implement ArrayBindable and return arrays of column values; the arrays are not optimized if there is only one column.
New interface Scalar represents a compiled way to execute a scalar expression (or expressions) and is easily built from a RexNode or AggregateCall.
Close apache/incubator-calcite#43
Project: http://git-wip-us.apache.org/repos/asf/incubator-calcite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-calcite/commit/66cfb120
Tree: http://git-wip-us.apache.org/repos/asf/incubator-calcite/tree/66cfb120
Diff: http://git-wip-us.apache.org/repos/asf/incubator-calcite/diff/66cfb120
Branch: refs/heads/master
Commit: 66cfb120fca58e4c809748e2bcef4d3d5a7c6b82
Parents: 4e18540
Author: Julian Hyde <jh...@apache.org>
Authored: Mon Dec 29 17:40:39 2014 -0800
Committer: Julian Hyde <jh...@apache.org>
Committed: Wed Jan 21 15:57:37 2015 -0800
----------------------------------------------------------------------
.../calcite/adapter/clone/ArrayTable.java | 21 +-
.../calcite/adapter/enumerable/AggImpState.java | 2 +-
.../calcite/adapter/enumerable/EnumUtils.java | 2 -
.../adapter/enumerable/EnumerableAggregate.java | 14 +-
.../adapter/enumerable/EnumerableBindable.java | 102 ++++
.../adapter/enumerable/EnumerableCalc.java | 2 +
.../adapter/enumerable/EnumerableCalcRule.java | 21 +-
.../enumerable/EnumerableConvention.java | 4 +
.../enumerable/EnumerableFilterRule.java | 20 +-
.../enumerable/EnumerableInterpretable.java | 205 +++++++
.../enumerable/EnumerableProjectRule.java | 20 +-
.../enumerable/EnumerableRelImplementor.java | 2 +-
.../adapter/enumerable/EnumerableSort.java | 4 +-
.../adapter/enumerable/EnumerableWindow.java | 32 +-
.../enumerable/EnumerableWindowRule.java | 4 +-
.../adapter/enumerable/JavaRowFormat.java | 1 -
.../adapter/enumerable/PhysTypeImpl.java | 17 +-
.../calcite/adapter/enumerable/RexImpTable.java | 12 +-
.../adapter/enumerable/RexToLixTranslator.java | 71 ++-
.../calcite/adapter/java/ReflectiveSchema.java | 36 +-
.../apache/calcite/adapter/jdbc/JdbcTable.java | 12 +-
.../calcite/interpreter/AggregateNode.java | 345 +++++++++--
.../calcite/interpreter/BindableConvention.java | 64 ++
.../apache/calcite/interpreter/BindableRel.java | 31 +
.../apache/calcite/interpreter/Bindables.java | 583 +++++++++++++++++++
.../org/apache/calcite/interpreter/Context.java | 8 +
.../apache/calcite/interpreter/FilterNode.java | 13 +-
.../interpreter/InterpretableConvention.java | 58 ++
.../interpreter/InterpretableConverter.java | 55 ++
.../calcite/interpreter/InterpretableRel.java | 55 ++
.../apache/calcite/interpreter/Interpreter.java | 182 +++---
.../calcite/interpreter/Interpreters.java | 50 ++
.../calcite/interpreter/JaninoRexCompiler.java | 191 ++++++
.../apache/calcite/interpreter/JoinNode.java | 76 +++
.../org/apache/calcite/interpreter/Nodes.java | 20 +-
.../NoneToBindableConverterRule.java | 44 ++
.../apache/calcite/interpreter/ProjectNode.java | 27 +-
.../org/apache/calcite/interpreter/Scalar.java | 1 +
.../apache/calcite/interpreter/ScanNode.java | 173 ------
.../apache/calcite/interpreter/SortNode.java | 47 +-
.../calcite/interpreter/TableScanNode.java | 174 ++++++
.../apache/calcite/interpreter/UnionNode.java | 58 ++
.../apache/calcite/interpreter/ValuesNode.java | 38 +-
.../apache/calcite/interpreter/WindowNode.java | 39 ++
.../org/apache/calcite/jdbc/CalcitePrepare.java | 5 +-
.../org/apache/calcite/plan/RelOptUtil.java | 40 ++
.../org/apache/calcite/plan/RelTraitSet.java | 12 +
.../org/apache/calcite/plan/volcano/RelSet.java | 5 +
.../calcite/plan/volcano/VolcanoPlanner.java | 30 +
.../calcite/prepare/CalciteMaterializer.java | 4 +-
.../calcite/prepare/CalcitePrepareImpl.java | 145 ++---
.../org/apache/calcite/prepare/Prepare.java | 23 +-
.../apache/calcite/prepare/RelOptTableImpl.java | 25 +-
.../calcite/rel/convert/ConverterRule.java | 38 +-
.../org/apache/calcite/rel/core/Window.java | 18 +
.../org/apache/calcite/rex/RexExecutorImpl.java | 5 +-
.../apache/calcite/runtime/ArrayBindable.java | 30 +
.../schema/impl/AggregateFunctionImpl.java | 2 +-
.../java/org/apache/calcite/sql/SqlKind.java | 4 +-
.../java/org/apache/calcite/tools/Programs.java | 2 +
.../org/apache/calcite/util/BuiltInMethod.java | 20 +-
.../main/java/org/apache/calcite/util/Util.java | 33 +-
.../calcite/sql/parser/SqlParserTest.java | 30 +-
.../apache/calcite/test/InterpreterTest.java | 37 +-
.../java/org/apache/calcite/test/JdbcTest.java | 1 +
.../org/apache/calcite/test/LatticeTest.java | 2 +-
core/src/test/resources/sql/misc.oq | 2 +-
.../linq4j/tree/ConditionalExpression.java | 5 +-
.../org/apache/calcite/linq4j/tree/Types.java | 9 +
.../calcite/adapter/spark/SparkHandlerImpl.java | 10 +-
.../calcite/adapter/spark/SparkRules.java | 8 +-
71 files changed, 2830 insertions(+), 651 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java b/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java
index 7c7ccf3..5d1c58a 100644
--- a/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java
+++ b/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java
@@ -16,7 +16,10 @@
*/
package org.apache.calcite.adapter.clone;
+import org.apache.calcite.DataContext;
import org.apache.calcite.adapter.java.AbstractQueryableTable;
+import org.apache.calcite.linq4j.AbstractEnumerable;
+import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.linq4j.Enumerator;
import org.apache.calcite.linq4j.Ord;
import org.apache.calcite.linq4j.QueryProvider;
@@ -25,6 +28,7 @@ import org.apache.calcite.linq4j.tree.Primitive;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelProtoDataType;
+import org.apache.calcite.schema.ScannableTable;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.Statistic;
import org.apache.calcite.schema.Statistics;
@@ -49,7 +53,7 @@ import java.util.List;
* values in the column; see {@link Representation} and
* {@link RepresentationType}.
*/
-class ArrayTable extends AbstractQueryableTable {
+class ArrayTable extends AbstractQueryableTable implements ScannableTable {
private final RelProtoDataType protoRowType;
private final Supplier<Content> supplier;
@@ -76,6 +80,15 @@ class ArrayTable extends AbstractQueryableTable {
return Statistics.of(content.size, keys);
}
+ public Enumerable<Object[]> scan(DataContext root) {
+ return new AbstractEnumerable<Object[]>() {
+ public Enumerator<Object[]> enumerator() {
+ final Content content = supplier.get();
+ return content.arrayEnumerator();
+ }
+ };
+ }
+
public <T> Queryable<T> asQueryable(final QueryProvider queryProvider,
SchemaPlus schema, String tableName) {
return new AbstractTableQueryable<T>(queryProvider, schema, this,
@@ -802,6 +815,10 @@ class ArrayTable extends AbstractQueryableTable {
}
}
+ public Enumerator<Object[]> arrayEnumerator() {
+ return new ArrayEnumerator(size, columns);
+ }
+
/** Enumerator over a table with a single column; each element
* returned is an object. */
private static class ObjectEnumerator implements Enumerator<Object> {
@@ -834,7 +851,7 @@ class ArrayTable extends AbstractQueryableTable {
/** Enumerator over a table with more than one column; each element
* returned is an array. */
- private static class ArrayEnumerator implements Enumerator {
+ private static class ArrayEnumerator implements Enumerator<Object[]> {
final int rowCount;
final List<Column> columns;
int i = -1;
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/AggImpState.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/AggImpState.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/AggImpState.java
index 1c8909a..b100cec 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/AggImpState.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/AggImpState.java
@@ -24,7 +24,7 @@ import java.util.List;
/**
* Represents internal state when implementing aggregate functions.
*/
-class AggImpState {
+public class AggImpState {
public final int aggIdx;
public final AggregateCall call;
public final AggImplementor implementor;
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
index d1948f1..17e0bf5 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
@@ -56,8 +56,6 @@ public class EnumUtils {
public static final String[] LEFT_RIGHT = {"left", "right"};
- static final boolean B = false;
-
/** Declares a method that overrides another method. */
public static MethodDeclaration overridingMethodDecl(Method method,
Iterable<ParameterExpression> parameters,
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableAggregate.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableAggregate.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableAggregate.java
index 532821f..38f3ee9 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableAggregate.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableAggregate.java
@@ -38,13 +38,11 @@ import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlAggFunction;
-import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;
-import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import java.lang.reflect.Type;
@@ -54,15 +52,7 @@ import java.util.List;
/** Implementation of {@link org.apache.calcite.rel.core.Aggregate} in
* {@link org.apache.calcite.adapter.enumerable.EnumerableConvention enumerable calling convention}. */
-public class EnumerableAggregate extends Aggregate
- implements EnumerableRel {
- private static final List<SqlAggFunction> SUPPORTED_AGGREGATIONS =
- ImmutableList.of(
- SqlStdOperatorTable.COUNT,
- SqlStdOperatorTable.MIN,
- SqlStdOperatorTable.MAX,
- SqlStdOperatorTable.SUM);
-
+public class EnumerableAggregate extends Aggregate implements EnumerableRel {
public EnumerableAggregate(
RelOptCluster cluster,
RelTraitSet traitSet,
@@ -209,7 +199,7 @@ public class EnumerableAggregate extends Aggregate
for (final AggImpState agg : aggs) {
agg.context =
new AggContext() {
- public org.apache.calcite.sql.SqlAggFunction aggregation() {
+ public SqlAggFunction aggregation() {
return agg.call.getAggregation();
}
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableBindable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableBindable.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableBindable.java
new file mode 100644
index 0000000..2ad069e
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableBindable.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.adapter.enumerable;
+
+import org.apache.calcite.DataContext;
+import org.apache.calcite.interpreter.BindableConvention;
+import org.apache.calcite.interpreter.BindableRel;
+import org.apache.calcite.interpreter.Node;
+import org.apache.calcite.interpreter.Row;
+import org.apache.calcite.interpreter.Sink;
+import org.apache.calcite.linq4j.Enumerable;
+import org.apache.calcite.linq4j.Enumerator;
+import org.apache.calcite.plan.ConventionTraitDef;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterImpl;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.runtime.ArrayBindable;
+import org.apache.calcite.runtime.Bindable;
+
+import com.google.common.collect.ImmutableMap;
+
+import java.util.List;
+
+/**
+ * Relational expression that converts an enumerable input to interpretable
+ * calling convention.
+ *
+ * @see org.apache.calcite.adapter.enumerable.EnumerableConvention
+ * @see org.apache.calcite.interpreter.BindableConvention
+ */
+public class EnumerableBindable extends ConverterImpl implements BindableRel {
+ protected EnumerableBindable(RelOptCluster cluster, RelNode input) {
+ super(cluster, ConventionTraitDef.INSTANCE,
+ cluster.traitSetOf(BindableConvention.INSTANCE), input);
+ }
+
+ @Override public EnumerableBindable copy(RelTraitSet traitSet,
+ List<RelNode> inputs) {
+ return new EnumerableBindable(getCluster(), sole(inputs));
+ }
+
+ public Class<Object[]> getElementType() {
+ return Object[].class;
+ }
+
+ public Enumerable<Object[]> bind(DataContext dataContext) {
+ final ImmutableMap<String, Object> map = ImmutableMap.of();
+ final Bindable bindable = EnumerableInterpretable.toBindable(map, null,
+ (EnumerableRel) getInput(), EnumerableRel.Prefer.ARRAY);
+ final ArrayBindable arrayBindable = EnumerableInterpretable.box(bindable);
+ return arrayBindable.bind(dataContext);
+ }
+
+ public Node implement(final InterpreterImplementor implementor) {
+ return new Node() {
+ public void run() throws InterruptedException {
+ final Sink sink =
+ implementor.relSinks.get(EnumerableBindable.this).get(0);
+ final Enumerable<Object[]> enumerable = bind(implementor.dataContext);
+ final Enumerator<Object[]> enumerator = enumerable.enumerator();
+ while (enumerator.moveNext()) {
+ sink.send(Row.asCopy(enumerator.current()));
+ }
+ }
+ };
+ }
+
+ /**
+ * Rule that converts any enumerable relational expression to bindable.
+ */
+ public static class EnumerableToBindableConverterRule extends ConverterRule {
+ public static final EnumerableToBindableConverterRule INSTANCE =
+ new EnumerableToBindableConverterRule();
+
+ private EnumerableToBindableConverterRule() {
+ super(EnumerableRel.class, EnumerableConvention.INSTANCE,
+ BindableConvention.INSTANCE, "EnumerableToBindableConverterRule");
+ }
+
+ @Override public RelNode convert(RelNode rel) {
+ return new EnumerableBindable(rel.getCluster(), rel);
+ }
+ }
+}
+
+// End EnumerableBindable.java
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalc.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalc.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalc.java
index b467163..b93256d 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalc.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalc.java
@@ -16,6 +16,7 @@
*/
package org.apache.calcite.adapter.enumerable;
+import org.apache.calcite.DataContext;
import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.linq4j.Enumerator;
import org.apache.calcite.linq4j.tree.BlockBuilder;
@@ -144,6 +145,7 @@ public class EnumerableCalc extends Calc implements EnumerableRel {
typeFactory,
builder3,
physType,
+ DataContext.ROOT,
new RexToLixTranslator.InputGetterImpl(
Collections.singletonList(
Pair.of(input, result.physType))),
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalcRule.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalcRule.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalcRule.java
index 35c83dd..793dcf8 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalcRule.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableCalcRule.java
@@ -17,11 +17,10 @@
package org.apache.calcite.adapter.enumerable;
import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.convert.ConverterRule;
import org.apache.calcite.rel.logical.LogicalCalc;
-import org.apache.calcite.rex.RexMultisetUtil;
-import org.apache.calcite.rex.RexProgram;
/**
* Rule to convert a {@link org.apache.calcite.rel.logical.LogicalCalc} to an
@@ -29,22 +28,14 @@ import org.apache.calcite.rex.RexProgram;
*/
class EnumerableCalcRule extends ConverterRule {
EnumerableCalcRule() {
- super(LogicalCalc.class, Convention.NONE, EnumerableConvention.INSTANCE,
- "EnumerableCalcRule");
+ // The predicate ensures that if there's a multiset, FarragoMultisetSplitter
+ // will work on it first.
+ super(LogicalCalc.class, RelOptUtil.CALC_PREDICATE, Convention.NONE,
+ EnumerableConvention.INSTANCE, "EnumerableCalcRule");
}
public RelNode convert(RelNode rel) {
final LogicalCalc calc = (LogicalCalc) rel;
-
- // If there's a multiset, let FarragoMultisetSplitter work on it
- // first.
- final RexProgram program = calc.getProgram();
- if (EnumUtils.B
- && RexMultisetUtil.containsMultiset(program)
- || program.containsAggs()) {
- return null;
- }
-
return new EnumerableCalc(
rel.getCluster(),
rel.getTraitSet().replace(EnumerableConvention.INSTANCE),
@@ -52,7 +43,7 @@ class EnumerableCalcRule extends ConverterRule {
calc.getInput(),
calc.getInput().getTraitSet()
.replace(EnumerableConvention.INSTANCE)),
- program,
+ calc.getProgram(),
calc.getCollationList());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableConvention.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableConvention.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableConvention.java
index a8ad841..ac1533a 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableConvention.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableConvention.java
@@ -29,6 +29,10 @@ import org.apache.calcite.plan.RelTraitDef;
public enum EnumerableConvention implements Convention {
INSTANCE;
+ /** Cost of an enumerable node versus implementing an equivalent node in a
+ * "typical" calling convention. */
+ public static final double COST_MULTIPLIER = 1.0d;
+
@Override public String toString() {
return getName();
}
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableFilterRule.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableFilterRule.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableFilterRule.java
index 81bbaf2..51fcdd4 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableFilterRule.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableFilterRule.java
@@ -17,11 +17,10 @@
package org.apache.calcite.adapter.enumerable;
import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.convert.ConverterRule;
import org.apache.calcite.rel.logical.LogicalFilter;
-import org.apache.calcite.rex.RexMultisetUtil;
-import org.apache.calcite.rex.RexOver;
/**
* Rule to convert a {@link org.apache.calcite.rel.logical.LogicalFilter} to an
@@ -29,24 +28,15 @@ import org.apache.calcite.rex.RexOver;
*/
class EnumerableFilterRule extends ConverterRule {
EnumerableFilterRule() {
- super(LogicalFilter.class, Convention.NONE, EnumerableConvention.INSTANCE,
- "EnumerableFilterRule");
+ super(LogicalFilter.class, RelOptUtil.FILTER_PREDICATE, Convention.NONE,
+ EnumerableConvention.INSTANCE, "EnumerableFilterRule");
}
public RelNode convert(RelNode rel) {
final LogicalFilter filter = (LogicalFilter) rel;
-
- if (EnumUtils.B
- && RexMultisetUtil.containsMultiset(filter.getCondition(), true)
- || RexOver.containsOver(filter.getCondition())) {
- return null;
- }
-
- return new EnumerableFilter(
- rel.getCluster(),
+ return new EnumerableFilter(rel.getCluster(),
rel.getTraitSet().replace(EnumerableConvention.INSTANCE),
- convert(
- filter.getInput(),
+ convert(filter.getInput(),
filter.getInput().getTraitSet()
.replace(EnumerableConvention.INSTANCE)),
filter.getCondition());
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java
new file mode 100644
index 0000000..19d5f2a
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.adapter.enumerable;
+
+import org.apache.calcite.DataContext;
+import org.apache.calcite.avatica.Helper;
+import org.apache.calcite.interpreter.InterpretableConvention;
+import org.apache.calcite.interpreter.InterpretableRel;
+import org.apache.calcite.interpreter.Interpreter;
+import org.apache.calcite.interpreter.Node;
+import org.apache.calcite.interpreter.Row;
+import org.apache.calcite.interpreter.Sink;
+import org.apache.calcite.jdbc.CalcitePrepare;
+import org.apache.calcite.linq4j.AbstractEnumerable;
+import org.apache.calcite.linq4j.Enumerable;
+import org.apache.calcite.linq4j.Enumerator;
+import org.apache.calcite.linq4j.tree.ClassDeclaration;
+import org.apache.calcite.linq4j.tree.Expressions;
+import org.apache.calcite.plan.ConventionTraitDef;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.prepare.CalcitePrepareImpl;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterImpl;
+import org.apache.calcite.runtime.ArrayBindable;
+import org.apache.calcite.runtime.Bindable;
+import org.apache.calcite.runtime.Hook;
+import org.apache.calcite.runtime.Typed;
+import org.apache.calcite.runtime.Utilities;
+import org.apache.calcite.util.Util;
+
+import org.codehaus.commons.compiler.CompileException;
+import org.codehaus.commons.compiler.CompilerFactoryFactory;
+import org.codehaus.commons.compiler.IClassBodyEvaluator;
+import org.codehaus.commons.compiler.ICompilerFactory;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Relational expression that converts an enumerable input to interpretable
+ * calling convention.
+ *
+ * @see EnumerableConvention
+ * @see org.apache.calcite.interpreter.BindableConvention
+ */
+public class EnumerableInterpretable extends ConverterImpl
+ implements InterpretableRel {
+ protected EnumerableInterpretable(RelOptCluster cluster, RelNode input) {
+ super(cluster, ConventionTraitDef.INSTANCE,
+ cluster.traitSetOf(InterpretableConvention.INSTANCE), input);
+ }
+
+ @Override public EnumerableInterpretable copy(RelTraitSet traitSet,
+ List<RelNode> inputs) {
+ return new EnumerableInterpretable(getCluster(), sole(inputs));
+ }
+
+ public Node implement(final InterpreterImplementor implementor) {
+ final Bindable bindable = toBindable(implementor.internalParameters,
+ implementor.spark, (EnumerableRel) getInput(),
+ EnumerableRel.Prefer.ARRAY);
+ final ArrayBindable arrayBindable = box(bindable);
+ final Enumerable<Object[]> enumerable =
+ arrayBindable.bind(implementor.dataContext);
+ return new EnumerableNode(enumerable, implementor.interpreter, this);
+ }
+
+ public static Bindable toBindable(Map<String, Object> parameters,
+ CalcitePrepare.SparkHandler spark, EnumerableRel rel,
+ EnumerableRel.Prefer prefer) {
+ EnumerableRelImplementor relImplementor =
+ new EnumerableRelImplementor(rel.getCluster().getRexBuilder(),
+ parameters);
+
+ final ClassDeclaration expr = relImplementor.implementRoot(rel, prefer);
+ String s = Expressions.toString(expr.memberDeclarations, "\n", false);
+
+ if (CalcitePrepareImpl.DEBUG) {
+ Util.debugCode(System.out, s);
+ }
+
+ Hook.JAVA_PLAN.run(s);
+
+ try {
+ if (spark != null && spark.enabled()) {
+ return spark.compile(expr, s);
+ } else {
+ return getBindable(expr, s, rel.getRowType().getFieldCount());
+ }
+ } catch (Exception e) {
+ throw Helper.INSTANCE.wrap("Error while compiling generated Java code:\n"
+ + s, e);
+ }
+ }
+
+ static ArrayBindable getArrayBindable(ClassDeclaration expr, String s,
+ int fieldCount) throws CompileException, IOException {
+ Bindable bindable = getBindable(expr, s, fieldCount);
+ return box(bindable);
+ }
+
+ static Bindable getBindable(ClassDeclaration expr, String s, int fieldCount)
+ throws CompileException, IOException {
+ ICompilerFactory compilerFactory;
+ try {
+ compilerFactory = CompilerFactoryFactory.getDefaultCompilerFactory();
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Unable to instantiate java compiler", e);
+ }
+ IClassBodyEvaluator cbe = compilerFactory.newClassBodyEvaluator();
+ cbe.setClassName(expr.name);
+ cbe.setExtendedClass(Utilities.class);
+ cbe.setImplementedInterfaces(
+ fieldCount == 1
+ ? new Class[] {Bindable.class, Typed.class}
+ : new Class[] {ArrayBindable.class});
+ cbe.setParentClassLoader(EnumerableInterpretable.class.getClassLoader());
+ if (CalcitePrepareImpl.DEBUG) {
+ // Add line numbers to the generated janino class
+ cbe.setDebuggingInformation(true, true, true);
+ }
+ return (Bindable) cbe.createInstance(new StringReader(s));
+ }
+
+ /** Converts a bindable over scalar values into an array bindable, with each
+ * row as an array of 1 element. */
+ static ArrayBindable box(final Bindable bindable) {
+ if (bindable instanceof ArrayBindable) {
+ return (ArrayBindable) bindable;
+ }
+ return new ArrayBindable() {
+ public Class<Object[]> getElementType() {
+ return Object[].class;
+ }
+
+ public Enumerable<Object[]> bind(DataContext dataContext) {
+ final Enumerable<?> enumerable = bindable.bind(dataContext);
+ return new AbstractEnumerable<Object[]>() {
+ public Enumerator<Object[]> enumerator() {
+ final Enumerator<?> enumerator = enumerable.enumerator();
+ return new Enumerator<Object[]>() {
+ public Object[] current() {
+ return new Object[] {enumerator.current()};
+ }
+
+ public boolean moveNext() {
+ return enumerator.moveNext();
+ }
+
+ public void reset() {
+ enumerator.reset();
+ }
+
+ public void close() {
+ enumerator.close();
+ }
+ };
+ }
+ };
+ }
+ };
+ }
+
+ /** Interpreter node that reads from an {@link Enumerable}.
+ *
+ * <p>From the interpreter's perspective, it is a leaf node. */
+ private static class EnumerableNode implements Node {
+ private final Enumerable<Object[]> enumerable;
+ private final Sink sink;
+
+ public EnumerableNode(Enumerable<Object[]> enumerable,
+ Interpreter interpreter, EnumerableInterpretable rel) {
+ this.enumerable = enumerable;
+ this.sink = interpreter.sink(rel);
+ }
+
+ public void run() throws InterruptedException {
+ final Enumerator<Object[]> enumerator = enumerable.enumerator();
+ while (enumerator.moveNext()) {
+ Object[] values = enumerator.current();
+ sink.send(Row.of(values));
+ }
+ }
+ }
+}
+
+// End EnumerableInterpretable.java
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableProjectRule.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableProjectRule.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableProjectRule.java
index 919856c..158365c 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableProjectRule.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableProjectRule.java
@@ -17,12 +17,11 @@
package org.apache.calcite.adapter.enumerable;
import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.convert.ConverterRule;
import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.logical.LogicalProject;
-import org.apache.calcite.rex.RexMultisetUtil;
-import org.apache.calcite.rex.RexOver;
/**
* Rule to convert a {@link org.apache.calcite.rel.logical.LogicalProject} to an
@@ -30,24 +29,15 @@ import org.apache.calcite.rex.RexOver;
*/
class EnumerableProjectRule extends ConverterRule {
EnumerableProjectRule() {
- super(LogicalProject.class, Convention.NONE, EnumerableConvention.INSTANCE,
- "EnumerableProjectRule");
+ super(LogicalProject.class, RelOptUtil.PROJECT_PREDICATE, Convention.NONE,
+ EnumerableConvention.INSTANCE, "EnumerableProjectRule");
}
public RelNode convert(RelNode rel) {
final LogicalProject project = (LogicalProject) rel;
-
- if (EnumUtils.B
- && RexMultisetUtil.containsMultiset(project.getProjects(), true)
- || RexOver.containsOver(project.getProjects(), null)) {
- return null;
- }
-
- return new EnumerableProject(
- rel.getCluster(),
+ return new EnumerableProject(rel.getCluster(),
rel.getTraitSet().replace(EnumerableConvention.INSTANCE),
- convert(
- project.getInput(),
+ convert(project.getInput(),
project.getInput().getTraitSet()
.replace(EnumerableConvention.INSTANCE)),
project.getProjects(),
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRelImplementor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRelImplementor.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRelImplementor.java
index d69fffd..4280a7e 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRelImplementor.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableRelImplementor.java
@@ -142,7 +142,7 @@ public class EnumerableRelImplementor extends JavaRelImplementor {
Expressions.list(root0_),
block));
memberDeclarations.add(
- Expressions.methodDecl(Modifier.PUBLIC, Type.class,
+ Expressions.methodDecl(Modifier.PUBLIC, Class.class,
BuiltInMethod.TYPED_GET_ELEMENT_TYPE.method.getName(),
Collections.<ParameterExpression>emptyList(),
Blocks.toFunctionBlock(
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableSort.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableSort.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableSort.java
index 7c6bec5..a20279c 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableSort.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableSort.java
@@ -71,8 +71,8 @@ public class EnumerableSort extends Sort implements EnumerableRel {
BuiltInMethod.ORDER_BY.method,
Expressions.list(
builder.append("keySelector", pair.left))
- .appendIfNotNull(builder.appendIfNotNull("comparator",
- pair.right)))));
+ .appendIfNotNull(
+ builder.appendIfNotNull("comparator", pair.right)))));
return implementor.result(physType, builder.toBlock());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindow.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindow.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindow.java
index aa0ada7..7650c62 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindow.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindow.java
@@ -38,7 +38,6 @@ import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.Window;
-import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rex.RexInputRef;
@@ -46,6 +45,7 @@ import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexWindowBound;
import org.apache.calcite.runtime.SortedMultiMap;
+import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;
@@ -75,18 +75,8 @@ public class EnumerableWindow extends Window implements EnumerableRel {
}
public RelOptCost computeSelfCost(RelOptPlanner planner) {
- // Cost is proportional to the number of rows and the number of
- // components (groups and aggregate functions). There is
- // no I/O cost.
- //
- // TODO #1. Add memory cost.
- // TODO #2. MIN and MAX have higher CPU cost than SUM and COUNT.
- final double rowsIn = RelMetadataQuery.getRowCount(getInput());
- int count = groups.size();
- for (Group group : groups) {
- count += group.aggCalls.size();
- }
- return planner.getCostFactory().makeCost(rowsIn, rowsIn * count, 0);
+ return super.computeSelfCost(planner)
+ .multiplyBy(EnumerableConvention.COST_MULTIPLIER);
}
/** Implementation of {@link RexToLixTranslator.InputGetter}
@@ -373,8 +363,9 @@ public class EnumerableWindow extends Window implements EnumerableRel {
builder4.append("totalRows", rowCountWhenNonEmpty);
} else {
frameRowCount =
- builder4.append("totalRows", Expressions.condition(hasRows,
- rowCountWhenNonEmpty, Expressions.constant(0)));
+ builder4.append("totalRows",
+ Expressions.condition(hasRows, rowCountWhenNonEmpty,
+ Expressions.constant(0)));
}
ParameterExpression actualStart = Expressions.parameter(
@@ -585,10 +576,11 @@ public class EnumerableWindow extends Window implements EnumerableRel {
}
//noinspection UnnecessaryLocalVariable
- Expression res = block.append("rowInFrame", Expressions.foldAnd(
- ImmutableList.of(hasRows,
- Expressions.greaterThanOrEqual(rowIndex, minIndex),
- Expressions.lessThanOrEqual(rowIndex, maxIndex))));
+ Expression res = block.append("rowInFrame",
+ Expressions.foldAnd(
+ ImmutableList.of(hasRows,
+ Expressions.greaterThanOrEqual(rowIndex, minIndex),
+ Expressions.lessThanOrEqual(rowIndex, maxIndex))));
return res;
}
@@ -757,7 +749,7 @@ public class EnumerableWindow extends Window implements EnumerableRel {
for (final AggImpState agg: aggs) {
agg.context =
new WinAggContext() {
- public org.apache.calcite.sql.SqlAggFunction aggregation() {
+ public SqlAggFunction aggregation() {
return agg.call.getAggregation();
}
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindowRule.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindowRule.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindowRule.java
index 58e199f..a295470 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindowRule.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableWindowRule.java
@@ -23,8 +23,8 @@ import org.apache.calcite.rel.convert.ConverterRule;
import org.apache.calcite.rel.logical.LogicalWindow;
/**
- * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalAggregate} to
- * an {@link org.apache.calcite.adapter.enumerable.EnumerableAggregate}.
+ * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalWindow} to
+ * an {@link org.apache.calcite.adapter.enumerable.EnumerableWindow}.
*/
class EnumerableWindowRule extends ConverterRule {
EnumerableWindowRule() {
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/JavaRowFormat.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/JavaRowFormat.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/JavaRowFormat.java
index 146f420..ac87aad 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/JavaRowFormat.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/JavaRowFormat.java
@@ -199,7 +199,6 @@ public enum JavaRowFormat {
Type javaRowClass(
JavaTypeFactory typeFactory,
RelDataType type) {
- assert type.getFieldCount() > 1;
return Object[].class;
}
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/PhysTypeImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/PhysTypeImpl.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/PhysTypeImpl.java
index 54b6424..59ec646 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/PhysTypeImpl.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/PhysTypeImpl.java
@@ -77,10 +77,19 @@ public class PhysTypeImpl implements PhysType {
JavaTypeFactory typeFactory,
RelDataType rowType,
JavaRowFormat format) {
- final JavaRowFormat format2 = format.optimize(rowType);
- final Type javaRowClass = format2.javaRowClass(typeFactory, rowType);
- return new PhysTypeImpl(
- typeFactory, rowType, javaRowClass, format2);
+ return of(typeFactory, rowType, format, true);
+ }
+
+ public static PhysType of(
+ JavaTypeFactory typeFactory,
+ RelDataType rowType,
+ JavaRowFormat format,
+ boolean optimize) {
+ if (optimize) {
+ format = format.optimize(rowType);
+ }
+ final Type javaRowClass = format.javaRowClass(typeFactory, rowType);
+ return new PhysTypeImpl(typeFactory, rowType, javaRowClass, format);
}
static PhysType of(
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
index 1d721f8..e1fd9b0 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
@@ -70,7 +70,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.apache.calcite.DataContext.ROOT;
import static org.apache.calcite.linq4j.tree.ExpressionType.Add;
import static org.apache.calcite.linq4j.tree.ExpressionType.AndAlso;
import static org.apache.calcite.linq4j.tree.ExpressionType.Divide;
@@ -1695,6 +1694,7 @@ public class RexImpTable {
return Expressions.constant(true);
}
final SqlOperator op = call.getOperator();
+ final Expression root = translator.getRoot();
if (op == CURRENT_USER
|| op == SESSION_USER
|| op == USER) {
@@ -1707,15 +1707,15 @@ public class RexImpTable {
// the empty string because a role has to be set explicitly.
return Expressions.constant("");
} else if (op == CURRENT_TIMESTAMP) {
- return Expressions.call(BuiltInMethod.CURRENT_TIMESTAMP.method, ROOT);
+ return Expressions.call(BuiltInMethod.CURRENT_TIMESTAMP.method, root);
} else if (op == CURRENT_TIME) {
- return Expressions.call(BuiltInMethod.CURRENT_TIME.method, ROOT);
+ return Expressions.call(BuiltInMethod.CURRENT_TIME.method, root);
} else if (op == CURRENT_DATE) {
- return Expressions.call(BuiltInMethod.CURRENT_DATE.method, ROOT);
+ return Expressions.call(BuiltInMethod.CURRENT_DATE.method, root);
} else if (op == LOCALTIMESTAMP) {
- return Expressions.call(BuiltInMethod.LOCAL_TIMESTAMP.method, ROOT);
+ return Expressions.call(BuiltInMethod.LOCAL_TIMESTAMP.method, root);
} else if (op == LOCALTIME) {
- return Expressions.call(BuiltInMethod.LOCAL_TIME.method, ROOT);
+ return Expressions.call(BuiltInMethod.LOCAL_TIME.method, root);
} else {
throw new AssertionError("unknown function " + op);
}
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/enumerable/RexToLixTranslator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/RexToLixTranslator.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexToLixTranslator.java
index 92bd345..aceafa5 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/RexToLixTranslator.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexToLixTranslator.java
@@ -51,6 +51,8 @@ import org.apache.calcite.util.NlsString;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;
+import com.google.common.collect.ImmutableMap;
+
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Type;
@@ -86,6 +88,7 @@ public class RexToLixTranslator {
final JavaTypeFactory typeFactory;
final RexBuilder builder;
private final RexProgram program;
+ private final Expression root;
private final RexToLixTranslator.InputGetter inputGetter;
private final BlockBuilder list;
private final Map<? extends RexNode, Boolean> exprNullableMap;
@@ -101,13 +104,9 @@ public class RexToLixTranslator {
}
}
- private RexToLixTranslator(
- RexProgram program,
- JavaTypeFactory typeFactory,
- InputGetter inputGetter,
- BlockBuilder list) {
- this(
- program, typeFactory, inputGetter, list,
+ private RexToLixTranslator(RexProgram program, JavaTypeFactory typeFactory,
+ Expression root, InputGetter inputGetter, BlockBuilder list) {
+ this(program, typeFactory, root, inputGetter, list,
Collections.<RexNode, Boolean>emptyMap(),
new RexBuilder(typeFactory));
}
@@ -115,29 +114,32 @@ public class RexToLixTranslator {
private RexToLixTranslator(
RexProgram program,
JavaTypeFactory typeFactory,
+ Expression root,
InputGetter inputGetter,
BlockBuilder list,
Map<RexNode, Boolean> exprNullableMap,
RexBuilder builder) {
- this(program, typeFactory, inputGetter, list, exprNullableMap, builder,
- null);
+ this(program, typeFactory, root, inputGetter, list, exprNullableMap,
+ builder, null);
}
private RexToLixTranslator(
RexProgram program,
JavaTypeFactory typeFactory,
+ Expression root,
InputGetter inputGetter,
BlockBuilder list,
Map<? extends RexNode, Boolean> exprNullableMap,
RexBuilder builder,
RexToLixTranslator parent) {
- this(program, typeFactory, inputGetter, list, exprNullableMap, builder,
- parent, null);
+ this(program, typeFactory, root, inputGetter, list, exprNullableMap,
+ builder, parent, null);
}
private RexToLixTranslator(
RexProgram program,
JavaTypeFactory typeFactory,
+ Expression root,
InputGetter inputGetter,
BlockBuilder list,
Map<? extends RexNode, Boolean> exprNullableMap,
@@ -146,6 +148,7 @@ public class RexToLixTranslator {
Function1<String, InputGetter> correlates) {
this.program = program;
this.typeFactory = typeFactory;
+ this.root = root;
this.inputGetter = inputGetter;
this.list = list;
this.exprNullableMap = exprNullableMap;
@@ -162,16 +165,17 @@ public class RexToLixTranslator {
* @param typeFactory Type factory
* @param list List of statements, populated with declarations
* @param outputPhysType Output type, or null
+ * @param root Root expression
* @param inputGetter Generates expressions for inputs
* @param correlates Provider of references to the values of correlated
* variables
* @return Sequence of expressions, optional condition
*/
- public static List<Expression> translateProjects(
- RexProgram program,
+ public static List<Expression> translateProjects(RexProgram program,
JavaTypeFactory typeFactory,
BlockBuilder list,
PhysType outputPhysType,
+ Expression root,
InputGetter inputGetter,
Function1<String, InputGetter> correlates) {
List<Type> storageTypes = null;
@@ -182,7 +186,7 @@ public class RexToLixTranslator {
storageTypes.add(outputPhysType.getJavaFieldType(i));
}
}
- return new RexToLixTranslator(program, typeFactory, inputGetter, list)
+ return new RexToLixTranslator(program, typeFactory, root, inputGetter, list)
.setCorrelates(correlates)
.translateList(program.getProjectList(), storageTypes);
}
@@ -190,7 +194,8 @@ public class RexToLixTranslator {
/** Creates a translator for translating aggregate functions. */
public static RexToLixTranslator forAggregation(JavaTypeFactory typeFactory,
BlockBuilder list, InputGetter inputGetter) {
- return new RexToLixTranslator(null, typeFactory, inputGetter, list);
+ final ParameterExpression root = DataContext.ROOT;
+ return new RexToLixTranslator(null, typeFactory, root, inputGetter, list);
}
Expression translate(RexNode expr) {
@@ -509,9 +514,7 @@ public class RexToLixTranslator {
}
return nullAs.handle(
convert(
- Expressions.call(
- DataContext.ROOT,
- BuiltInMethod.DATA_CONTEXT_GET.method,
+ Expressions.call(root, BuiltInMethod.DATA_CONTEXT_GET.method,
Expressions.constant("?" + expr.getIndex())),
storageType));
}
@@ -677,8 +680,9 @@ public class RexToLixTranslator {
if (program.getCondition() == null) {
return RexImpTable.TRUE_EXPR;
}
+ final ParameterExpression root = DataContext.ROOT;
RexToLixTranslator translator =
- new RexToLixTranslator(program, typeFactory, inputGetter, list);
+ new RexToLixTranslator(program, typeFactory, root, inputGetter, list);
translator = translator.setCorrelates(correlates);
return translator.translate(
program.getCondition(),
@@ -844,14 +848,6 @@ public class RexToLixTranslator {
return Expressions.convert_(operand, toType);
}
- private static <T> T elvis(T t0, T t1) {
- return t0 != null ? t0 : t1;
- }
-
- private static <T> T elvis(T t0, T t1, T t2) {
- return t0 != null ? t0 : t1 != null ? t1 : t2;
- }
-
public Expression translateConstructor(
List<RexNode> operandList, SqlKind kind) {
switch (kind) {
@@ -936,23 +932,21 @@ public class RexToLixTranslator {
/** Creates a read-only copy of this translator that records that a given
* expression is nullable. */
- public RexToLixTranslator setNullable(Map<? extends RexNode,
- Boolean> nullable) {
+ public RexToLixTranslator setNullable(
+ Map<? extends RexNode, Boolean> nullable) {
if (nullable == null || nullable.isEmpty()) {
return this;
}
- return new RexToLixTranslator(
- program, typeFactory, inputGetter, list, nullable, builder, this,
- correlates);
+ return new RexToLixTranslator(program, typeFactory, root, inputGetter, list,
+ nullable, builder, this, correlates);
}
public RexToLixTranslator setBlock(BlockBuilder block) {
if (block == list) {
return this;
}
- return new RexToLixTranslator(
- program, typeFactory, inputGetter, block,
- Collections.<RexNode, Boolean>emptyMap(), builder, this, correlates);
+ return new RexToLixTranslator(program, typeFactory, root, inputGetter,
+ block, ImmutableMap.<RexNode, Boolean>of(), builder, this, correlates);
}
public RexToLixTranslator setCorrelates(
@@ -960,8 +954,7 @@ public class RexToLixTranslator {
if (this.correlates == correlates) {
return this;
}
- return new RexToLixTranslator(
- program, typeFactory, inputGetter, list,
+ return new RexToLixTranslator(program, typeFactory, root, inputGetter, list,
Collections.<RexNode, Boolean>emptyMap(), builder, this, correlates);
}
@@ -983,6 +976,10 @@ public class RexToLixTranslator {
return null;
}
+ public Expression getRoot() {
+ return root;
+ }
+
/** Translates a field of an input to an expression. */
public interface InputGetter {
Expression field(BlockBuilder list, int index, Type storageType);
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/java/ReflectiveSchema.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/java/ReflectiveSchema.java b/core/src/main/java/org/apache/calcite/adapter/java/ReflectiveSchema.java
index 27b5151..a7155a1 100644
--- a/core/src/main/java/org/apache/calcite/adapter/java/ReflectiveSchema.java
+++ b/core/src/main/java/org/apache/calcite/adapter/java/ReflectiveSchema.java
@@ -16,11 +16,13 @@
*/
package org.apache.calcite.adapter.java;
+import org.apache.calcite.DataContext;
import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.linq4j.Enumerator;
import org.apache.calcite.linq4j.Linq4j;
import org.apache.calcite.linq4j.QueryProvider;
import org.apache.calcite.linq4j.Queryable;
+import org.apache.calcite.linq4j.function.Function1;
import org.apache.calcite.linq4j.tree.Expression;
import org.apache.calcite.linq4j.tree.Expressions;
import org.apache.calcite.linq4j.tree.Primitive;
@@ -28,6 +30,7 @@ import org.apache.calcite.linq4j.tree.Types;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.Function;
+import org.apache.calcite.schema.ScannableTable;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaFactory;
import org.apache.calcite.schema.SchemaPlus;
@@ -178,7 +181,7 @@ public class ReflectiveSchema
/** Table that is implemented by reading from a Java object. */
private static class ReflectiveTable
extends AbstractQueryableTable
- implements Table {
+ implements Table, ScannableTable {
private final Type elementType;
private final Enumerable enumerable;
@@ -196,6 +199,16 @@ public class ReflectiveSchema
return Statistics.UNKNOWN;
}
+ public Enumerable<Object[]> scan(DataContext root) {
+ if (elementType == Object[].class) {
+ //noinspection unchecked
+ return enumerable;
+ } else {
+ //noinspection unchecked
+ return enumerable.select(new FieldSelector((Class) elementType));
+ }
+ }
+
public <T> Queryable<T> asQueryable(QueryProvider queryProvider,
SchemaPlus schema, String tableName) {
return new AbstractTableQueryable<T>(queryProvider, schema, this,
@@ -323,6 +336,27 @@ public class ReflectiveSchema
schema.getParentSchema(), schema.getName()), field);
}
}
+
+ /** Function that returns an array of a given object's field values. */
+ private static class FieldSelector implements Function1<Object, Object[]> {
+ private final Field[] fields;
+
+ public FieldSelector(Class elementType) {
+ this.fields = elementType.getFields();
+ }
+
+ public Object[] apply(Object o) {
+ try {
+ final Object[] objects = new Object[fields.length];
+ for (int i = 0; i < fields.length; i++) {
+ objects[i] = fields[i].get(o);
+ }
+ return objects;
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
}
// End ReflectiveSchema.java
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcTable.java b/core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcTable.java
index 135568b..7e68c52 100644
--- a/core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcTable.java
+++ b/core/src/main/java/org/apache/calcite/adapter/jdbc/JdbcTable.java
@@ -16,6 +16,7 @@
*/
package org.apache.calcite.adapter.jdbc;
+import org.apache.calcite.DataContext;
import org.apache.calcite.adapter.java.AbstractQueryableTable;
import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.jdbc.CalciteConnection;
@@ -33,6 +34,7 @@ import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rel.type.RelProtoDataType;
import org.apache.calcite.runtime.ResultSetEnumerable;
+import org.apache.calcite.schema.ScannableTable;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.TranslatableTable;
@@ -61,7 +63,8 @@ import java.util.List;
* The resulting queryable can then be converted to a SQL query, which can be
* executed efficiently on the JDBC server.</p>
*/
-class JdbcTable extends AbstractQueryableTable implements TranslatableTable {
+class JdbcTable extends AbstractQueryableTable
+ implements TranslatableTable, ScannableTable {
private RelProtoDataType protoRowType;
private final JdbcSchema jdbcSchema;
private final String jdbcCatalogName;
@@ -156,6 +159,13 @@ class JdbcTable extends AbstractQueryableTable implements TranslatableTable {
return new JdbcTableQueryable<T>(queryProvider, schema, tableName);
}
+ public Enumerable<Object[]> scan(DataContext root) {
+ final JavaTypeFactory typeFactory = root.getTypeFactory();
+ final SqlString sql = generateSql();
+ return ResultSetEnumerable.of(jdbcSchema.getDataSource(), sql.getSql(),
+ JdbcUtils.ObjectArrayRowBuilder.factory(fieldClasses(typeFactory)));
+ }
+
/** Enumerable that returns the contents of a {@link JdbcTable} by connecting
* to the JDBC data source. */
private class JdbcTableQueryable<T> extends AbstractTableQueryable<T> {
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/interpreter/AggregateNode.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/interpreter/AggregateNode.java b/core/src/main/java/org/apache/calcite/interpreter/AggregateNode.java
index 145b44e..1dff33a 100644
--- a/core/src/main/java/org/apache/calcite/interpreter/AggregateNode.java
+++ b/core/src/main/java/org/apache/calcite/interpreter/AggregateNode.java
@@ -16,15 +16,40 @@
*/
package org.apache.calcite.interpreter;
+import org.apache.calcite.DataContext;
+import org.apache.calcite.adapter.enumerable.AggAddContext;
+import org.apache.calcite.adapter.enumerable.AggImpState;
+import org.apache.calcite.adapter.enumerable.JavaRowFormat;
+import org.apache.calcite.adapter.enumerable.PhysType;
+import org.apache.calcite.adapter.enumerable.PhysTypeImpl;
+import org.apache.calcite.adapter.enumerable.RexToLixTranslator;
+import org.apache.calcite.adapter.enumerable.impl.AggAddContextImpl;
+import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.interpreter.Row.RowBuilder;
+import org.apache.calcite.linq4j.tree.BlockBuilder;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.linq4j.tree.Expressions;
+import org.apache.calcite.linq4j.tree.ParameterExpression;
import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.schema.impl.AggregateFunctionImpl;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.Pair;
+import com.google.common.base.Supplier;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -36,9 +61,12 @@ public class AggregateNode extends AbstractSingleNode<Aggregate> {
private final List<Grouping> groups = Lists.newArrayList();
private final ImmutableBitSet unionGroups;
private final int outputRowLength;
+ private final ImmutableList<AccumulatorFactory> accumulatorFactories;
+ private final DataContext dataContext;
public AggregateNode(Interpreter interpreter, Aggregate rel) {
super(interpreter, rel);
+ this.dataContext = interpreter.getDataContext();
ImmutableBitSet union = ImmutableBitSet.of();
@@ -53,10 +81,15 @@ public class AggregateNode extends AbstractSingleNode<Aggregate> {
this.outputRowLength = unionGroups.cardinality()
+ (rel.indicator ? unionGroups.cardinality() : 0)
+ rel.getAggCallList().size();
+
+ ImmutableList.Builder<AccumulatorFactory> builder = ImmutableList.builder();
+ for (AggregateCall aggregateCall : rel.getAggCallList()) {
+ builder.add(getAccumulator(aggregateCall));
+ }
+ accumulatorFactories = builder.build();
}
public void run() throws InterruptedException {
-
Row r;
while ((r = source.receive()) != null) {
for (Grouping group : groups) {
@@ -69,53 +102,173 @@ public class AggregateNode extends AbstractSingleNode<Aggregate> {
}
}
- private AccumulatorList getNewAccumList() {
- AccumulatorList list = new AccumulatorList();
- for (AggregateCall call : rel.getAggCallList()) {
- list.add(getAccumulator(call));
+ private AccumulatorFactory getAccumulator(final AggregateCall call) {
+ if (call.getAggregation() == SqlStdOperatorTable.COUNT) {
+ return new AccumulatorFactory() {
+ public Accumulator get() {
+ return new CountAccumulator(call);
+ }
+ };
+ } else if (call.getAggregation() == SqlStdOperatorTable.SUM) {
+ return new UdaAccumulatorFactory(
+ AggregateFunctionImpl.create(IntSum.class), call);
+ } else {
+ final JavaTypeFactory typeFactory =
+ (JavaTypeFactory) rel.getCluster().getTypeFactory();
+ int stateOffset = 0;
+ final AggImpState agg = new AggImpState(0, call, false);
+ int stateSize = agg.state.size();
+
+ final BlockBuilder builder2 = new BlockBuilder();
+ final PhysType inputPhysType =
+ PhysTypeImpl.of(typeFactory, rel.getInput().getRowType(),
+ JavaRowFormat.ARRAY);
+ final RelDataTypeFactory.FieldInfoBuilder builder = typeFactory.builder();
+ for (Expression expression : agg.state) {
+ builder.add("a",
+ typeFactory.createJavaType((Class) expression.getType()));
+ }
+ final PhysType accPhysType =
+ PhysTypeImpl.of(typeFactory, builder.build(), JavaRowFormat.ARRAY);
+ final ParameterExpression inParameter =
+ Expressions.parameter(inputPhysType.getJavaRowType(), "in");
+ final ParameterExpression acc_ =
+ Expressions.parameter(accPhysType.getJavaRowType(), "acc");
+
+ List<Expression> accumulator =
+ new ArrayList<Expression>(stateSize);
+ for (int j = 0; j < stateSize; j++) {
+ accumulator.add(accPhysType.fieldReference(acc_, j + stateOffset));
+ }
+ agg.state = accumulator;
+
+ AggAddContext addContext =
+ new AggAddContextImpl(builder2, accumulator) {
+ public List<RexNode> rexArguments() {
+ List<RelDataTypeField> inputTypes =
+ inputPhysType.getRowType().getFieldList();
+ List<RexNode> args = new ArrayList<RexNode>();
+ for (Integer index : agg.call.getArgList()) {
+ args.add(
+ new RexInputRef(index, inputTypes.get(index).getType()));
+ }
+ return args;
+ }
+
+ public RexToLixTranslator rowTranslator() {
+ return RexToLixTranslator.forAggregation(typeFactory,
+ currentBlock(),
+ new RexToLixTranslator.InputGetterImpl(
+ Collections.singletonList(
+ Pair.of((Expression) inParameter, inputPhysType))))
+ .setNullable(currentNullables());
+ }
+ };
+
+ agg.implementor.implementAdd(agg.context, addContext);
+
+ final ParameterExpression context_ =
+ Expressions.parameter(Context.class, "context");
+ final ParameterExpression outputValues_ =
+ Expressions.parameter(Object[].class, "outputValues");
+ Scalar addScalar =
+ JaninoRexCompiler.baz(context_, outputValues_, builder2.toBlock());
+ return new ScalarAccumulatorDef(null, addScalar, null,
+ rel.getInput().getRowType().getFieldCount(), stateSize, dataContext);
}
- return list;
}
- private static Accumulator getAccumulator(final AggregateCall call) {
- String agg = call.getAggregation().getName();
+ /** Accumulator for calls to the COUNT function. */
+ private static class CountAccumulator implements Accumulator {
+ private final AggregateCall call;
+ long cnt;
- if (agg.equals("COUNT")) {
- return new Accumulator() {
- long cnt = 0;
+ public CountAccumulator(AggregateCall call) {
+ this.call = call;
+ cnt = 0;
+ }
- public void send(Row row) {
- boolean notNull = true;
- for (Integer i : call.getArgList()) {
- if (row.getObject(i) == null) {
- notNull = false;
- break;
- }
- }
- if (notNull) {
- cnt++;
- }
+ public void send(Row row) {
+ boolean notNull = true;
+ for (Integer i : call.getArgList()) {
+ if (row.getObject(i) == null) {
+ notNull = false;
+ break;
}
+ }
+ if (notNull) {
+ cnt++;
+ }
+ }
- public Object end() {
- return cnt;
- }
+ public Object end() {
+ return cnt;
+ }
+ }
- };
- } else {
- throw new UnsupportedOperationException(
- String.format("Aggregate doesn't currently support "
- + "the %s aggregate function.", agg));
+ /** Creates an {@link Accumulator}. */
+ private interface AccumulatorFactory extends Supplier<Accumulator> {
+ }
+
+ /** Accumulator powered by {@link Scalar} code fragments. */
+ private static class ScalarAccumulatorDef implements AccumulatorFactory {
+ final Scalar initScalar;
+ final Scalar addScalar;
+ final Scalar endScalar;
+ final Context sendContext;
+ final Context endContext;
+ final int rowLength;
+ final int accumulatorLength;
+
+ private ScalarAccumulatorDef(Scalar initScalar, Scalar addScalar,
+ Scalar endScalar, int rowLength, int accumulatorLength,
+ DataContext root) {
+ this.initScalar = initScalar;
+ this.addScalar = addScalar;
+ this.endScalar = endScalar;
+ this.accumulatorLength = accumulatorLength;
+ this.rowLength = rowLength;
+ this.sendContext = new Context(root);
+ this.sendContext.values = new Object[rowLength + accumulatorLength];
+ this.endContext = new Context(root);
+ this.endContext.values = new Object[accumulatorLength];
+ }
+
+ public Accumulator get() {
+ return new ScalarAccumulator(this, new Object[accumulatorLength]);
+ }
+ }
+
+ /** Accumulator powered by {@link Scalar} code fragments. */
+ private static class ScalarAccumulator implements Accumulator {
+ final ScalarAccumulatorDef def;
+ final Object[] values;
+
+ private ScalarAccumulator(ScalarAccumulatorDef def, Object[] values) {
+ this.def = def;
+ this.values = values;
}
+ public void send(Row row) {
+ System.arraycopy(row.getValues(), 0, def.sendContext.values, 0,
+ def.rowLength);
+ System.arraycopy(values, 0, def.sendContext.values, def.rowLength,
+ values.length);
+ def.addScalar.execute(def.sendContext, values);
+ }
+
+ public Object end() {
+ System.arraycopy(values, 0, def.endContext.values, 0, values.length);
+ return def.endScalar.execute(def.endContext);
+ }
}
/**
- * Internal class to track groupings
+ * Internal class to track groupings.
*/
private class Grouping {
- private ImmutableBitSet grouping;
- private Map<Row, AccumulatorList> accum = Maps.newHashMap();
+ private final ImmutableBitSet grouping;
+ private final Map<Row, AccumulatorList> accumulators = Maps.newHashMap();
private Grouping(ImmutableBitSet grouping) {
this.grouping = grouping;
@@ -129,15 +282,19 @@ public class AggregateNode extends AbstractSingleNode<Aggregate> {
}
Row key = builder.build();
- if (!accum.containsKey(key)) {
- accum.put(key, getNewAccumList());
+ if (!accumulators.containsKey(key)) {
+ AccumulatorList list = new AccumulatorList();
+ for (AccumulatorFactory factory : accumulatorFactories) {
+ list.add(factory.get());
+ }
+ accumulators.put(key, list);
}
- accum.get(key).send(row);
+ accumulators.get(key).send(row);
}
public void end(Sink sink) throws InterruptedException {
- for (Map.Entry<Row, AccumulatorList> e : accum.entrySet()) {
+ for (Map.Entry<Row, AccumulatorList> e : accumulators.entrySet()) {
final Row key = e.getKey();
final AccumulatorList list = e.getValue();
@@ -188,6 +345,120 @@ public class AggregateNode extends AbstractSingleNode<Aggregate> {
void send(Row row);
Object end();
}
+
+ /** Implementation of {@code SUM} over INTEGER values as a user-defined
+ * aggregate. */
+ public static class IntSum {
+ public IntSum() {
+ }
+ public int init() {
+ return 0;
+ }
+ public int add(int accumulator, int v) {
+ return accumulator + v;
+ }
+ public int merge(int accumulator0, int accumulator1) {
+ return accumulator0 + accumulator1;
+ }
+ public int result(int accumulator) {
+ return accumulator;
+ }
+ }
+
+ /** Implementation of {@code SUM} over BIGINT values as a user-defined
+ * aggregate. */
+ public static class LongSum {
+ public LongSum() {
+ }
+ public long init() {
+ return 0L;
+ }
+ public long add(long accumulator, int v) {
+ return accumulator + v;
+ }
+ public long merge(long accumulator0, long accumulator1) {
+ return accumulator0 + accumulator1;
+ }
+ public long result(long accumulator) {
+ return accumulator;
+ }
+ }
+
+ /** Accumulator factory based on a user-defined aggregate function. */
+ private static class UdaAccumulatorFactory implements AccumulatorFactory {
+ public final AggregateFunctionImpl aggFunction;
+ public final int argOrdinal;
+ public final Object instance;
+
+ public UdaAccumulatorFactory(AggregateFunctionImpl aggFunction,
+ AggregateCall call) {
+ this.aggFunction = aggFunction;
+ if (call.getArgList().size() != 1) {
+ throw new UnsupportedOperationException("in current implementation, "
+ + "aggregate must have precisely one argument");
+ }
+ argOrdinal = call.getArgList().get(0);
+ if (aggFunction.isStatic) {
+ instance = null;
+ } else {
+ try {
+ instance = aggFunction.declaringClass.newInstance();
+ } catch (InstantiationException e) {
+ throw Throwables.propagate(e);
+ } catch (IllegalAccessException e) {
+ throw Throwables.propagate(e);
+ }
+ }
+ }
+
+ public Accumulator get() {
+ return new UdaAccumulator(this);
+ }
+ }
+
+ /** Accumulator based upon a user-defined aggregate. */
+ private static class UdaAccumulator implements Accumulator {
+ private final UdaAccumulatorFactory factory;
+ private Object value;
+
+ public UdaAccumulator(UdaAccumulatorFactory factory) {
+ this.factory = factory;
+ try {
+ this.value = factory.aggFunction.initMethod.invoke(factory.instance);
+ } catch (IllegalAccessException e) {
+ throw Throwables.propagate(e);
+ } catch (InvocationTargetException e) {
+ throw Throwables.propagate(e);
+ }
+ }
+
+ public void send(Row row) {
+ final Object[] args = {value, row.getValues()[factory.argOrdinal]};
+ for (int i = 1; i < args.length; i++) {
+ if (args[i] == null) {
+ return; // one of the arguments is null; don't add to the total
+ }
+ }
+ try {
+ value = factory.aggFunction.addMethod.invoke(factory.instance, args);
+ } catch (IllegalAccessException e) {
+ throw Throwables.propagate(e);
+ } catch (InvocationTargetException e) {
+ throw Throwables.propagate(e);
+ }
+ }
+
+ public Object end() {
+ final Object[] args = {value};
+ try {
+ return factory.aggFunction.resultMethod.invoke(factory.instance, args);
+ } catch (IllegalAccessException e) {
+ throw Throwables.propagate(e);
+ } catch (InvocationTargetException e) {
+ throw Throwables.propagate(e);
+ }
+ }
+ }
}
// End AggregateNode.java
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/interpreter/BindableConvention.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/interpreter/BindableConvention.java b/core/src/main/java/org/apache/calcite/interpreter/BindableConvention.java
new file mode 100644
index 0000000..d74745c
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/interpreter/BindableConvention.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.interpreter;
+
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.ConventionTraitDef;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelTrait;
+import org.apache.calcite.plan.RelTraitDef;
+
+/**
+ * Calling convention that returns results as an
+ * {@link org.apache.calcite.linq4j.Enumerable} of object arrays.
+ *
+ * <p>The relational expression needs to implement
+ * {@link org.apache.calcite.runtime.ArrayBindable}.
+ * Unlike {@link org.apache.calcite.adapter.enumerable.EnumerableConvention},
+ * no code generation is required.
+ */
+public enum BindableConvention implements Convention {
+ INSTANCE;
+
+ /** Cost of a bindable node versus implementing an equivalent node in a
+ * "typical" calling convention. */
+ public static final double COST_MULTIPLIER = 2.0d;
+
+ @Override public String toString() {
+ return getName();
+ }
+
+ public Class getInterface() {
+ return BindableRel.class;
+ }
+
+ public String getName() {
+ return "BINDABLE";
+ }
+
+ public RelTraitDef getTraitDef() {
+ return ConventionTraitDef.INSTANCE;
+ }
+
+ public boolean subsumes(RelTrait trait) {
+ return this == trait;
+ }
+
+ public void register(RelOptPlanner planner) {}
+}
+
+// End BindableConvention.java
http://git-wip-us.apache.org/repos/asf/incubator-calcite/blob/66cfb120/core/src/main/java/org/apache/calcite/interpreter/BindableRel.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/calcite/interpreter/BindableRel.java b/core/src/main/java/org/apache/calcite/interpreter/BindableRel.java
new file mode 100644
index 0000000..7bde2e5
--- /dev/null
+++ b/core/src/main/java/org/apache/calcite/interpreter/BindableRel.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.interpreter;
+
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.runtime.ArrayBindable;
+
+/**
+ * Relational expression that can implement itself in Bindable
+ * convention.
+ *
+ * @see org.apache.calcite.interpreter.BindableConvention
+ */
+public interface BindableRel extends RelNode, ArrayBindable, InterpretableRel {
+}
+
+// End BindableRel.java