import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Override
public Void visitAggregate(Aggregate aggregate, List<Void> inputStreams) throws Exception {
  beginAggregateStage(aggregate);
  pw.println("        if (_data != null) {");
  pw.println("        List<Object> curGroupValues = getGroupValues(_data);");
  pw.println("        if (!correlatedGroupedValues.containsKey(curGroupValues)) {");
  pw.println("          correlatedGroupedValues.put(curGroupValues, new ArrayList<CorrelatedValues>());");
  pw.println("        }");
  pw.println("        correlatedGroupedValues.get(curGroupValues).add(_data);");
  pw.println("        if (!state.containsKey(curGroupValues)) {");
  pw.println("          state.put(curGroupValues, new HashMap<String, Object>());");
  pw.println("        }");
  pw.println("        Map<String, Object> accumulators = state.get(curGroupValues);");
  for (AggregateCall call : aggregate.getAggCallList()) {
    aggregate(call);
  }
  pw.println("        }");
  endStage();
  return null;
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
private boolean canPush(Aggregate aggregate, ImmutableBitSet rCols) {
  // If the filter references columns not in the group key, we cannot push
  final ImmutableBitSet groupKeys =
      ImmutableBitSet.range(0, aggregate.getGroupSet().cardinality());
  if (!groupKeys.contains(rCols)) {
    return false;
  }

  if (aggregate.indicator) {
    // If grouping sets are used, the filter can be pushed if
    // the columns referenced in the predicate are present in
    // all the grouping sets.
    for (ImmutableBitSet groupingSet : aggregate.getGroupSets()) {
      if (!groupingSet.contains(rCols)) {
        return false;
      }
    }
  }
  return true;
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Test public void testAggregateExtractProjectRuleWithFilter() {
  final String sql = "select sum(sal) filter (where empno = 40)\n"
      + "from emp";
  HepProgram pre = new HepProgramBuilder()
      .addRuleInstance(AggregateProjectMergeRule.INSTANCE)
      .build();
  // AggregateProjectMergeRule does not merges Project with Filter.
  // Force match Aggregate on top of Project once explicitly in unit test.
  final AggregateExtractProjectRule rule =
      new AggregateExtractProjectRule(
          operand(Aggregate.class,
              operand(Project.class, null,
                  new PredicateImpl<Project>() {
                    int matchCount = 0;

                    public boolean test(@Nullable Project project) {
                      return matchCount++ == 0;
                    }
                  },
                  none())),
          RelFactories.LOGICAL_BUILDER);
  sql(sql).withPre(pre).withRule(rule).checkUnchanged();
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
/**
 * Takes a bitmap representing a set of input references and extracts the
 * ones that reference the group by columns in an aggregate.
 *
 * @param groupKey the original bitmap
 * @param aggRel   the aggregate
 * @param childKey sets bits from groupKey corresponding to group by columns
 */
public static void setAggChildKeys(
    ImmutableBitSet groupKey,
    Aggregate aggRel,
    ImmutableBitSet.Builder childKey) {
  List<AggregateCall> aggCalls = aggRel.getAggCallList();
  for (int bit : groupKey) {
    if (bit < aggRel.getGroupCount()) {
      // group by column
      childKey.set(bit);
    } else {
      // aggregate column -- set a bit for each argument being
      // aggregated
      AggregateCall agg = aggCalls.get(bit
          - (aggRel.getGroupCount() + aggRel.getIndicatorCount()));
      for (Integer arg : agg.getArgList()) {
        childKey.set(arg);
      }
    }
  }
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
public Double getSelectivity(Aggregate rel, RelMetadataQuery mq,
    RexNode predicate) {
  final List<RexNode> notPushable = new ArrayList<>();
  final List<RexNode> pushable = new ArrayList<>();
  RelOptUtil.splitFilters(
      rel.getGroupSet(),
      predicate,
      pushable,
      notPushable);
  final RexBuilder rexBuilder = rel.getCluster().getRexBuilder();
  RexNode childPred =
      RexUtil.composeConjunction(rexBuilder, pushable, true);

  Double selectivity = mq.getSelectivity(rel.getInput(), childPred);
  if (selectivity == null) {
    return null;
  } else {
    RexNode pred =
        RexUtil.composeConjunction(rexBuilder, notPushable, true);
    return selectivity * RelMdUtil.guessSelectivity(pred);
  }
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
public Boolean areColumnsUnique(RelSubset rel, RelMetadataQuery mq,
    ImmutableBitSet columns, boolean ignoreNulls) {
  int nullCount = 0;
  for (RelNode rel2 : rel.getRels()) {
    if (rel2 instanceof Aggregate
        || rel2 instanceof Filter
        || rel2 instanceof Values
        || rel2 instanceof TableScan
        || simplyProjects(rel2, columns)) {
      try {
        final Boolean unique = mq.areColumnsUnique(rel2, columns, ignoreNulls);
        if (unique != null) {
          if (unique) {
            return true;
          }
        } else {
          ++nullCount;
        }
      } catch (CyclicMetadataException e) {
        // Ignore this relational expression; there will be non-cyclic ones
        // in this set.
      }
    }
  }
  return nullCount == 0 ? false : null;
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
private AggregateCall createAggregateCallWithBinding(
    RelDataTypeFactory typeFactory,
    SqlAggFunction aggFunction,
    RelDataType operandType,
    Aggregate oldAggRel,
    AggregateCall oldCall,
    int argOrdinal) {
  final Aggregate.AggCallBinding binding =
      new Aggregate.AggCallBinding(typeFactory, aggFunction,
          ImmutableList.of(operandType), oldAggRel.getGroupCount(),
          oldCall.filterArg >= 0);
  return AggregateCall.create(aggFunction,
      oldCall.isDistinct(),
      oldCall.isApproximate(),
      ImmutableIntList.of(argOrdinal),
      oldCall.filterArg,
      aggFunction.inferReturnType(binding),
      null);
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
public void onMatch(RelOptRuleCall call) {
  final Aggregate aggregate = call.rel(0);
  final RelNode input = call.rel(1);
  if (!aggregate.getAggCallList().isEmpty() || aggregate.indicator) {
    return;
  }
  final RelMetadataQuery mq = call.getMetadataQuery();
  if (!SqlFunctions.isTrue(mq.areColumnsUnique(input, aggregate.getGroupSet()))) {
    return;
  }
  // Distinct is "GROUP BY c1, c2" (where c1, c2 are a set of columns on
  // which the input is unique, i.e. contain a key) and has no aggregate
  // functions. It can be removed.
  final RelNode newInput = convert(input, aggregate.getTraitSet().simplify());

  // If aggregate was projecting a subset of columns, add a project for the
  // same effect.
  final RelBuilder relBuilder = call.builder();
  relBuilder.push(newInput);
  if (newInput.getRowType().getFieldCount()
      > aggregate.getRowType().getFieldCount()) {
    relBuilder.project(relBuilder.fields(aggregate.getGroupSet().asList()));
  }
  call.transformTo(relBuilder.build());
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
private boolean canPush(Aggregate aggregate, ImmutableBitSet rCols) {
  // If the filter references columns not in the group key, we cannot push
  final ImmutableBitSet groupKeys =
      ImmutableBitSet.range(0, aggregate.getGroupSet().cardinality());
  if (!groupKeys.contains(rCols)) {
    return false;
  }

  if (aggregate.getGroupType() != Group.SIMPLE) {
    // If grouping sets are used, the filter can be pushed if
    // the columns referenced in the predicate are present in
    // all the grouping sets.
    for (ImmutableBitSet groupingSet : aggregate.getGroupSets()) {
      if (!groupingSet.contains(rCols)) {
        return false;
      }
    }
  }
  return true;
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
public AggregateNode(Compiler compiler, Aggregate rel) {
  super(compiler, rel);
  this.dataContext = compiler.getDataContext();

  ImmutableBitSet union = ImmutableBitSet.of();

  if (rel.getGroupSets() != null) {
    for (ImmutableBitSet group : rel.getGroupSets()) {
      union = union.union(group);
      groups.add(new Grouping(group));
    }
  }

  this.unionGroups = union;
  this.outputRowLength = unionGroups.cardinality()
      + (rel.indicator ? unionGroups.cardinality() : 0)
      + rel.getAggCallList().size();

  ImmutableList.Builder<AccumulatorFactory> builder = ImmutableList.builder();
  for (AggregateCall aggregateCall : rel.getAggCallList()) {
    builder.add(getAccumulator(aggregateCall, false));
  }
  accumulatorFactories = builder.build();
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
/** Tests a singleton grouping set {2}, whose power set has only two elements,
 * { {2}, {} }. */
@Test public void testInduceGroupingType1() {
  final ImmutableBitSet groupSet = ImmutableBitSet.of(2);

  // Could be ROLLUP but we prefer CUBE
  List<ImmutableBitSet> groupSets = Lists.newArrayList();
  groupSets.add(groupSet);
  groupSets.add(ImmutableBitSet.of());
  assertEquals(Aggregate.Group.CUBE,
      Aggregate.Group.induce(groupSet, groupSets));

  groupSets = Lists.newArrayList();
  groupSets.add(ImmutableBitSet.of());
  assertEquals(Aggregate.Group.OTHER,
      Aggregate.Group.induce(groupSet, groupSets));

  groupSets = Lists.newArrayList();
  groupSets.add(groupSet);
  assertEquals(Aggregate.Group.SIMPLE,
      Aggregate.Group.induce(groupSet, groupSets));

  groupSets = Lists.newArrayList();
  assertEquals(Aggregate.Group.OTHER,
      Aggregate.Group.induce(groupSet, groupSets));
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Override
public Double getRowCount(Aggregate rel) {
  ImmutableBitSet groupKey = ImmutableBitSet.range(rel.getGroupCount());

  if (groupKey.isEmpty()) {
    return 1.0;
  } else {
    return super.getRowCount(rel);
  }
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Override
public Aggregate copy(RelTraitSet traitSet, RelNode input, boolean indicator, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
  try {
    return new DrillAggregateRel(getCluster(), traitSet, input, indicator, groupSet, groupSets, aggCalls);
  } catch (InvalidRelException e) {
    throw new AssertionError(e);
  }
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Override
public boolean matches(RelOptRuleCall call) {
  if (!super.matches(call)) {
    return false;
  }
  Aggregate oldAggRel = (Aggregate) call.rels[0];
  return containsAvgStddevVarCall(oldAggRel.getAggCallList());
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Override
public Aggregate copy(RelTraitSet traitSet, RelNode input, boolean indicator, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
  try {
    return new HashAggPrel(getCluster(), traitSet, input, indicator, groupSet, groupSets, aggCalls,
        this.getOperatorPhase());
  } catch (InvalidRelException e) {
    throw new AssertionError(e);
  }
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Override
public Aggregate copy(RelTraitSet traitSet, RelNode input, boolean indicator, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
  try {
    return new StreamAggPrel(getCluster(), traitSet, input, indicator, groupSet, groupSets, aggCalls, this.getOperatorPhase());
  } catch (InvalidRelException e) {
    throw new AssertionError(e);
  }
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
public Double getDistinctRowCount(Aggregate rel, RelMetadataQuery mq,
                                  ImmutableBitSet groupKey, RexNode predicate) {
  if (predicate == null || predicate.isAlwaysTrue()) {
    if (groupKey.isEmpty()) {
      return 1D;
    }
  }

  final ImmutableBitSet allGroupSet = rel.getGroupSet().union(groupKey);
  return getDistinctRowCountFromEstimateRowCount(rel.getInput(), mq, allGroupSet, predicate);
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Override
public Double getRowCount(Aggregate rel, RelMetadataQuery mq) {
  ImmutableBitSet groupKey = ImmutableBitSet.range(rel.getGroupCount());

  if (groupKey.isEmpty()) {
    return 1.0;
  } else {
    return rel.estimateRowCount(mq);
  }
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
@Test
public void testAggregationExpansionAddsDefaultMeasures() {
  // Set up the dimensions/measures in the test.
  final List<String> names = Arrays.asList("f_dimension_one", "f_measure_one");
  final List<SqlTypeName> types = Arrays.asList(SqlTypeName.TIMESTAMP, SqlTypeName.DOUBLE);
  final RelDataType[] rowTypes = getRowTypes(names, types);

  // Create the layout to be used.
  final int split = 1;
  final Layout layout = createLayout(names, types, split);

  // Test all the possible permutations crossed with nullable combinations.
  testExpandedLayouts(rowTypes, split, layout, new Command() {
    public void validate(RelDataType rowType, RelNode aggregation) {
      compareWithType(rowType, aggregation, names.get(0), SqlTypeName.DATE);

      final Aggregate aggregate = (Aggregate)aggregation;
      final Project child = (Project) aggregate.getInput();

      final List<AggregateCall> calls = aggregate.getAggCallList();
      assertTrue(calls.size() > 2);
      // check sum0(1)
      final AggregateCall sumCall = calls.get(calls.size() - 2);
      assertEquals(SqlKind.SUM0, sumCall.getAggregation().getKind());
      assertEquals(1, sumCall.getArgList().size());
      final int sumRefIndex = sumCall.getArgList().get(0);
      assertEquals(SqlKind.LITERAL, child.getProjects().get(sumRefIndex).getKind());

      // check count(1)
      final AggregateCall countCall = calls.get(calls.size() - 1);
      assertEquals(SqlKind.COUNT, countCall.getAggregation().getKind());
      assertEquals(1, countCall.getArgList().size());
      final int countRefIndex = sumCall.getArgList().get(0);
      assertEquals(SqlKind.LITERAL, child.getProjects().get(countRefIndex).getKind());
    }
  });
}
 

import org.apache.calcite.rel.core.Aggregate; //导入依赖的package包/类
private void testExpandedLayouts(RelDataType[] rowTypes, int split, Layout layout, Command command) {
  for (final RelDataType rowType : rowTypes) {
    when(schema.getTableForMember(anyListOf(String.class))).thenReturn(table);
    when(table.getRowType()).thenReturn(rowType);

    final RelNode rel = builder.scan("xyz").build();
    final LayoutExpander expander = new LayoutExpander(rel, true);
    final RelNode aggregation = expander.expand(layout);

    assertTrue(Aggregate.class.isAssignableFrom(aggregation.getClass()));

    // 2 default aggregations + number of dimensions
    int numFields = 2 + split;
    Collection<SqlAggFunction> emptyList = ImmutableList.of();

    // Determine number of measures based on measure type.
    for (LayoutField field :layout.getDetails().getMeasureFieldList()) {
      numFields += Optional
          .fromNullable(LayoutExpander.AGG_CALLS_PER_TYPE_WITH_MIN_MAX.get(TypeUtils.getSqlTypeFamily(field).get()))
          .or(emptyList)
          .size();
     }

    // 2 default aggregations + number dimensions + (4 * number measures)
    assertEquals(numFields, aggregation.getRowType().getFieldCount());

    command.validate(rowType, aggregation);
  }
}