A patch to improve replicated table update/delete handling.

This commit performs updates/deletes to replicated tables
based on either primary key or unique index under the following
conditions

1. The replicated table has either a primary key or a unique index defined.
2. The query is not changing the primary key itself.

Otherwise ctid is used, like we were using previously.

The patch was submitted by Abbas Butt.

Author: koichi-szk

src/backend/executor/nodeModifyTable.c

@@ -363,7 +363,12 @@ ExecDelete(ItemPointer tupleid,
 		   TupleTableSlot *planSlot,
 		   EPQState *epqstate,
 		   EState *estate,
+#ifndef PGXC
 		   bool canSetTag)
+#else
+		   bool canSetTag,
+		   TupleTableSlot *pslot)
+#endif
 {
 	ResultRelInfo *resultRelInfo;
 	Relation	resultRelationDesc;
@@ -455,7 +460,18 @@ ldelete:;
 #ifdef PGXC
 		if (IS_PGXC_COORDINATOR && resultRemoteRel)
 		{
-			slot = ExecProcNodeDMLInXC(estate, planSlot, NULL);
+			/*
+			 * Pass in the source plan slot, we may attributes other than ctid
+			 * in case delete is based on primay key or other unique column.
+			 * planSlot can't be passed in place of pslot here because of the
+			 * following test case failure
+			 * create table t2(a int, b int) distribute by replication;
+			 * insert into t2 values(1,2), (3,4), (5,6);
+			 * set enable_fast_query_shipping=false;
+			 * delete from t2 where a = 1;
+			 * ERROR:  operator does not exist: tid = integer
+			 */
+			slot = ExecProcNodeDMLInXC(estate, planSlot, pslot);
 		}
 		else
 		{
@@ -1007,10 +1023,11 @@ ExecModifyTable(ModifyTableState *node)
 	ResultRelInfo *saved_resultRelInfo;
 	ResultRelInfo *resultRelInfo;
 	PlanState  *subplanstate;
-#ifdef PGXC	
+#ifdef PGXC
 	PlanState  *remoterelstate;
 	PlanState  *saved_resultRemoteRel;
-#endif	
+	RemoteQuery		*step = NULL;
+#endif
 	JunkFilter *junkfilter;
 	TupleTableSlot *slot;
 	TupleTableSlot *planSlot;
@@ -1054,6 +1071,8 @@ ExecModifyTable(ModifyTableState *node)
 #ifdef PGXC
 	/* Initialize remote plan state */
 	remoterelstate = node->mt_remoterels[node->mt_whichplan];
+	if (!IS_PGXC_DATANODE && remoterelstate != NULL)
+		step = (RemoteQuery *)((RemoteQueryState *)remoterelstate)->ss.ps.plan;
 #endif
 	junkfilter = resultRelInfo->ri_junkFilter;
 
@@ -1177,8 +1196,14 @@ ExecModifyTable(ModifyTableState *node)
 			/*
 			 * apply the junkfilter if needed.
 			 */
+#ifndef PGXC
 			if (operation != CMD_DELETE)
+#else
+			if (operation != CMD_DELETE ||
+				(!IS_PGXC_DATANODE && step != NULL &&
+					!step->rq_use_pk_for_rep_change))
 				slot = ExecFilterJunk(junkfilter, slot);
+#endif
 		}
 #ifdef PGXC
 		estate->es_result_remoterel = remoterelstate;
@@ -1194,7 +1219,11 @@ ExecModifyTable(ModifyTableState *node)
 				break;
 			case CMD_DELETE:
 				slot = ExecDelete(tupleid, oldtuple, planSlot,
+#ifndef PGXC
 								&node->mt_epqstate, estate, node->canSetTag);
+#else
+								&node->mt_epqstate, estate, node->canSetTag, slot);
+#endif
 				break;
 			default:
 				elog(ERROR, "unknown operation");

src/backend/nodes/copyfuncs.c

@@ -1048,6 +1048,7 @@ _copyRemoteQuery(const RemoteQuery *from)
 	COPY_SCALAR_FIELD(has_row_marks);
 	COPY_SCALAR_FIELD(rq_save_command_id);
 	COPY_SCALAR_FIELD(rq_params_internal);
