processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java
/**
   * Performs aggregation for all of the aggregators.
   *
   * @return Total incremental memory in bytes required by this step of the
   * aggregation. The returned value is non-zero only if
   * {@link #useMaxMemoryEstimates} is false.
   */
Performs aggregation for all of the aggregators.
@return Total incremental memory in bytes required by this step of the
aggregation. The returned value is non-zero only if
#useMaxMemoryEstimates is false.
Performs aggregation for all of the aggregators . @return Total incremental memory in bytes required by this step of the aggregation . The returned value is non - zero only if #useMaxMemoryEstimates is false .
private long doAggregate(
      AggregatorFactory[] metrics,
      Aggregator[] aggs,
      ThreadLocal<InputRow> rowContainer,
      InputRow row,
      List<String> parseExceptionsHolder
  )
  {
    rowContainer.set(row);
    long totalIncrementalBytes = 0L;
    for (int i = 0; i < metrics.length; i++) {
      final Aggregator agg;
      if (preserveExistingMetrics && row instanceof MapBasedRow && ((MapBasedRow) row).getEvent().containsKey(metrics[i].getName())) {
        agg = aggs[i + metrics.length];
      } else {
        agg = aggs[i];
      }
      synchronized (agg) {
        try {
          if (useMaxMemoryEstimates) {
            agg.aggregate();
          } else {
            totalIncrementalBytes += agg.aggregateWithSize();
          }
        }
        catch (ParseException e) {
          if (preserveExistingMetrics) {
            log.warn(e, "Failing ingestion as preserveExistingMetrics is enabled but selector of aggregator[%s] recieved incompatible type.", metrics[i].getName());
            throw e;
          } else {
            log.debug(e, "Encountered parse error, skipping aggregator[%s].", metrics[i].getName());
            parseExceptionsHolder.add(e.getMessage());
          }
        }
      }
    }
    rowContainer.set(null);
    return totalIncrementalBytes;
  }
private long doAggregate ( AggregatorFactory [ ] metrics , Aggregator [ ] aggs , ThreadLocal < InputRow > rowContainer , InputRow row , List < String > parseExceptionsHolder ) { rowContainer . set ( row ) ; long totalIncrementalBytes = 0L ; for ( int i = 0 ; i < metrics . length ; i ++ ) { final Aggregator agg ; if ( preserveExistingMetrics && row instanceof MapBasedRow && ( ( MapBasedRow ) row ) . getEvent ( ) . containsKey ( metrics [ i ] . getName ( ) ) ) { agg = aggs [ i + metrics . length ] ; } else { agg = aggs [ i ] ; } synchronized ( agg ) { try { if ( useMaxMemoryEstimates ) { agg . aggregate ( ) ; } else { totalIncrementalBytes += agg . aggregateWithSize ( ) ; } } catch ( ParseException e ) { if ( preserveExistingMetrics ) { log . warn ( e , "Failing ingestion as preserveExistingMetrics is enabled but selector of aggregator[%s] recieved incompatible type." , metrics [ i ] . getName ( ) ) ; throw e ; } else { log . debug ( e , "Encountered parse error, skipping aggregator[%s]." , metrics [ i ] . getName ( ) ) ; parseExceptionsHolder . add ( e . getMessage ( ) ) ; } } } } rowContainer . set ( null ) ; return totalIncrementalBytes ; }
Performs aggregation for all of the aggregators.
Performs aggregation for all of the aggregators .
// "aggregate" can throw ParseExceptions if a selector expects something but gets something else.