Method reduce

query/aql_batchexecutor.go:219–254

reduce is to aggregate measures based on dimensions and aggregation function

()

// reduce is to aggregate measures based on dimensions and aggregation function
func (e *BatchExecutorImpl) reduce() {
	// init dimIndexVectorD for sorting and reducing
	if e.qc.OOPK.IsHLL() {
		initIndexVector(e.qc.OOPK.currentBatch.dimIndexVectorD[0].getPointer(), 0, e.qc.OOPK.currentBatch.resultSize, e.stream, e.qc.Device)
		initIndexVector(e.qc.OOPK.currentBatch.dimIndexVectorD[1].getPointer(), e.qc.OOPK.currentBatch.resultSize, e.qc.OOPK.currentBatch.resultSize+e.qc.OOPK.currentBatch.size, e.stream, e.qc.Device)
	} else if !e.qc.OOPK.UseHashReduction() {
		initIndexVector(e.qc.OOPK.currentBatch.dimIndexVectorD[0].getPointer(), 0, e.qc.OOPK.currentBatch.resultSize+e.qc.OOPK.currentBatch.size, e.stream, e.qc.Device)
	}

	if e.qc.OOPK.IsHLL() {
		e.qc.doProfile(func() {
			e.qc.OOPK.hllVectorD, e.qc.OOPK.hllDimRegIDCountD, e.qc.OOPK.hllVectorSize =
				e.qc.OOPK.currentBatch.hll(e.qc.OOPK.NumDimsPerDimWidth, e.isLastBatch, e.stream, e.qc.Device)
			e.qc.reportTimingForCurrentBatch(e.stream, &e.start, hllEvalTiming)
		}, "hll", e.stream)
	} else if e.qc.OOPK.UseHashReduction() {
		e.qc.doProfile(func() {
			e.qc.OOPK.currentBatch.hashReduce(
				e.qc.OOPK.NumDimsPerDimWidth, e.qc.OOPK.MeasureBytes, e.qc.OOPK.AggregateType, e.stream, e.qc.Device)
			e.qc.reportTimingForCurrentBatch(e.stream, &e.start, hashReduceEvalTiming)
		}, "hash_reduce", e.stream)
	} else {
		// sort by key.
		e.qc.doProfile(func() {
			e.qc.OOPK.currentBatch.sortByKey(e.qc.OOPK.NumDimsPerDimWidth, e.stream, e.qc.Device)
			e.qc.reportTimingForCurrentBatch(e.stream, &e.start, sortEvalTiming)
		}, "sort", e.stream)

		// reduce by key.
		e.qc.doProfile(func() {
			e.qc.OOPK.currentBatch.reduceByKey(e.qc.OOPK.NumDimsPerDimWidth, e.qc.OOPK.MeasureBytes, e.qc.OOPK.AggregateType, e.stream, e.qc.Device)
			e.qc.reportTimingForCurrentBatch(e.stream, &e.start, reduceEvalTiming)
		}, "reduce", e.stream)
	}
	cgoutils.WaitForCudaStream(e.stream, e.qc.Device)
}

func (e *BatchExecutorImpl) preExec(isLastBatch bool, start time.Time) {
	e.isLastBatch = isLastBatch