*** a/contrib/file_fdw/output/file_fdw.source --- b/contrib/file_fdw/output/file_fdw.source *************** *** 315,321 **** SELECT tableoid::regclass, * FROM p2; (0 rows) COPY pt FROM '@abs_srcdir@/data/list2.bad' with (format 'csv', delimiter ','); -- ERROR ! ERROR: cannot route inserted tuples to a foreign table CONTEXT: COPY pt, line 2: "1,qux" COPY pt FROM '@abs_srcdir@/data/list2.csv' with (format 'csv', delimiter ','); SELECT tableoid::regclass, * FROM pt; --- 315,321 ---- (0 rows) COPY pt FROM '@abs_srcdir@/data/list2.bad' with (format 'csv', delimiter ','); -- ERROR ! ERROR: cannot route copied tuples to a foreign table CONTEXT: COPY pt, line 2: "1,qux" COPY pt FROM '@abs_srcdir@/data/list2.csv' with (format 'csv', delimiter ','); SELECT tableoid::regclass, * FROM pt; *************** *** 342,351 **** SELECT tableoid::regclass, * FROM p2; (2 rows) INSERT INTO pt VALUES (1, 'xyzzy'); -- ERROR ! ERROR: cannot route inserted tuples to a foreign table INSERT INTO pt VALUES (2, 'xyzzy'); UPDATE pt set a = 1 where a = 2; -- ERROR ! ERROR: cannot route inserted tuples to a foreign table SELECT tableoid::regclass, * FROM pt; tableoid | a | b ----------+---+------- --- 342,351 ---- (2 rows) INSERT INTO pt VALUES (1, 'xyzzy'); -- ERROR ! ERROR: cannot route inserted tuples to foreign table "p1" INSERT INTO pt VALUES (2, 'xyzzy'); UPDATE pt set a = 1 where a = 2; -- ERROR ! ERROR: cannot route inserted tuples to foreign table "p1" SELECT tableoid::regclass, * FROM pt; tableoid | a | b ----------+---+------- *** a/contrib/postgres_fdw/expected/postgres_fdw.out --- b/contrib/postgres_fdw/expected/postgres_fdw.out *************** *** 7364,7369 **** NOTICE: drop cascades to foreign table bar2 --- 7364,7448 ---- drop table loct1; drop table loct2; -- =================================================================== + -- test tuple routing for foreign-table partitions + -- =================================================================== + create table pt (a int, b int, c text) partition by list (a); + create table loct1 (a int check (a in (1)), b int, c text, constraint locp1_pkey primary key (b)); + create table loct2 (b int, c text, a int check (a in (2)), constraint locp2_pkey primary key (b)); + create foreign table ptp1 partition of pt for values in (1) server loopback options (table_name 'loct1'); + create foreign table ptp2 partition of pt for values in (2) server loopback options (table_name 'loct2'); + insert into pt values (1, 1, 'foo'); + insert into pt values (1, 2, 'bar') returning *; + a | b | c + ---+---+----- + 1 | 2 | bar + (1 row) + + insert into pt values (2, 1, 'baz') returning *; + a | b | c + ---+---+----- + 2 | 1 | baz + (1 row) + + select tableoid::regclass, * FROM pt; + tableoid | a | b | c + ----------+---+---+----- + ptp1 | 1 | 1 | foo + ptp1 | 1 | 2 | bar + ptp2 | 2 | 1 | baz + (3 rows) + + select tableoid::regclass, * FROM ptp1; + tableoid | a | b | c + ----------+---+---+----- + ptp1 | 1 | 1 | foo + ptp1 | 1 | 2 | bar + (2 rows) + + select tableoid::regclass, * FROM ptp2; + tableoid | a | b | c + ----------+---+---+----- + ptp2 | 2 | 1 | baz + (1 row) + + insert into pt values (2, 1, 'baz'); + ERROR: duplicate key value violates unique constraint "locp2_pkey" + DETAIL: Key (b)=(1) already exists. + CONTEXT: Remote SQL command: INSERT INTO public.loct2(a, b, c) VALUES ($1, $2, $3) + insert into pt values (2, 1, 'baz') on conflict do nothing; + insert into pt values (2, 2, 'qux') on conflict do nothing returning *; + a | b | c + ---+---+----- + 2 | 2 | qux + (1 row) + + select tableoid::regclass, * FROM pt; + tableoid | a | b | c + ----------+---+---+----- + ptp1 | 1 | 1 | foo + ptp1 | 1 | 2 | bar + ptp2 | 2 | 1 | baz + ptp2 | 2 | 2 | qux + (4 rows) + + select tableoid::regclass, * FROM ptp1; + tableoid | a | b | c + ----------+---+---+----- + ptp1 | 1 | 1 | foo + ptp1 | 1 | 2 | bar + (2 rows) + + select tableoid::regclass, * FROM ptp2; + tableoid | a | b | c + ----------+---+---+----- + ptp2 | 2 | 1 | baz + ptp2 | 2 | 2 | qux + (2 rows) + + drop table pt; + drop table loct1; + drop table loct2; + -- =================================================================== -- test IMPORT FOREIGN SCHEMA -- =================================================================== CREATE SCHEMA import_source; *** a/contrib/postgres_fdw/postgres_fdw.c --- b/contrib/postgres_fdw/postgres_fdw.c *************** *** 319,324 **** static TupleTableSlot *postgresExecForeignDelete(EState *estate, --- 319,332 ---- TupleTableSlot *planSlot); static void postgresEndForeignModify(EState *estate, ResultRelInfo *resultRelInfo); + static void postgresBeginForeignRouting(ModifyTableState *mtstate, + ResultRelInfo *resultRelInfo, + int partition_index); + static TupleTableSlot *postgresExecForeignRouting(EState *estate, + ResultRelInfo *resultRelInfo, + TupleTableSlot *slot); + static void postgresEndForeignRouting(EState *estate, + ResultRelInfo *resultRelInfo); static int postgresIsForeignRelUpdatable(Relation rel); static bool postgresPlanDirectModify(PlannerInfo *root, ModifyTable *plan, *************** *** 473,478 **** postgres_fdw_handler(PG_FUNCTION_ARGS) --- 481,489 ---- routine->ExecForeignUpdate = postgresExecForeignUpdate; routine->ExecForeignDelete = postgresExecForeignDelete; routine->EndForeignModify = postgresEndForeignModify; + routine->BeginForeignRouting = postgresBeginForeignRouting; + routine->ExecForeignRouting = postgresExecForeignRouting; + routine->EndForeignRouting = postgresEndForeignRouting; routine->IsForeignRelUpdatable = postgresIsForeignRelUpdatable; routine->PlanDirectModify = postgresPlanDirectModify; routine->BeginDirectModify = postgresBeginDirectModify; *************** *** 1857,1862 **** postgresEndForeignModify(EState *estate, --- 1868,1986 ---- } /* + * postgresBeginForeignRouting + * Begin a row-routing operation on a foreign table + */ + static void + postgresBeginForeignRouting(ModifyTableState *mtstate, + ResultRelInfo *resultRelInfo, + int partition_index) + { + PgFdwModifyState *fmstate; + Relation rel = resultRelInfo->ri_RelationDesc; + TupleDesc tupdesc = RelationGetDescr(rel); + int attnum; + RangeTblEntry *rte; + Query *query; + PlannerInfo *root; + StringInfoData sql; + List *targetAttrs = NIL; + List *retrieved_attrs = NIL; + bool doNothing = false; + + initStringInfo(&sql); + + /* Set up largely-dummy planner state */ + rte = makeNode(RangeTblEntry); + rte->rtekind = RTE_RELATION; + rte->relid = RelationGetRelid(rel); + rte->relkind = RELKIND_FOREIGN_TABLE; + query = makeNode(Query); + query->commandType = CMD_INSERT; + query->resultRelation = 1; + query->rtable = list_make1(rte); + root = makeNode(PlannerInfo); + root->parse = query; + + /* We transmit all columns that are defined in the foreign table. */ + for (attnum = 1; attnum <= tupdesc->natts; attnum++) + { + Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1); + + if (!attr->attisdropped) + targetAttrs = lappend_int(targetAttrs, attnum); + } + + /* We only support DO NOTHING without an inference specification. */ + if (mtstate->mt_onconflict == ONCONFLICT_NOTHING) + doNothing = true; + else if (mtstate->mt_onconflict != ONCONFLICT_NONE) + elog(ERROR, "unexpected ON CONFLICT specification: %d", + (int) mtstate->mt_onconflict); + + /* Construct the SQL command string. */ + deparseInsertSql(&sql, root, 1, rel, targetAttrs, doNothing, + resultRelInfo->ri_returningList, &retrieved_attrs); + + /* Construct an execution state. */ + fmstate = create_fdw_modify_state(mtstate, + resultRelInfo, + CMD_INSERT, + 0, /* dummy subplan index */ + sql.data, + targetAttrs, + retrieved_attrs != NIL, + retrieved_attrs); + + resultRelInfo->ri_FdwState = fmstate; + } + + /* + * postgresExecForeignRouting + * Route one row to a foreign table + */ + static TupleTableSlot * + postgresExecForeignRouting(EState *estate, + ResultRelInfo *resultRelInfo, + TupleTableSlot *slot) + { + PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState; + const char **p_values; + int n_rows; + + /* Set up the prepared statement on the remote server, if we didn't yet */ + if (!fmstate->p_name) + prepare_foreign_modify(fmstate); + + /* Convert parameters needed by prepared statement to text form */ + p_values = convert_prep_stmt_params(fmstate, NULL, slot); + + /* Execute the prepared statement and fetch RETURNING tuple if any */ + n_rows = execute_prep_stmt(fmstate, p_values, slot); + + MemoryContextReset(fmstate->temp_cxt); + + /* Return NULL if nothing was inserted on the remote end */ + return (n_rows > 0) ? slot : NULL; + } + + /* + * postgresEndForeignRouting + * Finish a row-routing operation on a foreign table + */ + static void + postgresEndForeignRouting(EState *estate, + ResultRelInfo *resultRelInfo) + { + PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState; + + Assert(fmstate != NULL); + + /* Destroy the execution state. */ + finish_foreign_modify(fmstate); + } + + /* * postgresIsForeignRelUpdatable * Determine whether a foreign table supports INSERT, UPDATE and/or * DELETE. *** a/contrib/postgres_fdw/sql/postgres_fdw.sql --- b/contrib/postgres_fdw/sql/postgres_fdw.sql *************** *** 1759,1764 **** drop table loct1; --- 1759,1794 ---- drop table loct2; -- =================================================================== + -- test tuple routing for foreign-table partitions + -- =================================================================== + + create table pt (a int, b int, c text) partition by list (a); + create table loct1 (a int check (a in (1)), b int, c text, constraint locp1_pkey primary key (b)); + create table loct2 (b int, c text, a int check (a in (2)), constraint locp2_pkey primary key (b)); + create foreign table ptp1 partition of pt for values in (1) server loopback options (table_name 'loct1'); + create foreign table ptp2 partition of pt for values in (2) server loopback options (table_name 'loct2'); + + insert into pt values (1, 1, 'foo'); + insert into pt values (1, 2, 'bar') returning *; + insert into pt values (2, 1, 'baz') returning *; + + select tableoid::regclass, * FROM pt; + select tableoid::regclass, * FROM ptp1; + select tableoid::regclass, * FROM ptp2; + + insert into pt values (2, 1, 'baz'); + insert into pt values (2, 1, 'baz') on conflict do nothing; + insert into pt values (2, 2, 'qux') on conflict do nothing returning *; + + select tableoid::regclass, * FROM pt; + select tableoid::regclass, * FROM ptp1; + select tableoid::regclass, * FROM ptp2; + + drop table pt; + drop table loct1; + drop table loct2; + + -- =================================================================== -- test IMPORT FOREIGN SCHEMA -- =================================================================== *** a/src/backend/commands/copy.c --- b/src/backend/commands/copy.c *************** *** 2616,2626 **** CopyFrom(CopyState cstate) Assert(resultRelInfo != NULL); } ! /* We do not yet have a way to insert into a foreign partition */ if (resultRelInfo->ri_FdwRoutine) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), ! errmsg("cannot route inserted tuples to a foreign table"))); /* * For ExecInsertIndexTuples() to work on the partition's indexes --- 2616,2626 ---- Assert(resultRelInfo != NULL); } ! /* We do not yet have a way to copy into a foreign partition */ if (resultRelInfo->ri_FdwRoutine) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), ! errmsg("cannot route copied tuples to a foreign table"))); /* * For ExecInsertIndexTuples() to work on the partition's indexes *************** *** 2814,2820 **** CopyFrom(CopyState cstate) /* Close all the partitioned tables, leaf partitions, and their indices */ if (cstate->partition_tuple_routing) ! ExecCleanupTupleRouting(cstate->partition_tuple_routing); /* Close any trigger target relations */ ExecCleanUpTriggerState(estate); --- 2814,2820 ---- /* Close all the partitioned tables, leaf partitions, and their indices */ if (cstate->partition_tuple_routing) ! ExecCleanupTupleRouting(NULL, cstate->partition_tuple_routing); /* Close any trigger target relations */ ExecCleanUpTriggerState(estate); *** a/src/backend/executor/execMain.c --- b/src/backend/executor/execMain.c *************** *** 1172,1189 **** CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation) switch (operation) { case CMD_INSERT: - - /* - * If foreign partition to do tuple-routing for, skip the - * check; it's disallowed elsewhere. - */ if (resultRelInfo->ri_PartitionRoot) ! break; ! if (fdwroutine->ExecForeignInsert == NULL) ! ereport(ERROR, ! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), ! errmsg("cannot insert into foreign table \"%s\"", ! RelationGetRelationName(resultRel)))); if (fdwroutine->IsForeignRelUpdatable != NULL && (fdwroutine->IsForeignRelUpdatable(resultRel) & (1 << CMD_INSERT)) == 0) ereport(ERROR, --- 1172,1193 ---- switch (operation) { case CMD_INSERT: if (resultRelInfo->ri_PartitionRoot) ! { ! if (fdwroutine->ExecForeignRouting == NULL) ! ereport(ERROR, ! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), ! errmsg("cannot route inserted tuples to foreign table \"%s\"", ! RelationGetRelationName(resultRel)))); ! } ! else ! { ! if (fdwroutine->ExecForeignInsert == NULL) ! ereport(ERROR, ! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), ! errmsg("cannot insert into foreign table \"%s\"", ! RelationGetRelationName(resultRel)))); ! } if (fdwroutine->IsForeignRelUpdatable != NULL && (fdwroutine->IsForeignRelUpdatable(resultRel) & (1 << CMD_INSERT)) == 0) ereport(ERROR, *************** *** 1364,1369 **** InitResultRelInfo(ResultRelInfo *resultRelInfo, --- 1368,1374 ---- resultRelInfo->ri_PartitionCheck = partition_check; resultRelInfo->ri_PartitionRoot = partition_root; + resultRelInfo->ri_PartitionIsValid = false; } /* *** a/src/backend/executor/execPartition.c --- b/src/backend/executor/execPartition.c *************** *** 17,22 **** --- 17,23 ---- #include "catalog/pg_inherits_fn.h" #include "executor/execPartition.h" #include "executor/executor.h" + #include "foreign/fdwapi.h" #include "mb/pg_wchar.h" #include "miscadmin.h" #include "utils/lsyscache.h" *************** *** 162,173 **** ExecSetupPartitionTupleRouting(ModifyTableState *mtstate, Relation rel) 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); } } --- 163,175 ---- gettext_noop("could not convert row type")); /* ! * Also let the FDW init itself if this parition is foreign. */ ! if (leaf_part_rri->ri_FdwRoutine != NULL && ! leaf_part_rri->ri_FdwRoutine->BeginForeignRouting != NULL) ! leaf_part_rri->ri_FdwRoutine->BeginForeignRouting(mtstate, ! leaf_part_rri, ! i); } } *************** *** 342,354 **** ExecInitPartitionInfo(ModifyTableState *mtstate, estate->es_instrument); /* - * Verify result relation is a valid target for an INSERT. An UPDATE - * of a partition-key becomes a DELETE+INSERT operation, so this check - * is still required when the operation is CMD_UPDATE. - */ - CheckValidResultRel(leaf_part_rri, CMD_INSERT); - - /* * Since we've just initialized this ResultRelInfo, it's not in * any list attached to the estate as yet. Add it, so that it can * be found later. --- 344,349 ---- *************** *** 468,473 **** ExecInitPartitionInfo(ModifyTableState *mtstate, --- 463,469 ---- returningList = map_partition_varattnos(returningList, firstVarno, partrel, firstResultRel, NULL); + leaf_part_rri->ri_returningList = returningList; /* * Initialize the projection itself. *************** *** 498,503 **** ExecInitPartitionInfo(ModifyTableState *mtstate, --- 494,509 ---- MemoryContextSwitchTo(oldContext); + /* + * Also let the FDW init itself if this parition is foreign. + */ + if (mtstate && + leaf_part_rri->ri_FdwRoutine != NULL && + leaf_part_rri->ri_FdwRoutine->BeginForeignRouting != NULL) + leaf_part_rri->ri_FdwRoutine->BeginForeignRouting(mtstate, + leaf_part_rri, + partidx); + return leaf_part_rri; } *************** *** 603,609 **** ConvertPartitionTupleSlot(TupleConversionMap *map, * Close all the partitioned tables, leaf partitions, and their indices. */ void ! ExecCleanupTupleRouting(PartitionTupleRouting *proute) { int i; int subplan_index = 0; --- 609,616 ---- * Close all the partitioned tables, leaf partitions, and their indices. */ void ! ExecCleanupTupleRouting(ModifyTableState *node, ! PartitionTupleRouting *proute) { int i; int subplan_index = 0; *************** *** 632,637 **** ExecCleanupTupleRouting(PartitionTupleRouting *proute) --- 639,653 ---- continue; /* + * Allow any FDWs to shut down + */ + if (node && + resultRelInfo->ri_FdwRoutine != NULL && + resultRelInfo->ri_FdwRoutine->EndForeignRouting != NULL) + resultRelInfo->ri_FdwRoutine->EndForeignRouting(node->ps.state, + resultRelInfo); + + /* * If this result rel is one of the UPDATE subplan result rels, let * ExecEndPlan() close it. For INSERT or COPY, * proute->subplan_partition_offsets will always be NULL. Note that *** a/src/backend/executor/nodeModifyTable.c --- b/src/backend/executor/nodeModifyTable.c *************** *** 319,329 **** ExecInsert(ModifyTableState *mtstate, Assert(resultRelInfo != NULL); } ! /* We do not yet have a way to insert into a foreign partition */ ! if (resultRelInfo->ri_FdwRoutine) ! ereport(ERROR, ! (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), ! errmsg("cannot route inserted tuples to a foreign table"))); /* For ExecInsertIndexTuples() to work on the partition's indexes */ estate->es_result_relation_info = resultRelInfo; --- 319,333 ---- Assert(resultRelInfo != NULL); } ! /* ! * Verify the specified partition is a valid target for tuple routing ! * if not already done. ! */ ! if (!resultRelInfo->ri_PartitionIsValid) ! { ! CheckValidResultRel(resultRelInfo, CMD_INSERT); ! resultRelInfo->ri_PartitionIsValid = true; ! } /* For ExecInsertIndexTuples() to work on the partition's indexes */ estate->es_result_relation_info = resultRelInfo; *************** *** 433,442 **** ExecInsert(ModifyTableState *mtstate, /* * insert into foreign table: let the FDW do it */ ! slot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate, ! resultRelInfo, ! slot, ! planSlot); if (slot == NULL) /* "do nothing" */ return NULL; --- 437,451 ---- /* * insert into foreign table: let the FDW do it */ ! if (resultRelInfo->ri_PartitionRoot) ! slot = resultRelInfo->ri_FdwRoutine->ExecForeignRouting(estate, ! resultRelInfo, ! slot); ! else ! slot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate, ! resultRelInfo, ! slot, ! planSlot); if (slot == NULL) /* "do nothing" */ return NULL; *************** *** 2311,2316 **** ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) --- 2320,2326 ---- { List *rlist = (List *) lfirst(l); + resultRelInfo->ri_returningList = rlist; resultRelInfo->ri_projectReturning = ExecBuildProjectionInfo(rlist, econtext, slot, &mtstate->ps, resultRelInfo->ri_RelationDesc->rd_att); *************** *** 2569,2575 **** 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 --- 2579,2585 ---- /* 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 *************** *** 121,126 **** extern HeapTuple ConvertPartitionTupleSlot(TupleConversionMap *map, HeapTuple tuple, TupleTableSlot *new_slot, TupleTableSlot **p_my_slot); ! extern void ExecCleanupTupleRouting(PartitionTupleRouting *proute); #endif /* EXECPARTITION_H */ --- 121,127 ---- HeapTuple tuple, TupleTableSlot *new_slot, TupleTableSlot **p_my_slot); ! extern void ExecCleanupTupleRouting(ModifyTableState *node, ! PartitionTupleRouting *proute); #endif /* EXECPARTITION_H */ *** a/src/include/foreign/fdwapi.h --- b/src/include/foreign/fdwapi.h *************** *** 97,102 **** typedef TupleTableSlot *(*ExecForeignDelete_function) (EState *estate, --- 97,113 ---- typedef void (*EndForeignModify_function) (EState *estate, ResultRelInfo *rinfo); + typedef void (*BeginForeignRouting_function) (ModifyTableState *mtstate, + ResultRelInfo *rinfo, + int partition_index); + + typedef TupleTableSlot *(*ExecForeignRouting_function) (EState *estate, + ResultRelInfo *rinfo, + TupleTableSlot *slot); + + typedef void (*EndForeignRouting_function) (EState *estate, + ResultRelInfo *rinfo); + typedef int (*IsForeignRelUpdatable_function) (Relation rel); typedef bool (*PlanDirectModify_function) (PlannerInfo *root, *************** *** 204,209 **** typedef struct FdwRoutine --- 215,223 ---- ExecForeignUpdate_function ExecForeignUpdate; ExecForeignDelete_function ExecForeignDelete; EndForeignModify_function EndForeignModify; + BeginForeignRouting_function BeginForeignRouting; + ExecForeignRouting_function ExecForeignRouting; + EndForeignRouting_function EndForeignRouting; IsForeignRelUpdatable_function IsForeignRelUpdatable; PlanDirectModify_function PlanDirectModify; BeginDirectModify_function BeginDirectModify; *** a/src/include/nodes/execnodes.h --- b/src/include/nodes/execnodes.h *************** *** 403,408 **** typedef struct ResultRelInfo --- 403,411 ---- /* list of WithCheckOption expr states */ List *ri_WithCheckOptionExprs; + /* list of RETURNING expressions */ + List *ri_returningList; + /* array of constraint-checking expr states */ ExprState **ri_ConstraintExprs; *************** *** 426,431 **** typedef struct ResultRelInfo --- 429,437 ---- /* relation descriptor for root partitioned table */ Relation ri_PartitionRoot; + + /* true if valid target for tuple routing */ + bool ri_PartitionIsValid; } ResultRelInfo; /* ----------------