fe/src/main/java/com/cloudera/impala/analysis/AggregateInfo.java
/**
   * Create the info for an aggregation node that computes the second phase of of
   * DISTINCT aggregate functions.
   * (Refer to createDistinctAggInfo() for an explanation of the phases.)
   * - 'this' is the phase 1 aggregation
   * - grouping exprs are those of the original query (param origGroupingExprs)
   * - aggregate exprs for the DISTINCT agg fns: these are aggregating the grouping
   *   slots that were added to the original grouping slots in phase 1;
   *   count is mapped to count(*) and sum is mapped to sum
   * - other aggregate exprs: same as the non-DISTINCT merge case
   *   (count is mapped to sum, everything else stays the same)
   *
   * This call also creates the tuple descriptor and smap for the returned AggregateInfo.
   */
Create the info for an aggregation node that computes the second phase of of
DISTINCT aggregate functions.

This call also creates the tuple descriptor and smap for the returned AggregateInfo.
Create the info for an aggregation node that computes the second phase of of DISTINCT aggregate functions . This call also creates the tuple descriptor and smap for the returned AggregateInfo .
private void createSecondPhaseDistinctAggInfo(
      ArrayList<Expr> origGroupingExprs,
      ArrayList<FunctionCallExpr> distinctAggExprs, Analyzer analyzer)
      throws InternalException, AuthorizationException {
    Preconditions.checkState(secondPhaseDistinctAggInfo_ == null);
    Preconditions.checkState(!distinctAggExprs.isEmpty());
    TupleDescriptor inputDesc = aggTupleDesc_;
    ArrayList<FunctionCallExpr> secondPhaseAggExprs = Lists.newArrayList();
    for (FunctionCallExpr inputExpr: distinctAggExprs) {
      Preconditions.checkState(inputExpr.isAggregateFunction());
      FunctionCallExpr aggExpr = null;
      if (inputExpr.getFnName().getFunction().equals("count")) {
        Expr ifExpr = createCountDistinctAggExprParam(origGroupingExprs.size(),
            origGroupingExprs.size() + inputExpr.getChildren().size() - 1,
            inputDesc.getSlots());
        Preconditions.checkNotNull(ifExpr);
        try {
          ifExpr.analyze(analyzer);
        } catch (AnalysisException e) {
          throw new InternalException("Failed to analyze 'IF' function " +
              "in second phase count distinct aggregation.", e);
        }
        aggExpr = new FunctionCallExpr("count",
            new FunctionParams(Lists.newArrayList(ifExpr)));
      } else {
        Preconditions.checkState(inputExpr.getFnName().getFunction().equals("sum"));
        Expr aggExprParam =
            new SlotRef(inputDesc.getSlots().get(origGroupingExprs.size()));
        aggExpr = new FunctionCallExpr("sum",
            new FunctionParams(Lists.newArrayList(aggExprParam)));
      }
      secondPhaseAggExprs.add(aggExpr);
    }
    for (int i = 0; i < aggregateExprs_.size(); ++i) {
      FunctionCallExpr inputExpr = aggregateExprs_.get(i);
      Preconditions.checkState(inputExpr.isAggregateFunction());
      Expr aggExprParam =
          new SlotRef(inputDesc.getSlots().get(i + getGroupingExprs().size()));
      List<Expr> aggExprParamList = Lists.newArrayList(aggExprParam);
      FunctionCallExpr aggExpr = null;
      if (inputExpr.getFnName().getFunction().equals("count")) {
        aggExpr = new FunctionCallExpr("sum",
            new FunctionParams(aggExprParamList));
      } else {
        aggExpr = new FunctionCallExpr(inputExpr, new FunctionParams(aggExprParamList));
      }
      secondPhaseAggExprs.add(aggExpr);
    }
    Preconditions.checkState(
        secondPhaseAggExprs.size() == aggregateExprs_.size() + distinctAggExprs.size());
    for (FunctionCallExpr aggExpr: secondPhaseAggExprs) {
      try {
        aggExpr.analyze(analyzer);
        Preconditions.checkState(aggExpr.isAggregateFunction());
      } catch (AnalysisException e) {
        throw new InternalException(
            "error constructing merge aggregation node", e);
      }
    }
    secondPhaseDistinctAggInfo_ =
        new AggregateInfo(
          Expr.cloneList(origGroupingExprs, aggTupleSMap_), secondPhaseAggExprs, true);
    secondPhaseDistinctAggInfo_.aggTupleDesc_ =
        secondPhaseDistinctAggInfo_.createAggTupleDesc(analyzer);
    secondPhaseDistinctAggInfo_.createSecondPhaseDistinctAggSMap(this, distinctAggExprs);
    secondPhaseDistinctAggInfo_.createMergeAggInfo(analyzer);
  }
