Check for Motions before scan if requesting Explicit Redistribute Motion

Previously, `Explicit Redistribute Motion` was added if there were any motions
in the subplan, ignoring any motions in `InitPlans`.

This patch updates the logic: if an `Explicit Redistribute Motion` has no
motions underneath it's subtree, then the row to update must originate from the
same segment, and no `Explicit Redistribute Motion` is needed. If there are any
motions, `Explicit Redistribute Motion` should be added only if there is a
motion between the scan and the `ModifyTable` on the relation we are going to
update.

src/backend/optimizer/plan/createplan.c

@@ -170,6 +170,7 @@ static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual,
 static BitmapAnd *make_bitmap_and(List *bitmapplans);
 static BitmapOr *make_bitmap_or(List *bitmapplans);
 static List *flatten_grouping_list(List *groupcls);
+static bool is_scan_before_motions(Plan *plan, List *rtable, Oid relid);
 static void adjust_modifytable_flow(PlannerInfo *root, ModifyTable *node, List *is_split_updates);
 static Plan *prepare_sort_from_pathkeys(PlannerInfo *root,
 						   Plan *lefttree, List *pathkeys,
@@ -6471,6 +6472,88 @@ make_modifytable(PlannerInfo *root,
 	return node;
 }
 
+/*
+ * is_scan_before_motions
+ *	  Check if there's any Redistribute or Broadcast Motions before any scan for
+ *	  specific Oid
+ */
+static bool
+is_scan_before_motions(Plan *plan, List *rtable, Oid relid)
+{
+	RangeTblEntry *target_rte;
+	ListCell *lcr;
+
+	while (plan)
+	{
+		/*
+		 * If this is a Motion, check if Motion's relation Oid matches rte's
+		 * Oid. If it does, then there is a Motion node before the scan.
+		 */
+		if (IsA(plan, Motion))
+		{
+			Motion *motion = (Motion *) plan;
+
+			/* Check targetlist for Broadcast Motion and Redistribute Motion */
+			if ((motion->motionType == MOTIONTYPE_FIXED && motion->isBroadcast) ||
+				motion->motionType == MOTIONTYPE_HASH)
+			{
+				TargetEntry *tle;
+
+				foreach(lcr, plan->targetlist)
+				{
+					tle = (TargetEntry *) lfirst(lcr);
+					target_rte = rt_fetch(tle->resno, rtable);
+
+					if (target_rte->relid == relid)
+						/* There is a motion before scan */
+						return false;
+				}
+			}
+		}
+		/*
+		 * If this is a scan and it's relation Oid matches rte's Oid, we
+		 * encountered a scan before any Motions.
+		 */
+		else
+		{
+			Scan *scan;
+
+			switch (nodeTag(plan))
+			{
+				case T_SeqScan:
+				case T_DynamicSeqScan:
+				case T_ExternalScan:
+				case T_IndexScan:
+				case T_DynamicIndexScan:
+				case T_IndexOnlyScan:
+				case T_BitmapIndexScan:
+				case T_DynamicBitmapIndexScan:
+				case T_BitmapHeapScan:
+				case T_DynamicBitmapHeapScan:
+					scan = (Scan *) plan;
+					target_rte = rt_fetch(scan->scanrelid, rtable);
+
+					if (target_rte->relid == relid)
+						/* There wasn't any Motions before scan */
+						return true;
+					break;
+				default:
+					break;
+			}
+		}
+
+		/*
+		 * If this plan has a right subtree, check it for Motions too.
+		 */
+		if (plan->righttree && is_scan_before_motions(plan->righttree, rtable, relid))
+			return true;
+
+		plan = plan->lefttree;
+	}
+
+	return false;
+}
+
 /*
  * Set the Flow in a ModifyTable and its children correctly.
  *
@@ -6726,7 +6809,18 @@ adjust_modifytable_flow(PlannerInfo *root, ModifyTable *node, List *is_split_upd
 					node->action_col_idxes = lappend_int(node->action_col_idxes, -1);
 					node->ctid_col_idxes = lappend_int(node->ctid_col_idxes, -1);
 					node->oid_col_idxes = lappend_int(node->oid_col_idxes, 0);
-					request_explicit_motion(subplan, rti, root->glob->finalrtable);
+
+					/*
+					 * If an Explicit Motion has no Motions underneath it, then
+					 * the row to update must originate from the same segment,
+					 * and no Motion is needed.
+					 *
+					 * We elide the motion even if there are Motions, as long as
+					 * they are not between the scan on the target table and the
+					 * ModifyTable.
+					 */
+					if (!is_scan_before_motions(subplan, root->parse->rtable, rte->relid))
+						request_explicit_motion(subplan, rti, root->glob->finalrtable);
 				}
 			}
 			else if (targetPolicyType == POLICYTYPE_ENTRY)