*** a/contrib/postgres_fdw/deparse.c --- b/contrib/postgres_fdw/deparse.c *************** *** 188,197 **** is_foreign_expr(PlannerInfo *root, if (!foreign_expr_walker((Node *) expr, &glob_cxt, &loc_cxt)) return false; - /* Expressions examined here should be boolean, ie noncollatable */ - Assert(loc_cxt.collation == InvalidOid); - Assert(loc_cxt.state == FDW_COLLATE_NONE); - /* * An expression which includes any mutable functions can't be sent over * because its result is not stable. For example, sending now() remote --- 188,193 ---- *************** *** 927,932 **** deparseUpdateSql(StringInfo buf, PlannerInfo *root, --- 923,981 ---- } /* + * deparse remote UPDATE statement + * + * The statement text is appended to buf, and we also create an integer List + * of the columns being retrieved by RETURNING (if any), which is returned + * to *retrieved_attrs. + */ + void + deparseDirectUpdateSql(StringInfo buf, PlannerInfo *root, + Index rtindex, Relation rel, + List *remote_conds, + List *targetlist, + List *targetAttrs, List *returningList, + List **retrieved_attrs) + { + RelOptInfo *baserel = root->simple_rel_array[rtindex]; + List *params_list = NIL; + deparse_expr_cxt context; + bool first; + ListCell *lc; + + /* Set up context struct for recursion */ + context.root = root; + context.foreignrel = baserel; + context.buf = buf; + context.params_list = NULL; + + appendStringInfoString(buf, "UPDATE "); + deparseRelation(buf, rel); + appendStringInfoString(buf, " SET "); + + first = true; + foreach(lc, targetAttrs) + { + int attnum = lfirst_int(lc); + TargetEntry *tle = get_tle_by_resno(targetlist, attnum); + + if (!first) + appendStringInfoString(buf, ", "); + first = false; + + deparseColumnRef(buf, rtindex, attnum, root); + appendStringInfo(buf, " = "); + deparseExpr((Expr *) tle->expr, &context); + } + if (remote_conds) + appendWhereClause(buf, root, baserel, remote_conds, + true, ¶ms_list); + + deparseReturningList(buf, root, rtindex, rel, false, + returningList, retrieved_attrs); + } + + /* * deparse remote DELETE statement * * The statement text is appended to buf, and we also create an integer List *************** *** 949,954 **** deparseDeleteSql(StringInfo buf, PlannerInfo *root, --- 998,1030 ---- } /* + * deparse remote DELETE statement + * + * The statement text is appended to buf, and we also create an integer List + * of the columns being retrieved by RETURNING (if any), which is returned + * to *retrieved_attrs. + */ + void + deparseDirectDeleteSql(StringInfo buf, PlannerInfo *root, + Index rtindex, Relation rel, + List *remote_conds, + List *returningList, + List **retrieved_attrs) + { + RelOptInfo *baserel = root->simple_rel_array[rtindex]; + List *params_list = NIL; + + appendStringInfoString(buf, "DELETE FROM "); + deparseRelation(buf, rel); + if (remote_conds) + appendWhereClause(buf, root, baserel, remote_conds, + true, ¶ms_list); + + deparseReturningList(buf, root, rtindex, rel, false, + returningList, retrieved_attrs); + } + + /* * Add a RETURNING clause, if needed, to an INSERT/UPDATE/DELETE. */ static void *** a/contrib/postgres_fdw/expected/postgres_fdw.out --- b/contrib/postgres_fdw/expected/postgres_fdw.out *************** *** 992,998 **** INSERT INTO ft2 (c1,c2,c3) --- 992,1019 ---- (3 rows) INSERT INTO ft2 (c1,c2,c3) VALUES (1104,204,'ddd'), (1105,205,'eee'); + EXPLAIN (verbose, costs off) + UPDATE ft2 SET c2 = c2 + 300, c3 = c3 || '_update3' WHERE c1 % 10 = 3; + QUERY PLAN + ---------------------------------------------------------------------------------------------------------------------- + Update on public.ft2 + -> Foreign Scan on public.ft2 + Output: c1, c2, NULL::integer, c3, c4, c5, c6, c7, c8, ctid + Remote SQL: UPDATE "S 1"."T 1" SET c2 = (c2 + 300), c3 = (c3 || '_update3'::text) WHERE ((("C 1" % 10) = 3)) + (4 rows) + UPDATE ft2 SET c2 = c2 + 300, c3 = c3 || '_update3' WHERE c1 % 10 = 3; + EXPLAIN (verbose, costs off) + UPDATE ft2 SET c2 = c2 + 400, c3 = c3 || '_update7' WHERE c1 % 10 = 7 RETURNING *; + QUERY PLAN + ------------------------------------------------------------------------------------------------------------------------------------------------------------------ + Update on public.ft2 + Output: c1, c2, c3, c4, c5, c6, c7, c8 + -> Foreign Scan on public.ft2 + Output: c1, c2, NULL::integer, c3, c4, c5, c6, c7, c8, ctid + Remote SQL: UPDATE "S 1"."T 1" SET c2 = (c2 + 400), c3 = (c3 || '_update7'::text) WHERE ((("C 1" % 10) = 7)) RETURNING "C 1", c2, c3, c4, c5, c6, c7, c8 + (5 rows) + UPDATE ft2 SET c2 = c2 + 400, c3 = c3 || '_update7' WHERE c1 % 10 = 7 RETURNING *; c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 ------+-----+--------------------+------------------------------+--------------------------+----+------------+----- *************** *** 1102,1108 **** UPDATE ft2 SET c2 = c2 + 400, c3 = c3 || '_update7' WHERE c1 % 10 = 7 RETURNING EXPLAIN (verbose, costs off) UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT ! FROM ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 9; QUERY PLAN ------------------------------------------------------------------------------------------------------------------------------------------------------------------- Update on public.ft2 --- 1123,1129 ---- EXPLAIN (verbose, costs off) UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT ! FROM ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 9; -- can't be pushed down QUERY PLAN ------------------------------------------------------------------------------------------------------------------------------------------------------------------- Update on public.ft2 *************** *** 1123,1138 **** UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT FROM ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 9; EXPLAIN (verbose, costs off) DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; ! QUERY PLAN ! ---------------------------------------------------------------------------------------- Delete on public.ft2 Output: c1, c4 - Remote SQL: DELETE FROM "S 1"."T 1" WHERE ctid = $1 RETURNING "C 1", c4 -> Foreign Scan on public.ft2 Output: ctid ! Remote SQL: SELECT ctid FROM "S 1"."T 1" WHERE ((("C 1" % 10) = 5)) FOR UPDATE ! (6 rows) DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; c1 | c4 --- 1144,1171 ---- UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT FROM ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 9; EXPLAIN (verbose, costs off) + UPDATE ft2 SET c7 = DEFAULT WHERE c1 % 10 = 5 AND date(c4) = '1970-01-17'::date; -- can't be pushed down + QUERY PLAN + ----------------------------------------------------------------------------------------------------------------------- + Update on public.ft2 + Remote SQL: UPDATE "S 1"."T 1" SET c7 = $2 WHERE ctid = $1 + -> Foreign Scan on public.ft2 + Output: c1, c2, NULL::integer, c3, c4, c5, c6, 'ft2 '::character(10), c8, ctid + Filter: (date(ft2.c4) = '01-17-1970'::date) + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c8, ctid FROM "S 1"."T 1" WHERE ((("C 1" % 10) = 5)) FOR UPDATE + (6 rows) + + UPDATE ft2 SET c7 = DEFAULT WHERE c1 % 10 = 5 AND date(c4) = '1970-01-17'::date; + EXPLAIN (verbose, costs off) DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; ! QUERY PLAN ! -------------------------------------------------------------------------------------------- Delete on public.ft2 Output: c1, c4 -> Foreign Scan on public.ft2 Output: ctid ! Remote SQL: DELETE FROM "S 1"."T 1" WHERE ((("C 1" % 10) = 5)) RETURNING "C 1", c4 ! (5 rows) DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; c1 | c4 *************** *** 1243,1249 **** DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; (103 rows) EXPLAIN (verbose, costs off) ! DELETE FROM ft2 USING ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 2; QUERY PLAN ---------------------------------------------------------------------------------------------------------------------- Delete on public.ft2 --- 1276,1282 ---- (103 rows) EXPLAIN (verbose, costs off) ! DELETE FROM ft2 USING ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 2; -- can't be pushed down QUERY PLAN ---------------------------------------------------------------------------------------------------------------------- Delete on public.ft2 *************** *** 2227,2233 **** CONTEXT: Remote SQL command: INSERT INTO "S 1"."T 1"("C 1", c2, c3, c4, c5, c6, UPDATE ft1 SET c2 = -c2 WHERE c1 = 1; -- c2positive ERROR: new row for relation "T 1" violates check constraint "c2positive" DETAIL: Failing row contains (1, -1, 00001_trig_update, 1970-01-02 08:00:00+00, 1970-01-02 00:00:00, 1, 1 , foo). ! CONTEXT: Remote SQL command: UPDATE "S 1"."T 1" SET c2 = $2 WHERE ctid = $1 -- Test savepoint/rollback behavior select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; c2 | count --- 2260,2266 ---- UPDATE ft1 SET c2 = -c2 WHERE c1 = 1; -- c2positive ERROR: new row for relation "T 1" violates check constraint "c2positive" DETAIL: Failing row contains (1, -1, 00001_trig_update, 1970-01-02 08:00:00+00, 1970-01-02 00:00:00, 1, 1 , foo). ! CONTEXT: Remote SQL command: UPDATE "S 1"."T 1" SET c2 = (- c2) WHERE (("C 1" = 1)) -- Test savepoint/rollback behavior select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; c2 | count *************** *** 2386,2392 **** savepoint s3; update ft2 set c2 = -2 where c2 = 42 and c1 = 10; -- fail on remote side ERROR: new row for relation "T 1" violates check constraint "c2positive" DETAIL: Failing row contains (10, -2, 00010_trig_update_trig_update, 1970-01-11 08:00:00+00, 1970-01-11 00:00:00, 0, 0 , foo). ! CONTEXT: Remote SQL command: UPDATE "S 1"."T 1" SET c2 = $2 WHERE ctid = $1 rollback to savepoint s3; select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; c2 | count --- 2419,2425 ---- update ft2 set c2 = -2 where c2 = 42 and c1 = 10; -- fail on remote side ERROR: new row for relation "T 1" violates check constraint "c2positive" DETAIL: Failing row contains (10, -2, 00010_trig_update_trig_update, 1970-01-11 08:00:00+00, 1970-01-11 00:00:00, 0, 0 , foo). ! CONTEXT: Remote SQL command: UPDATE "S 1"."T 1" SET c2 = (-2) WHERE ((c2 = 42)) AND (("C 1" = 10)) rollback to savepoint s3; select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; c2 | count *** a/contrib/postgres_fdw/postgres_fdw.c --- b/contrib/postgres_fdw/postgres_fdw.c *************** *** 86,92 **** typedef struct PgFdwRelationInfo * planner to executor. Currently we store: * * 1) SELECT statement text to be sent to the remote server ! * 2) Integer list of attribute numbers retrieved by the SELECT * * These items are indexed with the enum FdwScanPrivateIndex, so an item * can be fetched with list_nth(). For example, to get the SELECT statement: --- 86,97 ---- * planner to executor. Currently we store: * * 1) SELECT statement text to be sent to the remote server ! * 2) List of restriction clauses that can be executed remotely ! * 3) Integer list of attribute numbers retrieved by the SELECT ! * 4) UPDATE/DELETE statement text to be sent to the remote server ! * 5) Boolean flag showing if we set the command es_processed ! * 6) Boolean flag showing if the remote query has a RETURNING clause ! * 7) Integer list of attribute numbers retrieved by RETURNING, if any * * These items are indexed with the enum FdwScanPrivateIndex, so an item * can be fetched with list_nth(). For example, to get the SELECT statement: *************** *** 96,105 **** enum FdwScanPrivateIndex { /* SQL statement to execute remotely (as a String node) */ FdwScanPrivateSelectSql, ! /* Integer list of attribute numbers retrieved by the SELECT */ ! FdwScanPrivateRetrievedAttrs }; /* * Similarly, this enum describes what's kept in the fdw_private list for * a ModifyTable node referencing a postgres_fdw foreign table. We store: --- 101,123 ---- { /* SQL statement to execute remotely (as a String node) */ FdwScanPrivateSelectSql, ! /* List of restriction clauses that can be executed remotely */ ! FdwScanPrivateRemoteConds, ! /* Integer list of attribute numbers retrieved by SELECT */ ! FdwScanPrivateRetrievedAttrsBySelect, ! /* UPDATE/DELETE statement to execute remotely (as a String node) */ ! FdwScanPrivateUpdateSql, ! /* set-processed flag (as an integer Value node) */ ! FdwScanPrivateSetProcessed, ! /* has-returning flag (as an integer Value node) */ ! FdwScanPrivateHasReturning, ! /* Integer list of attribute numbers retrieved by RETURNING */ ! FdwScanPrivateRetrievedAttrsByReturning }; + #define SelectFdwScanPrivateLength 3 + #define UpdateFdwScanPrivateLength 7 + /* * Similarly, this enum describes what's kept in the fdw_private list for * a ModifyTable node referencing a postgres_fdw foreign table. We store: *************** *** 131,138 **** typedef struct PgFdwScanState AttInMetadata *attinmeta; /* attribute datatype conversion metadata */ /* extracted fdw_private data */ ! char *query; /* text of SELECT command */ List *retrieved_attrs; /* list of retrieved attribute numbers */ /* for remote query execution */ PGconn *conn; /* connection for the scan */ --- 149,158 ---- AttInMetadata *attinmeta; /* attribute datatype conversion metadata */ /* extracted fdw_private data */ ! char *query; /* text of SELECT or UPDATE/DELETE command */ List *retrieved_attrs; /* list of retrieved attribute numbers */ + bool set_processed; /* do we set the command es_processed? */ + bool has_returning; /* is there a RETURNING clause? */ /* for remote query execution */ PGconn *conn; /* connection for the scan */ *************** *** 152,157 **** typedef struct PgFdwScanState --- 172,182 ---- int fetch_ct_2; /* Min(# of fetches done, 2) */ bool eof_reached; /* true if last fetch reached EOF */ + /* for update pushdown */ + bool update_is_pushed_down; /* is UPDATE/DELETE pushed down? */ + PGresult *result; /* result of an UPDATE/DELETE query */ + TupleTableSlot *rslot; /* slot containing the result tuple */ + /* working memory contexts */ MemoryContext batch_cxt; /* context holding current batch of tuples */ MemoryContext temp_cxt; /* context for per-tuple temporary data */ *************** *** 180,185 **** typedef struct PgFdwModifyState --- 205,214 ---- int p_nums; /* number of parameters to transmit */ FmgrInfo *p_flinfo; /* output conversion functions for them */ + /* for update pushdown */ + bool update_is_pushed_down; /* is UPDATE/DELETE pushed down? */ + PgFdwScanState *fsstate; /* execution state of a foreign scan */ + /* working memory context */ MemoryContext temp_cxt; /* context for per-tuple temporary data */ } PgFdwModifyState; *************** *** 308,319 **** static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel, static void create_cursor(ForeignScanState *node); static void fetch_more_data(ForeignScanState *node); static void close_cursor(PGconn *conn, unsigned int cursor_number); 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 int postgresAcquireSampleRowsFunc(Relation relation, int elevel, HeapTuple *rows, int targrows, double *totalrows, --- 337,365 ---- static void create_cursor(ForeignScanState *node); static void fetch_more_data(ForeignScanState *node); static void close_cursor(PGconn *conn, unsigned int cursor_number); + static bool update_is_pushdown_safe(PlannerInfo *root, + ModifyTable *plan, + Index resultRelation, + int subplan_index, + Relation rel, + List *targetAttrs); + static List *push_update_down(PlannerInfo *root, + ModifyTable *plan, + Index resultRelation, + int subplan_index, + Relation rel, + List *targetAttrs); 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(TupleTableSlot *slot, ! PGresult *res, ! int row, ! Relation rel, ! AttInMetadata *attinmeta, ! List *retrieved_attrs, ! MemoryContext temp_context); static int postgresAcquireSampleRowsFunc(Relation relation, int elevel, HeapTuple *rows, int targrows, double *totalrows, *************** *** 852,858 **** postgresGetForeignPlan(PlannerInfo *root, * Build the fdw_private list that will be available to the executor. * Items in the list must match enum FdwScanPrivateIndex, above. */ ! fdw_private = list_make2(makeString(sql.data), retrieved_attrs); /* --- 898,905 ---- * Build the fdw_private list that will be available to the executor. * Items in the list must match enum FdwScanPrivateIndex, above. */ ! fdw_private = list_make3(makeString(sql.data), ! remote_conds, retrieved_attrs); /* *************** *** 914,935 **** postgresBeginForeignScan(ForeignScanState *node, int eflags) server = GetForeignServer(table->serverid); user = GetUserMapping(userid, server->serverid); /* * Get connection to the foreign server. Connection manager will * establish new connection if necessary. */ fsstate->conn = GetConnection(server, user, false); - /* Assign a unique ID for my cursor */ - fsstate->cursor_number = GetCursorNumber(fsstate->conn); - fsstate->cursor_exists = false; - - /* Get private info created by planner functions. */ - fsstate->query = strVal(list_nth(fsplan->fdw_private, - FdwScanPrivateSelectSql)); - fsstate->retrieved_attrs = (List *) list_nth(fsplan->fdw_private, - FdwScanPrivateRetrievedAttrs); - /* Create contexts for batches of tuples and per-tuple temp workspace. */ fsstate->batch_cxt = AllocSetContextCreate(estate->es_query_cxt, "postgres_fdw tuple data", --- 961,993 ---- server = GetForeignServer(table->serverid); user = GetUserMapping(userid, server->serverid); + /* Get private info created by planner functions. */ + if (list_length(fsplan->fdw_private) == UpdateFdwScanPrivateLength) + { + fsstate->query = strVal(list_nth(fsplan->fdw_private, + FdwScanPrivateUpdateSql)); + fsstate->set_processed = intVal(list_nth(fsplan->fdw_private, + FdwScanPrivateSetProcessed)); + fsstate->has_returning = intVal(list_nth(fsplan->fdw_private, + FdwScanPrivateHasReturning)); + fsstate->retrieved_attrs = (List *) list_nth(fsplan->fdw_private, + FdwScanPrivateRetrievedAttrsByReturning); + } + else + { + Assert(list_length(fsplan->fdw_private) == SelectFdwScanPrivateLength); + fsstate->query = strVal(list_nth(fsplan->fdw_private, + FdwScanPrivateSelectSql)); + fsstate->retrieved_attrs = (List *) list_nth(fsplan->fdw_private, + FdwScanPrivateRetrievedAttrsBySelect); + } + /* * Get connection to the foreign server. Connection manager will * establish new connection if necessary. */ fsstate->conn = GetConnection(server, user, false); /* Create contexts for batches of tuples and per-tuple temp workspace. */ fsstate->batch_cxt = AllocSetContextCreate(estate->es_query_cxt, "postgres_fdw tuple data", *************** *** 945,950 **** postgresBeginForeignScan(ForeignScanState *node, int eflags) --- 1003,1037 ---- /* Get info we'll need for input data conversion. */ fsstate->attinmeta = TupleDescGetAttInMetadata(RelationGetDescr(fsstate->rel)); + /* + * If pushing update down, execute the statement, and check for success. + */ + if (list_length(fsplan->fdw_private) == UpdateFdwScanPrivateLength) + { + /* + * We don't use a PG_TRY block here, so be careful not to throw error + * without releasing the PGresult. + */ + fsstate->result = PQexec(fsstate->conn, fsstate->query); + if (PQresultStatus(fsstate->result) != + (fsstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK)) + pgfdw_report_error(ERROR, fsstate->result, fsstate->conn, true, + fsstate->query); + + /* Check number of rows affected. */ + if (fsstate->has_returning) + fsstate->num_tuples = PQntuples(fsstate->result); + else + fsstate->num_tuples = atoi(PQcmdTuples(fsstate->result)); + + fsstate->update_is_pushed_down = true; + return; + } + + /* Assign a unique ID for my cursor */ + fsstate->cursor_number = GetCursorNumber(fsstate->conn); + fsstate->cursor_exists = false; + /* Prepare for output conversion of parameters used in remote query. */ numParams = list_length(fsplan->fdw_exprs); fsstate->numParams = numParams; *************** *** 995,1000 **** postgresIterateForeignScan(ForeignScanState *node) --- 1082,1155 ---- TupleTableSlot *slot = node->ss.ss_ScanTupleSlot; /* + * If pushing update down, get the result of UPDATE/DELETE RETURNING. + */ + if (fsstate->update_is_pushed_down) + { + MemoryContext oldcontext; + + /* + * If the update query doesn't have a RETURNING clause, then there is + * nothing to do, so we just return an empty slot. + */ + if (!fsstate->has_returning) + { + /* + * Increment the command es_processed if necessary. (ModifyTable + * cannot do that by itself in this case.) + */ + if (fsstate->set_processed) + { + EState *estate = node->ss.ps.state; + + estate->es_processed += fsstate->num_tuples; + } + + /* + * Increment the tuple count for EXPLAIN ANALYZE if we are running + * it. (The command cannot do that by itself in this case.) + */ + if (node->ss.ps.instrument) + { + Instrumentation *instr = node->ss.ps.instrument; + + instr->ntuples += fsstate->num_tuples; + } + + return ExecClearTuple(slot); + } + + /* If we didn't get any tuples, must be end of data. */ + if (fsstate->next_tuple >= fsstate->num_tuples) + return ExecClearTuple(slot); + + /* OK, we'll store RETURNING tuples in the batch_cxt. */ + oldcontext = MemoryContextSwitchTo(fsstate->batch_cxt); + + /* Fetch the next tuple. */ + store_returning_result(slot, + fsstate->result, + fsstate->next_tuple, + fsstate->rel, + fsstate->attinmeta, + fsstate->retrieved_attrs, + fsstate->temp_cxt); + fsstate->rslot = slot; + fsstate->next_tuple++; + + MemoryContextSwitchTo(oldcontext); + + /* + * Return slot. Note that this is safe because we can avoid applying + * ExecQual to the tuple due to no local quals (see the comment for + * update_is_pushdown_safe) and because the tuple can be safely + * projected by ExecProject (see push_update_down) and would then be + * ignored in postgresExecForeignUpdate or postgresExecForeignDelete. + */ + return slot; + } + + /* * If this is the first call after Begin or ReScan, we need to create the * cursor on the remote side. */ *************** *** 1094,1099 **** postgresEndForeignScan(ForeignScanState *node) --- 1249,1261 ---- if (fsstate == NULL) return; + /* if pushing update down, nothing to do other than cleanup */ + if (fsstate->update_is_pushed_down) + { + if (fsstate->result) + PQclear(fsstate->result); + } + /* Close the cursor if open, to prevent accumulation of cursors */ if (fsstate->cursor_exists) close_cursor(fsstate->conn, fsstate->cursor_number); *************** *** 1167,1174 **** postgresPlanForeignModify(PlannerInfo *root, List *returningList = NIL; List *retrieved_attrs = NIL; - initStringInfo(&sql); - /* * Core code already has some lock on each rel being planned, so we can * use NoLock here. --- 1329,1334 ---- *************** *** 1210,1215 **** postgresPlanForeignModify(PlannerInfo *root, --- 1370,1399 ---- } /* + * For UPDATE/DELETE, if there are no local conditions or joins needed (see + * update_is_pushdown_safe for more details), we push the command down. + */ + if (operation == CMD_UPDATE || operation == CMD_DELETE) + { + /* Check whether it's safe to push the command down. */ + if (update_is_pushdown_safe(root, plan, + resultRelation, + subplan_index, + rel, targetAttrs)) + { + List *fdw_private; + + /* OK, modify plan so as to push the command down. */ + fdw_private = push_update_down(root, plan, + resultRelation, + subplan_index, + rel, targetAttrs); + heap_close(rel, NoLock); + return fdw_private; + } + } + + /* * Extract the relevant RETURNING list if any. */ if (plan->returningLists) *************** *** 1218,1223 **** postgresPlanForeignModify(PlannerInfo *root, --- 1402,1408 ---- /* * Construct the SQL command string. */ + initStringInfo(&sql); switch (operation) { case CMD_INSERT: *************** *** 1288,1293 **** postgresBeginForeignModify(ModifyTableState *mtstate, --- 1473,1514 ---- fmstate = (PgFdwModifyState *) palloc0(sizeof(PgFdwModifyState)); fmstate->rel = rel; + /* 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); + + /* + * if query is NULL, we are in update pushdown case. + */ + if (fmstate->query == NULL) + { + PlanState *node = mtstate->mt_plans[subplan_index]; + PgFdwScanState *fsstate; + + Assert(fmstate->target_attrs == NIL); + Assert(fmstate->has_returning == false); + Assert(fmstate->retrieved_attrs == NIL); + + Assert(nodeTag(node) == T_ForeignScanState); + fsstate = ((ForeignScanState *) node)->fdw_state; + Assert(fsstate->update_is_pushed_down); + + fmstate->update_is_pushed_down = true; + if (fsstate->has_returning) + { + fmstate->has_returning = true; + fmstate->fsstate = fsstate; + } + resultRelInfo->ri_FdwState = fmstate; + return; + } + /* * Identify which user to do the remote access as. This should match what * ExecCheckRTEPerms() does. *************** *** 1304,1319 **** postgresBeginForeignModify(ModifyTableState *mtstate, fmstate->conn = GetConnection(server, 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", --- 1525,1530 ---- *************** *** 1411,1417 **** postgresExecForeignInsert(EState *estate, { n_rows = PQntuples(res); if (n_rows > 0) ! store_returning_result(fmstate, slot, res); } else n_rows = atoi(PQcmdTuples(res)); --- 1622,1632 ---- { n_rows = PQntuples(res); if (n_rows > 0) ! store_returning_result(slot, res, 0, ! fmstate->rel, ! fmstate->attinmeta, ! fmstate->retrieved_attrs, ! fmstate->temp_cxt); } else n_rows = atoi(PQcmdTuples(res)); *************** *** 1442,1447 **** postgresExecForeignUpdate(EState *estate, --- 1657,1670 ---- PGresult *res; int n_rows; + /* Just return slot created by the ForeignScan, if pushing update down */ + if (fmstate->update_is_pushed_down) + { + Assert(fmstate->has_returning); + Assert(fmstate->fsstate->rslot); + return fmstate->fsstate->rslot; + } + /* Set up the prepared statement on the remote server, if we didn't yet */ if (!fmstate->p_name) prepare_foreign_modify(fmstate); *************** *** 1481,1487 **** postgresExecForeignUpdate(EState *estate, { n_rows = PQntuples(res); if (n_rows > 0) ! store_returning_result(fmstate, slot, res); } else n_rows = atoi(PQcmdTuples(res)); --- 1704,1714 ---- { n_rows = PQntuples(res); if (n_rows > 0) ! store_returning_result(slot, res, 0, ! fmstate->rel, ! fmstate->attinmeta, ! fmstate->retrieved_attrs, ! fmstate->temp_cxt); } else n_rows = atoi(PQcmdTuples(res)); *************** *** 1512,1517 **** postgresExecForeignDelete(EState *estate, --- 1739,1752 ---- PGresult *res; int n_rows; + /* Just return slot created by the ForeignScan, if pushing update down */ + if (fmstate->update_is_pushed_down) + { + Assert(fmstate->has_returning); + Assert(fmstate->fsstate->rslot); + return fmstate->fsstate->rslot; + } + /* Set up the prepared statement on the remote server, if we didn't yet */ if (!fmstate->p_name) prepare_foreign_modify(fmstate); *************** *** 1551,1557 **** postgresExecForeignDelete(EState *estate, { n_rows = PQntuples(res); if (n_rows > 0) ! store_returning_result(fmstate, slot, res); } else n_rows = atoi(PQcmdTuples(res)); --- 1786,1796 ---- { n_rows = PQntuples(res); if (n_rows > 0) ! store_returning_result(slot, res, 0, ! fmstate->rel, ! fmstate->attinmeta, ! fmstate->retrieved_attrs, ! fmstate->temp_cxt); } else n_rows = atoi(PQcmdTuples(res)); *************** *** 1579,1584 **** postgresEndForeignModify(EState *estate, --- 1818,1827 ---- if (fmstate == NULL) return; + /* If pushing update down, nothing to do */ + if (fmstate->update_is_pushed_down) + return; + /* If we created a prepared statement, destroy it */ if (fmstate->p_name) { *************** *** 1661,1667 **** postgresExplainForeignScan(ForeignScanState *node, ExplainState *es) if (es->verbose) { fdw_private = ((ForeignScan *) node->ss.ps.plan)->fdw_private; ! sql = strVal(list_nth(fdw_private, FdwScanPrivateSelectSql)); ExplainPropertyText("Remote SQL", sql, es); } } --- 1904,1918 ---- if (es->verbose) { fdw_private = ((ForeignScan *) node->ss.ps.plan)->fdw_private; ! if (list_length(fdw_private) == UpdateFdwScanPrivateLength) ! { ! sql = strVal(list_nth(fdw_private, FdwScanPrivateUpdateSql)); ! } ! else ! { ! Assert(list_length(fdw_private) == SelectFdwScanPrivateLength); ! sql = strVal(list_nth(fdw_private, FdwScanPrivateSelectSql)); ! } ExplainPropertyText("Remote SQL", sql, es); } } *************** *** 1682,1688 **** postgresExplainForeignModify(ModifyTableState *mtstate, char *sql = strVal(list_nth(fdw_private, FdwModifyPrivateUpdateSql)); ! ExplainPropertyText("Remote SQL", sql, es); } } --- 1933,1940 ---- char *sql = strVal(list_nth(fdw_private, FdwModifyPrivateUpdateSql)); ! if (sql != NULL) ! ExplainPropertyText("Remote SQL", sql, es); } } *************** *** 1911,1916 **** ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel, --- 2163,2377 ---- } /* + * Check whether it's safe to push the UPDATE/DELETE command down. + * + * Conditions checked here: + * + * 1. If the target relation has any BEFORE/AFTER row-level local triggers, we + * must not push the command down, since that breaks execution of the triggers. + * + * 2. If there are any local joins needed, we mustn't push the command down, + * because that breaks execution of the joins. + * + * 3. If there are any quals that can't be evaluated remotely, we mustn't push + * the command down, because that breaks evaluation of the quals. + * + * 4. In UPDATE, if it is unsafe to evaluate any expressions to assign to the + * target columns on the remote server, we must not push the command down. + */ + static bool + update_is_pushdown_safe(PlannerInfo *root, + ModifyTable *plan, + Index resultRelation, + int subplan_index, + Relation rel, + List *targetAttrs) + { + RelOptInfo *baserel = root->simple_rel_array[resultRelation]; + Plan *subplan = (Plan *) list_nth(plan->plans, subplan_index); + ListCell *lc; + + /* Check point 1 */ + if (rel->trigdesc && + (rel->trigdesc->trig_update_after_row || + rel->trigdesc->trig_update_before_row)) + return false; + + /* Check point 2 */ + if (nodeTag(subplan) != T_ForeignScan) + return false; + + /* Check point 3 */ + if (subplan->qual != NIL) + return false; + + /* Check point 4 */ + foreach(lc, targetAttrs) + { + int attnum = lfirst_int(lc); + TargetEntry *tle = get_tle_by_resno(subplan->targetlist, + attnum); + + if (!is_foreign_expr(root, baserel, (Expr *) tle->expr)) + return false; + } + + return true; + } + + /* + * Modify a plan so as to push the update command down. + */ + static List * + push_update_down(PlannerInfo *root, + ModifyTable *plan, + Index resultRelation, + int subplan_index, + Relation rel, + List *targetAttrs) + { + CmdType operation = plan->operation; + bool canSetTag = plan->canSetTag; + Plan *subplan = (Plan *) list_nth(plan->plans, subplan_index); + ForeignScan *fscan = (ForeignScan *) subplan; + StringInfoData sql; + List *remote_conds; + List *returningList = NIL; + List *retrieved_attrs = NIL; + List *new_tlist = NIL; + List *fdw_private; + + Assert(operation == CMD_UPDATE || operation == CMD_DELETE); + + initStringInfo(&sql); + + /* + * Extract the baserestrictinfo clauses that can be evaluated remotely. + */ + remote_conds = (List *) list_nth(fscan->fdw_private, + FdwScanPrivateRemoteConds); + + /* + * Extract the relevant RETURNING list if any. + */ + if (plan->returningLists) + returningList = (List *) list_nth(plan->returningLists, subplan_index); + + /* + * Construct the SQL command string. + */ + if (operation == CMD_UPDATE) + { + List *targetlist = subplan->targetlist; + + deparseDirectUpdateSql(&sql, root, resultRelation, rel, + remote_conds, + targetlist, + targetAttrs, + returningList, + &retrieved_attrs); + } + else + { + Assert(operation == CMD_DELETE); + + deparseDirectDeleteSql(&sql, root, resultRelation, rel, + remote_conds, + returningList, + &retrieved_attrs); + } + + /* + * Update the fdw_private list that will be available to the executor. + * Items in the list must match enum FdwScanPrivateIndex, above. + */ + fscan->fdw_private = lappend(fscan->fdw_private, makeString(sql.data)); + fscan->fdw_private = lappend(fscan->fdw_private, makeInteger(canSetTag)); + fscan->fdw_private = lappend(fscan->fdw_private, + makeInteger((retrieved_attrs != NIL))); + fscan->fdw_private = lappend(fscan->fdw_private, retrieved_attrs); + + /* + * Rrewrite the targetlist for an UPDATE command for safety of ExecProject. + * Note we ignore and do not reference result tuples in update pushdown case. + */ + if (operation == CMD_UPDATE) + { + ListCell *lc; + int attrno = 1; + int numattrs = RelationGetNumberOfAttributes(rel); + + foreach(lc, subplan->targetlist) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + + if (tle->resjunk) + { + new_tlist = lappend(new_tlist, tle); + continue; + } + + if (attrno > numattrs) + ereport(ERROR, + (errcode(ERRCODE_DATATYPE_MISMATCH), + errmsg("table row type and query-specified row type do not match"), + errdetail("Query has too many columns."))); + + if (!list_member_int(targetAttrs, attrno)) + new_tlist = lappend(new_tlist, tle); + else + { + Form_pg_attribute attr; + Oid atttype; + int32 atttypmod; + Oid attcollation; + Node *new_expr; + TargetEntry *new_tle; + + attr = rel->rd_att->attrs[attrno - 1]; + + Assert(!attr->attisdropped); + atttype = attr->atttypid; + atttypmod = attr->atttypmod; + attcollation = attr->attcollation; + + new_expr = (Node *) makeVar(resultRelation, + attrno, + atttype, + atttypmod, + attcollation, + 0); + + new_tle = makeTargetEntry((Expr *) new_expr, + attrno, + pstrdup(NameStr(attr->attname)), + false); + + new_tlist = lappend(new_tlist, new_tle); + } + + attrno++; + } + + if (attrno != numattrs + 1) + ereport(ERROR, + (errcode(ERRCODE_DATATYPE_MISMATCH), + errmsg("table row type and query-specified row type do not match"), + errdetail("Query has too few columns."))); + + subplan->targetlist = new_tlist; + } + + /* + * Build the fdw_private list that will be available to the executor. + * Items in the list must match enum FdwModifyPrivateIndex, above. + */ + fdw_private = list_make4(makeString(NULL), NIL, makeInteger(false), NIL); + + return fdw_private; + } + + /* * Create cursor for node's query with current parameter values. */ static void *************** *** 2258,2275 **** convert_prep_stmt_params(PgFdwModifyState *fmstate, * have PG_TRY blocks to ensure this happens. */ static void ! store_returning_result(PgFdwModifyState *fmstate, ! TupleTableSlot *slot, PGresult *res) { PG_TRY(); { HeapTuple newtup; ! newtup = make_tuple_from_result_row(res, 0, ! fmstate->rel, ! fmstate->attinmeta, ! fmstate->retrieved_attrs, ! fmstate->temp_cxt); /* tuple will be deleted when it is cleared from the slot */ ExecStoreTuple(newtup, slot, InvalidBuffer, true); } --- 2719,2741 ---- * have PG_TRY blocks to ensure this happens. */ static void ! store_returning_result(TupleTableSlot *slot, ! PGresult *res, ! int row, ! Relation rel, ! AttInMetadata *attinmeta, ! List *retrieved_attrs, ! MemoryContext temp_context) { PG_TRY(); { HeapTuple newtup; ! newtup = make_tuple_from_result_row(res, row, ! rel, ! attinmeta, ! retrieved_attrs, ! temp_context); /* tuple will be deleted when it is cleared from the slot */ ExecStoreTuple(newtup, slot, InvalidBuffer, true); } *** a/contrib/postgres_fdw/postgres_fdw.h --- b/contrib/postgres_fdw/postgres_fdw.h *************** *** 66,75 **** extern void deparseUpdateSql(StringInfo buf, PlannerInfo *root, --- 66,87 ---- Index rtindex, Relation rel, List *targetAttrs, List *returningList, List **retrieved_attrs); + extern void deparseDirectUpdateSql(StringInfo buf, PlannerInfo *root, + Index rtindex, Relation rel, + List *remote_conds, + List *targetlist, + List *targetAttrs, + List *returningList, + List **retrieved_attrs); extern void deparseDeleteSql(StringInfo buf, PlannerInfo *root, Index rtindex, Relation rel, List *returningList, List **retrieved_attrs); + extern void deparseDirectDeleteSql(StringInfo buf, PlannerInfo *root, + Index rtindex, Relation rel, + List *remote_conds, + List *returningList, + List **retrieved_attrs); extern void deparseAnalyzeSizeSql(StringInfo buf, Relation rel); extern void deparseAnalyzeSql(StringInfo buf, Relation rel, List **retrieved_attrs); *** a/contrib/postgres_fdw/sql/postgres_fdw.sql --- b/contrib/postgres_fdw/sql/postgres_fdw.sql *************** *** 314,331 **** INSERT INTO ft2 (c1,c2,c3) SELECT c1+1000,c2+100, c3 || c3 FROM ft2 LIMIT 20; INSERT INTO ft2 (c1,c2,c3) VALUES (1101,201,'aaa'), (1102,202,'bbb'), (1103,203,'ccc') RETURNING *; INSERT INTO ft2 (c1,c2,c3) VALUES (1104,204,'ddd'), (1105,205,'eee'); UPDATE ft2 SET c2 = c2 + 300, c3 = c3 || '_update3' WHERE c1 % 10 = 3; UPDATE ft2 SET c2 = c2 + 400, c3 = c3 || '_update7' WHERE c1 % 10 = 7 RETURNING *; EXPLAIN (verbose, costs off) UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT ! FROM ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 9; UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT FROM ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 9; EXPLAIN (verbose, costs off) DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; EXPLAIN (verbose, costs off) ! DELETE FROM ft2 USING ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 2; DELETE FROM ft2 USING ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 2; SELECT c1,c2,c3,c4 FROM ft2 ORDER BY c1; --- 314,338 ---- INSERT INTO ft2 (c1,c2,c3) VALUES (1101,201,'aaa'), (1102,202,'bbb'), (1103,203,'ccc') RETURNING *; INSERT INTO ft2 (c1,c2,c3) VALUES (1104,204,'ddd'), (1105,205,'eee'); + EXPLAIN (verbose, costs off) + UPDATE ft2 SET c2 = c2 + 300, c3 = c3 || '_update3' WHERE c1 % 10 = 3; UPDATE ft2 SET c2 = c2 + 300, c3 = c3 || '_update3' WHERE c1 % 10 = 3; + EXPLAIN (verbose, costs off) + UPDATE ft2 SET c2 = c2 + 400, c3 = c3 || '_update7' WHERE c1 % 10 = 7 RETURNING *; UPDATE ft2 SET c2 = c2 + 400, c3 = c3 || '_update7' WHERE c1 % 10 = 7 RETURNING *; EXPLAIN (verbose, costs off) UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT ! FROM ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 9; -- can't be pushed down UPDATE ft2 SET c2 = ft2.c2 + 500, c3 = ft2.c3 || '_update9', c7 = DEFAULT FROM ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 9; EXPLAIN (verbose, costs off) + UPDATE ft2 SET c7 = DEFAULT WHERE c1 % 10 = 5 AND date(c4) = '1970-01-17'::date; -- can't be pushed down + UPDATE ft2 SET c7 = DEFAULT WHERE c1 % 10 = 5 AND date(c4) = '1970-01-17'::date; + EXPLAIN (verbose, costs off) DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; EXPLAIN (verbose, costs off) ! DELETE FROM ft2 USING ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 2; -- can't be pushed down DELETE FROM ft2 USING ft1 WHERE ft1.c1 = ft2.c2 AND ft1.c1 % 10 = 2; SELECT c1,c2,c3,c4 FROM ft2 ORDER BY c1; *** a/doc/src/sgml/postgres-fdw.sgml --- b/doc/src/sgml/postgres-fdw.sgml *************** *** 414,419 **** --- 414,428 ---- WHERE clauses are not sent to the remote server unless they use only built-in data types, operators, and functions. Operators and functions in the clauses must be IMMUTABLE as well. + For an UPDATE or DELETE query, + postgres_fdw attempts to optimize the query execution by + sending the whole query to the remote server if there are no query + WHERE clauses that cannot be sent to the remote server, + no local joins for the query, or no BEFORE or AFTER + row-level local triggers on the target table. In UPDATE, + expressions to assign to target columns must use only built-in data types, + IMMUTABLE operators, and IMMUTABLE functions, + to reduce the risk of misexecution of the query.