(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;