Compile time optimizations (#134)

* identify compile time optimizations

* apply compile time optimizations before cacheing execution plan

* address PR comments

* removed apply limit and apply skip optimizations

src/commands/execution_ctx.c

@@ -9,6 +9,7 @@
 #include "../query_ctx.h"
 #include "../errors/errors.h"
 #include "../execution_plan/execution_plan_clone.h"
+#include "../execution_plan/optimizations/optimizer.h"
 
 static ExecutionType _GetExecutionTypeFromAST
 (
@@ -191,6 +192,9 @@ ExecutionCtx *ExecutionCtx_FromQuery
 			return NULL;
 		}
 
+		// apply compile time optimizations
+		Optimizer_CompileTimeOptimize(plan);
+
 		ExecutionCtx *exec_ctx = _ExecutionCtx_New(ast, plan, exec_type);
 		ret = Cache_SetGetValue(cache, q_str, exec_ctx);
 	} else {

src/execution_plan/execution_plan.c

@@ -376,7 +376,7 @@ ExecutionPlan *ExecutionPlan_FromTLS_AST(void) {
 void ExecutionPlan_PreparePlan(ExecutionPlan *plan) {
 	// Plan should be prepared only once.
 	ASSERT(!plan->prepared);
-	optimizePlan(plan);
+	Optimizer_RuntimeOptimize(plan);
 	plan->prepared = true;
 }
 

src/execution_plan/execution_plan_build/execution_plan_util.c

@@ -5,13 +5,16 @@
  */
 
 #include "execution_plan_util.h"
+#include "../ops/op_skip.h"
+#include "../ops/op_limit.h"
 
 // returns true if an operation in the op-tree rooted at `root` is eager
 bool ExecutionPlan_isEager
 (
     OpBase *root
 ) {
-	return ExecutionPlan_LocateOpMatchingTypes(root, EAGER_OPERATIONS, 6) != NULL;
+	return ExecutionPlan_LocateOpMatchingTypes(root, EAGER_OPERATIONS,
+			EAGER_OP_COUNT) != NULL;
 }
 
 OpBase *ExecutionPlan_LocateOpResolvingAlias
@@ -189,6 +192,62 @@ OpBase *ExecutionPlan_LocateReferencesExcludingOps
 	return resolving_op;
 }
 
+// scans plan from root via parent nodes until a limit operation is found
+// eager operation will terminate the scan
+// return true if a limit operation was found, in which case 'limit' is set
+// otherwise return false
+bool ExecutionPlan_ContainsLimit
+(
+	OpBase *root,    // root to start the scan from
+	uint64_t *limit  // limit value
+) {
+	ASSERT(root  != NULL);
+	ASSERT(limit != NULL);
+
+	while(root != NULL) {
+		// halt if we encounter an eager operation
+		if(ExecutionPlan_isEager(root)) return false;
+
+		// found a limit operation
+		if(root->type == OPType_LIMIT) {
+			*limit = ((const OpLimit*)root)->limit;
+			return true;
+		}
+
+		root = root->parent;
+	}
+
+	return false;
+}
+
+// scans plan from root via parent nodes until a skip operation is found
+// eager operation will terminate the scan
+// return true if a skip operation was found, in which case 'skip' is set
+// otherwise return false
+bool ExecutionPlan_ContainsSkip
+(
+	OpBase *root,   // root to start the scan from
+	uint64_t *skip  // skip value
+) {
+	ASSERT(root != NULL);
+	ASSERT(skip != NULL);
+
+	while(root != NULL) {
+		// halt if we encounter an eager operation
+		if(ExecutionPlan_isEager(root)) return false;
+
+		// found a skip operation
+		if(root->type == OPType_SKIP) {
+			*skip = ((const OpSkip*)root)->skip;
+			return true;
+		}
+
+		root = root->parent;
+	}
+
+	return false;
+}
+
 OpBase *ExecutionPlan_LocateReferences
 (
 	OpBase *root,
@@ -201,19 +260,25 @@ OpBase *ExecutionPlan_LocateReferences
 
 // populates `ops` with all operations with a type in `types` in an
 // execution plan, based at `root`
-static void _ExecutionPlan_CollectOpsMatchingTypes(OpBase *root, const OPType *types, int type_count,
-												  OpBase ***ops) {
+static void _ExecutionPlan_CollectOpsMatchingTypes
+(
+	OpBase *root,
+	const OPType *types,
+	int type_count,
+	OpBase ***ops
+) {
 	for(int i = 0; i < type_count; i++) {
-		// Check to see if the op's type matches any of the types we're searching for.
+		// check to see if the op's type matches any of the types provided
 		if(root->type == types[i]) {
 			array_append(*ops, root);
 			break;
 		}
 	}
 
 	for(int i = 0; i < root->childCount; i++) {
-		// Recursively visit children.
-		_ExecutionPlan_CollectOpsMatchingTypes(root->children[i], types, type_count, ops);
+		// recursively visit children
+		_ExecutionPlan_CollectOpsMatchingTypes(root->children[i], types,
+				type_count, ops);
 	}
 }
 
@@ -236,8 +301,7 @@ OpBase **ExecutionPlan_CollectOps
     OPType type
 ) {
 	OpBase **ops = array_new(OpBase *, 0);
-	const OPType type_arr[1] = {type};
-	_ExecutionPlan_CollectOpsMatchingTypes(root, type_arr, 1, &ops);
+	_ExecutionPlan_CollectOpsMatchingTypes(root, &type, 1, &ops);
 	return ops;
 }
 

src/execution_plan/execution_plan_build/execution_plan_util.h

@@ -85,6 +85,26 @@ OpBase *ExecutionPlan_LocateReferencesExcludingOps
     rax *refs_to_resolve
 );
 
+// scans plan from root via parent nodes until a limit operation is found
+// eager operation will terminate the scan
+// return true if a limit operation was found, in which case 'limit' is set
+// otherwise return false
+bool ExecutionPlan_ContainsLimit
+(
+	OpBase *root,    // root to start the scan from
+	uint64_t *limit  // limit value
+);
+
+// scans plan from root via parent nodes until a skip operation is found
+// eager operation will terminate the scan
+// return true if a skip operation was found, in which case 'skip' is set
+// otherwise return false
+bool ExecutionPlan_ContainsSkip
+(
+	OpBase *root,   // root to start the scan from
+	uint64_t *skip  // skip value
+);
+
 //------------------------------------------------------------------------------
 // ExecutionPlan_Collect API:
 // For collecting all matching operations in tree.

src/execution_plan/ops/op.c

@@ -238,6 +238,20 @@ bool OpBase_IsWriter
 	return op->writer;
 }
 
