Function NewAggregateOp

src/execution_plan/ops/op_aggregate.c:225–265

}

OpBase *NewAggregateOp
(
	const ExecutionPlan *plan,
	AR_ExpNode **exps
) {
	OpAggregate *op = rm_malloc(sizeof(OpAggregate));

	op->groups               = HashTableCreate(&_dt);
	op->group_iter           = NULL;

	OpBase_Init((OpBase *)op, OPType_AGGREGATE, "Aggregate", NULL,
			AggregateConsume, AggregateReset, NULL, AggregateClone,
			AggregateFree, false, plan);

	// expand hashtable to 2048 slots
	int res = HashTableExpand(op->groups, 2048);
	ASSERT(res == DICT_OK);

	// migrate each expression to the keys array or
	// the aggregations array as appropriate
	_migrate_expressions(op, exps);
	array_free(exps);

	// the projected record will associate values with their resolved name
	// to ensure that space is allocated for each entry
	op->record_offsets = array_new(uint, op->aggregate_count + op->key_count);
	for(uint i = 0; i < op->key_count; i++) {
		// store the index of each key expression
		int record_idx = OpBase_Modifies((OpBase *)op,
				op->key_exps[i]->resolved_name);
		array_append(op->record_offsets, record_idx);
	}
	for(uint i = 0; i < op->aggregate_count; i++) {
		// store the index of each aggregating expression
		int record_idx = OpBase_Modifies((OpBase *)op,
				op->aggregate_exps[i]->resolved_name);
		array_append(op->record_offsets, record_idx);
	}

	return (OpBase *)op;
}

static Record AggregateConsume
(