Function UpdateConsume

src/execution_plan/ops/op_update.c:77–123

}

static Record UpdateConsume
(
	OpBase *opBase
) {
	OpUpdate *op = (OpUpdate *)opBase;
	OpBase *child = op->op.children[0];
	Record r;

	// updates already performed
	if(op->updates_committed) return _handoff(op);

	while((r = OpBase_Consume(child))) {
		Record_PersistScalars(r);

		// evaluate update expressions
		raxSeek(&op->it, "^", NULL, 0);
		while(raxNext(&op->it)) {
			EntityUpdateEvalCtx *ctx = op->it.data;
			EvalEntityUpdates(op->gc, op->node_updates, op->edge_updates, r, ctx, true);
		}

		array_append(op->records, r);
	}
	
	uint node_updates_count = HashTableElemCount(op->node_updates);
	uint edge_updates_count = HashTableElemCount(op->edge_updates);

	if(node_updates_count > 0 || edge_updates_count > 0) {
		// done reading; we're not going to call Consume any longer
		// there might be operations like "Index Scan" that need to free the
		// index R/W lock - as such, free all ExecutionPlan operations up the chain.
		OpBase_PropagateReset(child);

		// lock everything
		QueryCtx_LockForCommit();

		CommitUpdates(op->gc, op->node_updates, ENTITY_NODE);
		CommitUpdates(op->gc, op->edge_updates, ENTITY_EDGE);
	}

	HashTableEmpty(op->node_updates, NULL);
	HashTableEmpty(op->edge_updates, NULL);

	op->updates_committed = true;

	return _handoff(op);
}

static OpBase *UpdateClone(const ExecutionPlan *plan, const OpBase *opBase) {
	ASSERT(opBase->type == OPType_UPDATE);