added id seek clone (RedisGraph#977)

* added id seek clone

* added better comment on the range clone

* Update op_node_by_id_seek.c

Co-authored-by: Roi Lipman <swilly22@users.noreply.github.com>

src/execution_plan/ops/op_node_by_id_seek.c

@@ -13,6 +13,7 @@ static OpResult NodeByIdSeekInit(OpBase *opBase);
 static Record NodeByIdSeekConsume(OpBase *opBase);
 static Record NodeByIdSeekConsumeFromChild(OpBase *opBase);
 static OpResult NodeByIdSeekReset(OpBase *opBase);
+static OpBase *NodeByIdSeekClone(const ExecutionPlan *plan, const OpBase *opBase);
 static void NodeByIdSeekFree(OpBase *opBase);
 
 static inline int NodeByIdSeekToString(const OpBase *ctx, char *buf, uint buf_len) {
@@ -35,22 +36,27 @@ OpBase *NewNodeByIdSeekOp(const ExecutionPlan *plan, const QGNode *n, UnsignedRa
 	op->child_record = NULL;
 
 	op->minId = id_range->include_min ? id_range->min : id_range->min + 1;
-
-	// The largest possible entity ID is the same as Graph_RequiredMatrixDim.
-	NodeID maxId = id_range->include_max ? id_range->max : id_range->max - 1;
-	op->maxId = MIN(Graph_RequiredMatrixDim(op->g) - 1, maxId);
+	/* The largest possible entity ID is the same as Graph_RequiredMatrixDim.
+	 * This value will be set on Init, to allow operation clone be independent
+	 * on the current graph size.*/
+	op->maxId = id_range->include_max ? id_range->max : id_range->max - 1;
 
 	op->currentId = op->minId;
 
 	OpBase_Init((OpBase *)op, OPType_NODE_BY_ID_SEEK, "NodeByIdSeek", NodeByIdSeekInit,
-				NodeByIdSeekConsume, NodeByIdSeekReset, NodeByIdSeekToString, NULL, NodeByIdSeekFree, false, plan);
+				NodeByIdSeekConsume, NodeByIdSeekReset, NodeByIdSeekToString, NodeByIdSeekClone, NodeByIdSeekFree,
+				false, plan);
 
 	op->nodeRecIdx = OpBase_Modifies((OpBase *)op, n->alias);
 
 	return (OpBase *)op;
 }
 
 static OpResult NodeByIdSeekInit(OpBase *opBase) {
+	assert(opBase->type == OPType_NODE_BY_ID_SEEK);
+	NodeByIdSeek *op = (NodeByIdSeek *)opBase;
+	// The largest possible entity ID is the same as Graph_RequiredMatrixDim.
+	op->maxId = MIN(Graph_RequiredMatrixDim(op->g) - 1, op->maxId);
 	if(opBase->childCount > 0) opBase->consume = NodeByIdSeekConsumeFromChild;
 	return OP_OK;
 }
@@ -130,6 +136,23 @@ static OpResult NodeByIdSeekReset(OpBase *ctx) {
 	return OP_OK;
 }
 
+static OpBase *NodeByIdSeekClone(const ExecutionPlan *plan, const OpBase *opBase) {
+	assert(opBase->type == OPType_NODE_BY_ID_SEEK);
+	NodeByIdSeek *op = (NodeByIdSeek *)op;
+	UnsignedRange range;
+	range.min = op->minId;
+	range.max = op->maxId;
+	/* In order to clone the range set at the original op, the range must be inclusive so the clone will set the exact range as the original.
+	 * During the call to NewNodeByIdSeekOp with the range, the following lines are executed:
+	 * op->minId = id_range->include_min ? id_range->min : id_range->min + 1;
+	 * op->maxId = id_range->include_max ? id_range->max : id_range->max - 1;
+	 * Since the range object min equals to the origin minId, and so is the max equals to the origin maxId, and the range is inclusive
+	 * the clone will set its values to be the same as in the origin. */
+	range.include_min = true;
+	range.include_max = true;
+	return NewNodeByIdSeekOp(plan, op->n, &range);
+}
+
 static void NodeByIdSeekFree(OpBase *opBase) {
 	NodeByIdSeek *op = (NodeByIdSeek *)opBase;
 	if(op->child_record) {