(2018/02/21 20:54), Etsuro Fujita wrote:
void
BeginForeignRouting();
Prepare for a tuple-routing operation on a foreign table. This is called
from ExecSetupPartitionTupleRouting and ExecInitPartitionInfo.
I modified execPartition.c so that this callback routine is called from
a single function that I added to execPartition.c and it is called the
first time the foreign partition is chose as the target partition to
route a tuple to. That removes CheckValidResultRel, the
tuple-conversion setup, and the FDW initialization for each UPDATE
subplan from ExecSetupPartitionTupleRouting, so it would minimize the
possibly-useless overhead in doing that function.
Changes other than that are:
* Fixed typo and revised code/comments
* Added more regression tests
* Added docs
Attached is a new version of the patch set.
Best regards,
Etsuro Fujita
*** a/contrib/postgres_fdw/postgres_fdw.c
--- b/contrib/postgres_fdw/postgres_fdw.c
***************
*** 375,386 **** static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel,
--- 375,398 ----
static void create_cursor(ForeignScanState *node);
static void fetch_more_data(ForeignScanState *node);
static void close_cursor(PGconn *conn, unsigned int cursor_number);
+ static PgFdwModifyState *create_fdw_modify_state(ModifyTableState *mtstate,
+ ResultRelInfo *resultRelInfo,
+ CmdType operation,
+ int subplan_index,
+ char *query,
+ List *target_attrs,
+ bool has_returning,
+ List *retrieved_attrs);
static void prepare_foreign_modify(PgFdwModifyState *fmstate);
static const char **convert_prep_stmt_params(PgFdwModifyState *fmstate,
ItemPointer tupleid,
TupleTableSlot *slot);
+ static int execute_prep_stmt(PgFdwModifyState *fmstate,
+ const char **p_values,
+ TupleTableSlot *slot);
static void store_returning_result(PgFdwModifyState *fmstate,
TupleTableSlot *slot, PGresult *res);
+ static void finish_foreign_modify(PgFdwModifyState *fmstate);
static List *build_remote_returning(Index rtindex, Relation rel,
List *returningList);
static void rebuild_fdw_scan_tlist(ForeignScan *fscan, List *tlist);
***************
*** 1678,1695 **** postgresBeginForeignModify(ModifyTableState *mtstate,
int eflags)
{
PgFdwModifyState *fmstate;
! EState *estate = mtstate->ps.state;
! CmdType operation = mtstate->operation;
! Relation rel = resultRelInfo->ri_RelationDesc;
! RangeTblEntry *rte;
! Oid userid;
! ForeignTable *table;
! UserMapping *user;
! AttrNumber n_params;
! Oid typefnoid;
! bool isvarlena;
! ListCell *lc;
! TupleDesc tupdesc = RelationGetDescr(rel);
/*
* Do nothing in EXPLAIN (no ANALYZE) case. resultRelInfo->ri_FdwState
--- 1690,1699 ----
int eflags)
{
PgFdwModifyState *fmstate;
! char *query;
! List *target_attrs;
! bool has_returning;
! List *retrieved_attrs;
/*
* Do nothing in EXPLAIN (no ANALYZE) case. resultRelInfo->ri_FdwState
***************
*** 1698,1779 **** postgresBeginForeignModify(ModifyTableState *mtstate,
if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
return;
- /* Begin constructing PgFdwModifyState. */
- fmstate = (PgFdwModifyState *) palloc0(sizeof(PgFdwModifyState));
- fmstate->rel = rel;
-
- /*
- * Identify which user to do the remote access as. This should match what
- * ExecCheckRTEPerms() does.
- */
- rte = rt_fetch(resultRelInfo->ri_RangeTableIndex, estate->es_range_table);
- userid = rte->checkAsUser ? rte->checkAsUser : GetUserId();
-
- /* Get info about foreign table. */
- table = GetForeignTable(RelationGetRelid(rel));
- user = GetUserMapping(userid, table->serverid);
-
- /* Open connection; report that we'll create a prepared statement. */
- fmstate->conn = GetConnection(user, true);
- fmstate->p_name = NULL; /* prepared statement not made yet */
-
/* Deconstruct fdw_private data. */
! fmstate->query = strVal(list_nth(fdw_private,
! FdwModifyPrivateUpdateSql));
! fmstate->target_attrs = (List *) list_nth(fdw_private,
! FdwModifyPrivateTargetAttnums);
! fmstate->has_returning = intVal(list_nth(fdw_private,
! FdwModifyPrivateHasReturning));
! fmstate->retrieved_attrs = (List *) list_nth(fdw_private,
! FdwModifyPrivateRetrievedAttrs);
!
! /* Create context for per-tuple temp workspace. */
! fmstate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
! "postgres_fdw temporary data",
! ALLOCSET_SMALL_SIZES);
!
! /* Prepare for input conversion of RETURNING results. */
! if (fmstate->has_returning)
! fmstate->attinmeta = TupleDescGetAttInMetadata(tupdesc);
!
! /* Prepare for output conversion of parameters used in prepared stmt. */
! n_params = list_length(fmstate->target_attrs) + 1;
! fmstate->p_flinfo = (FmgrInfo *) palloc0(sizeof(FmgrInfo) * n_params);
! fmstate->p_nums = 0;
!
! if (operation == CMD_UPDATE || operation == CMD_DELETE)
! {
! /* Find the ctid resjunk column in the subplan's result */
! Plan *subplan = mtstate->mt_plans[subplan_index]->plan;
!
! fmstate->ctidAttno = ExecFindJunkAttributeInTlist(subplan->targetlist,
! "ctid");
! if (!AttributeNumberIsValid(fmstate->ctidAttno))
! elog(ERROR, "could not find junk ctid column");
!
! /* First transmittable parameter will be ctid */
! getTypeOutputInfo(TIDOID, &typefnoid, &isvarlena);
! fmgr_info(typefnoid, &fmstate->p_flinfo[fmstate->p_nums]);
! fmstate->p_nums++;
! }
!
! if (operation == CMD_INSERT || operation == CMD_UPDATE)
! {
! /* Set up for remaining transmittable parameters */
! foreach(lc, fmstate->target_attrs)
! {
! int attnum = lfirst_int(lc);
! Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
! Assert(!attr->attisdropped);
!
! getTypeOutputInfo(attr->atttypid, &typefnoid, &isvarlena);
! fmgr_info(typefnoid, &fmstate->p_flinfo[fmstate->p_nums]);
! fmstate->p_nums++;
! }
! }
!
! Assert(fmstate->p_nums <= n_params);
resultRelInfo->ri_FdwState = fmstate;
}
--- 1702,1726 ----
if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
return;
/* Deconstruct fdw_private data. */
! query = strVal(list_nth(fdw_private,
! FdwModifyPrivateUpdateSql));
! target_attrs = (List *) list_nth(fdw_private,
! FdwModifyPrivateTargetAttnums);
! has_returning = intVal(list_nth(fdw_private,
! FdwModifyPrivateHasReturning));
! retrieved_attrs = (List *) list_nth(fdw_private,
! FdwModifyPrivateRetrievedAttrs);
! /* Construct an execution state. */
! fmstate = create_fdw_modify_state(mtstate,
! resultRelInfo,
! mtstate->operation,
! subplan_index,
! query,
! target_attrs,
! has_returning,
! retrieved_attrs);
resultRelInfo->ri_FdwState = fmstate;
}
***************
*** 1790,1796 **** postgresExecForeignInsert(EState *estate,
{
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 */
--- 1737,1742 ----
***************
*** 1800,1840 **** postgresExecForeignInsert(EState *estate,
/* 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);
--- 1746,1753 ----
/* Convert parameters needed by prepared statement to text form */
p_values = convert_prep_stmt_params(fmstate, NULL, slot);
! /* Execute the prepared statement and fetch RETURNING tuple if any */
! n_rows = execute_prep_stmt(fmstate, p_values, slot);
MemoryContextReset(fmstate->temp_cxt);
***************
*** 1856,1862 **** postgresExecForeignUpdate(EState *estate,
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 */
--- 1769,1774 ----
***************
*** 1876,1916 **** postgresExecForeignUpdate(EState *estate,
(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);
--- 1788,1795 ----
(ItemPointer) DatumGetPointer(datum),
slot);
! /* Execute the prepared statement and fetch RETURNING tuple if any */
! n_rows = execute_prep_stmt(fmstate, p_values, slot);
MemoryContextReset(fmstate->temp_cxt);
***************
*** 1932,1938 **** postgresExecForeignDelete(EState *estate,
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 */
--- 1811,1816 ----
***************
*** 1952,1992 **** postgresExecForeignDelete(EState *estate,
(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);
--- 1830,1837 ----
(ItemPointer) DatumGetPointer(datum),
NULL);
! /* Execute the prepared statement and fetch RETURNING tuple if any */
! n_rows = execute_prep_stmt(fmstate, p_values, slot);
MemoryContextReset(fmstate->temp_cxt);
***************
*** 2008,2035 **** postgresEndForeignModify(EState *estate,
if (fmstate == NULL)
return;
! /* If we created a prepared statement, destroy it */
! if (fmstate->p_name)
! {
! char sql[64];
! PGresult *res;
!
! snprintf(sql, sizeof(sql), "DEALLOCATE %s", fmstate->p_name);
!
! /*
! * We don't use a PG_TRY block here, so be careful not to throw error
! * without releasing the PGresult.
! */
! res = pgfdw_exec_query(fmstate->conn, sql);
! if (PQresultStatus(res) != PGRES_COMMAND_OK)
! pgfdw_report_error(ERROR, res, fmstate->conn, true, sql);
! PQclear(res);
! fmstate->p_name = NULL;
! }
!
! /* Release remote connection */
! ReleaseConnection(fmstate->conn);
! fmstate->conn = NULL;
}
/*
--- 1853,1860 ----
if (fmstate == NULL)
return;
! /* Destroy the execution state. */
! finish_foreign_modify(fmstate);
}
/*
***************
*** 3217,3222 **** close_cursor(PGconn *conn, unsigned int cursor_number)
--- 3042,3151 ----
}
/*
+ * create_fdw_modify_state
+ * Construct an execution state of a foreign insert/update/delete
+ * operation.
+ */
+ static PgFdwModifyState *
+ create_fdw_modify_state(ModifyTableState *mtstate,
+ ResultRelInfo *resultRelInfo,
+ CmdType operation,
+ int subplan_index,
+ char *query,
+ List *target_attrs,
+ bool has_returning,
+ List *retrieved_attrs)
+ {
+ PgFdwModifyState *fmstate;
+ EState *estate = mtstate->ps.state;
+ Relation rel = resultRelInfo->ri_RelationDesc;
+ RangeTblEntry *rte;
+ Oid userid;
+ ForeignTable *table;
+ UserMapping *user;
+ AttrNumber n_params;
+ Oid typefnoid;
+ bool isvarlena;
+ ListCell *lc;
+ TupleDesc tupdesc = RelationGetDescr(rel);
+
+ /* Begin constructing PgFdwModifyState. */
+ fmstate = (PgFdwModifyState *) palloc0(sizeof(PgFdwModifyState));
+ fmstate->rel = rel;
+
+ /*
+ * Identify which user to do the remote access as. This should match what
+ * ExecCheckRTEPerms() does.
+ */
+ rte = rt_fetch(resultRelInfo->ri_RangeTableIndex, estate->es_range_table);
+ userid = rte->checkAsUser ? rte->checkAsUser : GetUserId();
+
+ /* Get info about foreign table. */
+ table = GetForeignTable(RelationGetRelid(rel));
+ user = GetUserMapping(userid, table->serverid);
+
+ /* Open connection; report that we'll create a prepared statement. */
+ fmstate->conn = GetConnection(user, true);
+ fmstate->p_name = NULL; /* prepared statement not made yet */
+
+ /* Set remote query information. */
+ fmstate->query = query;
+ fmstate->target_attrs = target_attrs;
+ fmstate->has_returning = has_returning;
+ fmstate->retrieved_attrs = retrieved_attrs;
+
+ /* Create context for per-tuple temp workspace. */
+ fmstate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
+ "postgres_fdw temporary data",
+ ALLOCSET_SMALL_SIZES);
+
+ /* Prepare for input conversion of RETURNING results. */
+ if (fmstate->has_returning)
+ fmstate->attinmeta = TupleDescGetAttInMetadata(tupdesc);
+
+ /* Prepare for output conversion of parameters used in prepared stmt. */
+ n_params = list_length(fmstate->target_attrs) + 1;
+ fmstate->p_flinfo = (FmgrInfo *) palloc0(sizeof(FmgrInfo) * n_params);
+ fmstate->p_nums = 0;
+
+ if (operation == CMD_UPDATE || operation == CMD_DELETE)
+ {
+ /* Find the ctid resjunk column in the subplan's result */
+ Plan *subplan = mtstate->mt_plans[subplan_index]->plan;
+
+ fmstate->ctidAttno = ExecFindJunkAttributeInTlist(subplan->targetlist,
+ "ctid");
+ if (!AttributeNumberIsValid(fmstate->ctidAttno))
+ elog(ERROR, "could not find junk ctid column");
+
+ /* First transmittable parameter will be ctid */
+ getTypeOutputInfo(TIDOID, &typefnoid, &isvarlena);
+ fmgr_info(typefnoid, &fmstate->p_flinfo[fmstate->p_nums]);
+ fmstate->p_nums++;
+ }
+
+ if (operation == CMD_INSERT || operation == CMD_UPDATE)
+ {
+ /* Set up for remaining transmittable parameters */
+ foreach(lc, fmstate->target_attrs)
+ {
+ int attnum = lfirst_int(lc);
+ Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
+
+ Assert(!attr->attisdropped);
+
+ getTypeOutputInfo(attr->atttypid, &typefnoid, &isvarlena);
+ fmgr_info(typefnoid, &fmstate->p_flinfo[fmstate->p_nums]);
+ fmstate->p_nums++;
+ }
+ }
+
+ Assert(fmstate->p_nums <= n_params);
+
+ return fmstate;
+ }
+
+ /*
* prepare_foreign_modify
* Establish a prepared statement for execution of INSERT/UPDATE/DELETE
*/
***************
*** 3326,3331 **** convert_prep_stmt_params(PgFdwModifyState *fmstate,
--- 3255,3311 ----
}
/*
+ * execute_prep_stmt
+ * Execute the prepared statement and fetch RETURNING tuple if any
+ */
+ static int
+ execute_prep_stmt(PgFdwModifyState *fmstate,
+ const char **p_values,
+ TupleTableSlot *slot)
+ {
+ PGresult *res;
+ int n_rows;
+
+ /*
+ * 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);
+
+ return n_rows;
+ }
+
+ /*
* store_returning_result
* Store the result of a RETURNING clause
*
***************
*** 3359,3364 **** store_returning_result(PgFdwModifyState *fmstate,
--- 3339,3377 ----
}
/*
+ * finish_foreign_modify
+ * Release resources for a foreign insert/update/delete operation.
+ */
+ static void
+ finish_foreign_modify(PgFdwModifyState *fmstate)
+ {
+ Assert(fmstate != NULL);
+
+ /* If we created a prepared statement, destroy it */
+ if (fmstate->p_name)
+ {
+ char sql[64];
+ PGresult *res;
+
+ snprintf(sql, sizeof(sql), "DEALLOCATE %s", fmstate->p_name);
+
+ /*
+ * We don't use a PG_TRY block here, so be careful not to throw error
+ * without releasing the PGresult.
+ */
+ res = pgfdw_exec_query(fmstate->conn, sql);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ pgfdw_report_error(ERROR, res, fmstate->conn, true, sql);
+ PQclear(res);
+ fmstate->p_name = NULL;
+ }
+
+ /* Release remote connection */
+ ReleaseConnection(fmstate->conn);
+ fmstate->conn = NULL;
+ }
+
+ /*
* build_remote_returning
* Build a RETURNING targetlist of a remote query for performing an
* UPDATE/DELETE .. RETURNING on a join directly
*** a/contrib/file_fdw/output/file_fdw.source
--- b/contrib/file_fdw/output/file_fdw.source
***************
*** 315,321 **** SELECT tableoid::regclass, * FROM p2;
(0 rows)
COPY pt FROM '@abs_srcdir@/data/list2.bad' with (format 'csv', delimiter ','); -- ERROR
! ERROR: cannot route inserted tuples to a foreign table
CONTEXT: COPY pt, line 2: "1,qux"
COPY pt FROM '@abs_srcdir@/data/list2.csv' with (format 'csv', delimiter ',');
SELECT tableoid::regclass, * FROM pt;
--- 315,321 ----
(0 rows)
COPY pt FROM '@abs_srcdir@/data/list2.bad' with (format 'csv', delimiter ','); -- ERROR
! ERROR: cannot route copied tuples to a foreign table
CONTEXT: COPY pt, line 2: "1,qux"
COPY pt FROM '@abs_srcdir@/data/list2.csv' with (format 'csv', delimiter ',');
SELECT tableoid::regclass, * FROM pt;
***************
*** 342,351 **** SELECT tableoid::regclass, * FROM p2;
(2 rows)
INSERT INTO pt VALUES (1, 'xyzzy'); -- ERROR
! ERROR: cannot route inserted tuples to a foreign table
INSERT INTO pt VALUES (2, 'xyzzy');
UPDATE pt set a = 1 where a = 2; -- ERROR
! ERROR: cannot route inserted tuples to a foreign table
SELECT tableoid::regclass, * FROM pt;
tableoid | a | b
----------+---+-------
--- 342,351 ----
(2 rows)
INSERT INTO pt VALUES (1, 'xyzzy'); -- ERROR
! ERROR: cannot route inserted tuples to foreign table "p1"
INSERT INTO pt VALUES (2, 'xyzzy');
UPDATE pt set a = 1 where a = 2; -- ERROR
! ERROR: cannot route inserted tuples to foreign table "p1"
SELECT tableoid::regclass, * FROM pt;
tableoid | a | b
----------+---+-------
*** a/contrib/postgres_fdw/expected/postgres_fdw.out
--- b/contrib/postgres_fdw/expected/postgres_fdw.out
***************
*** 7364,7369 **** NOTICE: drop cascades to foreign table bar2
--- 7364,7510 ----
drop table loct1;
drop table loct2;
-- ===================================================================
+ -- test tuple routing for foreign-table partitions
+ -- ===================================================================
+ create table itrtest (a int, b int, c text) partition by list (a);
+ create table loct1 (a int check (a in (1)), b int primary key, c text);
+ create table loct2 (b int primary key, c text, a int check (a in (2)));
+ create foreign table remp1 (a int check (a in (1)), b int, c text) server loopback options (table_name 'loct1');
+ create foreign table remp2 (b int, c text, a int check (a in (2))) server loopback options (table_name 'loct2');
+ alter table itrtest attach partition remp1 for values in (1);
+ alter table itrtest attach partition remp2 for values in (2);
+ insert into itrtest values (1, 1, 'foo');
+ insert into itrtest values (1, 2, 'bar') returning *;
+ a | b | c
+ ---+---+-----
+ 1 | 2 | bar
+ (1 row)
+
+ insert into itrtest values (2, 1, 'baz') returning *;
+ a | b | c
+ ---+---+-----
+ 2 | 1 | baz
+ (1 row)
+
+ select tableoid::regclass, * FROM itrtest;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ remp1 | 1 | 1 | foo
+ remp1 | 1 | 2 | bar
+ remp2 | 2 | 1 | baz
+ (3 rows)
+
+ select tableoid::regclass, * FROM remp1;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ remp1 | 1 | 1 | foo
+ remp1 | 1 | 2 | bar
+ (2 rows)
+
+ select tableoid::regclass, * FROM remp2;
+ tableoid | b | c | a
+ ----------+---+-----+---
+ remp2 | 1 | baz | 2
+ (1 row)
+
+ insert into itrtest values (2, 1, 'baz');
+ ERROR: duplicate key value violates unique constraint "loct2_pkey"
+ DETAIL: Key (b)=(1) already exists.
+ CONTEXT: Remote SQL command: INSERT INTO public.loct2(b, c, a) VALUES ($1, $2, $3)
+ insert into itrtest values (2, 1, 'baz') on conflict do nothing;
+ insert into itrtest values (2, 2, 'qux') on conflict do nothing returning *;
+ a | b | c
+ ---+---+-----
+ 2 | 2 | qux
+ (1 row)
+
+ select tableoid::regclass, * FROM itrtest;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ remp1 | 1 | 1 | foo
+ remp1 | 1 | 2 | bar
+ remp2 | 2 | 1 | baz
+ remp2 | 2 | 2 | qux
+ (4 rows)
+
+ select tableoid::regclass, * FROM remp1;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ remp1 | 1 | 1 | foo
+ remp1 | 1 | 2 | bar
+ (2 rows)
+
+ select tableoid::regclass, * FROM remp2;
+ tableoid | b | c | a
+ ----------+---+-----+---
+ remp2 | 1 | baz | 2
+ remp2 | 2 | qux | 2
+ (2 rows)
+
+ drop table itrtest;
+ drop table loct1;
+ drop table loct2;
+ create table utrtest (a int, b int, c text) partition by list (a);
+ create table loct (a int check (a in (1)), b int, c text);
+ create foreign table remp (a int check (a in (1)), b int, c text) server loopback options (table_name 'loct');
+ create table locp (a int check (a in (2)), b int, c text);
+ alter table utrtest attach partition remp for values in (1);
+ alter table utrtest attach partition locp for values in (2);
+ insert into utrtest values (1, 1, 'foo');
+ insert into utrtest values (2, 2, 'qux');
+ select tableoid::regclass, * FROM utrtest;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ remp | 1 | 1 | foo
+ locp | 2 | 2 | qux
+ (2 rows)
+
+ select tableoid::regclass, * FROM remp;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ remp | 1 | 1 | foo
+ (1 row)
+
+ select tableoid::regclass, * FROM locp;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ locp | 2 | 2 | qux
+ (1 row)
+
+ -- It's not allowed to move a row from a partition that is foreign to another
+ update utrtest set a = 2 where c = 'foo' returning *;
+ ERROR: new row for relation "loct" violates check constraint "loct_a_check"
+ DETAIL: Failing row contains (2, 1, foo).
+ CONTEXT: Remote SQL command: UPDATE public.loct SET a = 2 WHERE ((c = 'foo'::text)) RETURNING a, b, c
+ -- But the reverse is allowed
+ update utrtest set a = 1 where c = 'qux' returning *;
+ a | b | c
+ ---+---+-----
+ 1 | 2 | qux
+ (1 row)
+
+ select tableoid::regclass, * FROM utrtest;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ remp | 1 | 1 | foo
+ remp | 1 | 2 | qux
+ (2 rows)
+
+ select tableoid::regclass, * FROM remp;
+ tableoid | a | b | c
+ ----------+---+---+-----
+ remp | 1 | 1 | foo
+ remp | 1 | 2 | qux
+ (2 rows)
+
+ select tableoid::regclass, * FROM locp;
+ tableoid | a | b | c
+ ----------+---+---+---
+ (0 rows)
+
+ drop table utrtest;
+ drop table loct;
+ -- ===================================================================
-- test IMPORT FOREIGN SCHEMA
-- ===================================================================
CREATE SCHEMA import_source;
*** a/contrib/postgres_fdw/postgres_fdw.c
--- b/contrib/postgres_fdw/postgres_fdw.c
***************
*** 319,324 **** static TupleTableSlot *postgresExecForeignDelete(EState *estate,
--- 319,332 ----
TupleTableSlot *planSlot);
static void postgresEndForeignModify(EState *estate,
ResultRelInfo *resultRelInfo);
+ static void postgresBeginForeignRouting(ModifyTableState *mtstate,
+ ResultRelInfo *resultRelInfo,
+ int partition_index);
+ static TupleTableSlot *postgresExecForeignRouting(EState *estate,
+ ResultRelInfo *resultRelInfo,
+ TupleTableSlot *slot);
+ static void postgresEndForeignRouting(EState *estate,
+ ResultRelInfo *resultRelInfo);
static int postgresIsForeignRelUpdatable(Relation rel);
static bool postgresPlanDirectModify(PlannerInfo *root,
ModifyTable *plan,
***************
*** 473,478 **** postgres_fdw_handler(PG_FUNCTION_ARGS)
--- 481,489 ----
routine->ExecForeignUpdate = postgresExecForeignUpdate;
routine->ExecForeignDelete = postgresExecForeignDelete;
routine->EndForeignModify = postgresEndForeignModify;
+ routine->BeginForeignRouting = postgresBeginForeignRouting;
+ routine->ExecForeignRouting = postgresExecForeignRouting;
+ routine->EndForeignRouting = postgresEndForeignRouting;
routine->IsForeignRelUpdatable = postgresIsForeignRelUpdatable;
routine->PlanDirectModify = postgresPlanDirectModify;
routine->BeginDirectModify = postgresBeginDirectModify;
***************
*** 1858,1863 **** postgresEndForeignModify(EState *estate,
--- 1869,1987 ----
}
/*
+ * postgresBeginForeignRouting
+ * Begin a tuple routing operation on a foreign table
+ */
+ static void
+ postgresBeginForeignRouting(ModifyTableState *mtstate,
+ ResultRelInfo *resultRelInfo,
+ int partition_index)
+ {
+ PgFdwModifyState *fmstate;
+ Relation rel = resultRelInfo->ri_RelationDesc;
+ RangeTblEntry *rte;
+ Query *query;
+ PlannerInfo *root;
+ TupleDesc tupdesc = RelationGetDescr(rel);
+ int attnum;
+ StringInfoData sql;
+ List *targetAttrs = NIL;
+ List *retrieved_attrs = NIL;
+ bool doNothing = false;
+
+ initStringInfo(&sql);
+
+ /* Set up largely-dummy planner state */
+ rte = makeNode(RangeTblEntry);
+ rte->rtekind = RTE_RELATION;
+ rte->relid = RelationGetRelid(rel);
+ rte->relkind = RELKIND_FOREIGN_TABLE;
+ query = makeNode(Query);
+ query->commandType = CMD_INSERT;
+ query->resultRelation = 1;
+ query->rtable = list_make1(rte);
+ root = makeNode(PlannerInfo);
+ root->parse = query;
+
+ /* We transmit all columns that are defined in the foreign table. */
+ for (attnum = 1; attnum <= tupdesc->natts; attnum++)
+ {
+ Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
+
+ if (!attr->attisdropped)
+ targetAttrs = lappend_int(targetAttrs, attnum);
+ }
+
+ /* We only support DO NOTHING without an inference specification. */
+ if (mtstate->mt_onconflict == ONCONFLICT_NOTHING)
+ doNothing = true;
+ else if (mtstate->mt_onconflict != ONCONFLICT_NONE)
+ elog(ERROR, "unexpected ON CONFLICT specification: %d",
+ (int) mtstate->mt_onconflict);
+
+ /* Construct the SQL command string. */
+ deparseInsertSql(&sql, root, 1, rel, targetAttrs, doNothing,
+ resultRelInfo->ri_returningList, &retrieved_attrs);
+
+ /* Construct an execution state. */
+ fmstate = create_fdw_modify_state(mtstate,
+ resultRelInfo,
+ CMD_INSERT,
+ 0, /* dummy subplan index */
+ sql.data,
+ targetAttrs,
+ retrieved_attrs != NIL,
+ retrieved_attrs);
+
+ resultRelInfo->ri_FdwState = fmstate;
+ }
+
+ /*
+ * postgresExecForeignRouting
+ * Route one tuple to a foreign table
+ */
+ static TupleTableSlot *
+ postgresExecForeignRouting(EState *estate,
+ ResultRelInfo *resultRelInfo,
+ TupleTableSlot *slot)
+ {
+ PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
+ const char **p_values;
+ 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 and fetch RETURNING tuple if any */
+ n_rows = execute_prep_stmt(fmstate, p_values, slot);
+
+ MemoryContextReset(fmstate->temp_cxt);
+
+ /* Return NULL if nothing was inserted on the remote end */
+ return (n_rows > 0) ? slot : NULL;
+ }
+
+ /*
+ * postgresEndForeignRouting
+ * Finish a tuple routing operation on a foreign table
+ */
+ static void
+ postgresEndForeignRouting(EState *estate,
+ ResultRelInfo *resultRelInfo)
+ {
+ PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
+
+ Assert(fmstate != NULL);
+
+ /* Destroy the execution state. */
+ finish_foreign_modify(fmstate);
+ }
+
+ /*
* postgresIsForeignRelUpdatable
* Determine whether a foreign table supports INSERT, UPDATE and/or
* DELETE.
*** a/contrib/postgres_fdw/sql/postgres_fdw.sql
--- b/contrib/postgres_fdw/sql/postgres_fdw.sql
***************
*** 1759,1764 **** drop table loct1;
--- 1759,1823 ----
drop table loct2;
-- ===================================================================
+ -- test tuple routing for foreign-table partitions
+ -- ===================================================================
+
+ create table itrtest (a int, b int, c text) partition by list (a);
+ create table loct1 (a int check (a in (1)), b int primary key, c text);
+ create table loct2 (b int primary key, c text, a int check (a in (2)));
+ create foreign table remp1 (a int check (a in (1)), b int, c text) server loopback options (table_name 'loct1');
+ create foreign table remp2 (b int, c text, a int check (a in (2))) server loopback options (table_name 'loct2');
+ alter table itrtest attach partition remp1 for values in (1);
+ alter table itrtest attach partition remp2 for values in (2);
+
+ insert into itrtest values (1, 1, 'foo');
+ insert into itrtest values (1, 2, 'bar') returning *;
+ insert into itrtest values (2, 1, 'baz') returning *;
+
+ select tableoid::regclass, * FROM itrtest;
+ select tableoid::regclass, * FROM remp1;
+ select tableoid::regclass, * FROM remp2;
+
+ insert into itrtest values (2, 1, 'baz');
+ insert into itrtest values (2, 1, 'baz') on conflict do nothing;
+ insert into itrtest values (2, 2, 'qux') on conflict do nothing returning *;
+
+ select tableoid::regclass, * FROM itrtest;
+ select tableoid::regclass, * FROM remp1;
+ select tableoid::regclass, * FROM remp2;
+
+ drop table itrtest;
+ drop table loct1;
+ drop table loct2;
+
+ create table utrtest (a int, b int, c text) partition by list (a);
+ create table loct (a int check (a in (1)), b int, c text);
+ create foreign table remp (a int check (a in (1)), b int, c text) server loopback options (table_name 'loct');
+ create table locp (a int check (a in (2)), b int, c text);
+ alter table utrtest attach partition remp for values in (1);
+ alter table utrtest attach partition locp for values in (2);
+
+ insert into utrtest values (1, 1, 'foo');
+ insert into utrtest values (2, 2, 'qux');
+
+ select tableoid::regclass, * FROM utrtest;
+ select tableoid::regclass, * FROM remp;
+ select tableoid::regclass, * FROM locp;
+
+ -- It's not allowed to move a row from a partition that is foreign to another
+ update utrtest set a = 2 where c = 'foo' returning *;
+
+ -- But the reverse is allowed
+ update utrtest set a = 1 where c = 'qux' returning *;
+
+ select tableoid::regclass, * FROM utrtest;
+ select tableoid::regclass, * FROM remp;
+ select tableoid::regclass, * FROM locp;
+
+ drop table utrtest;
+ drop table loct;
+
+ -- ===================================================================
-- test IMPORT FOREIGN SCHEMA
-- ===================================================================
*** a/doc/src/sgml/fdwhandler.sgml
--- b/doc/src/sgml/fdwhandler.sgml
***************
*** 690,695 **** EndForeignModify(EState *estate,
--- 690,795 ----
</para>
<para>
+ Tuples inserted into partitioned tables are routed to partitions. If
+ an FDW supports routable foreign-table partitions, it should also
+ provide the following callback functions.
+ </para>
+
+ <para>
+ <programlisting>
+ void
+ BeginForeignRouting(ModifyTableState *mtstate,
+ ResultRelInfo *rinfo,
+ int partition_index);
+ </programlisting>
+
+ Begin executing a tuple routing operation on a foreign table. This
+ routine is called the first time the foreign table is chosen as the
+ partition to route an inserted tuple to. It should perform any
+ initialization needed prior to the actual tuple routing. Subsequently,
+ <function>ExecForeignRouting</function> will be called for each tuple
+ to be routed.
+ </para>
+
+ <para>
+ <literal>mtstate</literal> is the overall state of the
+ <structname>ModifyTable</structname> plan node being executed; global data about
+ the plan and execution state is available via this structure.
+ <literal>rinfo</literal> is the <structname>ResultRelInfo</structname> struct describing
+ the target foreign table. (The <structfield>ri_FdwState</structfield> field of
+ <structname>ResultRelInfo</structname> is available for the FDW to store any
+ private state it needs for this operation.)
+ <literal>partition_index</literal> identifies which partition of
+ a partitioned table this is, counting from zero; use this if you want
+ to index into substructures of
+ <literal>mtstate->mt_partition_tuple_routing</literal>, which contain
+ information about each partition of the partitioned table.
+ </para>
+
+ <para>
+ If the <function>BeginForeignRouting</function> pointer is set to
+ <literal>NULL</literal>, no action is taken the first time the foreign
+ table is chosen as the partition.
+ </para>
+
+ <para>
+ <programlisting>
+ TupleTableSlot *
+ ExecForeignRouting(EState *estate,
+ ResultRelInfo *rinfo,
+ TupleTableSlot *slot);
+ </programlisting>
+
+ Route one tuple to the foreign table.
+ <literal>estate</literal> is global execution state for the query.
+ <literal>rinfo</literal> is the <structname>ResultRelInfo</structname> struct describing
+ the target foreign table.
+ <literal>slot</literal> contains the tuple to be routed; it will match the
+ row-type definition of the foreign table.
+ </para>
+
+ <para>
+ The return value is either a slot containing the data that was actually
+ routed (this might differ from the data supplied, for example as a
+ result of trigger actions), or NULL if no row was actually routed
+ (again, typically as a result of triggers). The passed-in
+ <literal>slot</literal> can be re-used for this purpose.
+ </para>
+
+ <para>
+ The data in the returned slot is used only if the <command>INSERT</command>
+ query has a <literal>RETURNING</literal> clause or the foreign table has
+ an <literal>AFTER ROW</literal> trigger. Triggers require all columns, but the
+ FDW could choose to optimize away returning some or all columns depending
+ on the contents of the <literal>RETURNING</literal> clause. Regardless, some
+ slot must be returned to indicate success, or the query's reported row
+ count will be wrong.
+ </para>
+
+ <para>
+ If the <function>ExecForeignRouting</function> pointer is set to
+ <literal>NULL</literal>, attempts to route rows to the foreign table will fail
+ with an error message.
+ </para>
+
+ <para>
+ <programlisting>
+ void
+ EndForeignRouting(EState *estate,
+ ResultRelInfo *rinfo);
+ </programlisting>
+
+ End the tuple routing and release resources. It is normally not important
+ to release palloc'd memory, but for example open files and connections
+ to remote servers should be cleaned up.
+ </para>
+
+ <para>
+ If the <function>EndForeignRouting</function> pointer is set to
+ <literal>NULL</literal>, no action is taken during executor shutdown.
+ </para>
+
+ <para>
<programlisting>
int
IsForeignRelUpdatable(Relation rel);
***************
*** 709,715 **** IsForeignRelUpdatable(Relation rel);
<literal>NULL</literal>, foreign tables are assumed to be insertable, updatable,
or deletable if the FDW provides <function>ExecForeignInsert</function>,
<function>ExecForeignUpdate</function>, or <function>ExecForeignDelete</function>
! respectively. This function is only needed if the FDW supports some
tables that are updatable and some that are not. (Even then, it's
permissible to throw an error in the execution routine instead of
checking in this function. However, this function is used to determine
--- 809,817 ----
<literal>NULL</literal>, foreign tables are assumed to be insertable, updatable,
or deletable if the FDW provides <function>ExecForeignInsert</function>,
<function>ExecForeignUpdate</function>, or <function>ExecForeignDelete</function>
! respectively. In that case, foreign tables are also assumed to be
! routable if the FDW provides <function>ExecForeignRouting</function>.
! This function is only needed if the FDW supports some
tables that are updatable and some that are not. (Even then, it's
permissible to throw an error in the execution routine instead of
checking in this function. However, this function is used to determine
*** a/doc/src/sgml/ref/update.sgml
--- b/doc/src/sgml/ref/update.sgml
***************
*** 291,296 **** UPDATE <replaceable class="parameter">count</replaceable>
--- 291,299 ----
concurrent <command>UPDATE</command> or <command>DELETE</command> on the
same row may miss this row. For details see the section
<xref linkend="ddl-partitioning-declarative-limitations"/>.
+ Currently, it is not allowed to move a row from a partition that is a
+ foreign table to another, but the reverse is allowed if the foreign table
+ is routable.
</para>
</refsect1>
*** a/src/backend/commands/copy.c
--- b/src/backend/commands/copy.c
***************
*** 2615,2625 **** CopyFrom(CopyState cstate)
Assert(resultRelInfo != NULL);
}
! /* We do not yet have a way to insert into a foreign partition */
! if (resultRelInfo->ri_FdwRoutine)
! ereport(ERROR,
! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! errmsg("cannot route inserted tuples to a foreign table")));
/*
* For ExecInsertIndexTuples() to work on the partition's indexes
--- 2615,2639 ----
Assert(resultRelInfo != NULL);
}
! /*
! * Verify the specified partition is a valid target for COPY if we
! * didn't yet.
! */
! if (!resultRelInfo->ri_PartitionIsValid)
! {
! /* We do not yet have a way to copy into a foreign partition */
! if (resultRelInfo->ri_FdwRoutine)
! ereport(ERROR,
! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! errmsg("cannot route copied tuples to a foreign table")));
! /* OK, do the remaining initilization work. */
! ExecInitPartitionExtraInfo(NULL, estate,
! proute,
! resultRelInfo,
! leaf_part_index);
!
! resultRelInfo->ri_PartitionIsValid = true;
! }
/*
* For ExecInsertIndexTuples() to work on the partition's indexes
***************
*** 2813,2819 **** CopyFrom(CopyState cstate)
/* Close all the partitioned tables, leaf partitions, and their indices */
if (cstate->partition_tuple_routing)
! ExecCleanupTupleRouting(cstate->partition_tuple_routing);
/* Close any trigger target relations */
ExecCleanUpTriggerState(estate);
--- 2827,2833 ----
/* Close all the partitioned tables, leaf partitions, and their indices */
if (cstate->partition_tuple_routing)
! ExecCleanupTupleRouting(NULL, cstate->partition_tuple_routing);
/* Close any trigger target relations */
ExecCleanUpTriggerState(estate);
*** a/src/backend/executor/execMain.c
--- b/src/backend/executor/execMain.c
***************
*** 1172,1189 **** CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation)
switch (operation)
{
case CMD_INSERT:
-
- /*
- * If foreign partition to do tuple-routing for, skip the
- * check; it's disallowed elsewhere.
- */
if (resultRelInfo->ri_PartitionRoot)
! break;
! if (fdwroutine->ExecForeignInsert == NULL)
! ereport(ERROR,
! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! errmsg("cannot insert into foreign table \"%s\"",
! RelationGetRelationName(resultRel))));
if (fdwroutine->IsForeignRelUpdatable != NULL &&
(fdwroutine->IsForeignRelUpdatable(resultRel) & (1 << CMD_INSERT)) == 0)
ereport(ERROR,
--- 1172,1193 ----
switch (operation)
{
case CMD_INSERT:
if (resultRelInfo->ri_PartitionRoot)
! {
! if (fdwroutine->ExecForeignRouting == NULL)
! ereport(ERROR,
! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! errmsg("cannot route inserted tuples to foreign table \"%s\"",
! RelationGetRelationName(resultRel))));
! }
! else
! {
! if (fdwroutine->ExecForeignInsert == NULL)
! ereport(ERROR,
! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! errmsg("cannot insert into foreign table \"%s\"",
! RelationGetRelationName(resultRel))));
! }
if (fdwroutine->IsForeignRelUpdatable != NULL &&
(fdwroutine->IsForeignRelUpdatable(resultRel) & (1 << CMD_INSERT)) == 0)
ereport(ERROR,
***************
*** 1364,1369 **** InitResultRelInfo(ResultRelInfo *resultRelInfo,
--- 1368,1374 ----
resultRelInfo->ri_PartitionCheck = partition_check;
resultRelInfo->ri_PartitionRoot = partition_root;
+ resultRelInfo->ri_PartitionIsValid = false;
}
/*
*** a/src/backend/executor/execPartition.c
--- b/src/backend/executor/execPartition.c
***************
*** 17,22 ****
--- 17,23 ----
#include "catalog/pg_inherits_fn.h"
#include "executor/execPartition.h"
#include "executor/executor.h"
+ #include "foreign/fdwapi.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "utils/lsyscache.h"
***************
*** 56,62 **** static char *ExecBuildSlotPartitionKeyDescription(Relation rel,
PartitionTupleRouting *
ExecSetupPartitionTupleRouting(ModifyTableState *mtstate, Relation rel)
{
- TupleDesc tupDesc = RelationGetDescr(rel);
List *leaf_parts;
ListCell *cell;
int i;
--- 57,62 ----
***************
*** 128,138 **** ExecSetupPartitionTupleRouting(ModifyTableState *mtstate, Relation rel)
if (update_rri_index < num_update_rri &&
RelationGetRelid(update_rri[update_rri_index].ri_RelationDesc) == leaf_oid)
{
- Relation partrel;
- TupleDesc part_tupdesc;
-
leaf_part_rri = &update_rri[update_rri_index];
- partrel = leaf_part_rri->ri_RelationDesc;
/*
* This is required in order to convert the partition's tuple to
--- 128,134 ----
***************
*** 146,168 **** ExecSetupPartitionTupleRouting(ModifyTableState *mtstate, Relation rel)
proute->subplan_partition_offsets[update_rri_index] = i;
update_rri_index++;
-
- part_tupdesc = RelationGetDescr(partrel);
-
- /*
- * Save a tuple conversion map to convert a tuple routed to this
- * partition from the parent's type to the partition's.
- */
- proute->parent_child_tupconv_maps[i] =
- convert_tuples_by_name(tupDesc, part_tupdesc,
- gettext_noop("could not convert row type"));
-
- /*
- * Verify result relation is a valid target for an INSERT. An
- * UPDATE of a partition-key becomes a DELETE+INSERT operation, so
- * this check is required even when the operation is CMD_UPDATE.
- */
- CheckValidResultRel(leaf_part_rri, CMD_INSERT);
}
proute->partitions[i] = leaf_part_rri;
--- 142,147 ----
***************
*** 336,348 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
estate->es_instrument);
/*
- * Verify result relation is a valid target for an INSERT. An UPDATE of a
- * partition-key becomes a DELETE+INSERT operation, so this check is still
- * required when the operation is CMD_UPDATE.
- */
- CheckValidResultRel(leaf_part_rri, CMD_INSERT);
-
- /*
* Since we've just initialized this ResultRelInfo, it's not in any list
* attached to the estate as yet. Add it, so that it can be found later.
*
--- 315,320 ----
***************
*** 459,464 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
--- 431,437 ----
returningList = map_partition_varattnos(returningList, firstVarno,
partrel, firstResultRel,
NULL);
+ leaf_part_rri->ri_returningList = returningList;
/*
* Initialize the projection itself.
***************
*** 478,495 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
Assert(proute->partitions[partidx] == NULL);
proute->partitions[partidx] = leaf_part_rri;
/*
! * Save a tuple conversion map to convert a tuple routed to this partition
! * from the parent's type to the partition's.
*/
proute->parent_child_tupconv_maps[partidx] =
! convert_tuples_by_name(RelationGetDescr(rootrel),
! RelationGetDescr(partrel),
gettext_noop("could not convert row type"));
! MemoryContextSwitchTo(oldContext);
! return leaf_part_rri;
}
/*
--- 451,500 ----
Assert(proute->partitions[partidx] == NULL);
proute->partitions[partidx] = leaf_part_rri;
+ MemoryContextSwitchTo(oldContext);
+
+ return leaf_part_rri;
+ }
+
+ /*
+ * ExecInitPartitionExtraInfo
+ * Do the remaining initialization work for the given partition
+ */
+ void
+ ExecInitPartitionExtraInfo(ModifyTableState *mtstate,
+ EState *estate,
+ PartitionTupleRouting *proute,
+ ResultRelInfo *resultRelInfo,
+ int partidx)
+ {
+ MemoryContext oldContext;
+
+ /*
+ * Switch into per-query memory context.
+ */
+ oldContext = MemoryContextSwitchTo(estate->es_query_cxt);
+
/*
! * Set up a tuple conversion map to convert a tuple routed to the given
! * partition from the parent's type to the partition's.
*/
proute->parent_child_tupconv_maps[partidx] =
! convert_tuples_by_name(RelationGetDescr(resultRelInfo->ri_PartitionRoot),
! RelationGetDescr(resultRelInfo->ri_RelationDesc),
gettext_noop("could not convert row type"));
! /*
! * If this is INSERT/UPDATE and the given partition is a foreign table,
! * let the FDW init itself for the partition.
! */
! if (mtstate &&
! resultRelInfo->ri_FdwRoutine != NULL &&
! resultRelInfo->ri_FdwRoutine->BeginForeignRouting != NULL)
! resultRelInfo->ri_FdwRoutine->BeginForeignRouting(mtstate,
! resultRelInfo,
! partidx);
! MemoryContextSwitchTo(oldContext);
}
/*
***************
*** 594,600 **** ConvertPartitionTupleSlot(TupleConversionMap *map,
* Close all the partitioned tables, leaf partitions, and their indices.
*/
void
! ExecCleanupTupleRouting(PartitionTupleRouting *proute)
{
int i;
int subplan_index = 0;
--- 599,606 ----
* Close all the partitioned tables, leaf partitions, and their indices.
*/
void
! ExecCleanupTupleRouting(ModifyTableState *node,
! PartitionTupleRouting *proute)
{
int i;
int subplan_index = 0;
***************
*** 623,628 **** ExecCleanupTupleRouting(PartitionTupleRouting *proute)
--- 629,643 ----
continue;
/*
+ * If this is INSERT/UPDATE, allow any FDWs to shut down
+ */
+ if (node &&
+ resultRelInfo->ri_FdwRoutine != NULL &&
+ resultRelInfo->ri_FdwRoutine->EndForeignRouting != NULL)
+ resultRelInfo->ri_FdwRoutine->EndForeignRouting(node->ps.state,
+ resultRelInfo);
+
+ /*
* If this result rel is one of the UPDATE subplan result rels, let
* ExecEndPlan() close it. For INSERT or COPY,
* proute->subplan_partition_offsets will always be NULL. Note that
*** a/src/backend/executor/nodeModifyTable.c
--- b/src/backend/executor/nodeModifyTable.c
***************
*** 319,329 **** ExecInsert(ModifyTableState *mtstate,
Assert(resultRelInfo != NULL);
}
! /* We do not yet have a way to insert into a foreign partition */
! if (resultRelInfo->ri_FdwRoutine)
! ereport(ERROR,
! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! errmsg("cannot route inserted tuples to a foreign table")));
/* For ExecInsertIndexTuples() to work on the partition's indexes */
estate->es_result_relation_info = resultRelInfo;
--- 319,340 ----
Assert(resultRelInfo != NULL);
}
! /*
! * Verify the specified partition is a valid target for INSERT if we
! * didn't yet.
! */
! if (!resultRelInfo->ri_PartitionIsValid)
! {
! CheckValidResultRel(resultRelInfo, CMD_INSERT);
!
! /* OK, do the remaining initilization work. */
! ExecInitPartitionExtraInfo(mtstate, estate,
! proute,
! resultRelInfo,
! leaf_part_index);
!
! resultRelInfo->ri_PartitionIsValid = true;
! }
/* For ExecInsertIndexTuples() to work on the partition's indexes */
estate->es_result_relation_info = resultRelInfo;
***************
*** 433,442 **** ExecInsert(ModifyTableState *mtstate,
/*
* insert into foreign table: let the FDW do it
*/
! slot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
! resultRelInfo,
! slot,
! planSlot);
if (slot == NULL) /* "do nothing" */
return NULL;
--- 444,458 ----
/*
* insert into foreign table: let the FDW do it
*/
! if (resultRelInfo->ri_PartitionRoot)
! slot = resultRelInfo->ri_FdwRoutine->ExecForeignRouting(estate,
! resultRelInfo,
! slot);
! else
! slot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
! resultRelInfo,
! slot,
! planSlot);
if (slot == NULL) /* "do nothing" */
return NULL;
***************
*** 2310,2315 **** ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
--- 2326,2332 ----
{
List *rlist = (List *) lfirst(l);
+ resultRelInfo->ri_returningList = rlist;
resultRelInfo->ri_projectReturning =
ExecBuildProjectionInfo(rlist, econtext, slot, &mtstate->ps,
resultRelInfo->ri_RelationDesc->rd_att);
***************
*** 2568,2574 **** ExecEndModifyTable(ModifyTableState *node)
/* Close all the partitioned tables, leaf partitions, and their indices */
if (node->mt_partition_tuple_routing)
! ExecCleanupTupleRouting(node->mt_partition_tuple_routing);
/*
* Free the exprcontext
--- 2585,2591 ----
/* Close all the partitioned tables, leaf partitions, and their indices */
if (node->mt_partition_tuple_routing)
! ExecCleanupTupleRouting(node, node->mt_partition_tuple_routing);
/*
* Free the exprcontext
*** a/src/include/executor/execPartition.h
--- b/src/include/executor/execPartition.h
***************
*** 118,123 **** extern ResultRelInfo *ExecInitPartitionInfo(ModifyTableState *mtstate,
--- 118,128 ----
ResultRelInfo *resultRelInfo,
PartitionTupleRouting *proute,
EState *estate, int partidx);
+ extern void ExecInitPartitionExtraInfo(ModifyTableState *mtstate,
+ EState *estate,
+ PartitionTupleRouting *proute,
+ ResultRelInfo *partRelInfo,
+ int partidx);
extern void ExecSetupChildParentMapForLeaf(PartitionTupleRouting *proute);
extern TupleConversionMap *TupConvMapForLeaf(PartitionTupleRouting *proute,
ResultRelInfo *rootRelInfo, int leaf_index);
***************
*** 125,130 **** extern HeapTuple ConvertPartitionTupleSlot(TupleConversionMap *map,
HeapTuple tuple,
TupleTableSlot *new_slot,
TupleTableSlot **p_my_slot);
! extern void ExecCleanupTupleRouting(PartitionTupleRouting *proute);
#endif /* EXECPARTITION_H */
--- 130,136 ----
HeapTuple tuple,
TupleTableSlot *new_slot,
TupleTableSlot **p_my_slot);
! extern void ExecCleanupTupleRouting(ModifyTableState *node,
! PartitionTupleRouting *proute);
#endif /* EXECPARTITION_H */
*** a/src/include/foreign/fdwapi.h
--- b/src/include/foreign/fdwapi.h
***************
*** 97,102 **** typedef TupleTableSlot *(*ExecForeignDelete_function) (EState *estate,
--- 97,113 ----
typedef void (*EndForeignModify_function) (EState *estate,
ResultRelInfo *rinfo);
+ typedef void (*BeginForeignRouting_function) (ModifyTableState *mtstate,
+ ResultRelInfo *rinfo,
+ int partition_index);
+
+ typedef TupleTableSlot *(*ExecForeignRouting_function) (EState *estate,
+ ResultRelInfo *rinfo,
+ TupleTableSlot *slot);
+
+ typedef void (*EndForeignRouting_function) (EState *estate,
+ ResultRelInfo *rinfo);
+
typedef int (*IsForeignRelUpdatable_function) (Relation rel);
typedef bool (*PlanDirectModify_function) (PlannerInfo *root,
***************
*** 204,209 **** typedef struct FdwRoutine
--- 215,223 ----
ExecForeignUpdate_function ExecForeignUpdate;
ExecForeignDelete_function ExecForeignDelete;
EndForeignModify_function EndForeignModify;
+ BeginForeignRouting_function BeginForeignRouting;
+ ExecForeignRouting_function ExecForeignRouting;
+ EndForeignRouting_function EndForeignRouting;
IsForeignRelUpdatable_function IsForeignRelUpdatable;
PlanDirectModify_function PlanDirectModify;
BeginDirectModify_function BeginDirectModify;
*** a/src/include/nodes/execnodes.h
--- b/src/include/nodes/execnodes.h
***************
*** 403,408 **** typedef struct ResultRelInfo
--- 403,411 ----
/* list of WithCheckOption expr states */
List *ri_WithCheckOptionExprs;
+ /* list of RETURNING expressions */
+ List *ri_returningList;
+
/* array of constraint-checking expr states */
ExprState **ri_ConstraintExprs;
***************
*** 426,431 **** typedef struct ResultRelInfo
--- 429,437 ----
/* relation descriptor for root partitioned table */
Relation ri_PartitionRoot;
+
+ /* true if valid target for tuple routing */
+ bool ri_PartitionIsValid;
} ResultRelInfo;
/* ----------------
