(2014/08/13 12:40), Etsuro Fujita wrote:
(2014/08/12 18:34), Shigeru Hanada wrote:
Issues addressed by Eitoku-san were fixed properly, but he found a bug
and a possible enhancement in the v2 patch.
* push-down check misses delete triggers
update_is_pushdown_safe() seems to have a bug that it misses the
existence of row-level delete trigger. DELETE statement executed
against a foreign table which has row-level delete trigger is pushed
down to remote, and consequently no row-level delete trigger is fired.
Ah, I noticed that the current code for that is not correct. Will fix.
Done.
* further optimization
Is there any chance to consider further optimization by passing the
operation type (UPDATE|DELETE) of undergoing statement to
update_is_pushdown_safe()? It seems safe to push down UPDATE
statement when the target foreign table has no update trigger even it
has a delete trigger (of course the opposite combination would be also
fine).
Good idea! Will improve that too.
Done.
* Documentation
The requirement of pushing down UPDATE/DELETE statements would not be
easy to understand for non-expert users, so it seems that there is a
room to enhance documentation. An idea is to define which expression
is safe to send to remote first (it might need to mention the
difference of semantics), and refer the definition from the place
describing the requirement of pushing-down for SELECT, UPDATE and
DELETE.
Yeah, I also think that it would not necessarily easy for the users to
understand which expression is safe to send. So I agree with that
enhancement, but ISTM that it would be better to do that as a separate
patch.
As above, I'd like to leave this as another patch.
Please find attached the updated version of the patch.
Thanks,
Best regards,
Etsuro Fujita
*** 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
***************
*** 998,1004 **** INSERT INTO ft2 (c1,c2,c3)
--- 998,1025 ----
(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; -- can be pushed down
+ 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 *; -- can be pushed down
+ 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
------+-----+--------------------+------------------------------+--------------------------+----+------------+-----
***************
*** 1108,1114 **** 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
--- 1129,1135 ----
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
***************
*** 1129,1144 **** 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
--- 1150,1164 ----
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; -- can be pushed down
! 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
***************
*** 1249,1255 **** 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
--- 1269,1275 ----
(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
***************
*** 2092,2097 **** SELECT c1,c2,c3,c4 FROM ft2 ORDER BY c1;
--- 2112,2130 ----
1104 | 204 | ddd |
(819 rows)
+ EXPLAIN (verbose, costs off)
+ UPDATE ft2 SET c7 = DEFAULT WHERE c1 % 10 = 0 AND date(c4) = '1970-01-01'::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-01-1970'::date)
+ Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c8, ctid FROM "S 1"."T 1" WHERE ((("C 1" % 10) = 0)) FOR UPDATE
+ (6 rows)
+
+ UPDATE ft2 SET c7 = DEFAULT WHERE c1 % 10 = 0 AND date(c4) = '1970-01-01'::date;
-- Test that trigger on remote table works as expected
CREATE OR REPLACE FUNCTION "S 1".F_BRTRIG() RETURNS trigger AS $$
BEGIN
***************
*** 2233,2239 **** 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
--- 2266,2272 ----
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
***************
*** 2392,2398 **** 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
--- 2425,2431 ----
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
***************
*** 2840,2845 **** NOTICE: NEW: (13,"test triggered !")
--- 2873,3082 ----
(0,27)
(1 row)
+ -- cleanup
+ DROP TRIGGER trig_row_before ON rem1;
+ DROP TRIGGER trig_row_after ON rem1;
+ -- Test update-pushdown functionality
+ -- Test with statement-level triggers
+ CREATE TRIGGER trig_stmt_before
+ BEFORE DELETE OR INSERT OR UPDATE ON rem1
+ FOR EACH STATEMENT EXECUTE PROCEDURE trigger_func();
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ QUERY PLAN
+ ----------------------------------------------------------
+ Update on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: f1, f2, ctid
+ Remote SQL: UPDATE public.loc1 SET f2 = ''::text
+ (4 rows)
+
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ QUERY PLAN
+ ---------------------------------------------
+ Delete on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: ctid
+ Remote SQL: DELETE FROM public.loc1
+ (4 rows)
+
+ DROP TRIGGER trig_stmt_before ON rem1;
+ CREATE TRIGGER trig_stmt_after
+ AFTER DELETE OR INSERT OR UPDATE ON rem1
+ FOR EACH STATEMENT EXECUTE PROCEDURE trigger_func();
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ QUERY PLAN
+ ----------------------------------------------------------
+ Update on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: f1, f2, ctid
+ Remote SQL: UPDATE public.loc1 SET f2 = ''::text
+ (4 rows)
+
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ QUERY PLAN
+ ---------------------------------------------
+ Delete on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: ctid
+ Remote SQL: DELETE FROM public.loc1
+ (4 rows)
+
+ DROP TRIGGER trig_stmt_after ON rem1;
+ -- Test with row-level ON INSERT triggers
+ CREATE TRIGGER trig_row_before_insert
+ BEFORE INSERT ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ QUERY PLAN
+ ----------------------------------------------------------
+ Update on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: f1, f2, ctid
+ Remote SQL: UPDATE public.loc1 SET f2 = ''::text
+ (4 rows)
+
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ QUERY PLAN
+ ---------------------------------------------
+ Delete on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: ctid
+ Remote SQL: DELETE FROM public.loc1
+ (4 rows)
+
+ DROP TRIGGER trig_row_before_insert ON rem1;
+ CREATE TRIGGER trig_row_after_insert
+ AFTER INSERT ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ QUERY PLAN
+ ----------------------------------------------------------
+ Update on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: f1, f2, ctid
+ Remote SQL: UPDATE public.loc1 SET f2 = ''::text
+ (4 rows)
+
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ QUERY PLAN
+ ---------------------------------------------
+ Delete on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: ctid
+ Remote SQL: DELETE FROM public.loc1
+ (4 rows)
+
+ DROP TRIGGER trig_row_after_insert ON rem1;
+ -- Test with row-level ON UPDATE triggers
+ CREATE TRIGGER trig_row_before_update
+ BEFORE UPDATE ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can't be pushed down
+ QUERY PLAN
+ ---------------------------------------------------------------------
+ Update on public.rem1
+ Remote SQL: UPDATE public.loc1 SET f2 = $2 WHERE ctid = $1
+ -> Foreign Scan on public.rem1
+ Output: f1, ''::text, ctid, rem1.*
+ Remote SQL: SELECT f1, f2, ctid FROM public.loc1 FOR UPDATE
+ (5 rows)
+
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ QUERY PLAN
+ ---------------------------------------------
+ Delete on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: ctid
+ Remote SQL: DELETE FROM public.loc1
+ (4 rows)
+
+ DROP TRIGGER trig_row_before_update ON rem1;
+ CREATE TRIGGER trig_row_after_update
+ AFTER UPDATE ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can't be pushed down
+ QUERY PLAN
+ -------------------------------------------------------------------------------
+ Update on public.rem1
+ Remote SQL: UPDATE public.loc1 SET f2 = $2 WHERE ctid = $1 RETURNING f1, f2
+ -> Foreign Scan on public.rem1
+ Output: f1, ''::text, ctid, rem1.*
+ Remote SQL: SELECT f1, f2, ctid FROM public.loc1 FOR UPDATE
+ (5 rows)
+
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ QUERY PLAN
+ ---------------------------------------------
+ Delete on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: ctid
+ Remote SQL: DELETE FROM public.loc1
+ (4 rows)
+
+ DROP TRIGGER trig_row_after_update ON rem1;
+ -- Test with row-level ON DELETE triggers
+ CREATE TRIGGER trig_row_before_delete
+ BEFORE DELETE ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ QUERY PLAN
+ ----------------------------------------------------------
+ Update on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: f1, f2, ctid
+ Remote SQL: UPDATE public.loc1 SET f2 = ''::text
+ (4 rows)
+
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can't be pushed down
+ QUERY PLAN
+ ---------------------------------------------------------------------
+ Delete on public.rem1
+ Remote SQL: DELETE FROM public.loc1 WHERE ctid = $1
+ -> Foreign Scan on public.rem1
+ Output: ctid, rem1.*
+ Remote SQL: SELECT f1, f2, ctid FROM public.loc1 FOR UPDATE
+ (5 rows)
+
+ DROP TRIGGER trig_row_before_delete ON rem1;
+ CREATE TRIGGER trig_row_after_delete
+ AFTER DELETE ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ QUERY PLAN
+ ----------------------------------------------------------
+ Update on public.rem1
+ -> Foreign Scan on public.rem1
+ Output: f1, f2, ctid
+ Remote SQL: UPDATE public.loc1 SET f2 = ''::text
+ (4 rows)
+
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can't be pushed down
+ QUERY PLAN
+ ------------------------------------------------------------------------
+ Delete on public.rem1
+ Remote SQL: DELETE FROM public.loc1 WHERE ctid = $1 RETURNING f1, f2
+ -> Foreign Scan on public.rem1
+ Output: ctid, rem1.*
+ Remote SQL: SELECT f1, f2, ctid FROM public.loc1 FOR UPDATE
+ (5 rows)
+
+ DROP TRIGGER trig_row_after_delete ON rem1;
-- ===================================================================
-- test IMPORT FOREIGN SCHEMA
-- ===================================================================
*** 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,994 ----
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)
--- 1004,1023 ----
/* Get info we'll need for input data conversion. */
fsstate->attinmeta = TupleDescGetAttInMetadata(RelationGetDescr(fsstate->rel));
+ /*
+ * If pushing update down, we've got no more to do.
+ */
+ if (list_length(fsplan->fdw_private) == UpdateFdwScanPrivateLength)
+ {
+ fsstate->num_tuples = -1; /* -1 means not set yet */
+ 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)
--- 1068,1164 ----
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 this is the first call after Begin, we need to execute the statement,
+ * and check for success.
+ */
+ if (fsstate->num_tuples == -1)
+ {
+ /*
+ * 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));
+ }
+
+ /*
+ * 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 count 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 necessary.
+ * (EXPLAIN ANALYZE 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.
*/
***************
*** 1036,1041 **** postgresReScanForeignScan(ForeignScanState *node)
--- 1200,1208 ----
char sql[64];
PGresult *res;
+ /* This shouldn't be called in update pushdown case. */
+ Assert(fsstate->update_is_pushed_down == false);
+
/* If we haven't created the cursor yet, nothing to do. */
if (!fsstate->cursor_exists)
return;
***************
*** 1094,1099 **** postgresEndForeignScan(ForeignScanState *node)
--- 1261,1273 ----
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.
--- 1341,1346 ----
***************
*** 1210,1215 **** postgresPlanForeignModify(PlannerInfo *root,
--- 1382,1411 ----
}
/*
+ * 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,
--- 1414,1420 ----
/*
* Construct the SQL command string.
*/
+ initStringInfo(&sql);
switch (operation)
{
case CMD_INSERT:
***************
*** 1288,1293 **** postgresBeginForeignModify(ModifyTableState *mtstate,
--- 1485,1526 ----
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 = (PgFdwScanState *) ((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",
--- 1537,1542 ----
***************
*** 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));
--- 1634,1644 ----
{
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,
--- 1669,1682 ----
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));
--- 1716,1726 ----
{
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,
--- 1751,1764 ----
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));
--- 1798,1808 ----
{
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,
--- 1830,1839 ----
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);
}
}
--- 1916,1930 ----
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);
}
}
--- 1945,1952 ----
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,
--- 2175,2394 ----
}
/*
+ * Check whether it's safe to push the UPDATE/DELETE command down.
+ *
+ * Conditions checked here:
+ *
+ * 1. If the target relation has any row-level local BEFORE/AFTER 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)
+ {
+ CmdType operation = plan->operation;
+ RelOptInfo *baserel = root->simple_rel_array[resultRelation];
+ Plan *subplan = (Plan *) list_nth(plan->plans, subplan_index);
+ ListCell *lc;
+
+ /* Check point 1 */
+ if (rel->trigdesc &&
+ ((operation == CMD_UPDATE &&
+ (rel->trigdesc->trig_update_after_row ||
+ rel->trigdesc->trig_update_before_row)) ||
+ (operation == CMD_DELETE &&
+ (rel->trigdesc->trig_delete_after_row ||
+ rel->trigdesc->trig_delete_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);
}
--- 2736,2758 ----
* 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
***************
*** 320,339 **** 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;
-- Test that trigger on remote table works as expected
CREATE OR REPLACE FUNCTION "S 1".F_BRTRIG() RETURNS trigger AS $$
--- 320,346 ----
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; -- can be pushed down
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 *; -- can be pushed down
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)
! DELETE FROM ft2 WHERE c1 % 10 = 5 RETURNING c1, c4; -- can be pushed down
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;
+ EXPLAIN (verbose, costs off)
+ UPDATE ft2 SET c7 = DEFAULT WHERE c1 % 10 = 0 AND date(c4) = '1970-01-01'::date; -- can't be pushed down
+ UPDATE ft2 SET c7 = DEFAULT WHERE c1 % 10 = 0 AND date(c4) = '1970-01-01'::date;
-- Test that trigger on remote table works as expected
CREATE OR REPLACE FUNCTION "S 1".F_BRTRIG() RETURNS trigger AS $$
***************
*** 616,621 **** UPDATE rem1 SET f2 = 'testo';
--- 623,711 ----
-- Test returning a system attribute
INSERT INTO rem1(f2) VALUES ('test') RETURNING ctid;
+ -- cleanup
+ DROP TRIGGER trig_row_before ON rem1;
+ DROP TRIGGER trig_row_after ON rem1;
+
+
+ -- Test update-pushdown functionality
+
+ -- Test with statement-level triggers
+ CREATE TRIGGER trig_stmt_before
+ BEFORE DELETE OR INSERT OR UPDATE ON rem1
+ FOR EACH STATEMENT EXECUTE PROCEDURE trigger_func();
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ DROP TRIGGER trig_stmt_before ON rem1;
+
+ CREATE TRIGGER trig_stmt_after
+ AFTER DELETE OR INSERT OR UPDATE ON rem1
+ FOR EACH STATEMENT EXECUTE PROCEDURE trigger_func();
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ DROP TRIGGER trig_stmt_after ON rem1;
+
+ -- Test with row-level ON INSERT triggers
+ CREATE TRIGGER trig_row_before_insert
+ BEFORE INSERT ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ DROP TRIGGER trig_row_before_insert ON rem1;
+
+ CREATE TRIGGER trig_row_after_insert
+ AFTER INSERT ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ DROP TRIGGER trig_row_after_insert ON rem1;
+
+ -- Test with row-level ON UPDATE triggers
+ CREATE TRIGGER trig_row_before_update
+ BEFORE UPDATE ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can't be pushed down
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ DROP TRIGGER trig_row_before_update ON rem1;
+
+ CREATE TRIGGER trig_row_after_update
+ AFTER UPDATE ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can't be pushed down
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can be pushed down
+ DROP TRIGGER trig_row_after_update ON rem1;
+
+ -- Test with row-level ON DELETE triggers
+ CREATE TRIGGER trig_row_before_delete
+ BEFORE DELETE ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can't be pushed down
+ DROP TRIGGER trig_row_before_delete ON rem1;
+
+ CREATE TRIGGER trig_row_after_delete
+ AFTER DELETE ON rem1
+ FOR EACH ROW EXECUTE PROCEDURE trigger_data(23,'skidoo');
+ EXPLAIN (verbose, costs off)
+ UPDATE rem1 set f2 = ''; -- can be pushed down
+ EXPLAIN (verbose, costs off)
+ DELETE FROM rem1; -- can't be pushed down
+ DROP TRIGGER trig_row_after_delete ON rem1;
+
-- ===================================================================
-- test IMPORT FOREIGN SCHEMA
-- ===================================================================
*** a/doc/src/sgml/postgres-fdw.sgml
--- b/doc/src/sgml/postgres-fdw.sgml
***************
*** 414,419 ****
--- 414,428 ----
<literal>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 <literal>IMMUTABLE</> as well.
+ For an <command>UPDATE</> or <command>DELETE</> query,
+ <filename>postgres_fdw</> attempts to optimize the query execution by
+ sending the whole query to the remote server if there are no query
+ <literal>WHERE</> clauses that cannot be sent to the remote server,
+ no local joins for the query, or no row-level local <literal>BEFORE</> or
+ <literal>AFTER</> triggers on the target table. In <command>UPDATE</>,
+ expressions to assign to target columns must use only built-in data types,
+ <literal>IMMUTABLE</> operators, and <literal>IMMUTABLE</> functions,
+ to reduce the risk of misexecution of the query.
</para>
<para>
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
