/*
 * 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(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(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);
  }
}