Refactor-postgres_fdw-s-DML-execution-hook-functions-v2.patch

*** a/contrib/postgres_fdw/postgres_fdw.c
--- b/contrib/postgres_fdw/postgres_fdw.c
***************
*** 390,395 **** static PgFdwModifyState *create_foreign_modify(EState *estate,
--- 390,400 ----
  					  List *target_attrs,
  					  bool has_returning,
  					  List *retrieved_attrs);
+ static TupleTableSlot *execute_foreign_modify(EState *estate,
+ 					   ResultRelInfo *resultRelInfo,
+ 					   CmdType operation,
+ 					   TupleTableSlot *slot,
+ 					   TupleTableSlot *planSlot);
  static void prepare_foreign_modify(PgFdwModifyState *fmstate);
  static const char **convert_prep_stmt_params(PgFdwModifyState *fmstate,
  						 ItemPointer tupleid,
***************
*** 1775,1832 **** postgresExecForeignInsert(EState *estate,
  						  TupleTableSlot *slot,
  						  TupleTableSlot *planSlot)
  {
! 	PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
! 	const char **p_values;
! 	PGresult   *res;
! 	int			n_rows;
! 
! 	/* Set up the prepared statement on the remote server, if we didn't yet */
! 	if (!fmstate->p_name)
! 		prepare_foreign_modify(fmstate);
! 
! 	/* Convert parameters needed by prepared statement to text form */
! 	p_values = convert_prep_stmt_params(fmstate, NULL, slot);
! 
! 	/*
! 	 * Execute the prepared statement.
! 	 */
! 	if (!PQsendQueryPrepared(fmstate->conn,
! 							 fmstate->p_name,
! 							 fmstate->p_nums,
! 							 p_values,
! 							 NULL,
! 							 NULL,
! 							 0))
! 		pgfdw_report_error(ERROR, NULL, fmstate->conn, false, fmstate->query);
! 
! 	/*
! 	 * Get the result, and check for success.
! 	 *
! 	 * We don't use a PG_TRY block here, so be careful not to throw error
! 	 * without releasing the PGresult.
! 	 */
! 	res = pgfdw_get_result(fmstate->conn, fmstate->query);
! 	if (PQresultStatus(res) !=
! 		(fmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK))
! 		pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query);
! 
! 	/* Check number of rows affected, and fetch RETURNING tuple if any */
! 	if (fmstate->has_returning)
! 	{
! 		n_rows = PQntuples(res);
! 		if (n_rows > 0)
! 			store_returning_result(fmstate, slot, res);
! 	}
! 	else
! 		n_rows = atoi(PQcmdTuples(res));
! 
! 	/* And clean up */
! 	PQclear(res);
! 
! 	MemoryContextReset(fmstate->temp_cxt);
! 
! 	/* Return NULL if nothing was inserted on the remote end */
! 	return (n_rows > 0) ? slot : NULL;
  }
  
  /*
--- 1780,1787 ----
  						  TupleTableSlot *slot,
  						  TupleTableSlot *planSlot)
  {
! 	return execute_foreign_modify(estate, resultRelInfo, CMD_INSERT,
! 								  slot, planSlot);
  }
  
  /*
***************
*** 1839,1908 **** postgresExecForeignUpdate(EState *estate,
  						  TupleTableSlot *slot,
  						  TupleTableSlot *planSlot)
  {
! 	PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
! 	Datum		datum;
! 	bool		isNull;
! 	const char **p_values;
! 	PGresult   *res;
! 	int			n_rows;
! 
! 	/* Set up the prepared statement on the remote server, if we didn't yet */
! 	if (!fmstate->p_name)
! 		prepare_foreign_modify(fmstate);
! 
! 	/* Get the ctid that was passed up as a resjunk column */
! 	datum = ExecGetJunkAttribute(planSlot,
! 								 fmstate->ctidAttno,
! 								 &isNull);
! 	/* shouldn't ever get a null result... */
! 	if (isNull)
! 		elog(ERROR, "ctid is NULL");
! 
! 	/* Convert parameters needed by prepared statement to text form */
! 	p_values = convert_prep_stmt_params(fmstate,
! 										(ItemPointer) DatumGetPointer(datum),
! 										slot);
! 
! 	/*
! 	 * Execute the prepared statement.
! 	 */
! 	if (!PQsendQueryPrepared(fmstate->conn,
! 							 fmstate->p_name,
! 							 fmstate->p_nums,
! 							 p_values,
! 							 NULL,
! 							 NULL,
! 							 0))
! 		pgfdw_report_error(ERROR, NULL, fmstate->conn, false, fmstate->query);
! 
! 	/*
! 	 * Get the result, and check for success.
! 	 *
! 	 * We don't use a PG_TRY block here, so be careful not to throw error
! 	 * without releasing the PGresult.
! 	 */
! 	res = pgfdw_get_result(fmstate->conn, fmstate->query);
! 	if (PQresultStatus(res) !=
! 		(fmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK))
! 		pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query);
! 
! 	/* Check number of rows affected, and fetch RETURNING tuple if any */
! 	if (fmstate->has_returning)
! 	{
! 		n_rows = PQntuples(res);
! 		if (n_rows > 0)
! 			store_returning_result(fmstate, slot, res);
! 	}
! 	else
! 		n_rows = atoi(PQcmdTuples(res));
! 
! 	/* And clean up */
! 	PQclear(res);
! 
! 	MemoryContextReset(fmstate->temp_cxt);
! 
! 	/* Return NULL if nothing was updated on the remote end */
! 	return (n_rows > 0) ? slot : NULL;
  }
  
  /*
--- 1794,1801 ----
  						  TupleTableSlot *slot,
  						  TupleTableSlot *planSlot)
  {
! 	return execute_foreign_modify(estate, resultRelInfo, CMD_UPDATE,
! 								  slot, planSlot);
  }
  
  /*
***************
*** 1915,1984 **** postgresExecForeignDelete(EState *estate,
  						  TupleTableSlot *slot,
  						  TupleTableSlot *planSlot)
  {
! 	PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
! 	Datum		datum;
! 	bool		isNull;
! 	const char **p_values;
! 	PGresult   *res;
! 	int			n_rows;
! 
! 	/* Set up the prepared statement on the remote server, if we didn't yet */
! 	if (!fmstate->p_name)
! 		prepare_foreign_modify(fmstate);
! 
! 	/* Get the ctid that was passed up as a resjunk column */
! 	datum = ExecGetJunkAttribute(planSlot,
! 								 fmstate->ctidAttno,
! 								 &isNull);
! 	/* shouldn't ever get a null result... */
! 	if (isNull)
! 		elog(ERROR, "ctid is NULL");
! 
! 	/* Convert parameters needed by prepared statement to text form */
! 	p_values = convert_prep_stmt_params(fmstate,
! 										(ItemPointer) DatumGetPointer(datum),
! 										NULL);
! 
! 	/*
! 	 * Execute the prepared statement.
! 	 */
! 	if (!PQsendQueryPrepared(fmstate->conn,
! 							 fmstate->p_name,
! 							 fmstate->p_nums,
! 							 p_values,
! 							 NULL,
! 							 NULL,
! 							 0))
! 		pgfdw_report_error(ERROR, NULL, fmstate->conn, false, fmstate->query);
! 
! 	/*
! 	 * Get the result, and check for success.
! 	 *
! 	 * We don't use a PG_TRY block here, so be careful not to throw error
! 	 * without releasing the PGresult.
! 	 */
! 	res = pgfdw_get_result(fmstate->conn, fmstate->query);
! 	if (PQresultStatus(res) !=
! 		(fmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK))
! 		pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query);
! 
! 	/* Check number of rows affected, and fetch RETURNING tuple if any */
! 	if (fmstate->has_returning)
! 	{
! 		n_rows = PQntuples(res);
! 		if (n_rows > 0)
! 			store_returning_result(fmstate, slot, res);
! 	}
! 	else
! 		n_rows = atoi(PQcmdTuples(res));
! 
! 	/* And clean up */
! 	PQclear(res);
! 
! 	MemoryContextReset(fmstate->temp_cxt);
! 
! 	/* Return NULL if nothing was deleted on the remote end */
! 	return (n_rows > 0) ? slot : NULL;
  }
  
  /*
--- 1808,1815 ----
  						  TupleTableSlot *slot,
  						  TupleTableSlot *planSlot)
  {
! 	return execute_foreign_modify(estate, resultRelInfo, CMD_DELETE,
! 								  slot, planSlot);
  }
  
  /*
***************
*** 3425,3430 **** create_foreign_modify(EState *estate,
--- 3256,3353 ----
  }
  
  /*
+  * execute_foreign_modify
+  *		Perform foreign-table modification as required, and fetch RETURNING
+  *		result if any.  (This is the shared guts of postgresExecForeignInsert,
+  *		postgresExecForeignUpdate, and postgresExecForeignDelete.)
+  */
+ static TupleTableSlot *
+ execute_foreign_modify(EState *estate,
+ 					   ResultRelInfo *resultRelInfo,
+ 					   CmdType operation,
+ 					   TupleTableSlot *slot,
+ 					   TupleTableSlot *planSlot)
+ {
+ 	PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
+ 	ItemPointer	ctid = NULL;
+ 	const char **p_values;
+ 	PGresult   *res;
+ 	int			n_rows;
+ 
+ 	/* The operation should be INSERT, UPDATE, or DELETE */
+ 	Assert(operation == CMD_INSERT ||
+ 		   operation == CMD_UPDATE ||
+ 		   operation == CMD_DELETE);
+ 
+ 	/* Set up the prepared statement on the remote server, if we didn't yet */
+ 	if (!fmstate->p_name)
+ 		prepare_foreign_modify(fmstate);
+ 
+ 	/*
+ 	 * For UPDATE/DELETE, get the ctid that was passed up as a resjunk column
+ 	 */
+ 	if (operation == CMD_UPDATE || operation == CMD_DELETE)
+ 	{
+ 		Datum		datum;
+ 		bool		isNull;
+ 
+ 		datum = ExecGetJunkAttribute(planSlot,
+ 									 fmstate->ctidAttno,
+ 									 &isNull);
+ 		/* shouldn't ever get a null result... */
+ 		if (isNull)
+ 			elog(ERROR, "ctid is NULL");
+ 		ctid = (ItemPointer) DatumGetPointer(datum);
+ 	}
+ 
+ 	/* Convert parameters needed by prepared statement to text form */
+ 	p_values = convert_prep_stmt_params(fmstate, ctid, slot);
+ 
+ 	/*
+ 	 * Execute the prepared statement.
+ 	 */
+ 	if (!PQsendQueryPrepared(fmstate->conn,
+ 							 fmstate->p_name,
+ 							 fmstate->p_nums,
+ 							 p_values,
+ 							 NULL,
+ 							 NULL,
+ 							 0))
+ 		pgfdw_report_error(ERROR, NULL, fmstate->conn, false, fmstate->query);
+ 
+ 	/*
+ 	 * Get the result, and check for success.
+ 	 *
+ 	 * We don't use a PG_TRY block here, so be careful not to throw error
+ 	 * without releasing the PGresult.
+ 	 */
+ 	res = pgfdw_get_result(fmstate->conn, fmstate->query);
+ 	if (PQresultStatus(res) !=
+ 		(fmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK))
+ 		pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query);
+ 
+ 	/* Check number of rows affected, and fetch RETURNING tuple if any */
+ 	if (fmstate->has_returning)
+ 	{
+ 		n_rows = PQntuples(res);
+ 		if (n_rows > 0)
+ 			store_returning_result(fmstate, slot, res);
+ 	}
+ 	else
+ 		n_rows = atoi(PQcmdTuples(res));
+ 
+ 	/* And clean up */
+ 	PQclear(res);
+ 
+ 	MemoryContextReset(fmstate->temp_cxt);
+ 
+ 	/*
+ 	 * Return NULL if nothing was inserted/updated/deleted on the remote end
+ 	 */
+ 	return (n_rows > 0) ? slot : NULL;
+ }
+ 
+ /*
   * prepare_foreign_modify
   *		Establish a prepared statement for execution of INSERT/UPDATE/DELETE
   */