private void createSecondPhaseDistinctAggInfo ( ArrayList < Expr > origGroupingExprs , ArrayList < FunctionCallExpr > distinctAggExprs , Analyzer analyzer ) throws InternalException , AuthorizationException { Preconditions . checkState ( secondPhaseDistinctAggInfo_ == null ) ; Preconditions . checkState ( ! distinctAggExprs . isEmpty ( ) ) ; TupleDescriptor inputDesc = aggTupleDesc_ ; ArrayList < FunctionCallExpr > secondPhaseAggExprs = Lists . newArrayList ( ) ; for ( FunctionCallExpr inputExpr : distinctAggExprs ) { Preconditions . checkState ( inputExpr . isAggregateFunction ( ) ) ; FunctionCallExpr aggExpr = null ; if ( inputExpr . getFnName ( ) . getFunction ( ) . equals ( "count" ) ) { Expr ifExpr = createCountDistinctAggExprParam ( origGroupingExprs . size ( ) , origGroupingExprs . size ( ) + inputExpr . getChildren ( ) . size ( ) - 1 , inputDesc . getSlots ( ) ) ; Preconditions . checkNotNull ( ifExpr ) ; try { ifExpr . analyze ( analyzer ) ; } catch ( AnalysisException e ) { throw new InternalException ( "Failed to analyze 'IF' function " + "in second phase count distinct aggregation." , e ) ; } aggExpr = new FunctionCallExpr ( "count" , new FunctionParams ( Lists . newArrayList ( ifExpr ) ) ) ; } else { Preconditions . checkState ( inputExpr . getFnName ( ) . getFunction ( ) . equals ( "sum" ) ) ; Expr aggExprParam = new SlotRef ( inputDesc . getSlots ( ) . get ( origGroupingExprs . size ( ) ) ) ; aggExpr = new FunctionCallExpr ( "sum" , new FunctionParams ( Lists . newArrayList ( aggExprParam ) ) ) ; } secondPhaseAggExprs . add ( aggExpr ) ; } for ( int i = 0 ; i < aggregateExprs_ . size ( ) ; ++ i ) { FunctionCallExpr inputExpr = aggregateExprs_ . get ( i ) ; Preconditions . checkState ( inputExpr . isAggregateFunction ( ) ) ; Expr aggExprParam = new SlotRef ( inputDesc . getSlots ( ) . get ( i + getGroupingExprs ( ) . size ( ) ) ) ; List < Expr > aggExprParamList = Lists . newArrayList ( aggExprParam ) ; FunctionCallExpr aggExpr = null ; if ( inputExpr . getFnName ( ) . getFunction ( ) . equals ( "count" ) ) { aggExpr = new FunctionCallExpr ( "sum" , new FunctionParams ( aggExprParamList ) ) ; } else { aggExpr = new FunctionCallExpr ( inputExpr , new FunctionParams ( aggExprParamList ) ) ; } secondPhaseAggExprs . add ( aggExpr ) ; } Preconditions . checkState ( secondPhaseAggExprs . size ( ) == aggregateExprs_ . size ( ) + distinctAggExprs . size ( ) ) ; for ( FunctionCallExpr aggExpr : secondPhaseAggExprs ) { try { aggExpr . analyze ( analyzer ) ; Preconditions . checkState ( aggExpr . isAggregateFunction ( ) ) ; } catch ( AnalysisException e ) { throw new InternalException ( "error constructing merge aggregation node" , e ) ; } } secondPhaseDistinctAggInfo_ = new AggregateInfo ( Expr . cloneList ( origGroupingExprs , aggTupleSMap_ ) , secondPhaseAggExprs , true ) ; secondPhaseDistinctAggInfo_ . aggTupleDesc_ = secondPhaseDistinctAggInfo_ . createAggTupleDesc ( analyzer ) ; secondPhaseDistinctAggInfo_ . createSecondPhaseDistinctAggSMap ( this , distinctAggExprs ) ; secondPhaseDistinctAggInfo_ . createMergeAggInfo ( analyzer ) ; }
Create the info for an aggregation node that computes the second phase of of
 DISTINCT aggregate functions.
Create the info for an aggregation node that computes the second phase of of DISTINCT aggregate functions .
// construct agg exprs for original DISTINCT aggregate functions // (these aren't part of aggExprs_) // COUNT(DISTINCT ...) -> // COUNT(IF(IsNull(<agg slot 1>), NULL, IF(IsNull(<agg slot 2>), NULL, ...))) // We need the nested IF to make sure that we do not count // column-value combinations if any of the distinct columns are NULL. // This behavior is consistent with MySQL. // SUM(DISTINCT <expr>) -> SUM(<last grouping slot>); // (MIN(DISTINCT ...) and MAX(DISTINCT ...) have their DISTINCT turned // off during analysis, and AVG() is changed to SUM()/COUNT()) // map all the remaining agg fns // we're aggregating an output slot of the 1st agg phase // TODO: remap types here. The inserted agg expr doesn't need to be the same // type. e.g. original expr takes bigint and returns bigint, but after // inserting the new aggexpr, it would be bigint -> string. string -> bigint. // we shouldn't see this