You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by fj...@apache.org on 2018/09/10 05:33:03 UTC
[incubator-druid] branch master updated: SQL: UNION ALL operator.
(#6314)
This is an automated email from the ASF dual-hosted git repository.
fjy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git
The following commit(s) were added to refs/heads/master by this push:
new 4669f08 SQL: UNION ALL operator. (#6314)
4669f08 is described below
commit 4669f0878fb91b0392bc46b898874fa348c51fb4
Author: Gian Merlino <gi...@gmail.com>
AuthorDate: Sun Sep 9 22:32:56 2018 -0700
SQL: UNION ALL operator. (#6314)
* SQL: UNION ALL operator.
* Remove unused import.
---
docs/content/querying/sql.md | 4 +
.../sql/calcite/planner/DruidConvertletTable.java | 2 +
.../apache/druid/sql/calcite/planner/Rules.java | 18 +-
.../org/apache/druid/sql/calcite/rel/DruidRel.java | 11 +-
.../druid/sql/calcite/rel/DruidUnionRel.java | 219 +++++++++++++++++++++
.../apache/druid/sql/calcite/rule/DruidRules.java | 33 ++--
.../druid/sql/calcite/rule/DruidSemiJoinRule.java | 17 +-
.../druid/sql/calcite/rule/DruidSortUnionRule.java | 79 ++++++++
.../druid/sql/calcite/rule/DruidUnionRule.java | 62 ++++++
.../apache/druid/sql/calcite/CalciteQueryTest.java | 64 +++++-
10 files changed, 478 insertions(+), 31 deletions(-)
diff --git a/docs/content/querying/sql.md b/docs/content/querying/sql.md
index 6942c88..e994347 100644
--- a/docs/content/querying/sql.md
+++ b/docs/content/querying/sql.md
@@ -44,6 +44,7 @@ FROM table
[ HAVING expr ]
[ ORDER BY expr [ ASC | DESC ], expr [ ASC | DESC ], ... ]
[ LIMIT limit ]
+[ UNION ALL <another query> ]
```
The FROM clause refers to either a Druid datasource, like `druid.foo`, an [INFORMATION_SCHEMA table](#retrieving-metadata), a
@@ -74,6 +75,9 @@ versions of Druid will support pushing down limits using the native GroupBy quer
adding a limit doesn't change performance very much, then it's likely that Druid didn't push down the limit for your
query.
+The "UNION ALL" operator can be used to fuse multiple queries together. Their results will be concatenated, and each
+query will run separately, back to back (not in parallel). Druid does not currently support "UNION" without "ALL".
+
Add "EXPLAIN PLAN FOR" to the beginning of any query to see how it would be run as a native Druid query. In this case,
the query will not actually be executed.
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidConvertletTable.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidConvertletTable.java
index 6e42108..443417b 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidConvertletTable.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidConvertletTable.java
@@ -68,6 +68,8 @@ public class DruidConvertletTable implements SqlRexConvertletTable
.add(SqlStdOperatorTable.SYMMETRIC_NOT_BETWEEN)
.add(SqlStdOperatorTable.ITEM)
.add(SqlStdOperatorTable.TIMESTAMP_ADD)
+ .add(SqlStdOperatorTable.UNION)
+ .add(SqlStdOperatorTable.UNION_ALL)
.build();
private final Map<SqlOperator, SqlRexConvertlet> table;
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/Rules.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/Rules.java
index c398af2..7fa66ef 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/Rules.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/Rules.java
@@ -20,15 +20,6 @@
package org.apache.druid.sql.calcite.planner;
import com.google.common.collect.ImmutableList;
-import org.apache.druid.sql.calcite.rel.QueryMaker;
-import org.apache.druid.sql.calcite.rule.CaseFilteredAggregatorRule;
-import org.apache.druid.sql.calcite.rule.DruidRelToBindableRule;
-import org.apache.druid.sql.calcite.rule.DruidRelToDruidRule;
-import org.apache.druid.sql.calcite.rule.DruidRules;
-import org.apache.druid.sql.calcite.rule.DruidSemiJoinRule;
-import org.apache.druid.sql.calcite.rule.DruidTableScanRule;
-import org.apache.druid.sql.calcite.rule.ProjectAggregatePruneUnusedCallRule;
-import org.apache.druid.sql.calcite.rule.SortCollapseRule;
import org.apache.calcite.interpreter.Bindables;
import org.apache.calcite.plan.RelOptLattice;
import org.apache.calcite.plan.RelOptMaterialization;
@@ -78,6 +69,15 @@ import org.apache.calcite.sql2rel.RelFieldTrimmer;
import org.apache.calcite.tools.Program;
import org.apache.calcite.tools.Programs;
import org.apache.calcite.tools.RelBuilder;
+import org.apache.druid.sql.calcite.rel.QueryMaker;
+import org.apache.druid.sql.calcite.rule.CaseFilteredAggregatorRule;
+import org.apache.druid.sql.calcite.rule.DruidRelToBindableRule;
+import org.apache.druid.sql.calcite.rule.DruidRelToDruidRule;
+import org.apache.druid.sql.calcite.rule.DruidRules;
+import org.apache.druid.sql.calcite.rule.DruidSemiJoinRule;
+import org.apache.druid.sql.calcite.rule.DruidTableScanRule;
+import org.apache.druid.sql.calcite.rule.ProjectAggregatePruneUnusedCallRule;
+import org.apache.druid.sql.calcite.rule.SortCollapseRule;
import java.util.List;
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidRel.java b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidRel.java
index db7ee54..bbf332a 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidRel.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidRel.java
@@ -20,9 +20,6 @@
package org.apache.druid.sql.calcite.rel;
import com.google.common.base.Throwables;
-import org.apache.druid.java.util.common.guava.Accumulator;
-import org.apache.druid.java.util.common.guava.Sequence;
-import org.apache.druid.sql.calcite.planner.PlannerContext;
import org.apache.calcite.DataContext;
import org.apache.calcite.interpreter.BindableRel;
import org.apache.calcite.interpreter.Node;
@@ -32,6 +29,9 @@ import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.AbstractRelNode;
+import org.apache.druid.java.util.common.guava.Accumulator;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
import javax.annotation.Nullable;
import java.util.List;
@@ -46,6 +46,11 @@ public abstract class DruidRel<T extends DruidRel> extends AbstractRelNode imple
this.queryMaker = queryMaker;
}
+ /**
+ * Returns the PartialDruidQuery associated with this DruidRel, and which can be built on top of. Returns null
+ * if this rel cannot be built on top of.
+ */
+ @Nullable
public abstract PartialDruidQuery getPartialDruidQuery();
/**
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidUnionRel.java b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidUnionRel.java
new file mode 100644
index 0000000..7b6fdde
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidUnionRel.java
@@ -0,0 +1,219 @@
+/*
+ * 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.druid.sql.calcite.rel;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.FluentIterable;
+import org.apache.calcite.interpreter.BindableConvention;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptCost;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Sequences;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class DruidUnionRel extends DruidRel<DruidUnionRel>
+{
+ private final RelDataType rowType;
+ private final List<RelNode> rels;
+ private final int limit;
+
+ private DruidUnionRel(
+ final RelOptCluster cluster,
+ final RelTraitSet traitSet,
+ final QueryMaker queryMaker,
+ final RelDataType rowType,
+ final List<RelNode> rels,
+ final int limit
+ )
+ {
+ super(cluster, traitSet, queryMaker);
+ this.rowType = rowType;
+ this.rels = rels;
+ this.limit = limit;
+ }
+
+ public static DruidUnionRel create(
+ final QueryMaker queryMaker,
+ final RelDataType rowType,
+ final List<RelNode> rels,
+ final int limit
+ )
+ {
+ Preconditions.checkState(rels.size() > 0, "rels must be nonempty");
+
+ return new DruidUnionRel(
+ rels.get(0).getCluster(),
+ rels.get(0).getTraitSet(),
+ queryMaker,
+ rowType,
+ new ArrayList<>(rels),
+ limit
+ );
+ }
+
+ @Override
+ @Nullable
+ public PartialDruidQuery getPartialDruidQuery()
+ {
+ return null;
+ }
+
+ @Override
+ public int getQueryCount()
+ {
+ return rels.stream().mapToInt(rel -> ((DruidRel) rel).getQueryCount()).sum();
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Sequence<Object[]> runQuery()
+ {
+ // Lazy: run each query in sequence, not all at once.
+ if (limit == 0) {
+ return Sequences.empty();
+ } else {
+ final Sequence baseSequence = Sequences.concat(
+ FluentIterable.from(rels).transform(rel -> ((DruidRel) rel).runQuery())
+ );
+
+ return limit > 0 ? baseSequence.limit(limit) : baseSequence;
+ }
+ }
+
+ @Override
+ public DruidUnionRel withPartialQuery(final PartialDruidQuery newQueryBuilder)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public DruidQuery toDruidQuery(final boolean finalizeAggregations)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public DruidQuery toDruidQueryForExplaining()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public DruidUnionRel asBindable()
+ {
+ return new DruidUnionRel(
+ getCluster(),
+ getTraitSet().replace(BindableConvention.INSTANCE),
+ getQueryMaker(),
+ rowType,
+ rels.stream().map(rel -> RelOptRule.convert(rel, BindableConvention.INSTANCE)).collect(Collectors.toList()),
+ limit
+ );
+ }
+
+ @Override
+ public DruidUnionRel asDruidConvention()
+ {
+ return new DruidUnionRel(
+ getCluster(),
+ getTraitSet().replace(DruidConvention.instance()),
+ getQueryMaker(),
+ rowType,
+ rels.stream().map(rel -> RelOptRule.convert(rel, DruidConvention.instance())).collect(Collectors.toList()),
+ limit
+ );
+ }
+
+ @Override
+ public List<RelNode> getInputs()
+ {
+ return rels;
+ }
+
+ @Override
+ public void replaceInput(int ordinalInParent, RelNode p)
+ {
+ rels.set(ordinalInParent, p);
+ }
+
+ @Override
+ public RelNode copy(final RelTraitSet traitSet, final List<RelNode> inputs)
+ {
+ return new DruidUnionRel(
+ getCluster(),
+ traitSet,
+ getQueryMaker(),
+ rowType,
+ inputs,
+ limit
+ );
+ }
+
+ @Override
+ public List<String> getDatasourceNames()
+ {
+ return rels.stream()
+ .flatMap(rel -> ((DruidRel<?>) rel).getDatasourceNames().stream())
+ .distinct()
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public RelWriter explainTerms(RelWriter pw)
+ {
+ super.explainTerms(pw);
+
+ for (int i = 0; i < rels.size(); i++) {
+ pw.input(StringUtils.format("input#%d", i), rels.get(i));
+ }
+
+ return pw.item("limit", limit);
+ }
+
+ @Override
+ protected RelDataType deriveRowType()
+ {
+ return rowType;
+ }
+
+ @Override
+ public RelOptCost computeSelfCost(final RelOptPlanner planner, final RelMetadataQuery mq)
+ {
+ return planner.getCostFactory().makeCost(rels.stream().mapToDouble(mq::getRowCount).sum(), 0, 0);
+ }
+
+ public int getLimit()
+ {
+ return limit;
+ }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidRules.java b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidRules.java
index 4acf982..07e414e 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidRules.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidRules.java
@@ -19,11 +19,8 @@
package org.apache.druid.sql.calcite.rule;
+import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableList;
-import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.sql.calcite.rel.DruidOuterQueryRel;
-import org.apache.druid.sql.calcite.rel.DruidRel;
-import org.apache.druid.sql.calcite.rel.PartialDruidQuery;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelOptRuleOperand;
@@ -32,12 +29,18 @@ import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.Filter;
import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.core.Sort;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.sql.calcite.rel.DruidOuterQueryRel;
+import org.apache.druid.sql.calcite.rel.DruidRel;
+import org.apache.druid.sql.calcite.rel.PartialDruidQuery;
import java.util.List;
import java.util.function.BiFunction;
public class DruidRules
{
+ public static final Predicate<DruidRel> CAN_BUILD_ON = druidRel -> druidRel.getPartialDruidQuery() != null;
+
private DruidRules()
{
// No instantiation.
@@ -90,7 +93,9 @@ public class DruidRules
DruidOuterQueryRule.FILTER_AGGREGATE,
DruidOuterQueryRule.FILTER_PROJECT_AGGREGATE,
DruidOuterQueryRule.PROJECT_AGGREGATE,
- DruidOuterQueryRule.AGGREGATE_SORT_PROJECT
+ DruidOuterQueryRule.AGGREGATE_SORT_PROJECT,
+ DruidUnionRule.instance(),
+ DruidSortUnionRule.instance()
);
}
@@ -106,7 +111,7 @@ public class DruidRules
)
{
super(
- operand(relClass, operand(DruidRel.class, any())),
+ operand(relClass, operand(DruidRel.class, null, CAN_BUILD_ON, any())),
StringUtils.format("%s(%s)", DruidQueryRule.class.getSimpleName(), stage)
);
this.stage = stage;
@@ -138,7 +143,7 @@ public class DruidRules
public abstract static class DruidOuterQueryRule extends RelOptRule
{
public static RelOptRule AGGREGATE = new DruidOuterQueryRule(
- operand(Aggregate.class, operand(DruidRel.class, any())),
+ operand(Aggregate.class, operand(DruidRel.class, null, CAN_BUILD_ON, any())),
"AGGREGATE"
)
{
@@ -160,7 +165,7 @@ public class DruidRules
};
public static RelOptRule FILTER_AGGREGATE = new DruidOuterQueryRule(
- operand(Aggregate.class, operand(Filter.class, operand(DruidRel.class, any()))),
+ operand(Aggregate.class, operand(Filter.class, operand(DruidRel.class, null, CAN_BUILD_ON, any()))),
"FILTER_AGGREGATE"
)
{
@@ -184,7 +189,10 @@ public class DruidRules
};
public static RelOptRule FILTER_PROJECT_AGGREGATE = new DruidOuterQueryRule(
- operand(Aggregate.class, operand(Project.class, operand(Filter.class, operand(DruidRel.class, any())))),
+ operand(
+ Aggregate.class,
+ operand(Project.class, operand(Filter.class, operand(DruidRel.class, null, CAN_BUILD_ON, any())))
+ ),
"FILTER_PROJECT_AGGREGATE"
)
{
@@ -210,7 +218,7 @@ public class DruidRules
};
public static RelOptRule PROJECT_AGGREGATE = new DruidOuterQueryRule(
- operand(Aggregate.class, operand(Project.class, operand(DruidRel.class, any()))),
+ operand(Aggregate.class, operand(Project.class, operand(DruidRel.class, null, CAN_BUILD_ON, any()))),
"PROJECT_AGGREGATE"
)
{
@@ -234,7 +242,10 @@ public class DruidRules
};
public static RelOptRule AGGREGATE_SORT_PROJECT = new DruidOuterQueryRule(
- operand(Project.class, operand(Sort.class, operand(Aggregate.class, operand(DruidRel.class, any())))),
+ operand(
+ Project.class,
+ operand(Sort.class, operand(Aggregate.class, operand(DruidRel.class, null, CAN_BUILD_ON, any())))
+ ),
"AGGREGATE_SORT_PROJECT"
)
{
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidSemiJoinRule.java b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidSemiJoinRule.java
index b2dffc7..97ea353 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidSemiJoinRule.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidSemiJoinRule.java
@@ -21,10 +21,6 @@ package org.apache.druid.sql.calcite.rule;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
-import org.apache.druid.sql.calcite.planner.PlannerConfig;
-import org.apache.druid.sql.calcite.rel.DruidRel;
-import org.apache.druid.sql.calcite.rel.DruidSemiJoin;
-import org.apache.druid.sql.calcite.rel.PartialDruidQuery;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelOptUtil;
@@ -37,6 +33,10 @@ import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.druid.sql.calcite.planner.PlannerConfig;
+import org.apache.druid.sql.calcite.rel.DruidRel;
+import org.apache.druid.sql.calcite.rel.DruidSemiJoin;
+import org.apache.druid.sql.calcite.rel.PartialDruidQuery;
/**
* Planner rule adapted from Calcite 1.11.0's SemiJoinRule.
@@ -59,7 +59,7 @@ public class DruidSemiJoinRule extends RelOptRule
};
private static final Predicate<DruidRel> IS_GROUP_BY = druidRel ->
- druidRel.getPartialDruidQuery().getAggregate() != null;
+ druidRel.getPartialDruidQuery() != null && druidRel.getPartialDruidQuery().getAggregate() != null;
private static final DruidSemiJoinRule INSTANCE = new DruidSemiJoinRule();
@@ -73,7 +73,12 @@ public class DruidSemiJoinRule extends RelOptRule
null,
IS_LEFT_OR_INNER,
some(
- operand(DruidRel.class, null, Predicates.not(IS_GROUP_BY), any()),
+ operand(
+ DruidRel.class,
+ null,
+ Predicates.and(DruidRules.CAN_BUILD_ON, Predicates.not(IS_GROUP_BY)),
+ any()
+ ),
operand(DruidRel.class, null, IS_GROUP_BY, any())
)
)
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidSortUnionRule.java b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidSortUnionRule.java
new file mode 100644
index 0000000..c350fef
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidSortUnionRule.java
@@ -0,0 +1,79 @@
+/*
+ * 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.druid.sql.calcite.rule;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.core.Sort;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.druid.sql.calcite.rel.DruidUnionRel;
+
+import java.util.Collections;
+
+public class DruidSortUnionRule extends RelOptRule
+{
+ private static final DruidSortUnionRule INSTANCE = new DruidSortUnionRule();
+
+ private DruidSortUnionRule()
+ {
+ super(operand(Sort.class, operand(DruidUnionRel.class, any())));
+ }
+
+ public static DruidSortUnionRule instance()
+ {
+ return INSTANCE;
+ }
+
+ @Override
+ public boolean matches(final RelOptRuleCall call)
+ {
+ // LIMIT, no ORDER BY
+ final Sort sort = call.rel(0);
+ return sort.collation.getFieldCollations().isEmpty() && sort.fetch != null;
+ }
+
+ @Override
+ public void onMatch(final RelOptRuleCall call)
+ {
+ final Sort sort = call.rel(0);
+ final DruidUnionRel unionRel = call.rel(1);
+
+ final int limit = RexLiteral.intValue(sort.fetch);
+ final int offset = sort.offset != null ? RexLiteral.intValue(sort.offset) : 0;
+
+ final DruidUnionRel newUnionRel = DruidUnionRel.create(
+ unionRel.getQueryMaker(),
+ unionRel.getRowType(),
+ unionRel.getInputs(),
+ unionRel.getLimit() >= 0 ? Math.min(limit + offset, unionRel.getLimit()) : limit + offset
+ );
+
+ if (offset == 0) {
+ call.transformTo(newUnionRel);
+ } else {
+ call.transformTo(
+ call.builder()
+ .push(newUnionRel)
+ .sortLimit(offset, -1, Collections.emptyList())
+ .build()
+ );
+ }
+ }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidUnionRule.java b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidUnionRule.java
new file mode 100644
index 0000000..1aefb98
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/rule/DruidUnionRule.java
@@ -0,0 +1,62 @@
+/*
+ * 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.druid.sql.calcite.rule;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Union;
+import org.apache.druid.sql.calcite.rel.DruidRel;
+import org.apache.druid.sql.calcite.rel.DruidUnionRel;
+
+import java.util.List;
+
+public class DruidUnionRule extends RelOptRule
+{
+ private static final DruidUnionRule INSTANCE = new DruidUnionRule();
+
+ private DruidUnionRule()
+ {
+ super(operand(Union.class, unordered(operand(DruidRel.class, any()))));
+ }
+
+ public static DruidUnionRule instance()
+ {
+ return INSTANCE;
+ }
+
+ @Override
+ public void onMatch(final RelOptRuleCall call)
+ {
+ final Union unionRel = call.rel(0);
+ final DruidRel someDruidRel = call.rel(1);
+ final List<RelNode> inputs = unionRel.getInputs();
+
+ if (unionRel.all) {
+ // Can only do UNION ALL.
+ call.transformTo(DruidUnionRel.create(
+ someDruidRel.getQueryMaker(),
+ unionRel.getRowType(),
+ inputs,
+ -1
+ ));
+ }
+ }
+}
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
index d7379de..8f9e0d4 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
@@ -23,6 +23,7 @@ import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
+import org.apache.calcite.plan.RelOptPlanner;
import org.apache.druid.common.config.NullHandling;
import org.apache.druid.hll.HLLCV1;
import org.apache.druid.java.util.common.DateTimes;
@@ -108,7 +109,6 @@ import org.apache.druid.sql.calcite.util.CalciteTests;
import org.apache.druid.sql.calcite.util.QueryLogHook;
import org.apache.druid.sql.calcite.util.SpecificSegmentsQuerySegmentWalker;
import org.apache.druid.sql.calcite.view.InProcessViewManager;
-import org.apache.calcite.plan.RelOptPlanner;
import org.hamcrest.CoreMatchers;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
@@ -188,7 +188,8 @@ public class CalciteQueryTest extends CalciteTestBase
return DateTimes.inferTzfromString("America/Los_Angeles");
}
};
- private static final PlannerConfig PLANNER_CONFIG_SEMI_JOIN_ROWS_LIMIT = new PlannerConfig() {
+ private static final PlannerConfig PLANNER_CONFIG_SEMI_JOIN_ROWS_LIMIT = new PlannerConfig()
+ {
@Override
public int getMaxSemiJoinRowsInMemory()
{
@@ -1618,6 +1619,65 @@ public class CalciteQueryTest extends CalciteTestBase
}
@Test
+ public void testUnionAll() throws Exception
+ {
+ testQuery(
+ "SELECT COUNT(*) FROM foo UNION ALL SELECT SUM(cnt) FROM foo UNION ALL SELECT COUNT(*) FROM foo",
+ ImmutableList.of(
+ Druids.newTimeseriesQueryBuilder()
+ .dataSource(CalciteTests.DATASOURCE1)
+ .intervals(QSS(Filtration.eternity()))
+ .granularity(Granularities.ALL)
+ .aggregators(AGGS(new CountAggregatorFactory("a0")))
+ .context(TIMESERIES_CONTEXT_DEFAULT)
+ .build(),
+ Druids.newTimeseriesQueryBuilder()
+ .dataSource(CalciteTests.DATASOURCE1)
+ .intervals(QSS(Filtration.eternity()))
+ .granularity(Granularities.ALL)
+ .aggregators(AGGS(new LongSumAggregatorFactory("a0", "cnt")))
+ .context(TIMESERIES_CONTEXT_DEFAULT)
+ .build(),
+ Druids.newTimeseriesQueryBuilder()
+ .dataSource(CalciteTests.DATASOURCE1)
+ .intervals(QSS(Filtration.eternity()))
+ .granularity(Granularities.ALL)
+ .aggregators(AGGS(new CountAggregatorFactory("a0")))
+ .context(TIMESERIES_CONTEXT_DEFAULT)
+ .build()
+ ),
+ ImmutableList.of(new Object[]{6L}, new Object[]{6L}, new Object[]{6L})
+ );
+ }
+
+ @Test
+ public void testUnionAllWithLimit() throws Exception
+ {
+ testQuery(
+ "SELECT * FROM ("
+ + "SELECT COUNT(*) FROM foo UNION ALL SELECT SUM(cnt) FROM foo UNION ALL SELECT COUNT(*) FROM foo"
+ + ") LIMIT 2",
+ ImmutableList.of(
+ Druids.newTimeseriesQueryBuilder()
+ .dataSource(CalciteTests.DATASOURCE1)
+ .intervals(QSS(Filtration.eternity()))
+ .granularity(Granularities.ALL)
+ .aggregators(AGGS(new CountAggregatorFactory("a0")))
+ .context(TIMESERIES_CONTEXT_DEFAULT)
+ .build(),
+ Druids.newTimeseriesQueryBuilder()
+ .dataSource(CalciteTests.DATASOURCE1)
+ .intervals(QSS(Filtration.eternity()))
+ .granularity(Granularities.ALL)
+ .aggregators(AGGS(new LongSumAggregatorFactory("a0", "cnt")))
+ .context(TIMESERIES_CONTEXT_DEFAULT)
+ .build()
+ ),
+ ImmutableList.of(new Object[]{6L}, new Object[]{6L})
+ );
+ }
+
+ @Test
public void testPruneDeadAggregators() throws Exception
{
// Test for ProjectAggregatePruneUnusedCallRule.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org