MaterializedViewAggregateRule.java
/*
* 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.rel.rules.materialize;
import org.apache.calcite.avatica.util.TimeUnitRange;
import org.apache.calcite.linq4j.Ord;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.plan.hep.HepPlanner;
import org.apache.calcite.plan.hep.HepProgram;
import org.apache.calcite.plan.hep.HepProgramBuilder;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.Filter;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.rules.AggregateProjectPullUpConstantsRule;
import org.apache.calcite.rel.rules.CoreRules;
import org.apache.calcite.rel.rules.FilterAggregateTransposeRule;
import org.apache.calcite.rel.rules.FilterProjectTransposeRule;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexPermuteInputsShuttle;
import org.apache.calcite.rex.RexSimplify;
import org.apache.calcite.rex.RexTableInputRef;
import org.apache.calcite.rex.RexTableInputRef.RelTableRef;
import org.apache.calcite.rex.RexUtil;
import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.sql.SqlFunction;
import org.apache.calcite.sql.fun.SqlMinMaxAggFunction;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.tools.RelBuilder.AggCall;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.mapping.Mapping;
import org.apache.calcite.util.mapping.MappingType;
import org.apache.calcite.util.mapping.Mappings;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.BiMap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Multimap;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.immutables.value.Value;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
/** Materialized view rewriting for aggregate.
*
* @param <C> Configuration type
*/
public abstract class MaterializedViewAggregateRule<C extends MaterializedViewAggregateRule.Config>
extends MaterializedViewRule<C> {
protected static final ImmutableList<TimeUnitRange> SUPPORTED_DATE_TIME_ROLLUP_UNITS =
ImmutableList.of(TimeUnitRange.YEAR, TimeUnitRange.QUARTER, TimeUnitRange.MONTH,
TimeUnitRange.DAY, TimeUnitRange.HOUR, TimeUnitRange.MINUTE,
TimeUnitRange.SECOND, TimeUnitRange.MILLISECOND, TimeUnitRange.MICROSECOND);
/** Creates a MaterializedViewAggregateRule. */
MaterializedViewAggregateRule(C config) {
super(config);
}
@Override protected boolean isValidPlan(@Nullable Project topProject, RelNode node,
RelMetadataQuery mq) {
if (!(node instanceof Aggregate)) {
return false;
}
Aggregate aggregate = (Aggregate) node;
if (aggregate.getGroupType() != Aggregate.Group.SIMPLE) {
// TODO: Rewriting with grouping sets not supported yet
return false;
}
return isValidRelNodePlan(aggregate.getInput(), mq);
}
@Override protected @Nullable ViewPartialRewriting compensateViewPartial(
RelBuilder relBuilder,
RexBuilder rexBuilder,
RelMetadataQuery mq,
RelNode input,
@Nullable Project topProject,
RelNode node,
Set<RelTableRef> queryTableRefs,
EquivalenceClasses queryEC,
@Nullable Project topViewProject,
RelNode viewNode,
Set<RelTableRef> viewTableRefs) {
// Modify view to join with missing tables and add Project on top to reorder columns.
// In turn, modify view plan to join with missing tables before Aggregate operator,
// change Aggregate operator to group by previous grouping columns and columns in
// attached tables, and add a final Project on top.
// We only need to add the missing tables on top of the view and view plan using
// a cartesian product.
// Then the rest of the rewriting algorithm can be executed in the same
// fashion, and if there are predicates between the existing and missing
// tables, the rewriting algorithm will enforce them.
final Set<RelTableRef> extraTableRefs = new HashSet<>();
for (RelTableRef tRef : queryTableRefs) {
if (!viewTableRefs.contains(tRef)) {
// Add to extra tables if table is not part of the view
extraTableRefs.add(tRef);
}
}
Multimap<Class<? extends RelNode>, RelNode> nodeTypes = mq.getNodeTypes(node);
if (nodeTypes == null) {
return null;
}
Collection<RelNode> tableScanNodes = nodeTypes.get(TableScan.class);
if (tableScanNodes == null) {
return null;
}
List<RelNode> newRels = new ArrayList<>();
for (RelTableRef tRef : extraTableRefs) {
int i = 0;
for (RelNode relNode : tableScanNodes) {
TableScan scan = (TableScan) relNode;
if (tRef.getQualifiedName().equals(scan.getTable().getQualifiedName())) {
if (tRef.getEntityNumber() == i++) {
newRels.add(relNode);
break;
}
}
}
}
assert extraTableRefs.size() == newRels.size();
relBuilder.push(input);
for (RelNode newRel : newRels) {
// Add to the view
relBuilder.push(newRel);
relBuilder.join(JoinRelType.INNER, rexBuilder.makeLiteral(true));
}
final RelNode newView = relBuilder.build();
final Aggregate aggregateViewNode = (Aggregate) viewNode;
relBuilder.push(aggregateViewNode.getInput());
int offset = 0;
for (RelNode newRel : newRels) {
// Add to the view plan
relBuilder.push(newRel);
relBuilder.join(JoinRelType.INNER, rexBuilder.makeLiteral(true));
offset += newRel.getRowType().getFieldCount();
}
// Modify aggregate: add grouping columns
ImmutableBitSet.Builder groupSet = ImmutableBitSet.builder();
groupSet.addAll(aggregateViewNode.getGroupSet());
groupSet.addAll(
ImmutableBitSet.range(
aggregateViewNode.getInput().getRowType().getFieldCount(),
aggregateViewNode.getInput().getRowType().getFieldCount() + offset));
final Aggregate newViewNode =
aggregateViewNode.copy(aggregateViewNode.getTraitSet(),
relBuilder.build(), groupSet.build(), null,
aggregateViewNode.getAggCallList());
relBuilder.push(newViewNode);
List<RexNode> nodes = new ArrayList<>();
List<String> fieldNames = new ArrayList<>();
if (topViewProject != null) {
// Insert existing expressions (and shift aggregation arguments),
// then append rest of columns
Mappings.TargetMapping shiftMapping =
Mappings.createShiftMapping(newViewNode.getRowType().getFieldCount(),
0, 0, aggregateViewNode.getGroupCount(),
newViewNode.getGroupCount(), aggregateViewNode.getGroupCount(),
aggregateViewNode.getAggCallList().size());
for (int i = 0; i < topViewProject.getProjects().size(); i++) {
nodes.add(
topViewProject.getProjects().get(i).accept(
new RexPermuteInputsShuttle(shiftMapping, newViewNode)));
fieldNames.add(topViewProject.getRowType().getFieldNames().get(i));
}
for (int i = aggregateViewNode.getRowType().getFieldCount();
i < newViewNode.getRowType().getFieldCount(); i++) {
int idx = i - aggregateViewNode.getAggCallList().size();
nodes.add(rexBuilder.makeInputRef(newViewNode, idx));
fieldNames.add(newViewNode.getRowType().getFieldNames().get(idx));
}
} else {
// Original grouping columns, aggregation columns, then new grouping columns
for (int i = 0; i < newViewNode.getRowType().getFieldCount(); i++) {
int idx;
if (i < aggregateViewNode.getGroupCount()) {
idx = i;
} else if (i < aggregateViewNode.getRowType().getFieldCount()) {
idx = i + offset;
} else {
idx = i - aggregateViewNode.getAggCallList().size();
}
nodes.add(rexBuilder.makeInputRef(newViewNode, idx));
fieldNames.add(newViewNode.getRowType().getFieldNames().get(idx));
}
}
relBuilder.project(nodes, fieldNames, true);
final Project newTopViewProject = (Project) relBuilder.build();
return ViewPartialRewriting.of(newView, newTopViewProject, newViewNode);
}
@Override protected @Nullable RelNode rewriteQuery(
RelBuilder relBuilder,
RexBuilder rexBuilder,
RexSimplify simplify,
RelMetadataQuery mq,
RexNode compensationColumnsEquiPred,
RexNode otherCompensationPred,
@Nullable Project topProject,
RelNode node,
BiMap<RelTableRef, RelTableRef> queryToViewTableMapping,
EquivalenceClasses viewEC, EquivalenceClasses queryEC) {
Aggregate aggregate = (Aggregate) node;
// Our target node is the node below the root, which should have the maximum
// number of available expressions in the tree in order to maximize our
// number of rewritings.
// If the program is available, we execute it to maximize rewriting opportunities.
// For instance, a program might pull up all the expressions that are below the
// aggregate so we can introduce compensation filters easily. This is important
// depending on the planner strategy.
RelNode newAggregateInput = aggregate.getInput(0);
RelNode target = aggregate.getInput(0);
HepProgram unionRewritingPullProgram = config.unionRewritingPullProgram();
if (unionRewritingPullProgram != null) {
final HepPlanner tmpPlanner = new HepPlanner(unionRewritingPullProgram);
tmpPlanner.setRoot(newAggregateInput);
newAggregateInput = tmpPlanner.findBestExp();
target = newAggregateInput.getInput(0);
}
// We need to check that all columns required by compensating predicates
// are contained in the query.
List<RexNode> queryExprs = extractReferences(rexBuilder, target);
if (!compensationColumnsEquiPred.isAlwaysTrue()) {
RexNode newCompensationColumnsEquiPred =
rewriteExpression(rexBuilder, mq, target, target, queryExprs,
queryToViewTableMapping, queryEC, false,
compensationColumnsEquiPred);
if (newCompensationColumnsEquiPred == null) {
// Skip it
return null;
}
compensationColumnsEquiPred = newCompensationColumnsEquiPred;
}
// For the rest, we use the query equivalence classes
if (!otherCompensationPred.isAlwaysTrue()) {
RexNode newOtherCompensationPred =
rewriteExpression(rexBuilder, mq, target, target, queryExprs,
queryToViewTableMapping, viewEC, true,
otherCompensationPred);
if (newOtherCompensationPred == null) {
// Skip it
return null;
}
otherCompensationPred = newOtherCompensationPred;
}
final RexNode queryCompensationPred =
RexUtil.not(
RexUtil.composeConjunction(rexBuilder,
ImmutableList.of(compensationColumnsEquiPred,
otherCompensationPred)));
// Generate query rewriting.
RelNode rewrittenPlan = relBuilder
.push(target)
.filter(simplify.simplifyUnknownAsFalse(queryCompensationPred))
.build();
if (config.unionRewritingPullProgram() != null) {
return aggregate.copy(aggregate.getTraitSet(),
ImmutableList.of(
newAggregateInput.copy(newAggregateInput.getTraitSet(),
ImmutableList.of(rewrittenPlan))));
}
return aggregate.copy(aggregate.getTraitSet(), ImmutableList.of(rewrittenPlan));
}
@Override protected @Nullable RelNode createUnion(RelBuilder relBuilder, RexBuilder rexBuilder,
@Nullable RelNode topProject, RelNode unionInputQuery, RelNode unionInputView) {
// Union
relBuilder.push(unionInputQuery);
relBuilder.push(unionInputView);
relBuilder.union(true);
List<RexNode> exprList = new ArrayList<>(relBuilder.peek().getRowType().getFieldCount());
List<String> nameList = new ArrayList<>(relBuilder.peek().getRowType().getFieldCount());
for (int i = 0; i < relBuilder.peek().getRowType().getFieldCount(); i++) {
// We can take unionInputQuery as it is query based.
RelDataTypeField field = unionInputQuery.getRowType().getFieldList().get(i);
exprList.add(
rexBuilder.ensureType(
field.getType(),
rexBuilder.makeInputRef(relBuilder.peek(), i),
true));
nameList.add(field.getName());
}
relBuilder.project(exprList, nameList);
// Rollup aggregate
Aggregate aggregate = (Aggregate) unionInputQuery;
final ImmutableBitSet groupSet = ImmutableBitSet.range(aggregate.getGroupCount());
final List<AggCall> aggregateCalls = new ArrayList<>();
for (int i = 0; i < aggregate.getAggCallList().size(); i++) {
AggregateCall aggCall = aggregate.getAggCallList().get(i);
if (aggCall.isDistinct()) {
// Cannot ROLLUP distinct
return null;
}
SqlAggFunction rollupAgg = aggCall.getAggregation().getRollup();
if (rollupAgg == null) {
// Cannot rollup this aggregate, bail out
return null;
}
final RexInputRef operand =
rexBuilder.makeInputRef(relBuilder.peek(),
aggregate.getGroupCount() + i);
aggregateCalls.add(
relBuilder.aggregateCall(aggCall.getParserPosition(), rollupAgg, operand)
.distinct(aggCall.isDistinct())
.approximate(aggCall.isApproximate())
.as(aggCall.name));
}
RelNode prevNode = relBuilder.peek();
RelNode result = relBuilder
.aggregate(relBuilder.groupKey(groupSet), aggregateCalls)
.build();
if (prevNode == result && groupSet.cardinality() != result.getRowType().getFieldCount()) {
// Aggregate was not inserted but we need to prune columns
result = relBuilder
.push(result)
.project(relBuilder.fields(groupSet))
.build();
}
if (topProject != null) {
// Top project
return topProject.copy(topProject.getTraitSet(), ImmutableList.of(result));
}
// Result
return result;
}
@Override protected @Nullable RelNode rewriteView(
RelBuilder relBuilder,
RexBuilder rexBuilder,
RexSimplify simplify,
RelMetadataQuery mq,
MatchModality matchModality,
boolean unionRewriting,
RelNode input,
@Nullable Project topProject,
RelNode node,
@Nullable Project topViewProject0,
RelNode viewNode,
BiMap<RelTableRef, RelTableRef> queryToViewTableMapping,
EquivalenceClasses queryEC) {
final Aggregate queryAggregate = (Aggregate) node;
final Aggregate viewAggregate = (Aggregate) viewNode;
// Get group by references and aggregate call input references needed
final ImmutableBitSet.Builder indexes = ImmutableBitSet.builder();
final ImmutableBitSet references;
if (topProject != null && !unionRewriting) {
// We have a Project on top, gather only what is needed
final RelOptUtil.InputFinder inputFinder =
new RelOptUtil.InputFinder(new LinkedHashSet<>());
inputFinder.visitEach(topProject.getProjects());
references = inputFinder.build();
for (int i = 0; i < queryAggregate.getGroupCount(); i++) {
indexes.set(queryAggregate.getGroupSet().nth(i));
}
for (int i = 0; i < queryAggregate.getAggCallList().size(); i++) {
if (references.get(queryAggregate.getGroupCount() + i)) {
for (int inputIdx : queryAggregate.getAggCallList().get(i).getArgList()) {
indexes.set(inputIdx);
}
}
}
} else {
// No project on top, all of them are needed
for (int i = 0; i < queryAggregate.getGroupCount(); i++) {
indexes.set(queryAggregate.getGroupSet().nth(i));
}
for (AggregateCall queryAggCall : queryAggregate.getAggCallList()) {
for (int inputIdx : queryAggCall.getArgList()) {
indexes.set(inputIdx);
}
}
references = null;
}
// Create mapping from query columns to view columns
final List<RexNode> rollupNodes = new ArrayList<>();
final Multimap<Integer, Integer> m =
generateMapping(rexBuilder, simplify, mq,
queryAggregate.getInput(), viewAggregate.getInput(), indexes.build(),
queryToViewTableMapping, queryEC, rollupNodes);
if (m == null) {
// Bail out
return null;
}
// We could map all expressions. Create aggregate mapping.
@SuppressWarnings("unused")
int viewAggregateAdditionalFieldCount = rollupNodes.size();
int viewInputFieldCount = viewAggregate.getInput().getRowType().getFieldCount();
int viewInputDifferenceViewFieldCount =
viewAggregate.getRowType().getFieldCount() - viewInputFieldCount;
int viewAggregateTotalFieldCount =
viewAggregate.getRowType().getFieldCount() + rollupNodes.size();
boolean forceRollup = false;
Mapping aggregateMapping =
Mappings.create(MappingType.FUNCTION,
queryAggregate.getRowType().getFieldCount(),
viewAggregateTotalFieldCount);
for (int i = 0; i < queryAggregate.getGroupCount(); i++) {
Collection<Integer> c = m.get(queryAggregate.getGroupSet().nth(i));
for (int j : c) {
if (j >= viewAggregate.getInput().getRowType().getFieldCount()) {
// This is one of the rollup columns
aggregateMapping.set(i, j + viewInputDifferenceViewFieldCount);
forceRollup = true;
} else {
int targetIdx = viewAggregate.getGroupSet().indexOf(j);
if (targetIdx == -1) {
continue;
}
aggregateMapping.set(i, targetIdx);
}
break;
}
if (aggregateMapping.getTargetOpt(i) == -1) {
// It is not part of group by, we bail out
return null;
}
}
boolean containsDistinctAgg = false;
for (Ord<AggregateCall> ord : Ord.zip(queryAggregate.getAggCallList())) {
if (references != null
&& !references.get(queryAggregate.getGroupCount() + ord.i)) {
// Ignore
continue;
}
final AggregateCall queryAggCall = ord.e;
if (queryAggCall.filterArg >= 0) {
// Not supported currently
return null;
}
List<Integer> queryAggCallIndexes = new ArrayList<>();
for (int aggCallIdx : queryAggCall.getArgList()) {
queryAggCallIndexes.add(m.get(aggCallIdx).iterator().next());
}
for (int j = 0; j < viewAggregate.getAggCallList().size(); j++) {
AggregateCall viewAggCall = viewAggregate.getAggCallList().get(j);
if (queryAggCall.getAggregation().getKind() != viewAggCall.getAggregation().getKind()
|| queryAggCall.isDistinct() != viewAggCall.isDistinct()
|| queryAggCall.getArgList().size() != viewAggCall.getArgList().size()
|| queryAggCall.getType() != viewAggCall.getType()
|| viewAggCall.filterArg >= 0) {
// Continue
continue;
}
if (!queryAggCallIndexes.equals(viewAggCall.getArgList())) {
// Continue
continue;
}
aggregateMapping.set(queryAggregate.getGroupCount() + ord.i,
viewAggregate.getGroupCount() + j);
if (queryAggCall.isDistinct()) {
containsDistinctAgg = true;
}
break;
}
}
// To simplify things, create an identity topViewProject if not present.
final Project topViewProject = topViewProject0 != null
? topViewProject0
: (Project) relBuilder.push(viewNode)
.project(relBuilder.fields(), ImmutableList.of(), true)
.build();
// Generate result rewriting
final List<RexNode> additionalViewExprs = new ArrayList<>();
// Multimap is required since a column in the materialized view's project
// could map to multiple columns in the target query
final ImmutableMultimap<Integer, Integer> rewritingMapping;
relBuilder.push(input);
// We create view expressions that will be used in a Project on top of the
// view in case we need to rollup the expression
final List<RexNode> inputViewExprs = new ArrayList<>(relBuilder.fields());
if (forceRollup
|| queryAggregate.getGroupCount() != viewAggregate.getGroupCount()
|| matchModality == MatchModality.VIEW_PARTIAL) {
if (containsDistinctAgg) {
// Cannot rollup DISTINCT aggregate
return null;
}
// Target is coarser level of aggregation. Generate an aggregate.
final ImmutableMultimap.Builder<Integer, Integer> rewritingMappingB =
ImmutableMultimap.builder();
final ImmutableBitSet.Builder groupSetB = ImmutableBitSet.builder();
for (int i = 0; i < queryAggregate.getGroupCount(); i++) {
final int targetIdx = aggregateMapping.getTargetOpt(i);
if (targetIdx == -1) {
// No matching group by column, we bail out
return null;
}
if (targetIdx >= viewAggregate.getRowType().getFieldCount()) {
RexNode targetNode =
rollupNodes.get(targetIdx - viewInputFieldCount
- viewInputDifferenceViewFieldCount);
// We need to rollup this expression
final Multimap<RexNode, Integer> exprsLineage = ArrayListMultimap.create();
for (int r : RelOptUtil.InputFinder.bits(targetNode)) {
final int j = find(viewNode, r);
final int k = find(topViewProject, j);
if (k < 0) {
// No matching column needed for computed expression, bail out
return null;
}
final RexInputRef ref =
relBuilder.with(viewNode.getInput(0), b -> b.field(r));
exprsLineage.put(ref, k);
}
// We create the new node pointing to the index
groupSetB.set(inputViewExprs.size());
rewritingMappingB.put(inputViewExprs.size(), i);
additionalViewExprs.add(
new RexInputRef(targetIdx, targetNode.getType()));
// We need to create the rollup expression
RexNode rollupExpression =
requireNonNull(shuttleReferences(rexBuilder, targetNode, exprsLineage),
() -> "shuttleReferences produced null for targetNode="
+ targetNode + ", exprsLineage=" + exprsLineage);
inputViewExprs.add(rollupExpression);
} else {
// This expression should be referenced directly
final int k = find(topViewProject, targetIdx);
if (k < 0) {
// No matching group by column, we bail out
return null;
}
groupSetB.set(k);
rewritingMappingB.put(k, i);
}
}
final ImmutableBitSet groupSet = groupSetB.build();
final List<AggCall> aggregateCalls = new ArrayList<>();
for (Ord<AggregateCall> ord : Ord.zip(queryAggregate.getAggCallList())) {
final int sourceIdx = queryAggregate.getGroupCount() + ord.i;
if (references != null && !references.get(sourceIdx)) {
// Ignore
continue;
}
final int targetIdx =
aggregateMapping.getTargetOpt(sourceIdx);
if (targetIdx < 0) {
// No matching aggregation column, we bail out
return null;
}
final int k = find(topViewProject, targetIdx);
if (k < 0) {
// No matching aggregation column, we bail out
return null;
}
final AggregateCall queryAggCall = ord.e;
SqlAggFunction rollupAgg = queryAggCall.getAggregation().getRollup();
if (rollupAgg == null) {
// Cannot rollup this aggregate, bail out
return null;
}
rewritingMappingB.put(k,
queryAggregate.getGroupCount() + aggregateCalls.size());
final RexInputRef operand = rexBuilder.makeInputRef(input, k);
aggregateCalls.add(
relBuilder.aggregateCall(queryAggCall.getParserPosition(), rollupAgg, operand)
.approximate(queryAggCall.isApproximate())
.distinct(queryAggCall.isDistinct())
.as(queryAggCall.name));
}
// Create aggregate on top of input
final RelNode prevNode = relBuilder.peek();
if (inputViewExprs.size() > prevNode.getRowType().getFieldCount()) {
relBuilder.project(inputViewExprs);
}
relBuilder
.aggregate(relBuilder.groupKey(groupSet), aggregateCalls);
if (prevNode == relBuilder.peek()
&& groupSet.cardinality() != relBuilder.peek().getRowType().getFieldCount()) {
// Aggregate was not inserted but we need to prune columns
relBuilder.project(relBuilder.fields(groupSet));
}
// We introduce a project on top, as group by columns order is lost.
// Multimap is required since a column in the materialized view's project
// could map to multiple columns in the target query.
rewritingMapping = rewritingMappingB.build();
final ImmutableMultimap<Integer, Integer> inverseMapping = rewritingMapping.inverse();
final List<RexNode> projects = new ArrayList<>();
final ImmutableBitSet.Builder addedProjects = ImmutableBitSet.builder();
for (int i = 0; i < queryAggregate.getGroupCount(); i++) {
final int pos = groupSet.indexOf(inverseMapping.get(i).iterator().next());
addedProjects.set(pos);
projects.add(relBuilder.field(pos));
}
final ImmutableBitSet projectedCols = addedProjects.build();
// We add aggregate functions that are present in result to projection list
for (int i = 0; i < relBuilder.peek().getRowType().getFieldCount(); i++) {
if (!projectedCols.get(i)) {
projects.add(relBuilder.field(i));
}
}
relBuilder.project(projects);
} else {
rewritingMapping = null;
}
// Add query expressions on top. We first map query expressions to view
// expressions. Once we have done that, if the expression is contained
// and we have introduced already an operator on top of the input node,
// we use the mapping to resolve the position of the expression in the
// node.
final RelDataType topRowType;
final List<RexNode> topExprs = new ArrayList<>();
if (topProject != null && !unionRewriting) {
topExprs.addAll(topProject.getProjects());
topRowType = topProject.getRowType();
} else {
// Add all
for (int pos = 0; pos < queryAggregate.getRowType().getFieldCount(); pos++) {
topExprs.add(rexBuilder.makeInputRef(queryAggregate, pos));
}
topRowType = queryAggregate.getRowType();
}
// Available in view.
final Multimap<RexNode, Integer> viewExprs = ArrayListMultimap.create();
addAllIndexed(viewExprs, topViewProject.getProjects());
addAllIndexed(viewExprs, additionalViewExprs);
final List<RexNode> rewrittenExprs = new ArrayList<>(topExprs.size());
for (RexNode expr : topExprs) {
// First map through the aggregate
final RexNode e2 = shuttleReferences(rexBuilder, expr, aggregateMapping);
if (e2 == null) {
// Cannot map expression
return null;
}
// Next map through the last project
final RexNode e3 =
shuttleReferences(rexBuilder, e2, viewExprs,
relBuilder.peek(), rewritingMapping);
if (e3 == null) {
// Cannot map expression
return null;
}
rewrittenExprs.add(e3);
}
return relBuilder
.project(rewrittenExprs)
.convert(topRowType, false)
.build();
}
private static <K> void addAllIndexed(Multimap<K, Integer> multimap,
Iterable<? extends K> list) {
for (K k : list) {
multimap.put(k, multimap.size());
}
}
/** Given a relational expression with a single input (such as a Project or
* Aggregate) and the ordinal of an input field, returns the ordinal of the
* output field that references the input field. Or -1 if the field is not
* propagated.
*
* <p>For example, if {@code rel} is {@code Project(c0, c2)} (on input with
* columns (c0, c1, c2)), then {@code find(rel, 2)} returns 1 (c2);
* {@code find(rel, 1)} returns -1 (because c1 is not projected).
*
* <p>If {@code rel} is {@code Aggregate([0, 2], sum(1))}, then
* {@code find(rel, 2)} returns 1, and {@code find(rel, 1)} returns -1.
*
* @param rel Relational expression
* @param ref Ordinal of output field
* @return Ordinal of input field, or -1
*/
private static int find(RelNode rel, int ref) {
if (rel instanceof Project) {
Project project = (Project) rel;
for (Ord<RexNode> p : Ord.zip(project.getProjects())) {
if (p.e instanceof RexInputRef
&& ((RexInputRef) p.e).getIndex() == ref) {
return p.i;
}
}
}
if (rel instanceof Aggregate) {
Aggregate aggregate = (Aggregate) rel;
int k = aggregate.getGroupSet().indexOf(ref);
if (k >= 0) {
return k;
}
}
return -1;
}
/**
* Mapping from node expressions to target expressions.
*
* <p>If any of the expressions cannot be mapped, we return null.
*/
protected @Nullable Multimap<Integer, Integer> generateMapping(
RexBuilder rexBuilder,
RexSimplify simplify,
RelMetadataQuery mq,
RelNode node,
RelNode target,
ImmutableBitSet positions,
BiMap<RelTableRef, RelTableRef> tableMapping,
EquivalenceClasses sourceEC,
List<RexNode> additionalExprs) {
checkArgument(additionalExprs.isEmpty());
Multimap<Integer, Integer> m = ArrayListMultimap.create();
Map<RexTableInputRef, Set<RexTableInputRef>> equivalenceClassesMap =
sourceEC.getEquivalenceClassesMap();
Multimap<RexNode, Integer> exprsLineage = ArrayListMultimap.create();
final List<RexNode> timestampExprs = new ArrayList<>();
for (int i = 0; i < target.getRowType().getFieldCount(); i++) {
Set<RexNode> s = mq.getExpressionLineage(target, rexBuilder.makeInputRef(target, i));
if (s == null) {
// Bail out
continue;
}
// We only support project - filter - join, thus it should map to
// a single expression
final RexNode e = Iterables.getOnlyElement(s);
// Rewrite expr to be expressed on query tables
final RexNode simplified = simplify.simplifyUnknownAsFalse(e);
final RexNode expr =
RexUtil.swapTableColumnReferences(rexBuilder, simplified,
tableMapping.inverse(), equivalenceClassesMap);
exprsLineage.put(expr, i);
SqlTypeName sqlTypeName = expr.getType().getSqlTypeName();
if (sqlTypeName == SqlTypeName.TIMESTAMP
|| sqlTypeName == SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE
|| sqlTypeName == SqlTypeName.TIMESTAMP_TZ) {
timestampExprs.add(expr);
}
}
// If this is a column of TIMESTAMP (WITH LOCAL TIME ZONE)
// type, we add the possible rollup columns too.
// This way we will be able to match FLOOR(ts to HOUR) to
// FLOOR(ts to DAY) via FLOOR(FLOOR(ts to HOUR) to DAY)
for (RexNode timestampExpr : timestampExprs) {
for (TimeUnitRange value : SUPPORTED_DATE_TIME_ROLLUP_UNITS) {
final SqlFunction[] functions =
{getCeilSqlFunction(value), getFloorSqlFunction(value)};
for (SqlFunction function : functions) {
final RexNode call =
rexBuilder.makeCall(function,
timestampExpr, rexBuilder.makeFlag(value));
// References self-row
final RexNode rewrittenCall =
shuttleReferences(rexBuilder, call, exprsLineage);
if (rewrittenCall == null) {
continue;
}
// We add the CEIL or FLOOR expression to the additional
// expressions, replacing the child expression by the position that
// it references
additionalExprs.add(rewrittenCall);
// Then we simplify the expression and we add it to the expressions
// lineage so we can try to find a match.
final RexNode simplified =
simplify.simplifyUnknownAsFalse(call);
exprsLineage.put(simplified,
target.getRowType().getFieldCount() + additionalExprs.size() - 1);
}
}
}
for (int i : positions) {
Set<RexNode> s = mq.getExpressionLineage(node, rexBuilder.makeInputRef(node, i));
if (s == null) {
// Bail out
return null;
}
// We only support project - filter - join, thus it should map to
// a single expression
final RexNode e = Iterables.getOnlyElement(s);
// Rewrite expr to be expressed on query tables
final RexNode simplified = simplify.simplifyUnknownAsFalse(e);
RexNode targetExpr =
RexUtil.swapColumnReferences(rexBuilder, simplified,
equivalenceClassesMap);
final Collection<Integer> c = exprsLineage.get(targetExpr);
if (!c.isEmpty()) {
for (Integer j : c) {
m.put(i, j);
}
} else {
// If we did not find the expression, try to navigate it
RexNode rewrittenTargetExpr =
shuttleReferences(rexBuilder, targetExpr, exprsLineage);
if (rewrittenTargetExpr == null) {
// Some expressions were not present
return null;
}
m.put(i, target.getRowType().getFieldCount() + additionalExprs.size());
additionalExprs.add(rewrittenTargetExpr);
}
}
return m;
}
/**
* Get ceil function datetime.
*/
protected SqlFunction getCeilSqlFunction(TimeUnitRange flag) {
return SqlStdOperatorTable.CEIL;
}
/**
* Get floor function datetime.
*/
protected SqlFunction getFloorSqlFunction(TimeUnitRange flag) {
return SqlStdOperatorTable.FLOOR;
}
/**
* Get rollup aggregation function.
*/
@Deprecated // to be removed before 2.0
protected @Nullable SqlAggFunction getRollup(SqlAggFunction aggregation) {
if (aggregation == SqlStdOperatorTable.SUM
|| aggregation == SqlStdOperatorTable.SUM0
|| aggregation instanceof SqlMinMaxAggFunction
|| aggregation == SqlStdOperatorTable.ANY_VALUE) {
return aggregation;
} else if (aggregation == SqlStdOperatorTable.COUNT) {
return SqlStdOperatorTable.SUM0;
} else {
return null;
}
}
@Override public Pair<@Nullable RelNode, RelNode> pushFilterToOriginalViewPlan(RelBuilder builder,
@Nullable RelNode topViewProject, RelNode viewNode, RexNode cond) {
// We add (and push) the filter to the view plan before triggering the rewriting.
// This is useful in case some of the columns can be folded to same value after
// filter is added.
HepProgramBuilder pushFiltersProgram = new HepProgramBuilder();
if (topViewProject != null) {
pushFiltersProgram.addRuleInstance(config.filterProjectTransposeRule());
}
pushFiltersProgram
.addRuleInstance(config.filterAggregateTransposeRule())
.addRuleInstance(config.aggregateProjectPullUpConstantsRule())
.addRuleInstance(config.projectMergeRule());
final HepPlanner tmpPlanner = new HepPlanner(pushFiltersProgram.build());
// Now that the planner is created, push the node
RelNode topNode = builder
.push(topViewProject != null ? topViewProject : viewNode)
.filter(cond).build();
tmpPlanner.setRoot(topNode);
topNode = tmpPlanner.findBestExp();
RelNode resultTopViewProject = null;
RelNode resultViewNode = null;
while (topNode != null) {
if (topNode instanceof Project) {
if (resultTopViewProject != null) {
// Both projects could not be merged, we will bail out
return Pair.of(topViewProject, viewNode);
}
resultTopViewProject = topNode;
topNode = topNode.getInput(0);
} else if (topNode instanceof Aggregate) {
resultViewNode = topNode;
topNode = null;
} else {
// We move to the child
topNode = topNode.getInput(0);
}
}
return Pair.of(resultTopViewProject, requireNonNull(resultViewNode, "resultViewNode"));
}
/**
* Rule configuration.
*/
public interface Config extends MaterializedViewRule.Config {
/** Instance of rule to push filter through project. */
@Value.Default default RelOptRule filterProjectTransposeRule() {
return CoreRules.FILTER_PROJECT_TRANSPOSE.config
.withRelBuilderFactory(relBuilderFactory())
.as(FilterProjectTransposeRule.Config.class)
.withOperandFor(Filter.class, filter ->
!RexUtil.containsCorrelation(filter.getCondition()),
Project.class, project -> true)
.withCopyFilter(true)
.withCopyProject(true)
.toRule();
}
/** Sets {@link #filterProjectTransposeRule()}. */
Config withFilterProjectTransposeRule(RelOptRule rule);
/** Instance of rule to push filter through aggregate. */
@Value.Default default RelOptRule filterAggregateTransposeRule() {
return CoreRules.FILTER_AGGREGATE_TRANSPOSE.config
.withRelBuilderFactory(relBuilderFactory())
.as(FilterAggregateTransposeRule.Config.class)
.withOperandFor(Filter.class, Aggregate.class)
.toRule();
}
/** Sets {@link #filterAggregateTransposeRule()}. */
Config withFilterAggregateTransposeRule(RelOptRule rule);
/** Instance of rule to pull up constants into aggregate. */
@Value.Default default RelOptRule aggregateProjectPullUpConstantsRule() {
return AggregateProjectPullUpConstantsRule.Config.DEFAULT
.withRelBuilderFactory(relBuilderFactory())
.withDescription("AggFilterPullUpConstants")
.as(AggregateProjectPullUpConstantsRule.Config.class)
.withOperandFor(Aggregate.class, Filter.class)
.toRule();
}
/** Sets {@link #aggregateProjectPullUpConstantsRule()}. */
Config withAggregateProjectPullUpConstantsRule(RelOptRule rule);
/** Instance of rule to merge project operators. */
@Value.Default default RelOptRule projectMergeRule() {
return CoreRules.PROJECT_MERGE.config
.withRelBuilderFactory(relBuilderFactory())
.toRule();
}
/** Sets {@link #projectMergeRule()}. */
Config withProjectMergeRule(RelOptRule rule);
}
}