Function ValueHashJoinConsume

src/execution_plan/ops/op_value_hash_join.c:259–324

Produce a record by joining records coming from the left and right hand side of this operation

// records coming from the left and right hand side
// of this operation
static Record ValueHashJoinConsume
(
	OpBase *opBase
) {
	OpValueHashJoin *op = (OpValueHashJoin *)opBase;
	OpBase *right_child = op->op.children[1];

	// eager, pull from left branch until depleted
	if(op->cached_records == NULL) {
		_cache_records(op);
		// sort cache on intersect node ID
		_sort_cached_records(op);
	}

	// try to produce a record:
	// given a right hand side record R,
	// evaluate V = exp on R,
	// see if there are any cached records
	// which V evaluated to V:
	// X in cached_records and X[idx] = V
	// return merged record:
	// X merged with R

	Record l;
	if(op->number_of_intersections > 0) {
		while((l = _get_intersecting_record(op))) {
			// clone cached record before merging rhs
			Record c = OpBase_CloneRecord(l);
			Record_Merge(c, op->rhs_rec);
			return c;
		}
	}

	// if we're here there are no more
	// left hand side records which intersect with R
	// discard R
	if(op->rhs_rec) OpBase_DeleteRecord(op->rhs_rec);

	// try to get new right hand side record
	// which intersect with a left hand side record
	while(true) {
		// pull from right branch
		op->rhs_rec = right_child->consume(right_child);
		if(!op->rhs_rec) return NULL;

		// get value on which we're intersecting
		SIValue v = AR_EXP_Evaluate(op->rhs_exp, op->rhs_rec);

		bool found_intersection = _set_intersection_idx(op, v);
		SIValue_Free(v);
		// no intersection, discard R
		if(!found_intersection) {
			OpBase_DeleteRecord(op->rhs_rec);
			continue;
		}

		// found atleast one intersecting record
		l = _get_intersecting_record(op);