Function CallSubqueryConsumeEager

src/execution_plan/ops/op_call_subquery.c:174–230

eagerly consume and aggregate all the records from the lhs (if exists). pass the aggregated record-list to the ArgumentList operation(s) after aggregating, return records to caller. if the subquery is returning, return the consumed record(s) from the body. otherwise, return the input record(s)

// return the consumed record(s) from the body. otherwise, return the input
// record(s)
static Record CallSubqueryConsumeEager
(
	OpBase *opBase  // operation
) {
	OpCallSubquery *op = (OpCallSubquery *)opBase;

	// if eager consumption has already occurred, don't consume again
	if(!op->first) {
		return _handoff_eager(op);
	}

	op->first = false;

	ASSERT(op->records == NULL);
	op->records = array_new(Record, 1);
	// eagerly consume all records from lhs if exists or create a
	// dummy-record, and place them\it in op->records
	Record r;
	if(op->lhs) {
		while((r = OpBase_Consume(op->lhs))) {
			array_append(op->records, r);
		}
		// propagate reset to lhs, to release RediSearch index locks (if any)
		OpBase_PropagateReset(op->lhs);
	} else {
		r = OpBase_CreateRecord((OpBase *)op);
		array_append(op->records, r);
	}

	_plant_records_ArgumentLists(op);

	int n_branches = (int)array_len(op->feeders.argumentLists);
	if(op->is_returning) {
		// give the last branch the original records
		ArgumentList_AddRecordList(
			op->feeders.argumentLists[n_branches - 1], op->records);

		// responsibility for the records is passed to the argumentList op(s)
		op->records = NULL;
	} else {
		// give the last branch a clone of the original record(s)
		Record *records_clone;
		array_clone_with_cb(records_clone, op->records,
			OpBase_DeepCloneRecord);
		ArgumentList_AddRecordList(
			op->feeders.argumentLists[n_branches - 1], records_clone);
	}

	if(!op->is_returning) {
		// deplete body and discard records
		while((r = OpBase_Consume(op->body))) {
			OpBase_DeleteRecord(r);
		}
	}

	return _handoff_eager(op);
}