Function ApplyConsume

src/execution_plan/ops/op_apply.c:50–92

}

static Record ApplyConsume(OpBase *opBase) {
	Apply *op = (Apply *)opBase;

	while(true) {
		if(op->r == NULL) {
			// retrieve a Record from the bound branch
			op->r = OpBase_Consume(op->bound_branch);
			if(op->r == NULL) {
				return NULL; // Bound branch and this op are depleted.
			}

			// collect record for future freeing
			array_append(op->records, op->r);

			// Successfully pulled a new Record, propagate to the top of the RHS branch.
			if(op->op_arg) {
				Argument_AddRecord(op->op_arg, OpBase_CloneRecord(op->r));
			}
		}

		// pull a Record from the RHS branch
		Record rhs_record = OpBase_Consume(op->rhs_branch);

		if(rhs_record == NULL) {
			// RHS branch depleted for the current bound Record
			// free it and loop back to retrieve a new one
			op->r = NULL;
			// reset the RHS branch
			OpBase_PropagateReset(op->rhs_branch);
			continue;
		}

		// clone the bound Record and merge the RHS Record into it
		Record r = OpBase_CloneRecord(op->r);
		Record_Merge(r, rhs_record);
		// delete the RHS record, as it has been merged into r
		OpBase_DeleteRecord(rhs_record);

		return r;
	}

	return NULL;
}

static OpResult ApplyReset(OpBase *opBase) {
	Apply *op = (Apply *)opBase;