Only cache local plans when reusing a distributed plan

src/backend/distributed/executor/citus_custom_scan.c

@@ -189,6 +189,12 @@ CitusBeginScan(CustomScanState *node, EState *estate, int eflags)
 	{
 		CitusBeginModifyScan(node, estate, eflags);
 	}
+
+	/*
+	 * In case of a prepared statement, we will see this distributed plan again
+	 * on the next execution with a higher usage counter.
+	 */
+	distributedPlan->numberOfTimesExecuted++;
 }
 
 

src/backend/distributed/planner/local_plan_cache.c

@@ -139,6 +139,14 @@ GetCachedLocalPlan(Task *task, DistributedPlan *distributedPlan)
 bool
 IsLocalPlanCachingSupported(Job *currentJob, DistributedPlan *originalDistributedPlan)
 {
+	if (originalDistributedPlan->numberOfTimesExecuted < 1)
+	{
+		/*
+		 * Only cache if a plan is being reused (via a prepared statement).
+		 */
+		return false;
+	}
+
 	if (!currentJob->deferredPruning)
 	{
 		/*

src/backend/distributed/utils/citus_copyfuncs.c

@@ -135,6 +135,7 @@ CopyNodeDistributedPlan(COPYFUNC_ARGS)
 	COPY_NODE_FIELD(subPlanList);
 	COPY_NODE_FIELD(usedSubPlanNodeList);
 	COPY_SCALAR_FIELD(fastPathRouterPlan);
+	COPY_SCALAR_FIELD(numberOfTimesExecuted);
 	COPY_NODE_FIELD(planningError);
 }
 

src/backend/distributed/utils/citus_outfuncs.c

@@ -198,6 +198,7 @@ OutDistributedPlan(OUTFUNC_ARGS)
 	WRITE_NODE_FIELD(subPlanList);
 	WRITE_NODE_FIELD(usedSubPlanNodeList);
 	WRITE_BOOL_FIELD(fastPathRouterPlan);
+	WRITE_UINT_FIELD(numberOfTimesExecuted);
 
 	WRITE_NODE_FIELD(planningError);
 }

src/include/distributed/multi_physical_planner.h

@@ -448,6 +448,9 @@ typedef struct DistributedPlan
 	 */
 	bool fastPathRouterPlan;
 
+	/* number of times this plan has been used (as a prepared statement) */
+	uint32 numberOfTimesExecuted;
+
 	/*
 	 * NULL if this a valid plan, an error description otherwise. This will
 	 * e.g. be set if SQL features are present that a planner doesn't support,