+	COPY_SCALAR_FIELD(rq_save_command_id);
 
 	return newnode;
 }

src/backend/nodes/outfuncs.c

@@ -492,6 +492,7 @@ _outRemoteQuery(StringInfo str, const RemoteQuery *node)
 	WRITE_NODE_FIELD(query_var_tlist);
 	WRITE_BOOL_FIELD(rq_save_command_id);
 	WRITE_BOOL_FIELD(rq_params_internal);
+	WRITE_BOOL_FIELD(rq_save_command_id);
 }
 
 static void

src/backend/optimizer/plan/pgxcplan.c

@@ -48,6 +48,7 @@
 #include "pgxc/pgxcnode.h"
 #include "pgxc/execRemote.h"
 #include "rewrite/rewriteManip.h"
+#include "rewrite/rewriteHandler.h"
 #include "utils/builtins.h"
 #include "utils/rel.h"
 #include "utils/lsyscache.h"
@@ -1073,6 +1074,9 @@ pgxc_build_dml_statement(PlannerInfo *root, CmdType cmdtype,
 	int				col_att = 0;
 	int				ctid_param_num;
 	ListCell		*lc;
+	bool			can_use_pk_for_rep_change = false;
+	int16			*indexed_col_numbers = NULL;
+	int				index_col_count = 0;
 
 	/* Make sure we are dealing with DMLs */
 	if (cmdtype != CMD_UPDATE &&
@@ -1134,6 +1138,28 @@ pgxc_build_dml_statement(PlannerInfo *root, CmdType cmdtype,
 
 	query_to_deparse->jointree = makeNode(FromExpr);
 
+	can_use_pk_for_rep_change = (cmdtype == CMD_UPDATE || cmdtype == CMD_DELETE) &&
+				IsRelationReplicated(GetRelationLocInfo(res_rel->relid));
+
+	if (can_use_pk_for_rep_change)
+	{
+		index_col_count = pgxc_find_unique_index(res_rel->relid,
+												&indexed_col_numbers);
+		if (index_col_count <= 0)
+			can_use_pk_for_rep_change = false;
+
+		if (can_use_pk_for_rep_change)
+		{
+			if (is_pk_being_changed(root->parse, indexed_col_numbers,
+									index_col_count))
+			{
+				can_use_pk_for_rep_change = false;
+			}
+		}
+	}
+
+	rqplan->rq_use_pk_for_rep_change = can_use_pk_for_rep_change;
+
 	/*
 	 * Prepare a param list for INSERT queries
 	 * While doing so note the position of ctid, xc_node_id in source data
@@ -1146,7 +1172,8 @@ pgxc_build_dml_statement(PlannerInfo *root, CmdType cmdtype,
 		col_att++;
 
 		/* The position of ctid/xc_node_id is not required for INSERT */
-		if (tle->resjunk && (cmdtype == CMD_UPDATE || cmdtype == CMD_DELETE))
+		if (!can_use_pk_for_rep_change && tle->resjunk &&
+			(cmdtype == CMD_UPDATE || cmdtype == CMD_DELETE))
 		{
 			Var *v = (Var *)tle->expr;
 
@@ -1232,42 +1259,89 @@ pgxc_build_dml_statement(PlannerInfo *root, CmdType cmdtype,
 		 * the n table attributes, followed by ctid and optionally node_id. So
 		 * we know that the ctid has to be n + 1.
 		 */
-		ctid_param_num = natts + 1;
+		if (!can_use_pk_for_rep_change)
+		{
+			ctid_param_num = natts + 1;
+		}
 	}
-	else if (cmdtype == CMD_DELETE)
+	if (cmdtype == CMD_DELETE)
 	{
-		/*
-		 * Since there is no data to update, the first param is going to be
-		 * ctid.
-		 */
-		ctid_param_num = 1;
+		if (!can_use_pk_for_rep_change)
+		{
+			/*
+			 * Since there is no data to update, the first param is going to be
+			 * ctid.
+			 */
+			ctid_param_num = 1;
+		}
 	}
 
 	/* Add quals like ctid = $4 AND xc_node_id = $6 to the UPDATE/DELETE query */
 	if (cmdtype == CMD_UPDATE || cmdtype == CMD_DELETE)
 	{
-		if (!ctid_found)
-			elog(ERROR, "Source data plan's target list does not contain ctid colum");
-
 		/*
-		 * Beware, the ordering of ctid and node_id is important ! ctid should
-		 * be followed by node_id, not vice-versa, so as to be consistent with
-		 * the data row to be generated while binding the parameters for the
-		 * update statement.
+		 * If it is not replicated, we can use CTID, otherwise we need
+		 * to use a defined primary key
 		 */
-		pgxc_dml_add_qual_to_query(query_to_deparse, ctid_param_num,
-						SelfItemPointerAttributeNumber, resultRelationIndex);
+		if (!can_use_pk_for_rep_change)
+		{
+			if (!ctid_found)
+				elog(ERROR, "Source data plan's target list does not contain ctid colum");
 
-		if (node_id_found)
+			/*
+			 * Beware, the ordering of ctid and node_id is important ! ctid should
+			 * be followed by node_id, not vice-versa, so as to be consistent with
+			 * the data row to be generated while binding the parameters for the
+			 * update statement.
+			 */
+			pgxc_dml_add_qual_to_query(query_to_deparse, ctid_param_num,
+							SelfItemPointerAttributeNumber, resultRelationIndex);
+
+			if (node_id_found)
+			{
+				pgxc_dml_add_qual_to_query(query_to_deparse, ctid_param_num + 1,
+								XC_NodeIdAttributeNumber, resultRelationIndex);
+			}
+		}
+		else
 		{
-			pgxc_dml_add_qual_to_query(query_to_deparse, ctid_param_num + 1,
-							XC_NodeIdAttributeNumber, resultRelationIndex);
+			/*
+			 * Add all the columns of the primary key or unique index
+			 * in the where clause of update / delete on the replicated table
+			 */
+			int i;
+			for (i = 0; i < index_col_count; i++)
+			{
+				int			pkattno = indexed_col_numbers[i];
 
-			query_to_deparse->jointree->quals = (Node *)make_andclause(
-								(List *)query_to_deparse->jointree->quals);
+				col_att = 0;
+				foreach(elt, sourceTargetList)
+				{
+					TargetEntry	*tle = lfirst(elt);
+					Var *v;
+			
+					col_att++;
+			
+					v = (Var *)tle->expr;
+		
+					if (v->varno == resultRelationIndex &&
+						v->varattno == pkattno)
+					{
+						break;
+					}
+				}
+
+				pgxc_dml_add_qual_to_query(query_to_deparse, col_att,
+							pkattno, resultRelationIndex);
+			}
 		}
+		query_to_deparse->jointree->quals = (Node *)make_andclause(
+						(List *)query_to_deparse->jointree->quals);
 	}
 
+	if (indexed_col_numbers != NULL)
+		pfree(indexed_col_numbers);
+
 	/* pgxc_add_returning_list copied returning list in base_tlist */
 	if (rqplan->base_tlist)
 		query_to_deparse->returningList = list_copy(rqplan->base_tlist);

src/backend/pgxc/pool/execRemote.c

@@ -4956,7 +4956,6 @@ void AtEOXact_DBCleanup(bool isCommit)
  * and the ctid/nodeid values to be supplied for the WHERE clause of the
  * query. The data values are present in dataSlot whereas the ctid/nodeid
  * are available in sourceSlot as junk attributes.
- * For DELETEs, the dataSlot is NULL.
  * sourceSlot is used only to retrieve ctid/nodeid, so it does not get
  * used for INSERTs, although it will never be NULL.
  * The slots themselves are undisturbed.
@@ -5054,10 +5053,6 @@ SetDataRowForIntParams(JunkFilter *junkfilter,
 		appendBinaryStringInfo(&buf, (char *) &params_nbo, sizeof(params_nbo));
 	}
 
-	/*
-	 * The data attributes would not be present for DELETE. In such case,
-	 * dataSlot will be NULL.
-	 */
 	if (dataSlot)
 	{
 		TupleDesc	 	tdesc = dataSlot->tts_tupleDescriptor;