(2018/04/04 19:31), Etsuro Fujita wrote:
Attached is an updated version of the patch set:
* As before, patch foreign-routing-fdwapi-4.patch is created on top of
patch postgres-fdw-refactoring-4.patch and the bug-fix patch [1].
I did a bit of cleanup and comment-rewording to patch
foreign-routing-fdwapi-4.patch. Attached is a new version of the patch
set. I renamed the postgres_fdw refactoring patch but no changes to
that patch.
Best regards,
Etsuro Fujita
*** a/contrib/postgres_fdw/postgres_fdw.c
--- b/contrib/postgres_fdw/postgres_fdw.c
***************
*** 376,387 **** static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel,
--- 376,396 ----
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_foreign_modify(EState *estate,
+ ResultRelInfo *resultRelInfo,
+ CmdType operation,
+ Plan *subplan,
+ 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 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);
***************
*** 1681,1698 **** 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
***************
*** 1701,1782 **** 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_foreign_modify(mtstate->ps.state,
! resultRelInfo,
! mtstate->operation,
! mtstate->mt_plans[subplan_index]->plan,
! query,
! target_attrs,
! has_returning,
! retrieved_attrs);
resultRelInfo->ri_FdwState = fmstate;
}
***************
*** 2011,2038 **** 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;
}
/*
--- 1955,1962 ----
if (fmstate == NULL)
return;
! /* Destroy the execution state */
! finish_foreign_modify(fmstate);
}
/*
***************
*** 3229,3234 **** close_cursor(PGconn *conn, unsigned int cursor_number)
--- 3153,3261 ----
}
/*
+ * create_foreign_modify
+ * Construct an execution state of a foreign insert/update/delete
+ * operation
+ */
+ static PgFdwModifyState *
+ create_foreign_modify(EState *estate,
+ ResultRelInfo *resultRelInfo,
+ CmdType operation,
+ Plan *subplan,
+ char *query,
+ List *target_attrs,
+ bool has_returning,
+ List *retrieved_attrs)
+ {
+ PgFdwModifyState *fmstate;
+ Relation rel = resultRelInfo->ri_RelationDesc;
+ TupleDesc tupdesc = RelationGetDescr(rel);
+ RangeTblEntry *rte;
+ Oid userid;
+ ForeignTable *table;
+ UserMapping *user;
+ AttrNumber n_params;
+ Oid typefnoid;
+ bool isvarlena;
+ ListCell *lc;
+
+ /* 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 up 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)
+ {
+ Assert(subplan != NULL);
+
+ /* Find the ctid resjunk column in the subplan's result */
+ 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
*/
***************
*** 3371,3376 **** store_returning_result(PgFdwModifyState *fmstate,
--- 3398,3436 ----
}
/*
+ * 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/input/file_fdw.source
--- b/contrib/file_fdw/input/file_fdw.source
***************
*** 136,141 **** DELETE FROM agg_csv WHERE a = 100;
--- 136,146 ----
-- but this should be allowed
SELECT * FROM agg_csv FOR UPDATE;
+ -- copy from isn't supported either
+ COPY agg_csv FROM STDIN;
+ 12 3.4
+ \.
+
-- constraint exclusion tests
\t on
EXPLAIN (VERBOSE, COSTS FALSE) SELECT * FROM agg_csv WHERE a < 0;
*** a/contrib/file_fdw/output/file_fdw.source
--- b/contrib/file_fdw/output/file_fdw.source
***************
*** 221,226 **** SELECT * FROM agg_csv FOR UPDATE;
--- 221,229 ----
42 | 324.78
(3 rows)
+ -- copy from isn't supported either
+ COPY agg_csv FROM STDIN;
+ ERROR: cannot insert into foreign table "agg_csv"
-- constraint exclusion tests
\t on
EXPLAIN (VERBOSE, COSTS FALSE) SELECT * FROM agg_csv WHERE a < 0;
***************
*** 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;
--- 318,324 ----
(0 rows)
COPY pt FROM '@abs_srcdir@/data/list2.bad' with (format 'csv', delimiter ','); -- ERROR
! ERROR: cannot insert into foreign table "p1"
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
----------+---+-------
--- 345,354 ----
(2 rows)
INSERT INTO pt VALUES (1, 'xyzzy'); -- ERROR
! ERROR: cannot insert into foreign table "p1"
INSERT INTO pt VALUES (2, 'xyzzy');
UPDATE pt set a = 1 where a = 2; -- ERROR
! ERROR: cannot insert into 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
***************
*** 7371,7376 **** NOTICE: drop cascades to foreign table bar2
--- 7371,7710 ----
drop table loct1;
drop table loct2;
-- ===================================================================
+ -- test tuple routing for foreign-table partitions
+ -- ===================================================================
+ -- Test insert tuple routing
+ create table itrtest (a int, b text) partition by list (a);
+ create table loct1 (a int check (a in (1)), b text);
+ create foreign table remp1 (a int check (a in (1)), b text) server loopback options (table_name 'loct1');
+ create table loct2 (a int check (a in (2)), b text);
+ create foreign table remp2 (b 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, 'foo');
+ insert into itrtest values (1, 'bar') returning *;
+ a | b
+ ---+-----
+ 1 | bar
+ (1 row)
+
+ insert into itrtest values (2, 'baz');
+ insert into itrtest values (2, 'qux') returning *;
+ a | b
+ ---+-----
+ 2 | qux
+ (1 row)
+
+ insert into itrtest values (1, 'test1'), (2, 'test2') returning *;
+ a | b
+ ---+-------
+ 1 | test1
+ 2 | test2
+ (2 rows)
+
+ select tableoid::regclass, * FROM itrtest;
+ tableoid | a | b
+ ----------+---+-------
+ remp1 | 1 | foo
+ remp1 | 1 | bar
+ remp1 | 1 | test1
+ remp2 | 2 | baz
+ remp2 | 2 | qux
+ remp2 | 2 | test2
+ (6 rows)
+
+ select tableoid::regclass, * FROM remp1;
+ tableoid | a | b
+ ----------+---+-------
+ remp1 | 1 | foo
+ remp1 | 1 | bar
+ remp1 | 1 | test1
+ (3 rows)
+
+ select tableoid::regclass, * FROM remp2;
+ tableoid | b | a
+ ----------+-------+---
+ remp2 | baz | 2
+ remp2 | qux | 2
+ remp2 | test2 | 2
+ (3 rows)
+
+ delete from itrtest;
+ create unique index loct1_idx on loct1 (a);
+ -- DO NOTHING without an inference specification is supported
+ insert into itrtest values (1, 'foo') on conflict do nothing returning *;
+ a | b
+ ---+-----
+ 1 | foo
+ (1 row)
+
+ insert into itrtest values (1, 'foo') on conflict do nothing returning *;
+ a | b
+ ---+---
+ (0 rows)
+
+ -- But other cases are not supported
+ insert into itrtest values (1, 'bar') on conflict (a) do nothing;
+ ERROR: there is no unique or exclusion constraint matching the ON CONFLICT specification
+ insert into itrtest values (1, 'bar') on conflict (a) do update set b = excluded.b;
+ ERROR: there is no unique or exclusion constraint matching the ON CONFLICT specification
+ select tableoid::regclass, * FROM itrtest;
+ tableoid | a | b
+ ----------+---+-----
+ remp1 | 1 | foo
+ (1 row)
+
+ drop table itrtest;
+ drop table loct1;
+ drop table loct2;
+ -- Test update tuple routing
+ create table utrtest (a int, b text) partition by list (a);
+ create table loct (a int check (a in (1)), b text);
+ create foreign table remp (a int check (a in (1)), b text) server loopback options (table_name 'loct');
+ create table locp (a int check (a in (2)), b 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, 'foo');
+ insert into utrtest values (2, 'qux');
+ select tableoid::regclass, * FROM utrtest;
+ tableoid | a | b
+ ----------+---+-----
+ remp | 1 | foo
+ locp | 2 | qux
+ (2 rows)
+
+ select tableoid::regclass, * FROM remp;
+ tableoid | a | b
+ ----------+---+-----
+ remp | 1 | foo
+ (1 row)
+
+ select tableoid::regclass, * FROM locp;
+ tableoid | a | b
+ ----------+---+-----
+ locp | 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 b = 'foo' returning *;
+ ERROR: new row for relation "loct" violates check constraint "loct_a_check"
+ DETAIL: Failing row contains (2, foo).
+ CONTEXT: remote SQL command: UPDATE public.loct SET a = 2 WHERE ((b = 'foo'::text)) RETURNING a, b
+ -- But the reverse is allowed
+ update utrtest set a = 1 where b = 'qux' returning *;
+ a | b
+ ---+-----
+ 1 | qux
+ (1 row)
+
+ select tableoid::regclass, * FROM utrtest;
+ tableoid | a | b
+ ----------+---+-----
+ remp | 1 | foo
+ remp | 1 | qux
+ (2 rows)
+
+ select tableoid::regclass, * FROM remp;
+ tableoid | a | b
+ ----------+---+-----
+ remp | 1 | foo
+ remp | 1 | qux
+ (2 rows)
+
+ select tableoid::regclass, * FROM locp;
+ tableoid | a | b
+ ----------+---+---
+ (0 rows)
+
+ -- The executor should not let unexercised FDWs shut down
+ update utrtest set a = 1 where b = 'foo';
+ drop table utrtest;
+ drop table loct;
+ -- Test copy tuple routing
+ create table ctrtest (a int, b text) partition by list (a);
+ create table loct1 (a int check (a in (1)), b text);
+ create foreign table remp1 (a int check (a in (1)), b text) server loopback options (table_name 'loct1');
+ create table loct2 (a int check (a in (2)), b text);
+ create foreign table remp2 (b text, a int check (a in (2))) server loopback options (table_name 'loct2');
+ alter table ctrtest attach partition remp1 for values in (1);
+ alter table ctrtest attach partition remp2 for values in (2);
+ copy ctrtest from stdin;
+ select tableoid::regclass, * FROM ctrtest;
+ tableoid | a | b
+ ----------+---+-----
+ remp1 | 1 | foo
+ remp2 | 2 | qux
+ (2 rows)
+
+ select tableoid::regclass, * FROM remp1;
+ tableoid | a | b
+ ----------+---+-----
+ remp1 | 1 | foo
+ (1 row)
+
+ select tableoid::regclass, * FROM remp2;
+ tableoid | b | a
+ ----------+-----+---
+ remp2 | qux | 2
+ (1 row)
+
+ -- Copying into foreign partitions directly should work as well
+ copy remp1 from stdin;
+ select tableoid::regclass, * FROM remp1;
+ tableoid | a | b
+ ----------+---+-----
+ remp1 | 1 | foo
+ remp1 | 1 | bar
+ (2 rows)
+
+ drop table ctrtest;
+ drop table loct1;
+ drop table loct2;
+ -- ===================================================================
+ -- test COPY FROM
+ -- ===================================================================
+ create table loc2 (f1 int, f2 text);
+ alter table loc2 set (autovacuum_enabled = 'false');
+ create foreign table rem2 (f1 int, f2 text) server loopback options(table_name 'loc2');
+ -- Test basic functionality
+ copy rem2 from stdin;
+ select * from rem2;
+ f1 | f2
+ ----+-----
+ 1 | foo
+ 2 | bar
+ (2 rows)
+
+ delete from rem2;
+ -- Test check constraints
+ alter table loc2 add constraint loc2_f1positive check (f1 >= 0);
+ alter foreign table rem2 add constraint rem2_f1positive check (f1 >= 0);
+ -- check constraint is enforced on the remote side, not locally
+ copy rem2 from stdin;
+ copy rem2 from stdin; -- ERROR
+ ERROR: new row for relation "loc2" violates check constraint "loc2_f1positive"
+ DETAIL: Failing row contains (-1, xyzzy).
+ CONTEXT: remote SQL command: INSERT INTO public.loc2(f1, f2) VALUES ($1, $2)
+ COPY rem2, line 1: "-1 xyzzy"
+ select * from rem2;
+ f1 | f2
+ ----+-----
+ 1 | foo
+ 2 | bar
+ (2 rows)
+
+ alter foreign table rem2 drop constraint rem2_f1positive;
+ alter table loc2 drop constraint loc2_f1positive;
+ delete from rem2;
+ -- Test local triggers
+ create trigger trig_stmt_before before insert on rem2
+ for each statement execute procedure trigger_func();
+ create trigger trig_stmt_after after insert on rem2
+ for each statement execute procedure trigger_func();
+ create trigger trig_row_before before insert on rem2
+ for each row execute procedure trigger_data(23,'skidoo');
+ create trigger trig_row_after after insert on rem2
+ for each row execute procedure trigger_data(23,'skidoo');
+ copy rem2 from stdin;
+ NOTICE: trigger_func(<NULL>) called: action = INSERT, when = BEFORE, level = STATEMENT
+ NOTICE: trig_row_before(23, skidoo) BEFORE ROW INSERT ON rem2
+ NOTICE: NEW: (1,foo)
+ NOTICE: trig_row_before(23, skidoo) BEFORE ROW INSERT ON rem2
+ NOTICE: NEW: (2,bar)
+ NOTICE: trig_row_after(23, skidoo) AFTER ROW INSERT ON rem2
+ NOTICE: NEW: (1,foo)
+ NOTICE: trig_row_after(23, skidoo) AFTER ROW INSERT ON rem2
+ NOTICE: NEW: (2,bar)
+ NOTICE: trigger_func(<NULL>) called: action = INSERT, when = AFTER, level = STATEMENT
+ select * from rem2;
+ f1 | f2
+ ----+-----
+ 1 | foo
+ 2 | bar
+ (2 rows)
+
+ drop trigger trig_row_before on rem2;
+ drop trigger trig_row_after on rem2;
+ drop trigger trig_stmt_before on rem2;
+ drop trigger trig_stmt_after on rem2;
+ delete from rem2;
+ create trigger trig_row_before_insert before insert on rem2
+ for each row execute procedure trig_row_before_insupdate();
+ -- The new values are concatenated with ' triggered !'
+ copy rem2 from stdin;
+ select * from rem2;
+ f1 | f2
+ ----+-----------------
+ 1 | foo triggered !
+ 2 | bar triggered !
+ (2 rows)
+
+ drop trigger trig_row_before_insert on rem2;
+ delete from rem2;
+ create trigger trig_null before insert on rem2
+ for each row execute procedure trig_null();
+ -- Nothing happens
+ copy rem2 from stdin;
+ select * from rem2;
+ f1 | f2
+ ----+----
+ (0 rows)
+
+ drop trigger trig_null on rem2;
+ delete from rem2;
+ -- Test remote triggers
+ create trigger trig_row_before_insert before insert on loc2
+ for each row execute procedure trig_row_before_insupdate();
+ -- The new values are concatenated with ' triggered !'
+ copy rem2 from stdin;
+ select * from rem2;
+ f1 | f2
+ ----+-----------------
+ 1 | foo triggered !
+ 2 | bar triggered !
+ (2 rows)
+
+ drop trigger trig_row_before_insert on loc2;
+ delete from rem2;
+ create trigger trig_null before insert on loc2
+ for each row execute procedure trig_null();
+ -- Nothing happens
+ copy rem2 from stdin;
+ select * from rem2;
+ f1 | f2
+ ----+----
+ (0 rows)
+
+ drop trigger trig_null on loc2;
+ delete from rem2;
+ -- Test a combination of local and remote triggers
+ create trigger rem2_trig_row_before before insert on rem2
+ for each row execute procedure trigger_data(23,'skidoo');
+ create trigger rem2_trig_row_after after insert on rem2
+ for each row execute procedure trigger_data(23,'skidoo');
+ create trigger loc2_trig_row_before_insert before insert on loc2
+ for each row execute procedure trig_row_before_insupdate();
+ copy rem2 from stdin;
+ NOTICE: rem2_trig_row_before(23, skidoo) BEFORE ROW INSERT ON rem2
+ NOTICE: NEW: (1,foo)
+ NOTICE: rem2_trig_row_before(23, skidoo) BEFORE ROW INSERT ON rem2
+ NOTICE: NEW: (2,bar)
+ NOTICE: rem2_trig_row_after(23, skidoo) AFTER ROW INSERT ON rem2
+ NOTICE: NEW: (1,"foo triggered !")
+ NOTICE: rem2_trig_row_after(23, skidoo) AFTER ROW INSERT ON rem2
+ NOTICE: NEW: (2,"bar triggered !")
+ select * from rem2;
+ f1 | f2
+ ----+-----------------
+ 1 | foo triggered !
+ 2 | bar triggered !
+ (2 rows)
+
+ drop trigger rem2_trig_row_before on rem2;
+ drop trigger rem2_trig_row_after on rem2;
+ drop trigger loc2_trig_row_before_insert on loc2;
+ delete from rem2;
+ -- ===================================================================
-- 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,328 ----
TupleTableSlot *planSlot);
static void postgresEndForeignModify(EState *estate,
ResultRelInfo *resultRelInfo);
+ static void postgresBeginForeignInsert(ModifyTableState *mtstate,
+ ResultRelInfo *resultRelInfo);
+ static void postgresEndForeignInsert(EState *estate,
+ ResultRelInfo *resultRelInfo);
static int postgresIsForeignRelUpdatable(Relation rel);
static bool postgresPlanDirectModify(PlannerInfo *root,
ModifyTable *plan,
***************
*** 473,478 **** postgres_fdw_handler(PG_FUNCTION_ARGS)
--- 477,484 ----
routine->ExecForeignUpdate = postgresExecForeignUpdate;
routine->ExecForeignDelete = postgresExecForeignDelete;
routine->EndForeignModify = postgresEndForeignModify;
+ routine->BeginForeignInsert = postgresBeginForeignInsert;
+ routine->EndForeignInsert = postgresEndForeignInsert;
routine->IsForeignRelUpdatable = postgresIsForeignRelUpdatable;
routine->PlanDirectModify = postgresPlanDirectModify;
routine->BeginDirectModify = postgresBeginDirectModify;
***************
*** 1960,1965 **** postgresEndForeignModify(EState *estate,
--- 1966,2061 ----
}
/*
+ * postgresBeginForeignInsert
+ * Begin an insert operation on a foreign table
+ */
+ static void
+ postgresBeginForeignInsert(ModifyTableState *mtstate,
+ ResultRelInfo *resultRelInfo)
+ {
+ PgFdwModifyState *fmstate;
+ Plan *plan = mtstate->ps.plan;
+ 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);
+ }
+
+ /* Check if we add the ON CONFLICT clause to the remote query. */
+ if (plan)
+ {
+ OnConflictAction onConflictAction = ((ModifyTable *) plan)->onConflictAction;
+
+ /* We only support DO NOTHING without an inference specification. */
+ if (onConflictAction == ONCONFLICT_NOTHING)
+ doNothing = true;
+ else if (onConflictAction != ONCONFLICT_NONE)
+ elog(ERROR, "unexpected ON CONFLICT specification: %d",
+ (int) onConflictAction);
+ }
+
+ /* Construct the SQL command string. */
+ deparseInsertSql(&sql, root, 1, rel, targetAttrs, doNothing,
+ resultRelInfo->ri_returningList, &retrieved_attrs);
+
+ /* Construct an execution state. */
+ fmstate = create_foreign_modify(mtstate->ps.state,
+ resultRelInfo,
+ CMD_INSERT,
+ NULL,
+ sql.data,
+ targetAttrs,
+ retrieved_attrs != NIL,
+ retrieved_attrs);
+
+ resultRelInfo->ri_FdwState = fmstate;
+ }
+
+ /*
+ * postgresEndForeignInsert
+ * Finish an insert operation on a foreign table
+ */
+ static void
+ postgresEndForeignInsert(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
***************
*** 1768,1773 **** drop table loct1;
--- 1768,2010 ----
drop table loct2;
-- ===================================================================
+ -- test tuple routing for foreign-table partitions
+ -- ===================================================================
+
+ -- Test insert tuple routing
+ create table itrtest (a int, b text) partition by list (a);
+ create table loct1 (a int check (a in (1)), b text);
+ create foreign table remp1 (a int check (a in (1)), b text) server loopback options (table_name 'loct1');
+ create table loct2 (a int check (a in (2)), b text);
+ create foreign table remp2 (b 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, 'foo');
+ insert into itrtest values (1, 'bar') returning *;
+ insert into itrtest values (2, 'baz');
+ insert into itrtest values (2, 'qux') returning *;
+ insert into itrtest values (1, 'test1'), (2, 'test2') returning *;
+
+ select tableoid::regclass, * FROM itrtest;
+ select tableoid::regclass, * FROM remp1;
+ select tableoid::regclass, * FROM remp2;
+
+ delete from itrtest;
+
+ create unique index loct1_idx on loct1 (a);
+
+ -- DO NOTHING without an inference specification is supported
+ insert into itrtest values (1, 'foo') on conflict do nothing returning *;
+ insert into itrtest values (1, 'foo') on conflict do nothing returning *;
+
+ -- But other cases are not supported
+ insert into itrtest values (1, 'bar') on conflict (a) do nothing;
+ insert into itrtest values (1, 'bar') on conflict (a) do update set b = excluded.b;
+
+ select tableoid::regclass, * FROM itrtest;
+
+ drop table itrtest;
+ drop table loct1;
+ drop table loct2;
+
+ -- Test update tuple routing
+ create table utrtest (a int, b text) partition by list (a);
+ create table loct (a int check (a in (1)), b text);
+ create foreign table remp (a int check (a in (1)), b text) server loopback options (table_name 'loct');
+ create table locp (a int check (a in (2)), b 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, 'foo');
+ insert into utrtest values (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 b = 'foo' returning *;
+
+ -- But the reverse is allowed
+ update utrtest set a = 1 where b = 'qux' returning *;
+
+ select tableoid::regclass, * FROM utrtest;
+ select tableoid::regclass, * FROM remp;
+ select tableoid::regclass, * FROM locp;
+
+ -- The executor should not let unexercised FDWs shut down
+ update utrtest set a = 1 where b = 'foo';
+
+ drop table utrtest;
+ drop table loct;
+
+ -- Test copy tuple routing
+ create table ctrtest (a int, b text) partition by list (a);
+ create table loct1 (a int check (a in (1)), b text);
+ create foreign table remp1 (a int check (a in (1)), b text) server loopback options (table_name 'loct1');
+ create table loct2 (a int check (a in (2)), b text);
+ create foreign table remp2 (b text, a int check (a in (2))) server loopback options (table_name 'loct2');
+ alter table ctrtest attach partition remp1 for values in (1);
+ alter table ctrtest attach partition remp2 for values in (2);
+
+ copy ctrtest from stdin;
+ 1 foo
+ 2 qux
+ \.
+
+ select tableoid::regclass, * FROM ctrtest;
+ select tableoid::regclass, * FROM remp1;
+ select tableoid::regclass, * FROM remp2;
+
+ -- Copying into foreign partitions directly should work as well
+ copy remp1 from stdin;
+ 1 bar
+ \.
+
+ select tableoid::regclass, * FROM remp1;
+
+ drop table ctrtest;
+ drop table loct1;
+ drop table loct2;
+
+ -- ===================================================================
+ -- test COPY FROM
+ -- ===================================================================
+
+ create table loc2 (f1 int, f2 text);
+ alter table loc2 set (autovacuum_enabled = 'false');
+ create foreign table rem2 (f1 int, f2 text) server loopback options(table_name 'loc2');
+
+ -- Test basic functionality
+ copy rem2 from stdin;
+ 1 foo
+ 2 bar
+ \.
+ select * from rem2;
+
+ delete from rem2;
+
+ -- Test check constraints
+ alter table loc2 add constraint loc2_f1positive check (f1 >= 0);
+ alter foreign table rem2 add constraint rem2_f1positive check (f1 >= 0);
+
+ -- check constraint is enforced on the remote side, not locally
+ copy rem2 from stdin;
+ 1 foo
+ 2 bar
+ \.
+ copy rem2 from stdin; -- ERROR
+ -1 xyzzy
+ \.
+ select * from rem2;
+
+ alter foreign table rem2 drop constraint rem2_f1positive;
+ alter table loc2 drop constraint loc2_f1positive;
+
+ delete from rem2;
+
+ -- Test local triggers
+ create trigger trig_stmt_before before insert on rem2
+ for each statement execute procedure trigger_func();
+ create trigger trig_stmt_after after insert on rem2
+ for each statement execute procedure trigger_func();
+ create trigger trig_row_before before insert on rem2
+ for each row execute procedure trigger_data(23,'skidoo');
+ create trigger trig_row_after after insert on rem2
+ for each row execute procedure trigger_data(23,'skidoo');
+
+ copy rem2 from stdin;
+ 1 foo
+ 2 bar
+ \.
+ select * from rem2;
+
+ drop trigger trig_row_before on rem2;
+ drop trigger trig_row_after on rem2;
+ drop trigger trig_stmt_before on rem2;
+ drop trigger trig_stmt_after on rem2;
+
+ delete from rem2;
+
+ create trigger trig_row_before_insert before insert on rem2
+ for each row execute procedure trig_row_before_insupdate();
+
+ -- The new values are concatenated with ' triggered !'
+ copy rem2 from stdin;
+ 1 foo
+ 2 bar
+ \.
+ select * from rem2;
+
+ drop trigger trig_row_before_insert on rem2;
+
+ delete from rem2;
+
+ create trigger trig_null before insert on rem2
+ for each row execute procedure trig_null();
+
+ -- Nothing happens
+ copy rem2 from stdin;
+ 1 foo
+ 2 bar
+ \.
+ select * from rem2;
+
+ drop trigger trig_null on rem2;
+
+ delete from rem2;
+
+ -- Test remote triggers
+ create trigger trig_row_before_insert before insert on loc2
+ for each row execute procedure trig_row_before_insupdate();
+
+ -- The new values are concatenated with ' triggered !'
+ copy rem2 from stdin;
+ 1 foo
+ 2 bar
+ \.
+ select * from rem2;
+
+ drop trigger trig_row_before_insert on loc2;
+
+ delete from rem2;
+
+ create trigger trig_null before insert on loc2
+ for each row execute procedure trig_null();
+
+ -- Nothing happens
+ copy rem2 from stdin;
+ 1 foo
+ 2 bar
+ \.
+ select * from rem2;
+
+ drop trigger trig_null on loc2;
+
+ delete from rem2;
+
+ -- Test a combination of local and remote triggers
+ create trigger rem2_trig_row_before before insert on rem2
+ for each row execute procedure trigger_data(23,'skidoo');
+ create trigger rem2_trig_row_after after insert on rem2
+ for each row execute procedure trigger_data(23,'skidoo');
+ create trigger loc2_trig_row_before_insert before insert on loc2
+ for each row execute procedure trig_row_before_insupdate();
+
+ copy rem2 from stdin;
+ 1 foo
+ 2 bar
+ \.
+ select * from rem2;
+
+ drop trigger rem2_trig_row_before on rem2;
+ drop trigger rem2_trig_row_after on rem2;
+ drop trigger loc2_trig_row_before_insert on loc2;
+
+ delete from rem2;
+
+ -- ===================================================================
-- test IMPORT FOREIGN SCHEMA
-- ===================================================================
*** a/doc/src/sgml/ddl.sgml
--- b/doc/src/sgml/ddl.sgml
***************
*** 3037,3047 **** VALUES ('Albany', NULL, NULL, 'NY');
</para>
<para>
! Partitions can also be foreign tables
! (see <xref linkend="sql-createforeigntable"/>),
! although these have some limitations that normal tables do not. For
! example, data inserted into the partitioned table is not routed to
! foreign table partitions.
</para>
<para>
--- 3037,3045 ----
</para>
<para>
! Partitions can also be foreign tables, although they have some limitations
! that normal tables do not; see <xref linkend="sql-createforeigntable"> for
! more information.
</para>
<para>
*** a/doc/src/sgml/fdwhandler.sgml
--- b/doc/src/sgml/fdwhandler.sgml
***************
*** 695,700 **** EndForeignModify(EState *estate,
--- 695,766 ----
</para>
<para>
+ Tuples inserted into a partitioned table by <command>INSERT</command> or
+ <command>COPY FROM</command> are routed to partitions. If an FDW
+ supports routable foreign-table partitions, it should also provide the
+ following callback functions. These functions are also called when
+ <command>COPY FROM</command> is executed on a foreign table.
+ </para>
+
+ <para>
+ <programlisting>
+ void
+ BeginForeignInsert(ModifyTableState *mtstate,
+ ResultRelInfo *rinfo);
+ </programlisting>
+
+ Begin executing an insert operation on a foreign table. This routine is
+ called right before the first tuple is inserted into the foreign table
+ in both cases when it is the partition chosen for tuple routing and the
+ target specified in a <command>COPY FROM</command> command. It should
+ perform any initialization needed prior to the actual insertion.
+ Subsequently, <function>ExecForeignInsert</function> will be called for
+ each tuple to be inserted into the foreign table.
+ </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.)
+ </para>
+
+ <para>
+ When this is called by a <command>COPY FROM</command> command, the
+ plan-related global data in <literal>mtstate</literal> is not provided
+ and the <literal>planSlot</literal> parameter of
+ <function>ExecForeignInsert</function> subsequently called for each
+ inserted tuple is <literal>NULL</literal>, whether the foreign table is
+ the partition chosen for tuple routing or the target specified in the
+ command.
+ </para>
+
+ <para>
+ If the <function>BeginForeignInsert</function> pointer is set to
+ <literal>NULL</literal>, no action is taken for the initialization.
+ </para>
+
+ <para>
+ <programlisting>
+ void
+ EndForeignInsert(EState *estate,
+ ResultRelInfo *rinfo);
+ </programlisting>
+
+ End the insert operation 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>EndForeignInsert</function> pointer is set to
+ <literal>NULL</literal>, no action is taken for the termination.
+ </para>
+
+ <para>
<programlisting>
int
IsForeignRelUpdatable(Relation rel);
*** a/doc/src/sgml/ref/copy.sgml
--- b/doc/src/sgml/ref/copy.sgml
***************
*** 402,409 **** COPY <replaceable class="parameter">count</replaceable>
</para>
<para>
! <command>COPY FROM</command> can be used with plain tables and with views
! that have <literal>INSTEAD OF INSERT</literal> triggers.
</para>
<para>
--- 402,410 ----
</para>
<para>
! <command>COPY FROM</command> can be used with plain, foreign, or
! partitioned tables or with views that have
! <literal>INSTEAD OF INSERT</literal> triggers.
</para>
<para>
*** 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
***************
*** 29,34 ****
--- 29,35 ----
#include "commands/trigger.h"
#include "executor/execPartition.h"
#include "executor/executor.h"
+ #include "foreign/fdwapi.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "mb/pg_wchar.h"
***************
*** 2284,2289 **** CopyFrom(CopyState cstate)
--- 2285,2291 ----
ResultRelInfo *resultRelInfo;
ResultRelInfo *saved_resultRelInfo = NULL;
EState *estate = CreateExecutorState(); /* for ExecConstraints() */
+ ModifyTableState *mtstate;
ExprContext *econtext;
TupleTableSlot *myslot;
MemoryContext oldcontext = CurrentMemoryContext;
***************
*** 2305,2315 **** CopyFrom(CopyState cstate)
Assert(cstate->rel);
/*
! * The target must be a plain relation or have an INSTEAD OF INSERT row
! * trigger. (Currently, such triggers are only allowed on views, so we
! * only hint about them in the view case.)
*/
if (cstate->rel->rd_rel->relkind != RELKIND_RELATION &&
cstate->rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE &&
!(cstate->rel->trigdesc &&
cstate->rel->trigdesc->trig_insert_instead_row))
--- 2307,2318 ----
Assert(cstate->rel);
/*
! * The target must be a plain, foreign, or partitioned relation, or have
! * an INSTEAD OF INSERT row trigger. (Currently, such triggers are only
! * allowed on views, so we only hint about them in the view case.)
*/
if (cstate->rel->rd_rel->relkind != RELKIND_RELATION &&
+ cstate->rel->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
cstate->rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE &&
!(cstate->rel->trigdesc &&
cstate->rel->trigdesc->trig_insert_instead_row))
***************
*** 2325,2335 **** CopyFrom(CopyState cstate)
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot copy to materialized view \"%s\"",
RelationGetRelationName(cstate->rel))));
- else if (cstate->rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
- ereport(ERROR,
- (errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("cannot copy to foreign table \"%s\"",
- RelationGetRelationName(cstate->rel))));
else if (cstate->rel->rd_rel->relkind == RELKIND_SEQUENCE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
--- 2328,2333 ----
***************
*** 2436,2441 **** CopyFrom(CopyState cstate)
--- 2434,2442 ----
NULL,
0);
+ /* Verify the named relation is a valid target for INSERT */
+ CheckValidResultRel(resultRelInfo, CMD_INSERT);
+
ExecOpenIndices(resultRelInfo, false);
estate->es_result_relations = resultRelInfo;
***************
*** 2448,2453 **** CopyFrom(CopyState cstate)
--- 2449,2469 ----
/* Triggers might need a slot as well */
estate->es_trig_tuple_slot = ExecInitExtraTupleSlot(estate, NULL);
+ /*
+ * Set up a ModifyTableState so we can let FDW(s) init themselves for
+ * foreign-table result relation(s).
+ */
+ mtstate = makeNode(ModifyTableState);
+ mtstate->ps.plan = NULL;
+ mtstate->ps.state = estate;
+ mtstate->operation = CMD_INSERT;
+ mtstate->resultRelInfo = estate->es_result_relations;
+
+ if (resultRelInfo->ri_FdwRoutine != NULL &&
+ resultRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL)
+ resultRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate,
+ resultRelInfo);
+
/* Prepare to catch AFTER triggers. */
AfterTriggerBeginQuery();
***************
*** 2489,2499 **** CopyFrom(CopyState cstate)
* expressions. Such triggers or expressions might query the table we're
* inserting to, and act differently if the tuples that have already been
* processed and prepared for insertion are not there. We also can't do
! * it if the table is partitioned.
*/
if ((resultRelInfo->ri_TrigDesc != NULL &&
(resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
resultRelInfo->ri_TrigDesc->trig_insert_instead_row)) ||
cstate->partition_tuple_routing != NULL ||
cstate->volatile_defexprs)
{
--- 2505,2516 ----
* expressions. Such triggers or expressions might query the table we're
* inserting to, and act differently if the tuples that have already been
* processed and prepared for insertion are not there. We also can't do
! * it if the table is foreign or partitioned.
*/
if ((resultRelInfo->ri_TrigDesc != NULL &&
(resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
resultRelInfo->ri_TrigDesc->trig_insert_instead_row)) ||
+ resultRelInfo->ri_FdwRoutine != NULL ||
cstate->partition_tuple_routing != NULL ||
cstate->volatile_defexprs)
{
***************
*** 2608,2626 **** CopyFrom(CopyState cstate)
resultRelInfo = proute->partitions[leaf_part_index];
if (resultRelInfo == NULL)
{
! resultRelInfo = ExecInitPartitionInfo(NULL,
saved_resultRelInfo,
proute, estate,
leaf_part_index);
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
*/
--- 2625,2637 ----
resultRelInfo = proute->partitions[leaf_part_index];
if (resultRelInfo == NULL)
{
! resultRelInfo = ExecInitPartitionInfo(mtstate,
saved_resultRelInfo,
proute, estate,
leaf_part_index);
Assert(resultRelInfo != NULL);
}
/*
* For ExecInsertIndexTuples() to work on the partition's indexes
*/
***************
*** 2708,2716 **** CopyFrom(CopyState cstate)
resultRelInfo->ri_TrigDesc->trig_insert_before_row))
check_partition_constr = false;
! /* Check the constraints of the tuple */
! if (resultRelInfo->ri_RelationDesc->rd_att->constr ||
! check_partition_constr)
ExecConstraints(resultRelInfo, slot, estate, true);
if (useHeapMultiInsert)
--- 2719,2731 ----
resultRelInfo->ri_TrigDesc->trig_insert_before_row))
check_partition_constr = false;
! /*
! * If the target is a plain table, check the constraints of
! * the tuple.
! */
! if (resultRelInfo->ri_FdwRoutine == NULL &&
! (resultRelInfo->ri_RelationDesc->rd_att->constr ||
! check_partition_constr))
ExecConstraints(resultRelInfo, slot, estate, true);
if (useHeapMultiInsert)
***************
*** 2742,2751 **** CopyFrom(CopyState cstate)
{
List *recheckIndexes = NIL;
! /* OK, store the tuple and create index entries for it */
! heap_insert(resultRelInfo->ri_RelationDesc, tuple, mycid,
! hi_options, bistate);
if (resultRelInfo->ri_NumIndices > 0)
recheckIndexes = ExecInsertIndexTuples(slot,
&(tuple->t_self),
--- 2757,2788 ----
{
List *recheckIndexes = NIL;
! /* OK, store the tuple */
! if (resultRelInfo->ri_FdwRoutine != NULL)
! {
! slot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
! resultRelInfo,
! slot,
! NULL);
!
! if (slot == NULL) /* "do nothing" */
! goto next_tuple;
!
! /* FDW might have changed tuple */
! tuple = ExecMaterializeSlot(slot);
+ /*
+ * AFTER ROW Triggers might reference the tableoid
+ * column, so initialize t_tableOid before evaluating
+ * them.
+ */
+ tuple->t_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
+ }
+ else
+ heap_insert(resultRelInfo->ri_RelationDesc, tuple,
+ mycid, hi_options, bistate);
+
+ /* And create index entries for it */
if (resultRelInfo->ri_NumIndices > 0)
recheckIndexes = ExecInsertIndexTuples(slot,
&(tuple->t_self),
***************
*** 2763,2775 **** CopyFrom(CopyState cstate)
}
/*
! * We count only tuples not suppressed by a BEFORE INSERT trigger;
! * this is the same definition used by execMain.c for counting
! * tuples inserted by an INSERT command.
*/
processed++;
}
/* Restore the saved ResultRelInfo */
if (saved_resultRelInfo)
{
--- 2800,2813 ----
}
/*
! * We count only tuples not suppressed by a BEFORE INSERT trigger
! * or FDW; this is the same definition used by nodeModifyTable.c
! * for counting tuples inserted by an INSERT command.
*/
processed++;
}
+ next_tuple:
/* Restore the saved ResultRelInfo */
if (saved_resultRelInfo)
{
***************
*** 2810,2820 **** CopyFrom(CopyState cstate)
ExecResetTupleTable(estate->es_tupleTable, false);
ExecCloseIndices(resultRelInfo);
/* 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);
--- 2848,2864 ----
ExecResetTupleTable(estate->es_tupleTable, false);
+ /* Allow the FDW to shut down */
+ if (resultRelInfo->ri_FdwRoutine != NULL &&
+ resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL)
+ resultRelInfo->ri_FdwRoutine->EndForeignInsert(estate,
+ resultRelInfo);
+
ExecCloseIndices(resultRelInfo);
/* Close all the partitioned tables, leaf partitions, and their indices */
if (cstate->partition_tuple_routing)
! ExecCleanupTupleRouting(mtstate, cstate->partition_tuple_routing);
/* Close any trigger target relations */
ExecCleanUpTriggerState(estate);
*** a/src/backend/executor/execMain.c
--- b/src/backend/executor/execMain.c
***************
*** 1179,1191 **** 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),
--- 1179,1184 ----
***************
*** 1378,1383 **** InitResultRelInfo(ResultRelInfo *resultRelInfo,
--- 1371,1377 ----
resultRelInfo->ri_PartitionCheck = partition_check;
resultRelInfo->ri_PartitionRoot = partition_root;
+ resultRelInfo->ri_PartitionReadyForRouting = false;
}
/*
*** a/src/backend/executor/execPartition.c
--- b/src/backend/executor/execPartition.c
***************
*** 18,23 ****
--- 18,24 ----
#include "catalog/pg_type.h"
#include "executor/execPartition.h"
#include "executor/executor.h"
+ #include "foreign/fdwapi.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
***************
*** 55,66 **** static List *adjust_partition_tlist(List *tlist, TupleConversionMap *map);
* see ExecInitPartitionInfo. However, if the function is invoked for update
* tuple routing, caller would already have initialized ResultRelInfo's for
* some of the partitions, which are reused and assigned to their respective
! * slot in the aforementioned array.
*/
PartitionTupleRouting *
ExecSetupPartitionTupleRouting(ModifyTableState *mtstate, Relation rel)
{
- TupleDesc tupDesc = RelationGetDescr(rel);
List *leaf_parts;
ListCell *cell;
int i;
--- 56,68 ----
* see ExecInitPartitionInfo. However, if the function is invoked for update
* tuple routing, caller would already have initialized ResultRelInfo's for
* some of the partitions, which are reused and assigned to their respective
! * slot in the aforementioned array. For such partitions, we delay setting
! * up objects such as TupleConversionMap until those are actually chosen as
! * the partitions to route tuples to. See ExecPrepareTupleRouting.
*/
PartitionTupleRouting *
ExecSetupPartitionTupleRouting(ModifyTableState *mtstate, Relation rel)
{
List *leaf_parts;
ListCell *cell;
int i;
***************
*** 141,151 **** 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
--- 143,149 ----
***************
*** 159,181 **** 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;
--- 157,162 ----
***************
*** 342,351 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
PartitionTupleRouting *proute,
EState *estate, int partidx)
{
Relation rootrel = resultRelInfo->ri_RelationDesc,
partrel;
ResultRelInfo *leaf_part_rri;
- ModifyTable *node = mtstate ? (ModifyTable *) mtstate->ps.plan : NULL;
MemoryContext oldContext;
/*
--- 323,332 ----
PartitionTupleRouting *proute,
EState *estate, int partidx)
{
+ ModifyTable *node = (ModifyTable *) mtstate->ps.plan;
Relation rootrel = resultRelInfo->ri_RelationDesc,
partrel;
ResultRelInfo *leaf_part_rri;
MemoryContext oldContext;
/*
***************
*** 369,379 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
leaf_part_rri->ri_PartitionLeafIndex = partidx;
! /*
! * 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);
/*
--- 350,356 ----
leaf_part_rri->ri_PartitionLeafIndex = partidx;
! /* Verify the specified partition is a valid target for INSERT */
CheckValidResultRel(leaf_part_rri, CMD_INSERT);
/*
***************
*** 389,394 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
--- 366,376 ----
leaf_part_rri);
/*
+ * Set up information needed for routing tuples to the specified partition
+ */
+ ExecInitRoutingInfo(mtstate, estate, proute, leaf_part_rri, partidx);
+
+ /*
* Open partition indices. The user may have asked to check for conflicts
* within this leaf partition and do "nothing" instead of throwing an
* error. Be prepared in that case by initializing the index information
***************
*** 493,498 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
--- 475,481 ----
returningList = map_partition_varattnos(returningList, firstVarno,
partrel, firstResultRel,
NULL);
+ leaf_part_rri->ri_returningList = returningList;
/*
* Initialize the projection itself.
***************
*** 510,524 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
}
/*
- * 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"));
-
- /*
* If there is an ON CONFLICT clause, initialize state for it.
*/
if (node && node->onConflictAction != ONCONFLICT_NONE)
--- 493,498 ----
***************
*** 654,659 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
--- 628,635 ----
}
}
+ leaf_part_rri->ri_PartitionReadyForRouting = true;
+
Assert(proute->partitions[partidx] == NULL);
proute->partitions[partidx] = leaf_part_rri;
***************
*** 747,752 **** ExecInitPartitionInfo(ModifyTableState *mtstate,
--- 723,766 ----
}
/*
+ * ExecInitRoutingInfo
+ * Set up information needed for routing tuples to a leaf partition
+ */
+ void
+ ExecInitRoutingInfo(ModifyTableState *mtstate,
+ EState *estate,
+ PartitionTupleRouting *proute,
+ ResultRelInfo *partRelInfo,
+ 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
+ * partition from the parent's type to the partition's.
+ */
+ proute->parent_child_tupconv_maps[partidx] =
+ convert_tuples_by_name(RelationGetDescr(partRelInfo->ri_PartitionRoot),
+ RelationGetDescr(partRelInfo->ri_RelationDesc),
+ gettext_noop("could not convert row type"));
+
+ /*
+ * If the partition is a foreign table, let the FDW init itself for
+ * routing tuples to the partition.
+ */
+ if (partRelInfo->ri_FdwRoutine != NULL &&
+ partRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL)
+ partRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, partRelInfo);
+
+ MemoryContextSwitchTo(oldContext);
+ }
+
+ /*
* ExecSetupChildParentMapForLeaf -- Initialize the per-leaf-partition
* child-to-root tuple conversion map array.
*
***************
*** 848,854 **** ConvertPartitionTupleSlot(TupleConversionMap *map,
* Close all the partitioned tables, leaf partitions, and their indices.
*/
void
! ExecCleanupTupleRouting(PartitionTupleRouting *proute)
{
int i;
int subplan_index = 0;
--- 862,869 ----
* Close all the partitioned tables, leaf partitions, and their indices.
*/
void
! ExecCleanupTupleRouting(ModifyTableState *mtstate,
! PartitionTupleRouting *proute)
{
int i;
int subplan_index = 0;
***************
*** 876,881 **** ExecCleanupTupleRouting(PartitionTupleRouting *proute)
--- 891,903 ----
if (resultRelInfo == NULL)
continue;
+ /* Allow any FDWs to shut down if they've been exercised */
+ if (resultRelInfo->ri_PartitionReadyForRouting &&
+ resultRelInfo->ri_FdwRoutine != NULL &&
+ resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL)
+ resultRelInfo->ri_FdwRoutine->EndForeignInsert(mtstate->ps.state,
+ resultRelInfo);
+
/*
* If this result rel is one of the UPDATE subplan result rels, let
* ExecEndPlan() close it. For INSERT or COPY,
*** a/src/backend/executor/nodeModifyTable.c
--- b/src/backend/executor/nodeModifyTable.c
***************
*** 1831,1841 **** ExecPrepareTupleRouting(ModifyTableState *mtstate,
proute, estate,
partidx);
! /* We do not yet have a way to insert into a foreign partition */
! if (partrel->ri_FdwRoutine)
! ereport(ERROR,
! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! errmsg("cannot route inserted tuples to a foreign table")));
/*
* Make it look like we are inserting into the partition.
--- 1831,1856 ----
proute, estate,
partidx);
! /*
! * Verify the partition is a valid target for INSERT if we didn't yet.
! *
! * Note: an UPDATE of a partition key invokes an INSERT that moves the
! * tuple to a new partition. This check would be applied to a subplan
! * partition of such an UPDATE that is chosen as the partition to move
! * the tuple to. The reason we do this check here rather than in
! * ExecSetupPartitionTupleRouting is to avoid aborting such an UPDATE
! * unnecessarily due to non-routable subplan partitions that may not be
! * chosen for update tuple movement after all.
! */
! if (!partrel->ri_PartitionReadyForRouting)
! {
! CheckValidResultRel(partrel, CMD_INSERT);
!
! /* Set up information needed for routing tuples to the partition */
! ExecInitRoutingInfo(mtstate, estate, proute, partrel, partidx);
!
! partrel->ri_PartitionReadyForRouting = true;
! }
/*
* Make it look like we are inserting into the partition.
***************
*** 2536,2541 **** ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
--- 2551,2557 ----
{
List *rlist = (List *) lfirst(l);
+ resultRelInfo->ri_returningList = rlist;
resultRelInfo->ri_projectReturning =
ExecBuildProjectionInfo(rlist, econtext, slot, &mtstate->ps,
resultRelInfo->ri_RelationDesc->rd_att);
***************
*** 2931,2937 **** 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
--- 2947,2953 ----
/* 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
***************
*** 119,124 **** extern ResultRelInfo *ExecInitPartitionInfo(ModifyTableState *mtstate,
--- 119,129 ----
ResultRelInfo *resultRelInfo,
PartitionTupleRouting *proute,
EState *estate, int partidx);
+ extern void ExecInitRoutingInfo(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);
***************
*** 126,131 **** extern HeapTuple ConvertPartitionTupleSlot(TupleConversionMap *map,
HeapTuple tuple,
TupleTableSlot *new_slot,
TupleTableSlot **p_my_slot);
! extern void ExecCleanupTupleRouting(PartitionTupleRouting *proute);
#endif /* EXECPARTITION_H */
--- 131,137 ----
HeapTuple tuple,
TupleTableSlot *new_slot,
TupleTableSlot **p_my_slot);
! extern void ExecCleanupTupleRouting(ModifyTableState *mtstate,
! PartitionTupleRouting *proute);
#endif /* EXECPARTITION_H */
*** a/src/include/foreign/fdwapi.h
--- b/src/include/foreign/fdwapi.h
***************
*** 98,103 **** typedef TupleTableSlot *(*ExecForeignDelete_function) (EState *estate,
--- 98,109 ----
typedef void (*EndForeignModify_function) (EState *estate,
ResultRelInfo *rinfo);
+ typedef void (*BeginForeignInsert_function) (ModifyTableState *mtstate,
+ ResultRelInfo *rinfo);
+
+ typedef void (*EndForeignInsert_function) (EState *estate,
+ ResultRelInfo *rinfo);
+
typedef int (*IsForeignRelUpdatable_function) (Relation rel);
typedef bool (*PlanDirectModify_function) (PlannerInfo *root,
***************
*** 205,210 **** typedef struct FdwRoutine
--- 211,218 ----
ExecForeignUpdate_function ExecForeignUpdate;
ExecForeignDelete_function ExecForeignDelete;
EndForeignModify_function EndForeignModify;
+ BeginForeignInsert_function BeginForeignInsert;
+ EndForeignInsert_function EndForeignInsert;
IsForeignRelUpdatable_function IsForeignRelUpdatable;
PlanDirectModify_function PlanDirectModify;
BeginDirectModify_function BeginDirectModify;
*** a/src/include/nodes/execnodes.h
--- b/src/include/nodes/execnodes.h
***************
*** 444,449 **** typedef struct ResultRelInfo
--- 444,452 ----
/* for removing junk attributes from tuples */
JunkFilter *ri_junkFilter;
+ /* list of RETURNING expressions */
+ List *ri_returningList;
+
/* for computing a RETURNING list */
ProjectionInfo *ri_projectReturning;
***************
*** 462,467 **** typedef struct ResultRelInfo
--- 465,473 ----
/* relation descriptor for root partitioned table */
Relation ri_PartitionRoot;
+ /* true if ready for tuple routing */
+ bool ri_PartitionReadyForRouting;
+
int ri_PartitionLeafIndex;
/* for running MERGE on this result relation */
MergeState *ri_mergeState;
