@SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public Object iterate(int currentRow, LeadLagInfo leadLagInfo) throws HiveException {
      // // Currently avg(distinct) not supported in PartitionEvaluator
      if (((AbstractGenericUDAFAverageEvaluator)wrappedEvaluator).isWindowingDistinct()) {
        return super.iterate(currentRow, leadLagInfo);
      }

      Range currentRange = PTFRangeUtil.getRange(winFrame, currentRow, partition, nullsLast);
      if (currentRow == 0 ||  // Reset for the new partition
          avgAgg.prevRange == null ||
          currentRange.getSize() <= currentRange.getDiff(avgAgg.prevRange)) {
        Object[] partial = calcPartialResult(currentRange.iterator(), leadLagInfo);
        avgAgg.prevRange = currentRange;
        avgAgg.empty = false;
        avgAgg.prevSum = (ResultType)partial[1];
        avgAgg.prevCount = (long)partial[0];
      } else {
        // Given the previous range and the current range, calculate the new sum
        // from the previous sum and the difference to save the computation.
        Range r1 = newRange(avgAgg.prevRange.start, currentRange.start, partition);
        Range r2 = newRange(avgAgg.prevRange.end, currentRange.end, partition);
        Object[] partial1 = calcPartialResult(r1.iterator(), leadLagInfo);
        Object[] partial2 = calcPartialResult(r2.iterator(), leadLagInfo);
        ResultType sum = typeOperation.add(typeOperation.minus(avgAgg.prevSum, (ResultType)partial1[1]), (ResultType)partial2[1]);
        long count = avgAgg.prevCount - (long)partial1[0]+ (long)partial2[0];

        avgAgg.prevRange = currentRange;
        avgAgg.prevSum = sum;
        avgAgg.prevCount = count;
      }

      return typeOperation.div(avgAgg.prevSum, avgAgg.prevCount);
    }
Calculate the partial result sum + count giving a parition range
@return a 2-element Object array of [count long, sum ResultType]
ql/src/java/org/apache/hadoop/hive/ql/udf/ptf/BasePartitionEvaluator.java