From 5f5a0434227debd8ca7ee406e8cb1997edff14a3 Mon Sep 17 00:00:00 2001 From: Alexander Pyhalov Date: Mon, 14 Feb 2022 12:01:26 +0300 Subject: [PATCH] postgres_fdw could pass lock wait policy to foreign server --- contrib/postgres_fdw/deparse.c | 12 +- .../postgres_fdw/expected/postgres_fdw.out | 110 +++++++++++++++++- contrib/postgres_fdw/option.c | 4 +- contrib/postgres_fdw/postgres_fdw.c | 4 + contrib/postgres_fdw/postgres_fdw.h | 1 + contrib/postgres_fdw/sql/postgres_fdw.sql | 18 +++ doc/src/sgml/postgres-fdw.sgml | 16 +++ 7 files changed, 161 insertions(+), 4 deletions(-) diff --git a/contrib/postgres_fdw/deparse.c b/contrib/postgres_fdw/deparse.c index bf12eac0288..f55afd461f0 100644 --- a/contrib/postgres_fdw/deparse.c +++ b/contrib/postgres_fdw/deparse.c @@ -1429,8 +1429,7 @@ deparseLockingClause(deparse_expr_cxt *context) * For now, just ignore any [NO] KEY specification, since (a) * it's not clear what that means for a remote table that we * don't have complete information about, and (b) it wouldn't - * work anyway on older remote servers. Likewise, we don't - * worry about NOWAIT. + * work anyway on older remote servers. */ switch (rc->strength) { @@ -1451,6 +1450,15 @@ deparseLockingClause(deparse_expr_cxt *context) if (bms_membership(rel->relids) == BMS_MULTIPLE && rc->strength != LCS_NONE) appendStringInfo(buf, " OF %s%d", REL_ALIAS_PREFIX, relid); + + /* Lock behavior */ + if (fpinfo->deparse_wait_policy && rc->strength != LCS_NONE) + { + if (rc->waitPolicy == LockWaitSkip) + appendStringInfoString(buf, " SKIP LOCKED"); + else if (rc->waitPolicy == LockWaitError) + appendStringInfoString(buf, " NOWAIT"); + } } } } diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index b2e02caefe4..b33d61b8656 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -389,6 +389,35 @@ SELECT * FROM ft1 t1 WHERE c1 = 102 FOR SHARE; 102 | 2 | 00102 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo (1 row) +-- test wait policy specification +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft1 t1 WHERE c1 = 101 FOR UPDATE NOWAIT; + QUERY PLAN +----------------------------------------------------------------------------------------------------------------- + Foreign Scan on public.ft1 t1 + Output: c1, c2, c3, c4, c5, c6, c7, c8, t1.* + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" = 101)) FOR UPDATE NOWAIT +(3 rows) + +SELECT * FROM ft1 t1 WHERE c1 = 101 FOR UPDATE NOWAIT; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +-----+----+-------+------------------------------+--------------------------+----+------------+----- + 101 | 1 | 00101 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft1 t1 WHERE c1 = 102 FOR SHARE SKIP LOCKED; + QUERY PLAN +--------------------------------------------------------------------------------------------------------------------- + Foreign Scan on public.ft1 t1 + Output: c1, c2, c3, c4, c5, c6, c7, c8, t1.* + Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" = 102)) FOR SHARE SKIP LOCKED +(3 rows) + +SELECT * FROM ft1 t1 WHERE c1 = 102 FOR SHARE SKIP LOCKED; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +-----+----+-------+------------------------------+--------------------------+----+------------+----- + 102 | 2 | 00102 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo +(1 row) + -- aggregate SELECT COUNT(*) FROM ft1 t1; count @@ -1862,6 +1891,59 @@ SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t 110 | 110 (10 rows) +-- test wait policy specification +EXPLAIN (VERBOSE, COSTS OFF) +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE SKIP LOCKED; + QUERY PLAN +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Foreign Scan + Output: t1.c1, t2.c1, t1.c3, t1.*, t2.* + Relations: (public.ft1 t1) INNER JOIN (public.ft2 t2) + Remote SQL: SELECT r1."C 1", r2."C 1", r1.c3, CASE WHEN (r1.*)::text IS NOT NULL THEN ROW(r1."C 1", r1.c2, r1.c3, r1.c4, r1.c5, r1.c6, r1.c7, r1.c8) END, CASE WHEN (r2.*)::text IS NOT NULL THEN ROW(r2."C 1", r2.c2, r2.c3, r2.c4, r2.c5, r2.c6, r2.c7, r2.c8) END FROM ("S 1"."T 1" r1 INNER JOIN "S 1"."T 1" r2 ON (((r1."C 1" = r2."C 1")))) ORDER BY r1.c3 ASC NULLS LAST, r1."C 1" ASC NULLS LAST LIMIT 10::bigint OFFSET 100::bigint FOR UPDATE OF r1 SKIP LOCKED FOR UPDATE OF r2 SKIP LOCKED +(4 rows) + +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE SKIP LOCKED; + c1 | c1 +-----+----- + 101 | 101 + 102 | 102 + 103 | 103 + 104 | 104 + 105 | 105 + 106 | 106 + 107 | 107 + 108 | 108 + 109 | 109 + 110 | 110 +(10 rows) + +ALTER SERVER loopback OPTIONS (ADD deparse_wait_policy 'false'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE SKIP LOCKED; + QUERY PLAN +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Foreign Scan + Output: t1.c1, t2.c1, t1.c3, t1.*, t2.* + Relations: (public.ft1 t1) INNER JOIN (public.ft2 t2) + Remote SQL: SELECT r1."C 1", r2."C 1", r1.c3, CASE WHEN (r1.*)::text IS NOT NULL THEN ROW(r1."C 1", r1.c2, r1.c3, r1.c4, r1.c5, r1.c6, r1.c7, r1.c8) END, CASE WHEN (r2.*)::text IS NOT NULL THEN ROW(r2."C 1", r2.c2, r2.c3, r2.c4, r2.c5, r2.c6, r2.c7, r2.c8) END FROM ("S 1"."T 1" r1 INNER JOIN "S 1"."T 1" r2 ON (((r1."C 1" = r2."C 1")))) ORDER BY r1.c3 ASC NULLS LAST, r1."C 1" ASC NULLS LAST LIMIT 10::bigint OFFSET 100::bigint FOR UPDATE OF r1 FOR UPDATE OF r2 +(4 rows) + +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE SKIP LOCKED; + c1 | c1 +-----+----- + 101 | 101 + 102 | 102 + 103 | 103 + 104 | 104 + 105 | 105 + 106 | 106 + 107 | 107 + 108 | 108 + 109 | 109 + 110 | 110 +(10 rows) + +ALTER SERVER loopback OPTIONS (DROP deparse_wait_policy); -- join two tables with FOR SHARE clause EXPLAIN (VERBOSE, COSTS OFF) SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR SHARE OF t1; @@ -1913,6 +1995,32 @@ SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t 110 | 110 (10 rows) +-- test wait policy specification +EXPLAIN (VERBOSE, COSTS OFF) +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE NOWAIT; + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ + Foreign Scan + Output: t1.c1, t2.c1, t1.c3, t1.*, t2.* + Relations: (public.ft1 t1) INNER JOIN (public.ft2 t2) + Remote SQL: SELECT r1."C 1", r2."C 1", r1.c3, CASE WHEN (r1.*)::text IS NOT NULL THEN ROW(r1."C 1", r1.c2, r1.c3, r1.c4, r1.c5, r1.c6, r1.c7, r1.c8) END, CASE WHEN (r2.*)::text IS NOT NULL THEN ROW(r2."C 1", r2.c2, r2.c3, r2.c4, r2.c5, r2.c6, r2.c7, r2.c8) END FROM ("S 1"."T 1" r1 INNER JOIN "S 1"."T 1" r2 ON (((r1."C 1" = r2."C 1")))) ORDER BY r1.c3 ASC NULLS LAST, r1."C 1" ASC NULLS LAST LIMIT 10::bigint OFFSET 100::bigint FOR UPDATE OF r1 NOWAIT FOR UPDATE OF r2 NOWAIT +(4 rows) + +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE NOWAIT; + c1 | c1 +-----+----- + 101 | 101 + 102 | 102 + 103 | 103 + 104 | 104 + 105 | 105 + 106 | 106 + 107 | 107 + 108 | 108 + 109 | 109 + 110 | 110 +(10 rows) + -- join in CTE EXPLAIN (VERBOSE, COSTS OFF) WITH t (c1_1, c1_3, c2_1) AS MATERIALIZED (SELECT t1.c1, t1.c3, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1)) SELECT c1_1, c2_1 FROM t ORDER BY c1_3, c1_1 OFFSET 100 LIMIT 10; @@ -9509,7 +9617,7 @@ DO $d$ END; $d$; ERROR: invalid option "password" -HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, keep_connections +HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, deparse_wait_policy, keep_connections CONTEXT: SQL statement "ALTER SERVER loopback_nopw OPTIONS (ADD password 'dummypw')" PL/pgSQL function inline_code_block line 3 at EXECUTE -- If we add a password for our user mapping instead, we should get a different diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c index fc3ce6a53a2..ec92c007862 100644 --- a/contrib/postgres_fdw/option.c +++ b/contrib/postgres_fdw/option.c @@ -121,7 +121,8 @@ postgres_fdw_validator(PG_FUNCTION_ARGS) strcmp(def->defname, "updatable") == 0 || strcmp(def->defname, "truncatable") == 0 || strcmp(def->defname, "async_capable") == 0 || - strcmp(def->defname, "keep_connections") == 0) + strcmp(def->defname, "keep_connections") == 0 || + strcmp(def->defname, "deparse_wait_policy") == 0) { /* these accept only boolean values */ (void) defGetBoolean(def); @@ -249,6 +250,7 @@ InitPgFdwOptions(void) /* async_capable is available on both server and table */ {"async_capable", ForeignServerRelationId, false}, {"async_capable", ForeignTableRelationId, false}, + {"deparse_wait_policy", ForeignServerRelationId, false}, {"keep_connections", ForeignServerRelationId, false}, {"password_required", UserMappingRelationId, false}, diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 56654844e8f..ead4824e8f3 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -650,6 +650,7 @@ postgresGetForeignRelSize(PlannerInfo *root, fpinfo->shippable_extensions = NIL; fpinfo->fetch_size = 100; fpinfo->async_capable = false; + fpinfo->deparse_wait_policy = true; apply_server_options(fpinfo); apply_table_options(fpinfo); @@ -5858,6 +5859,8 @@ apply_server_options(PgFdwRelationInfo *fpinfo) (void) parse_int(defGetString(def), &fpinfo->fetch_size, 0, NULL); else if (strcmp(def->defname, "async_capable") == 0) fpinfo->async_capable = defGetBoolean(def); + else if (strcmp(def->defname, "deparse_wait_policy") == 0) + fpinfo->deparse_wait_policy = defGetBoolean(def); } } @@ -5916,6 +5919,7 @@ merge_fdw_options(PgFdwRelationInfo *fpinfo, fpinfo->use_remote_estimate = fpinfo_o->use_remote_estimate; fpinfo->fetch_size = fpinfo_o->fetch_size; fpinfo->async_capable = fpinfo_o->async_capable; + fpinfo->deparse_wait_policy = fpinfo_o->deparse_wait_policy; /* Merge the table level options from either side of the join. */ if (fpinfo_i) diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index 8ae79e97e44..e4882e70fa4 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -80,6 +80,7 @@ typedef struct PgFdwRelationInfo Cost fdw_tuple_cost; List *shippable_extensions; /* OIDs of shippable extensions */ bool async_capable; + bool deparse_wait_policy; /* Cached catalog information. */ ForeignTable *table; diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index e050639b572..fdd55d76b6d 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -274,6 +274,11 @@ EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft1 t1 WHERE c1 = 101 FOR UPDATE; SELECT * FROM ft1 t1 WHERE c1 = 101 FOR UPDATE; EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft1 t1 WHERE c1 = 102 FOR SHARE; SELECT * FROM ft1 t1 WHERE c1 = 102 FOR SHARE; +-- test wait policy specification +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft1 t1 WHERE c1 = 101 FOR UPDATE NOWAIT; +SELECT * FROM ft1 t1 WHERE c1 = 101 FOR UPDATE NOWAIT; +EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft1 t1 WHERE c1 = 102 FOR SHARE SKIP LOCKED; +SELECT * FROM ft1 t1 WHERE c1 = 102 FOR SHARE SKIP LOCKED; -- aggregate SELECT COUNT(*) FROM ft1 t1; -- subquery @@ -566,6 +571,15 @@ SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t EXPLAIN (VERBOSE, COSTS OFF) SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE; SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE; +-- test wait policy specification +EXPLAIN (VERBOSE, COSTS OFF) +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE SKIP LOCKED; +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE SKIP LOCKED; +ALTER SERVER loopback OPTIONS (ADD deparse_wait_policy 'false'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE SKIP LOCKED; +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE SKIP LOCKED; +ALTER SERVER loopback OPTIONS (DROP deparse_wait_policy); -- join two tables with FOR SHARE clause EXPLAIN (VERBOSE, COSTS OFF) SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR SHARE OF t1; @@ -573,6 +587,10 @@ SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t EXPLAIN (VERBOSE, COSTS OFF) SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR SHARE; SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR SHARE; +-- test wait policy specification +EXPLAIN (VERBOSE, COSTS OFF) +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE NOWAIT; +SELECT t1.c1, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10 FOR UPDATE NOWAIT; -- join in CTE EXPLAIN (VERBOSE, COSTS OFF) WITH t (c1_1, c1_3, c2_1) AS MATERIALIZED (SELECT t1.c1, t1.c3, t2.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1)) SELECT c1_1, c2_1 FROM t ORDER BY c1_3, c1_1 OFFSET 100 LIMIT 10; diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml index 2bb31f11255..8eb194d1417 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -341,6 +341,22 @@ OPTIONS (ADD password_required 'false'); + + deparse_wait_policy (bool) + + + This option determines if SKIP LOCKED or + NOWAIT policy specified in + SELECT FOR UPDATE or SELECT FOR SHARE + queries should be passed to the remote server. + This option can only be specified for foreign servers, not per-table. + + + By default it's true. Note that such statements are not supported + by old PostgreSQL versions. + + + extensions (string) -- 2.25.1