From ede0f93dd44542a1ac401d073e7988c4e61822ce Mon Sep 17 00:00:00 2001 From: Julien Tachoires Date: Tue, 22 Oct 2024 23:12:44 +0200 Subject: [PATCH 4/6] Set ReorderBuffer compression via subscription opt The CREATE/ALTER SUBSCRIPTION commands now support a new option named "spill_compression" that will be used to select the compression method applied to the logical changes spilled on disk during decoding. The default value is "off", meaning no compression will be applied. Supported values are: "off", "on", "pglz", "lz4", and "zstd". --- doc/src/sgml/ref/create_subscription.sgml | 24 +++ src/backend/catalog/pg_subscription.c | 6 + src/backend/catalog/system_views.sql | 3 +- src/backend/commands/subscriptioncmds.c | 31 +++- .../libpqwalreceiver/libpqwalreceiver.c | 5 + src/backend/replication/logical/logical.c | 4 + .../replication/logical/reorderbuffer.c | 74 ++++++++- src/backend/replication/logical/worker.c | 13 +- src/backend/replication/pgoutput/pgoutput.c | 28 ++++ src/bin/pg_dump/pg_dump.c | 18 +- src/bin/pg_dump/pg_dump.h | 1 + src/bin/pg_dump/t/002_pg_dump.pl | 4 +- src/bin/psql/describe.c | 7 +- src/bin/psql/tab-complete.in.c | 6 +- src/include/catalog/pg_subscription.h | 4 + src/include/replication/logical.h | 2 + src/include/replication/pgoutput.h | 1 + .../replication/reorderbuffer_compression.h | 4 + src/include/replication/walreceiver.h | 1 + src/test/regress/expected/subscription.out | 156 +++++++++--------- src/test/regress/sql/subscription.sql | 4 + 21 files changed, 305 insertions(+), 91 deletions(-) diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index 6cf7d4f9a1..175965c4a4 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -435,6 +435,30 @@ CREATE SUBSCRIPTION subscription_name + + + spill_compression (enum) + + + Specifies whether the decoded changes that eventually need to be + temporarily written on disk by the publisher are compressed or not. + Default value is off meaning no data compression + involved. Setting spill_compression to + on or pglz means that the + decoded changes are compressed using the internal + PGLZ compression algorithm. + + + + If the PostgreSQL server running the + publisher node supports the external compression libraries + LZ4 or + Zstandard, + spill_compression can be set respectively to + lz4 or zstd. + + + diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index 89bf5ec933..e2bd881ab0 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -141,6 +141,12 @@ GetSubscription(Oid subid, bool missing_ok) /* Is the subscription owner a superuser? */ sub->ownersuperuser = superuser_arg(sub->owner); + /* Get spill_compression */ + datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, + tup, + Anum_pg_subscription_subspillcompression); + sub->spill_compression = TextDatumGetCString(datum); + ReleaseSysCache(tup); return sub; diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 3456b821bc..a087223677 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1358,7 +1358,8 @@ REVOKE ALL ON pg_subscription FROM public; GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled, subbinary, substream, subtwophasestate, subdisableonerr, subpasswordrequired, subrunasowner, subfailover, - subslotname, subsynccommit, subpublications, suborigin) + subslotname, subsynccommit, subpublications, suborigin, + subspillcompression) ON pg_subscription TO public; CREATE VIEW pg_stat_subscription_stats AS diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 03e97730e7..2c9f1eae4b 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -40,6 +40,7 @@ #include "replication/logicallauncher.h" #include "replication/logicalworker.h" #include "replication/origin.h" +#include "replication/reorderbuffer_compression.h" #include "replication/slot.h" #include "replication/walreceiver.h" #include "replication/walsender.h" @@ -73,6 +74,7 @@ #define SUBOPT_FAILOVER 0x00002000 #define SUBOPT_LSN 0x00004000 #define SUBOPT_ORIGIN 0x00008000 +#define SUBOPT_SPILL_COMPRESSION 0x00010000 /* check if the 'val' has 'bits' set */ #define IsSet(val, bits) (((val) & (bits)) == (bits)) @@ -100,6 +102,7 @@ typedef struct SubOpts bool failover; char *origin; XLogRecPtr lsn; + char *spill_compression; } SubOpts; static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); @@ -164,6 +167,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->failover = false; if (IsSet(supported_opts, SUBOPT_ORIGIN)) opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY); + if (IsSet(supported_opts, SUBOPT_SPILL_COMPRESSION)) + opts->spill_compression = "off"; /* Parse options */ foreach(lc, stmt_options) @@ -357,6 +362,18 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->specified_opts |= SUBOPT_LSN; opts->lsn = lsn; } + else if (IsSet(supported_opts, SUBOPT_SPILL_COMPRESSION) && + strcmp(defel->defname, "spill_compression") == 0) + { + if (IsSet(opts->specified_opts, SUBOPT_SPILL_COMPRESSION)) + errorConflictingDefElem(defel, pstate); + + opts->specified_opts |= SUBOPT_SPILL_COMPRESSION; + opts->spill_compression = defGetString(defel); + + ReorderBufferValidateCompressionMethod(opts->spill_compression, + ERROR); + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -563,7 +580,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY | SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT | SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED | - SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN); + SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN | + SUBOPT_SPILL_COMPRESSION); parse_subscription_options(pstate, stmt->options, supported_opts, &opts); /* @@ -683,6 +701,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, publicationListToArray(publications); values[Anum_pg_subscription_suborigin - 1] = CStringGetTextDatum(opts.origin); + values[Anum_pg_subscription_subspillcompression - 1] = + CStringGetTextDatum(opts.spill_compression); tup = heap_form_tuple(RelationGetDescr(rel), values, nulls); @@ -1165,7 +1185,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED | SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | - SUBOPT_ORIGIN); + SUBOPT_ORIGIN | SUBOPT_SPILL_COMPRESSION); parse_subscription_options(pstate, stmt->options, supported_opts, &opts); @@ -1332,6 +1352,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, replaces[Anum_pg_subscription_suborigin - 1] = true; } + if (IsSet(opts.specified_opts, SUBOPT_SPILL_COMPRESSION)) + { + values[Anum_pg_subscription_subspillcompression - 1] = + CStringGetTextDatum(opts.spill_compression); + replaces[Anum_pg_subscription_subspillcompression - 1] = true; + } + update_tuple = true; break; } diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 97f957cd87..20f5e4a83e 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -620,6 +620,11 @@ libpqrcv_startstreaming(WalReceiverConn *conn, PQserverVersion(conn->streamConn) >= 140000) appendStringInfoString(&cmd, ", binary 'true'"); + if (options->proto.logical.spill_compression && + PQserverVersion(conn->streamConn) >= 180000) + appendStringInfo(&cmd, ", spill_compression '%s'", + options->proto.logical.spill_compression); + appendStringInfoChar(&cmd, ')'); } else diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 3fe1774a1e..54fbbe6fea 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -36,6 +36,7 @@ #include "replication/decode.h" #include "replication/logical.h" #include "replication/reorderbuffer.h" +#include "replication/reorderbuffer_compression.h" #include "replication/slotsync.h" #include "replication/snapbuild.h" #include "storage/proc.h" @@ -298,6 +299,9 @@ StartupDecodingContext(List *output_plugin_options, ctx->fast_forward = fast_forward; + /* No spill files compression by default */ + ctx->spill_compression_method = REORDER_BUFFER_NO_COMPRESSION; + MemoryContextSwitchTo(old_context); return ctx; diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 17f8208000..9407466393 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -219,7 +219,7 @@ static const Size max_changes_in_memory = 4096; /* XXX for restore only */ int debug_logical_replication_streaming = DEBUG_LOGICAL_REP_STREAMING_BUFFERED; /* Compression strategy for spilled data. */ -int logical_decoding_spill_compression = REORDER_BUFFER_ZSTD_COMPRESSION; +int logical_decoding_spill_compression = REORDER_BUFFER_NO_COMPRESSION; /* --------------------------------------- * primary reorderbuffer support routines @@ -438,6 +438,15 @@ ReorderBufferFree(ReorderBuffer *rb) ReorderBufferCleanupSerializedTXNs(NameStr(MyReplicationSlot->data.name)); } +/* Returns spill files compression method */ +static inline uint8 +ReorderBufferSpillCompressionMethod(ReorderBuffer *rb) +{ + LogicalDecodingContext *ctx = rb->private_data; + + return ctx->spill_compression_method; +} + /* * Get an unused, possibly preallocated, ReorderBufferTXN. */ @@ -459,7 +468,7 @@ ReorderBufferGetTXN(ReorderBuffer *rb) txn->command_id = InvalidCommandId; txn->output_plugin_private = NULL; txn->compressor_state = ReorderBufferNewCompressorState(rb->context, - logical_decoding_spill_compression); + ReorderBufferSpillCompressionMethod(rb)); return txn; } @@ -498,7 +507,7 @@ ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) } ReorderBufferFreeCompressorState(rb->context, - logical_decoding_spill_compression, + ReorderBufferSpillCompressionMethod(rb), txn->compressor_state); /* Reset the toast hash */ @@ -4015,7 +4024,7 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn, } /* Inplace ReorderBuffer content compression before writing it on disk */ - ReorderBufferCompress(rb, &ondisk, logical_decoding_spill_compression, + ReorderBufferCompress(rb, &ondisk, ReorderBufferSpillCompressionMethod(rb), sz, txn->compressor_state); errno = 0; @@ -5697,3 +5706,60 @@ ReorderBufferDecompress(ReorderBuffer *rb, char *data, break; } } + +/* + * According to a given compression method (as string representation), returns + * the corresponding ReorderBufferCompressionMethod + */ +ReorderBufferCompressionMethod +ReorderBufferParseCompressionMethod(const char *method) +{ + if (pg_strcasecmp(method, "on") == 0) + return REORDER_BUFFER_PGLZ_COMPRESSION; + else if (pg_strcasecmp(method, "pglz") == 0) + return REORDER_BUFFER_PGLZ_COMPRESSION; + else if (pg_strcasecmp(method, "off") == 0) + return REORDER_BUFFER_NO_COMPRESSION; +#ifdef USE_LZ4 + else if (pg_strcasecmp(method, "lz4") == 0) + return REORDER_BUFFER_LZ4_COMPRESSION; +#endif +#ifdef USE_ZSTD + else if (pg_strcasecmp(method, "zstd") == 0) + return REORDER_BUFFER_ZSTD_COMPRESSION; +#endif + else + return REORDER_BUFFER_INVALID_COMPRESSION; +} + +/* + * Check whether the passed compression method is valid and report errors at + * elevel. + * + * As this validation is intended to be executed on subscriber side, then we + * actually don't know if the server running the publisher supports external + * compression libraries. We only check if the compression method is + * potentially supported. The real validation is done by the publisher when + * the replication starts, an error is then triggered if the compression method + * is not supported. + */ +void +ReorderBufferValidateCompressionMethod(const char *method, int elevel) +{ + bool valid = false; + char methods[5][5] = {"on", "off", "pglz", "lz4", "zstd"}; + + for (int i = 0; i < 5; i++) + { + if (pg_strcasecmp(method, methods[i]) == 0) + { + valid = true; + break; + } + } + + if (!valid) + ereport(elevel, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("compression method \"%s\" not valid", method))); +} diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 925dff9cc4..32b38b94dd 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -4021,7 +4021,8 @@ maybe_reread_subscription(void) newsub->passwordrequired != MySubscription->passwordrequired || strcmp(newsub->origin, MySubscription->origin) != 0 || newsub->owner != MySubscription->owner || - !equal(newsub->publications, MySubscription->publications)) + !equal(newsub->publications, MySubscription->publications) || + strcmp(newsub->spill_compression, MySubscription->spill_compression) != 0) { if (am_parallel_apply_worker()) ereport(LOG, @@ -4469,6 +4470,16 @@ set_stream_options(WalRcvStreamOptions *options, MyLogicalRepWorker->parallel_apply = false; } + if (server_version >= 180000 && + MySubscription->stream == LOGICALREP_STREAM_OFF && + MySubscription->spill_compression != NULL) + { + options->proto.logical.spill_compression = + pstrdup(MySubscription->spill_compression); + } + else + options->proto.logical.spill_compression = NULL; + options->proto.logical.twophase = false; options->proto.logical.origin = pstrdup(MySubscription->origin); } diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 00e7024563..521b646bb6 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -27,6 +27,7 @@ #include "replication/logicalproto.h" #include "replication/origin.h" #include "replication/pgoutput.h" +#include "replication/reorderbuffer_compression.h" #include "utils/builtins.h" #include "utils/inval.h" #include "utils/lsyscache.h" @@ -283,11 +284,13 @@ parse_output_parameters(List *options, PGOutputData *data) bool streaming_given = false; bool two_phase_option_given = false; bool origin_option_given = false; + bool spill_compression_option_given = false; data->binary = false; data->streaming = LOGICALREP_STREAM_OFF; data->messages = false; data->two_phase = false; + data->spill_compression_method = REORDER_BUFFER_NO_COMPRESSION; foreach(lc, options) { @@ -396,6 +399,28 @@ parse_output_parameters(List *options, PGOutputData *data) errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("unrecognized origin value: \"%s\"", origin)); } + else if (strcmp(defel->defname, "spill_compression") == 0) + { + uint8 method; + char *method_str; + + if (spill_compression_option_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + spill_compression_option_given = true; + + method_str = defGetString(defel); + method = ReorderBufferParseCompressionMethod(method_str); + + if (method == REORDER_BUFFER_INVALID_COMPRESSION) + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("invalid spill files compression method: \"%s\"", + method_str)); + + data->spill_compression_method = method; + } else elog(ERROR, "unrecognized pgoutput option: %s", defel->defname); } @@ -508,6 +533,9 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, data->publications = NIL; publications_valid = false; + /* Init spill files compression method */ + ctx->spill_compression_method = data->spill_compression_method; + /* * Register callback for pg_publication if we didn't already do that * during some previous call in this process. diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index d8c6330732..0b9e31ff2f 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -4850,6 +4850,7 @@ getSubscriptions(Archive *fout) int i_suboriginremotelsn; int i_subenabled; int i_subfailover; + int i_subspillcompression; int i, ntups; @@ -4922,10 +4923,17 @@ getSubscriptions(Archive *fout) if (fout->remoteVersion >= 170000) appendPQExpBufferStr(query, - " s.subfailover\n"); + " s.subfailover,\n"); else appendPQExpBuffer(query, - " false AS subfailover\n"); + " false AS subfailover,\n"); + + if (fout->remoteVersion >= 180000) + appendPQExpBufferStr(query, + " s.subspillcompression\n"); + else + appendPQExpBuffer(query, + " 'off' AS subspillcompression\n"); appendPQExpBufferStr(query, "FROM pg_subscription s\n"); @@ -4965,6 +4973,7 @@ getSubscriptions(Archive *fout) i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn"); i_subenabled = PQfnumber(res, "subenabled"); i_subfailover = PQfnumber(res, "subfailover"); + i_subspillcompression = PQfnumber(res, "subspillcompression"); subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo)); @@ -5011,6 +5020,8 @@ getSubscriptions(Archive *fout) pg_strdup(PQgetvalue(res, i, i_subenabled)); subinfo[i].subfailover = pg_strdup(PQgetvalue(res, i, i_subfailover)); + subinfo[i].subspillcompression = + pg_strdup(PQgetvalue(res, i, i_subspillcompression)); /* Decide whether we want to dump it */ selectDumpableObject(&(subinfo[i].dobj), fout); @@ -5259,6 +5270,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo) if (pg_strcasecmp(subinfo->suborigin, LOGICALREP_ORIGIN_ANY) != 0) appendPQExpBuffer(query, ", origin = %s", subinfo->suborigin); + if (strcmp(subinfo->subspillcompression, "off") != 0) + appendPQExpBuffer(query, ", spill_compression = %s", subinfo->subspillcompression); + appendPQExpBufferStr(query, ");\n"); /* diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index 9f907ed5ad..ecbf2c2e27 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -673,6 +673,7 @@ typedef struct _SubscriptionInfo char *suborigin; char *suboriginremotelsn; char *subfailover; + char *subspillcompression; } SubscriptionInfo; /* diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl index ac60829d68..4b7f75db8f 100644 --- a/src/bin/pg_dump/t/002_pg_dump.pl +++ b/src/bin/pg_dump/t/002_pg_dump.pl @@ -3001,9 +3001,9 @@ my %tests = ( create_order => 50, create_sql => 'CREATE SUBSCRIPTION sub2 CONNECTION \'dbname=doesnotexist\' PUBLICATION pub1 - WITH (connect = false, origin = none, streaming = off);', + WITH (connect = false, origin = none, spill_compression = on, streaming = off);', regexp => qr/^ - \QCREATE SUBSCRIPTION sub2 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub2', streaming = off, origin = none);\E + \QCREATE SUBSCRIPTION sub2 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub2', streaming = off, origin = none, spill_compression = on);\E /xm, like => { %full_runs, section_post_data => 1, }, }, diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c index 363a66e718..9735d0b099 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -6543,7 +6543,7 @@ describeSubscriptions(const char *pattern, bool verbose) printQueryOpt myopt = pset.popt; static const bool translate_columns[] = {false, false, false, false, false, false, false, false, false, false, false, false, false, false, - false}; + false, false}; if (pset.sversion < 100000) { @@ -6623,6 +6623,11 @@ describeSubscriptions(const char *pattern, bool verbose) appendPQExpBuffer(&buf, ", subskiplsn AS \"%s\"\n", gettext_noop("Skip LSN")); + + if (pset.sversion >= 180000) + appendPQExpBuffer(&buf, + ", subspillcompression AS \"%s\"\n", + gettext_noop("Spill files compression")); } /* Only display subscriptions in current database. */ diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index 1be0056af7..20969ec562 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -2280,7 +2280,8 @@ match_previous_words(int pattern_id, else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "SET", "(")) COMPLETE_WITH("binary", "disable_on_error", "failover", "origin", "password_required", "run_as_owner", "slot_name", - "streaming", "synchronous_commit", "two_phase"); + "spill_compression", "streaming", "synchronous_commit", + "two_phase"); /* ALTER SUBSCRIPTION SKIP ( */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "SKIP", "(")) COMPLETE_WITH("lsn"); @@ -3675,7 +3676,8 @@ match_previous_words(int pattern_id, COMPLETE_WITH("binary", "connect", "copy_data", "create_slot", "disable_on_error", "enabled", "failover", "origin", "password_required", "run_as_owner", "slot_name", - "streaming", "synchronous_commit", "two_phase"); + "spill_compression", "streaming", "synchronous_commit", + "two_phase"); /* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */ diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index b25f3fea56..2171a1f7f0 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -113,6 +113,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW /* Only publish data originating from the specified origin */ text suborigin BKI_DEFAULT(LOGICALREP_ORIGIN_ANY); + + /* Spill files compression algorithm */ + text subspillcompression BKI_FORCE_NOT_NULL; #endif } FormData_pg_subscription; @@ -157,6 +160,7 @@ typedef struct Subscription List *publications; /* List of publication names to subscribe to */ char *origin; /* Only publish data originating from the * specified origin */ + char *spill_compression; /* Spill files compression algorithm */ } Subscription; /* Disallow streaming in-progress transactions. */ diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index aff38e8d04..75c17866c3 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -112,6 +112,8 @@ typedef struct LogicalDecodingContext /* Do we need to process any change in fast_forward mode? */ bool processing_required; + /* Compression method used to compress spill files */ + uint8 spill_compression_method; } LogicalDecodingContext; diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h index 89f94e1147..eabcca62af 100644 --- a/src/include/replication/pgoutput.h +++ b/src/include/replication/pgoutput.h @@ -33,6 +33,7 @@ typedef struct PGOutputData bool messages; bool two_phase; bool publish_no_origin; + uint8 spill_compression_method; } PGOutputData; #endif /* PGOUTPUT_H */ diff --git a/src/include/replication/reorderbuffer_compression.h b/src/include/replication/reorderbuffer_compression.h index 240c188f00..1df508be91 100644 --- a/src/include/replication/reorderbuffer_compression.h +++ b/src/include/replication/reorderbuffer_compression.h @@ -24,6 +24,7 @@ /* ReorderBuffer on disk compression algorithms */ typedef enum ReorderBufferCompressionMethod { + REORDER_BUFFER_INVALID_COMPRESSION, REORDER_BUFFER_NO_COMPRESSION, REORDER_BUFFER_PGLZ_COMPRESSION, REORDER_BUFFER_LZ4_COMPRESSION, @@ -117,6 +118,9 @@ typedef struct ZSTDStreamingCompressorState extern void *lz4_NewCompressorState(MemoryContext context); extern void lz4_FreeCompressorState(MemoryContext context, void *compressor_state); +extern ReorderBufferCompressionMethod ReorderBufferParseCompressionMethod(const char *method); +extern void ReorderBufferValidateCompressionMethod(const char *method, + int elevel); extern void lz4_StreamingCompressData(MemoryContext context, char *src, Size src_size, char *dst, Size *dst_size, void *compressor_state); diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 132e789948..b759b5807d 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -186,6 +186,7 @@ typedef struct * prepare time */ char *origin; /* Only publish data originating from the * specified origin */ + char *spill_compression; /* Spill files compression algo */ } logical; } proto; } WalRcvStreamOptions; diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 1443e1d929..029d42f358 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub4 SET (origin = any); \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) DROP SUBSCRIPTION regress_testsub3; @@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar'; ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false); @@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname'); ALTER SUBSCRIPTION regress_testsub SET (password_required = false); ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | off | dbname=regress_doesnotexist2 | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | off | dbname=regress_doesnotexist2 | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET (password_required = true); @@ -176,10 +176,10 @@ ERROR: unrecognized subscription parameter: "create_slot" -- ok ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345'); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/12345 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/12345 | off (1 row) -- ok - with lsn = NONE @@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE); ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0'); ERROR: invalid WAL location (LSN): 0/0 \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/0 | off (1 row) BEGIN; @@ -222,11 +222,15 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = local); ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar); ERROR: invalid value for parameter "synchronous_commit": "foobar" HINT: Available values: local, remote_write, remote_apply, on, off. +ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = pglz); +ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = off); +ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = foobar); +ERROR: compression method "foobar" not valid \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+---------- - regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | local | dbname=regress_doesnotexist2 | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+------------------------- + regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | local | dbname=regress_doesnotexist2 | 0/0 | off (1 row) -- rename back to keep the rest simple @@ -255,19 +259,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET (binary = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) DROP SUBSCRIPTION regress_testsub; @@ -279,27 +283,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) -- fail - publication already exists @@ -314,10 +318,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); ERROR: publication "testpub1" is already in subscription "regress_testsub" \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) -- fail - publication used more than once @@ -332,10 +336,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub" -- ok - delete publications ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) DROP SUBSCRIPTION regress_testsub; @@ -371,19 +375,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) -- we can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -393,10 +397,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -409,18 +413,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+---------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN | Spill files compression +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | off | dbname=regress_doesnotexist | 0/0 | off (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 007c9e7037..368696f430 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -140,6 +140,10 @@ ALTER SUBSCRIPTION regress_testsub RENAME TO regress_testsub_foo; ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = local); ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar); +ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = pglz); +ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = off); +ALTER SUBSCRIPTION regress_testsub_foo SET (spill_compression = foobar); + \dRs+ -- rename back to keep the rest simple -- 2.43.0