+// indicates if the operation is an eager operation
+bool OpBase_IsEager
+(
+	const OpBase *op
+) {
+	ASSERT(op != NULL);
+
+	for(int i = 0; i < EAGER_OP_COUNT; i++) {
+		if(op->type == EAGER_OPERATIONS[i]) return true;
+	}
+
+	return false;
+}
+
 void OpBase_UpdateConsume
 (
 	OpBase *op,

src/execution_plan/ops/op.h

@@ -255,6 +255,12 @@ bool OpBase_IsWriter
 	OpBase *op
 );
 
+// indicates if the operation is an eager operation
+bool OpBase_IsEager
+(
+	const OpBase *op
+);
+
 // update operation consume function
 void OpBase_UpdateConsume
 (

src/execution_plan/ops/op_cond_var_len_traverse.c

@@ -67,25 +67,35 @@ static inline void _setTraverseDirection(CondVarLenTraverse *op, const QGEdge *e
 	}
 }
 
-static inline void CondVarLenTraverseToString(const OpBase *ctx, sds *buf) {
+static inline void CondVarLenTraverseToString
+(
+	const OpBase *ctx,
+	sds *buf
+) {
 	// TODO: tmp, improve TraversalToString
 	CondVarLenTraverse *op = (CondVarLenTraverse *)ctx;
 	AlgebraicExpression_Optimize(&op->ae);
 	TraversalToString(ctx, buf, op->ae);
 }
 
-void CondVarLenTraverseOp_ExpandInto(CondVarLenTraverse *op) {
-	// Expand into doesn't performs any modifications.
+void CondVarLenTraverseOp_ExpandInto
+(
+	CondVarLenTraverse *op
+) {
+	// expand into doesn't performs any modifications
 	array_clear(op->op.modifies);
 	op->expandInto = true;
 	op->op.type = OPType_CONDITIONAL_VAR_LEN_TRAVERSE_EXPAND_INTO;
 	op->op.name = "Conditional Variable Length Traverse (Expand Into)";
 }
 
-inline void CondVarLenTraverseOp_SetFilter(CondVarLenTraverse *op,
-										   FT_FilterNode *ft) {
-	ASSERT(op != NULL);
-	ASSERT(ft != NULL);
+inline void CondVarLenTraverseOp_SetFilter
+(
+	CondVarLenTraverse *op,
+	FT_FilterNode *ft
+) {
+	ASSERT(op     != NULL);
+	ASSERT(ft     != NULL);
 	ASSERT(op->ft == NULL);
 
 	op->ft = ft;
@@ -326,12 +336,33 @@ static OpResult CondVarLenTraverseReset(OpBase *ctx) {
 	return OP_OK;
 }
 
-static OpBase *CondVarLenTraverseClone(const ExecutionPlan *plan, const OpBase *opBase) {
-	ASSERT(opBase->type == OPType_CONDITIONAL_VAR_LEN_TRAVERSE);
+static OpBase *CondVarLenTraverseClone
+(
+	const ExecutionPlan *plan,
+	const OpBase *opBase
+) {
+	ASSERT(opBase->type == OPType_CONDITIONAL_VAR_LEN_TRAVERSE ||
+		   opBase->type == OPType_CONDITIONAL_VAR_LEN_TRAVERSE_EXPAND_INTO);
+
+	// clone "conditional var length traversal" operation
+	// handels the case where the operation had been modified into an
+	// "conditional var len expand into"
 	CondVarLenTraverse *op = (CondVarLenTraverse *) opBase;
-	OpBase *op_clone = NewCondVarLenTraverseOp(plan, QueryCtx_GetGraph(),
-											   AlgebraicExpression_Clone(op->ae));
-	return op_clone;
+	OpBase *clone = NewCondVarLenTraverseOp(plan, QueryCtx_GetGraph(),
+			AlgebraicExpression_Clone(op->ae));
+
+	// clone filter tree
+	if(op->ft != NULL) {
+		CondVarLenTraverseOp_SetFilter((CondVarLenTraverse*)clone,
+				FilterTree_Clone(op->ft));
+	}
+
+	// switch to expand into
+	if(opBase->type == OPType_CONDITIONAL_VAR_LEN_TRAVERSE_EXPAND_INTO) {
+		CondVarLenTraverseOp_ExpandInto((CondVarLenTraverse*) clone);
+	}
+
+	return clone;
 }
 
 static void CondVarLenTraverseFree(OpBase *ctx) {

src/execution_plan/ops/op_conditional_traverse.c

@@ -4,10 +4,11 @@
  * the Server Side Public License v1 (SSPLv1).
  */
 
-#include "op_conditional_traverse.h"
 #include "RG.h"
-#include "shared/print_functions.h"
 #include "../../query_ctx.h"
+#include "shared/print_functions.h"
+#include "op_conditional_traverse.h"
+#include "../execution_plan_build/execution_plan_util.h"
 
 // default number of records to accumulate before traversing
 #define BATCH_SIZE 16
@@ -109,11 +110,14 @@ OpBase *NewCondTraverseOp
 
 static OpResult CondTraverseInit(OpBase *opBase) {
 	OpCondTraverse *op = (OpCondTraverse *)opBase;
-	// Create 'records' with this Init function as 'record_cap'
-	// might be set during optimization time (applyLimit)
-	// If cap greater than BATCH_SIZE is specified,
-	// use BATCH_SIZE as the value.
+
+	// in case this operation is restricted by a limit
+	// set record_cap to the specified limit
+	ExecutionPlan_ContainsLimit(opBase, &op->record_cap);
+
+	// record_cap should not be greater than BATCH_SIZE
 	if(op->record_cap > BATCH_SIZE) op->record_cap = BATCH_SIZE;
+
 	op->records = rm_calloc(op->record_cap, sizeof(Record));
 
 	return OP_OK;

src/execution_plan/ops/op_conditional_traverse.h

@@ -24,8 +24,8 @@ typedef struct {
 	RG_MatrixTupleIter iter;    // Iterator over M.
 	int srcNodeIdx;             // Source node index into record.
 	int destNodeIdx;            // Destination node index into record.
-	uint record_count;          // Number of held records.
-	uint record_cap;            // Max number of records to process.
+	uint64_t record_count;      // Number of held records.
+	uint64_t record_cap;        // Max number of records to process.
 	Record *records;            // Array of records.
 	Record r;                   // Currently selected record.
 } OpCondTraverse;

src/execution_plan/ops/op_edge_by_index_scan.c

@@ -87,11 +87,14 @@ static OpResult EdgeIndexScanInit
 
 	// source and destination nodes may or may not already be resolved
 	// missing nodes will be resolved by this operation
-	const  char  *src_alias   =  QGNode_Alias(QGEdge_Src(op->edge));
-	const  char  *dest_alias  =  QGNode_Alias(QGEdge_Dest(op->edge));
+	const char *src_alias  = QGNode_Alias(QGEdge_Src(op->edge));
+	const char *dest_alias = QGNode_Alias(QGEdge_Dest(op->edge));
 
-	op->srcAware   =  OpBase_ChildrenAware((OpBase *)op, src_alias, &op->srcRecIdx);
-	op->destAware  =  OpBase_ChildrenAware((OpBase *)op, dest_alias, &op->destRecIdx);
+	op->srcAware = OpBase_ChildrenAware((OpBase *)op, src_alias,
+			&op->srcRecIdx);
+
+	op->destAware = OpBase_ChildrenAware((OpBase *)op, dest_alias,
+			&op->destRecIdx);
 
 	if(!op->srcAware) {
 		op->srcRecIdx = OpBase_Modifies((OpBase *)op, src_alias);

src/execution_plan/ops/op_expand_into.c

@@ -5,8 +5,9 @@
  */
 
 #include "op_expand_into.h"
-#include "shared/print_functions.h"
 #include "../../query_ctx.h"
+#include "shared/print_functions.h"
+#include "../execution_plan_build/execution_plan_util.h"
 
 // default number of records to accumulate before traversing
 #define BATCH_SIZE 16
@@ -131,6 +132,10 @@ static OpResult ExpandIntoInit
 ) {
 	OpExpandInto *op = (OpExpandInto *)opBase;
 
+	// in case this operation is restricted by a limit
+	// set record_cap to the specified limit
+	ExecutionPlan_ContainsLimit(opBase, &op->record_cap);
+
 	// see if we can optimize by avoiding matrix multiplication
 	// if the algebraic expression passed in is just a single operand
 	// there's no need to compute F and perform F*X, we can simply inspect X
@@ -161,10 +166,7 @@ static OpResult ExpandIntoInit
 		}
 	}
 
-	// create 'records' within this Init function as 'record_cap'
-	// might be set during optimization time (applyLimit)
-	// If cap greater than BATCH_SIZE is specified,
-	// use BATCH_SIZE as the value.
+	// allocate record buffer, limited to a maximum of BATCH_SIZE
 	if(op->record_cap > BATCH_SIZE) op->record_cap = BATCH_SIZE;
 
 	op->records = rm_calloc(op->record_cap, sizeof(Record));

src/execution_plan/ops/op_expand_into.h

@@ -23,8 +23,8 @@ typedef struct {
 	int srcNodeIdx;             // source node index into record
 	int destNodeIdx;            // destination node index into record
 	bool single_operand;        // expression contains a single operand
-	uint record_count;          // number of held records
-	uint record_cap;            // max number of records to process
+	uint64_t record_count;      // number of held records
+	uint64_t record_cap;        // max number of records to process
 	Record *records;            // array of records
 	Record r;                   // currently selected record
 } OpExpandInto;