From 2c7ea57e56ca5f668c32d4266e0a3e45b455bef5 Mon Sep 17 00:00:00 2001 From: Tomas Vondra Date: Thu, 7 Apr 2022 18:13:13 +0200 Subject: Revert "Logical decoding of sequences" This reverts a sequence of commits, implementing features related to logical decoding and replication of sequences: - 0da92dc530c9251735fc70b20cd004d9630a1266 - 80901b32913ffa59bf157a4d88284b2b3a7511d9 - b779d7d8fdae088d70da5ed9fcd8205035676df3 - d5ed9da41d96988d905b49bebb273a9b2d6e2915 - a180c2b34de0989269fdb819bff241a249bf5380 - 75b1521dae1ff1fde17fda2e30e591f2e5d64b6a - 2d2232933b02d9396113662e44dca5f120d6830e - 002c9dd97a0c874fd1693a570383e2dd38cd40d5 - 05843b1aa49df2ecc9b97c693b755bd1b6f856a9 The implementation has issues, mostly due to combining transactional and non-transactional behavior of sequences. It's not clear how this could be fixed, but it'll require reworking significant part of the patch. Discussion: https://postgr.es/m/95345a19-d508-63d1-860a-f5c2f41e8d40@enterprisedb.com --- contrib/test_decoding/test_decoding.c | 87 ----------------------------------- 1 file changed, 87 deletions(-) (limited to 'contrib/test_decoding/test_decoding.c') diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c index c7a87f5fe5b..08d366a594e 100644 --- a/contrib/test_decoding/test_decoding.c +++ b/contrib/test_decoding/test_decoding.c @@ -35,7 +35,6 @@ typedef struct bool include_timestamp; bool skip_empty_xacts; bool only_local; - bool include_sequences; } TestDecodingData; /* @@ -77,10 +76,6 @@ static void pg_decode_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message); -static void pg_decode_sequence(LogicalDecodingContext *ctx, - ReorderBufferTXN *txn, XLogRecPtr sequence_lsn, - Relation rel, bool transactional, - int64 last_value, int64 log_cnt, bool is_called); static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid); @@ -121,10 +116,6 @@ static void pg_decode_stream_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, const char *message); -static void pg_decode_stream_sequence(LogicalDecodingContext *ctx, - ReorderBufferTXN *txn, XLogRecPtr sequence_lsn, - Relation rel, bool transactional, - int64 last_value, int64 log_cnt, bool is_called); static void pg_decode_stream_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelations, Relation relations[], @@ -150,7 +141,6 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->filter_by_origin_cb = pg_decode_filter; cb->shutdown_cb = pg_decode_shutdown; cb->message_cb = pg_decode_message; - cb->sequence_cb = pg_decode_sequence; cb->filter_prepare_cb = pg_decode_filter_prepare; cb->begin_prepare_cb = pg_decode_begin_prepare_txn; cb->prepare_cb = pg_decode_prepare_txn; @@ -163,7 +153,6 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->stream_commit_cb = pg_decode_stream_commit; cb->stream_change_cb = pg_decode_stream_change; cb->stream_message_cb = pg_decode_stream_message; - cb->stream_sequence_cb = pg_decode_stream_sequence; cb->stream_truncate_cb = pg_decode_stream_truncate; } @@ -184,7 +173,6 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, data->include_xids = true; data->include_timestamp = false; data->skip_empty_xacts = false; - data->include_sequences = true; data->only_local = false; ctx->output_plugin_private = data; @@ -277,17 +265,6 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, errmsg("could not parse value \"%s\" for parameter \"%s\"", strVal(elem->arg), elem->defname))); } - else if (strcmp(elem->defname, "include-sequences") == 0) - { - - if (elem->arg == NULL) - data->include_sequences = false; - else if (!parse_bool(strVal(elem->arg), &data->include_sequences)) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("could not parse value \"%s\" for parameter \"%s\"", - strVal(elem->arg), elem->defname))); - } else { ereport(ERROR, @@ -779,38 +756,6 @@ pg_decode_message(LogicalDecodingContext *ctx, OutputPluginWrite(ctx, true); } -static void -pg_decode_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, - XLogRecPtr sequence_lsn, Relation rel, - bool transactional, - int64 last_value, int64 log_cnt, bool is_called) -{ - TestDecodingData *data = ctx->output_plugin_private; - TestDecodingTxnData *txndata = txn->output_plugin_private; - - if (!data->include_sequences) - return; - - /* output BEGIN if we haven't yet, but only for the transactional case */ - if (transactional) - { - if (data->skip_empty_xacts && !txndata->xact_wrote_changes) - { - pg_output_begin(ctx, data, txn, false); - } - txndata->xact_wrote_changes = true; - } - - OutputPluginPrepareWrite(ctx, true); - appendStringInfoString(ctx->out, "sequence "); - appendStringInfoString(ctx->out, - quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(rel))), - RelationGetRelationName(rel))); - appendStringInfo(ctx->out, ": transactional:%d last_value: " INT64_FORMAT " log_cnt: " INT64_FORMAT " is_called:%d", - transactional, last_value, log_cnt, is_called); - OutputPluginWrite(ctx, true); -} - static void pg_decode_stream_start(LogicalDecodingContext *ctx, ReorderBufferTXN *txn) @@ -1010,38 +955,6 @@ pg_decode_stream_message(LogicalDecodingContext *ctx, OutputPluginWrite(ctx, true); } -static void -pg_decode_stream_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, - XLogRecPtr sequence_lsn, Relation rel, - bool transactional, - int64 last_value, int64 log_cnt, bool is_called) -{ - TestDecodingData *data = ctx->output_plugin_private; - TestDecodingTxnData *txndata = txn->output_plugin_private; - - if (!data->include_sequences) - return; - - /* output BEGIN if we haven't yet, but only for the transactional case */ - if (transactional) - { - if (data->skip_empty_xacts && !txndata->xact_wrote_changes) - { - pg_output_begin(ctx, data, txn, false); - } - txndata->xact_wrote_changes = true; - } - - OutputPluginPrepareWrite(ctx, true); - appendStringInfoString(ctx->out, "streaming sequence "); - appendStringInfoString(ctx->out, - quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(rel))), - RelationGetRelationName(rel))); - appendStringInfo(ctx->out, ": transactional:%d last_value: " INT64_FORMAT " log_cnt: " INT64_FORMAT " is_called:%d", - transactional, last_value, log_cnt, is_called); - OutputPluginWrite(ctx, true); -} - /* * In streaming mode, we don't display the detailed information of Truncate. * See pg_decode_stream_change. -- cgit v1.2.3