Remove forward_changesets from pglogical_output

ringerc · ringerc · commit 1b28c01d8f96 · 2016-01-20T13:25:23.000+08:00
Use a hook if you want this functionality.
diff --git a/contrib/pglogical_output/README.md b/contrib/pglogical_output/README.md
@@ -222,42 +222,32 @@ It's possible to use `pglogical_output` to cascade replication between multiple
 PostgreSQL servers, in combination with an appropriate client to apply the
 changes to the downstreams.
 
-There are three forwarding modes:
+There are two forwarding modes:
 
 * Forward everything. Transactions are replicated whether they were made directly
   on the immediate upstream or some other node upstream of it. All rows from all
   transactions are sent.
 
-  Selected by setting `forward_changesets` to true (default) and not setting a
-  row or transaction filter hook.
-
-* No forwarding. Only transactions applied immediately on the upstream node are
-  forwarded. Transactions with any non-local origin are skipped. All rows from
-  locally originated transactions are sent.
-
-  Selected by setting `forward_changesets` to false. Remember to confirm by
-  checking the startup reply message.
+  Selected by not setting a row or transaction filter hook.
 
 * Filtered forwarding. Transactions are replicated unless a client-supplied
   transaction filter hook says to skip this transaction. Row changes are
   replicated unless the client-supplied row filter hook (if provided) says to
   skip that row.
 
-  Selected by setting `forward_changesets` to `true` and installing a
-  transaction and/or row filter hook (see "hooks").
-
-If the upstream server is 9.5 or newer and `forward_changesets` is enabled, the
-server will enable changeset origin information. It will set
-`forward_changeset_origins` to true in the startup reply message to indicate
-this. It will then send changeset origin messages after the `BEGIN` for each
-transaction, per the protocol documentation. Origin messages are omitted for
-transactions originating directly on the immediate upstream to save bandwidth.
-If `forward_changeset_origins` is true then transactions without an origin are
-always from the immediate upstream that’s running the decoding plugin.
-
-Note that changeset forwarding may be forced to on if not requested by some
-servers, so the client _should_ check the forward_changesets and
-`forward_changeset_origins` params in the startup reply message.
+  Selected by installing a transaction and/or row filter hook (see "hooks").
+
+If the upstream server is 9.5 or newer the server will enable changeset origin
+information. It will set `forward_changeset_origins` to true in the startup
+reply message to indicate this. It will then send changeset origin messages
+after the `BEGIN` for each transaction, per the protocol documentation. Origin
+messages are omitted for transactions originating directly on the immediate
+upstream to save bandwidth.  If `forward_changeset_origins` is true then
+transactions without an origin are always from the immediate upstream that’s
+running the decoding plugin.
+
+Note that 9.4 servers can't expose replication origin information so they pass
+zero to the row filter hook and don't call the transaction filter hook.
 
 Clients may use this facility to form arbitrarily complex topologies when
 combined with hooks to determine which transactions are forwarded. An obvious
diff --git a/contrib/pglogical_output/doc/protocol.txt b/contrib/pglogical_output/doc/protocol.txt
@@ -393,8 +393,7 @@ Since all parameter values are sent as strings, the value types given below spec
 |binary|_set of parameters, specified separately_|See “_the __‘binary’__ parameters_” below, and “_Parameters relating to exchange of binary values_”
 |database_encoding|string|The native text encoding of the database the plugin is running in
 |encoding|string|Field values for textual data will be in this encoding in native protocol text, binary or internal representation. For the native protocol this is currently always the same as `database_encoding`. For text-mode json protocol this is always the same as `client_encoding`.
-|forward_changesets|bool|Specifies that all transactions, not just those originating on the upstream, will be forwarded. See “_Changeset forwarding_”.
-|forward_changeset_origins|bool|Tells the client that the server will send changeset origin information. Independent of forward_changesets. See “_Changeset forwarding_” for details.
+|forward_changeset_origins|bool|Tells the client that the server will send changeset origin information. See “_Changeset forwarding_” for details.
 |no_txinfo|bool|Requests that variable transaction info such as XIDs, LSNs, and timestamps be omitted from output. Mainly for tests. Currently ignored for protos other than json.
 |===
 
@@ -472,7 +471,6 @@ Because these versions are expected to be incremented, to make it clear that the
 |*Key*|*Type*|*Default*|*Notes*
 
 |expected_encoding|string|null|The text encoding the downstream expects field values to be in. Applies to text, binary and internal representations of field values in native format. Has no effect on other protocol content. If specified, the upstream must honour it. For json protocol, must be unset or match `client_encoding`. (Current plugin versions ERROR if this is set for the native protocol and not equal to the upstream database's encoding).
-|forward_changesets|bool|false|Request that all transactions, not just those originating on the upstream, be forwarded. See “_Changeset forwarding_”.
 |want_coltypes|boolean|false|The client wants to receive data type information about columns.
 |===
 
diff --git a/contrib/pglogical_output/expected/basic_json.out b/contrib/pglogical_output/expected/basic_json.out
@@ -83,16 +83,15 @@ SELECT * FROM get_startup_params();
  coltypes                         | "f"
  database_encoding                | "UTF8"
  encoding                         | "UTF8"
- forward_changeset_origins        | "f"
- forward_changesets               | "f"
+ forward_changeset_origins        | "t"
  hooks.row_filter_enabled         | "f"
  hooks.shutdown_hook_enabled      | "f"
  hooks.startup_hook_enabled       | "f"
  hooks.transaction_filter_enabled | "f"
  max_proto_version                | "1"
  min_proto_version                | "1"
  no_txinfo                        | "t"
-(19 rows)
+(18 rows)
 
 SELECT * FROM get_queued_data();
                                                                              data                                                                             
diff --git a/contrib/pglogical_output/expected/basic_json_1.out b/contrib/pglogical_output/expected/basic_json_1.out
@@ -68,7 +68,7 @@ FROM pg_logical_slot_peek_changes('regression_slot',
 	'no_txinfo', 't');
                                                                                                                                                                                                                                                                                                                                                                                          data                                                                                                                                                                                                                                                                                                                                                                                         
 --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
- {"action":"S", "params": {"max_proto_version":"1","min_proto_version":"1","coltypes":"f","pg_version_num":"90405","pg_version":"9.4.5","pg_catversion":"201409291","database_encoding":"UTF8","encoding":"UTF8","forward_changesets":"t","forward_changeset_origins":"f","binary.internal_basetypes":"f","binary.binary_basetypes":"f","binary.basetypes_major_version":"904","binary.sizeof_int":"4","binary.sizeof_long":"8","binary.sizeof_datum":"8","binary.maxalign":"8","binary.bigendian":"f","binary.float4_byval":"t","binary.float8_byval":"t","binary.integer_datetimes":"t","binary.binary_pg_version":"904","no_txinfo":"t","hooks.startup_hook_enabled":"f","hooks.shutdown_hook_enabled":"f","hooks.row_filter_enabled":"f","hooks.transaction_filter_enabled":"f"}}
+ {"action":"S", "params": {"max_proto_version":"1","min_proto_version":"1","coltypes":"f","pg_version_num":"90405","pg_version":"9.4.5","pg_catversion":"201409291","database_encoding":"UTF8","encoding":"UTF8","forward_changeset_origins":"t","binary.internal_basetypes":"f","binary.binary_basetypes":"f","binary.basetypes_major_version":"904","binary.sizeof_int":"4","binary.sizeof_long":"8","binary.sizeof_datum":"8","binary.maxalign":"8","binary.bigendian":"f","binary.float4_byval":"t","binary.float8_byval":"t","binary.integer_datetimes":"t","binary.binary_pg_version":"904","no_txinfo":"t","hooks.startup_hook_enabled":"f","hooks.shutdown_hook_enabled":"f","hooks.row_filter_enabled":"f","hooks.transaction_filter_enabled":"f"}}
  {"action":"B", "has_catalog_changes":"f"}
  {"action":"I","relation":["public","demo"],"newtuple":{"seq":1,"tx":"textval","ts":null,"jsb":null,"js":null,"ba":null}}
  {"action":"C"}
diff --git a/contrib/pglogical_output/expected/hooks_json.out b/contrib/pglogical_output/expected/hooks_json.out
@@ -64,16 +64,15 @@ SELECT * FROM get_startup_params();
  coltypes                         | "f"
  database_encoding                | "UTF8"
  encoding                         | "UTF8"
- forward_changeset_origins        | "f"
- forward_changesets               | "f"
+ forward_changeset_origins        | "t"
  hooks.row_filter_enabled         | "t"
  hooks.shutdown_hook_enabled      | "t"
  hooks.startup_hook_enabled       | "t"
  hooks.transaction_filter_enabled | "t"
  max_proto_version                | "1"
  min_proto_version                | "1"
  no_txinfo                        | "t"
-(19 rows)
+(18 rows)
 
 SELECT * FROM get_queued_data();
                                       data                                       
@@ -134,16 +133,15 @@ SELECT * FROM get_startup_params();
  coltypes                         | "f"
  database_encoding                | "UTF8"
  encoding                         | "UTF8"
- forward_changeset_origins        | "f"
- forward_changesets               | "f"
+ forward_changeset_origins        | "t"
  hooks.row_filter_enabled         | "t"
  hooks.shutdown_hook_enabled      | "t"
  hooks.startup_hook_enabled       | "t"
  hooks.transaction_filter_enabled | "t"
  max_proto_version                | "1"
  min_proto_version                | "1"
  no_txinfo                        | "t"
-(19 rows)
+(18 rows)
 
 SELECT * FROM get_queued_data();
                                       data                                      
diff --git a/contrib/pglogical_output/expected/hooks_json_1.out b/contrib/pglogical_output/expected/hooks_json_1.out
@@ -49,7 +49,7 @@ FROM pg_logical_slot_peek_changes('regression_slot',
 	'no_txinfo', 't');
                                                                                                                                                                                                                                                                                                                                                                                          data                                                                                                                                                                                                                                                                                                                                                                                         
 --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
- {"action":"S", "params": {"max_proto_version":"1","min_proto_version":"1","coltypes":"f","pg_version_num":"90405","pg_version":"9.4.5","pg_catversion":"201409291","database_encoding":"UTF8","encoding":"UTF8","forward_changesets":"t","forward_changeset_origins":"f","binary.internal_basetypes":"f","binary.binary_basetypes":"f","binary.basetypes_major_version":"904","binary.sizeof_int":"4","binary.sizeof_long":"8","binary.sizeof_datum":"8","binary.maxalign":"8","binary.bigendian":"f","binary.float4_byval":"t","binary.float8_byval":"t","binary.integer_datetimes":"t","binary.binary_pg_version":"904","no_txinfo":"t","hooks.startup_hook_enabled":"t","hooks.shutdown_hook_enabled":"t","hooks.row_filter_enabled":"t","hooks.transaction_filter_enabled":"t"}}
+ {"action":"S", "params": {"max_proto_version":"1","min_proto_version":"1","coltypes":"f","pg_version_num":"90405","pg_version":"9.4.5","pg_catversion":"201409291","database_encoding":"UTF8","encoding":"UTF8","forward_changeset_origins":"t","binary.internal_basetypes":"f","binary.binary_basetypes":"f","binary.basetypes_major_version":"904","binary.sizeof_int":"4","binary.sizeof_long":"8","binary.sizeof_datum":"8","binary.maxalign":"8","binary.bigendian":"f","binary.float4_byval":"t","binary.float8_byval":"t","binary.integer_datetimes":"t","binary.binary_pg_version":"904","no_txinfo":"t","hooks.startup_hook_enabled":"t","hooks.shutdown_hook_enabled":"t","hooks.row_filter_enabled":"t","hooks.transaction_filter_enabled":"t"}}
  {"action":"B", "has_catalog_changes":"f"}
  {"action":"C"}
  {"action":"B", "has_catalog_changes":"f"}
@@ -89,7 +89,7 @@ FROM pg_logical_slot_peek_changes('regression_slot',
 	'no_txinfo', 't');
                                                                                                                                                                                                                                                                                                                                                                                          data                                                                                                                                                                                                                                                                                                                                                                                         
 --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
- {"action":"S", "params": {"max_proto_version":"1","min_proto_version":"1","coltypes":"f","pg_version_num":"90405","pg_version":"9.4.5","pg_catversion":"201409291","database_encoding":"UTF8","encoding":"UTF8","forward_changesets":"t","forward_changeset_origins":"f","binary.internal_basetypes":"f","binary.binary_basetypes":"f","binary.basetypes_major_version":"904","binary.sizeof_int":"4","binary.sizeof_long":"8","binary.sizeof_datum":"8","binary.maxalign":"8","binary.bigendian":"f","binary.float4_byval":"t","binary.float8_byval":"t","binary.integer_datetimes":"t","binary.binary_pg_version":"904","no_txinfo":"t","hooks.startup_hook_enabled":"t","hooks.shutdown_hook_enabled":"t","hooks.row_filter_enabled":"t","hooks.transaction_filter_enabled":"t"}}
+ {"action":"S", "params": {"max_proto_version":"1","min_proto_version":"1","coltypes":"f","pg_version_num":"90405","pg_version":"9.4.5","pg_catversion":"201409291","database_encoding":"UTF8","encoding":"UTF8","forward_changeset_origins":"t","binary.internal_basetypes":"f","binary.binary_basetypes":"f","binary.basetypes_major_version":"904","binary.sizeof_int":"4","binary.sizeof_long":"8","binary.sizeof_datum":"8","binary.maxalign":"8","binary.bigendian":"f","binary.float4_byval":"t","binary.float8_byval":"t","binary.integer_datetimes":"t","binary.binary_pg_version":"904","no_txinfo":"t","hooks.startup_hook_enabled":"t","hooks.shutdown_hook_enabled":"t","hooks.row_filter_enabled":"t","hooks.transaction_filter_enabled":"t"}}
  {"action":"B", "has_catalog_changes":"f"}
  {"action":"I","relation":["public","test_filter"],"newtuple":{"id":1}}
  {"action":"I","relation":["public","test_filter"],"newtuple":{"id":2}}
diff --git a/contrib/pglogical_output/pglogical_config.c b/contrib/pglogical_output/pglogical_config.c
@@ -70,7 +70,6 @@ enum {
 	PARAM_BINARY_WANT_BINARY_BASETYPES,
 	PARAM_BINARY_BASETYPES_MAJOR_VERSION,
 	PARAM_PG_VERSION,
-	PARAM_FORWARD_CHANGESETS,
 	PARAM_HOOKS_SETUP_FUNCTION,
 	PARAM_NO_TXINFO
 } OutputPluginParamKey;
@@ -97,7 +96,6 @@ static OutputPluginParam param_lookup[] = {
 	{"binary.want_binary_basetypes", PARAM_BINARY_WANT_BINARY_BASETYPES},
 	{"binary.basetypes_major_version", PARAM_BINARY_BASETYPES_MAJOR_VERSION},
 	{"pg_version", PARAM_PG_VERSION},
-	{"forward_changesets", PARAM_FORWARD_CHANGESETS},
 	{"hooks.setup_function", PARAM_HOOKS_SETUP_FUNCTION},
 	{"no_txinfo", PARAM_NO_TXINFO},
 	{NULL, PARAM_UNRECOGNISED}
@@ -210,18 +208,6 @@ process_parameters_v1(List *options, PGLogicalOutputData *data)
 				data->client_pg_version = DatumGetUInt32(val);
 				break;
 
-			case PARAM_FORWARD_CHANGESETS:
-				/*
-				 * Check to see if the client asked for changeset forwarding
-				 *
-				 * Note that we cannot support this on 9.4. We'll tell the client
-				 * in the startup reply message.
-				 */
-				val = get_param_value(elem, false, OUTPUT_PARAM_TYPE_BOOL);
-				data->client_forward_changesets_set = true;
-				data->client_forward_changesets = DatumGetBool(val);
-				break;
-
 			case PARAM_BINARY_WANT_INTERNAL_BASETYPES:
 				/* check if we want to use internal data representation */
 				val = get_param_value(elem, false, OUTPUT_PARAM_TYPE_BOOL);
@@ -444,8 +430,6 @@ prepare_startup_message(PGLogicalOutputData *data)
 
 	l = add_startup_msg_s(l, "encoding", (char*)pg_encoding_to_char(data->field_datum_encoding));
 
-	l = add_startup_msg_b(l, "forward_changesets",
-			data->forward_changesets);
 	l = add_startup_msg_b(l, "forward_changeset_origins",
 			data->forward_changeset_origins);
 
diff --git a/contrib/pglogical_output/pglogical_output.c b/contrib/pglogical_output/pglogical_output.c
diff --git a/contrib/pglogical_output/pglogical_output.h b/contrib/pglogical_output/pglogical_output.h
