src/test/subscription/t/026_stats.pl - cloudberry - Git at Google


 # Copyright (c) 2021-2023, PostgreSQL Global Development Group

 # Tests for subscription stats.
 use strict;
 use warnings;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
 use Test::More;

 # Create publisher node.
 my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
 $node_publisher->init(allows_streaming => 'logical');
 $node_publisher->start;

 # Create subscriber node.
 my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
 $node_subscriber->init(allows_streaming => 'logical');
 $node_subscriber->start;


 sub create_sub_pub_w_errors
 {
 	my ($node_publisher, $node_subscriber, $db, $table_name) = @_;
 	# Initial table setup on both publisher and subscriber. On subscriber we
 	# create the same tables but with primary keys. Also, insert some data that
 	# will conflict with the data replicated from publisher later.
 	$node_publisher->safe_psql(
 		$db,
 		qq[
 	BEGIN;
 	CREATE TABLE $table_name(a int);
 	INSERT INTO $table_name VALUES (1);
 	COMMIT;
 	]);
 	$node_subscriber->safe_psql(
 		$db,
 		qq[
 	BEGIN;
 	CREATE TABLE $table_name(a int primary key);
 	INSERT INTO $table_name VALUES (1);
 	COMMIT;
 	]);

 	# Set up publication.
 	my $pub_name = $table_name . '_pub';
 	my $publisher_connstr = $node_publisher->connstr . qq( dbname=$db);

 	$node_publisher->safe_psql($db,
 		qq(CREATE PUBLICATION $pub_name FOR TABLE $table_name));

 	# Create subscription. The tablesync for table on subscription will enter into
 	# infinite error loop due to violating the unique constraint.
 	my $sub_name = $table_name . '_sub';
 	$node_subscriber->safe_psql($db,
 		qq(CREATE SUBSCRIPTION $sub_name CONNECTION '$publisher_connstr' PUBLICATION $pub_name)
 	);

 	$node_publisher->wait_for_catchup($sub_name);

 	# Wait for the tablesync error to be reported.
 	$node_subscriber->poll_query_until(
 		$db,
 		qq[
 	SELECT sync_error_count > 0
 	FROM pg_stat_subscription_stats
 	WHERE subname = '$sub_name'
 	])
 	  or die
 	  qq(Timed out while waiting for tablesync errors for subscription '$sub_name');

 	# Truncate test_tab1 so that tablesync worker can continue.
 	$node_subscriber->safe_psql($db, qq(TRUNCATE $table_name));

 	# Wait for initial tablesync to finish.
 	$node_subscriber->poll_query_until(
 		$db,
 		qq[
 	SELECT count(1) = 1 FROM pg_subscription_rel
 	WHERE srrelid = '$table_name'::regclass AND srsubstate in ('r', 's')
 	])
 	  or die
 	  qq(Timed out while waiting for subscriber to synchronize data for table '$table_name'.);

 	# Check test table on the subscriber has one row.
 	my $result =
 	  $node_subscriber->safe_psql($db, qq(SELECT a FROM $table_name));
 	is($result, qq(1), qq(Check that table '$table_name' now has 1 row.));

 	# Insert data to test table on the publisher, raising an error on the
 	# subscriber due to violation of the unique constraint on test table.
 	$node_publisher->safe_psql($db, qq(INSERT INTO $table_name VALUES (1)));

 	# Wait for the apply error to be reported.
 	$node_subscriber->poll_query_until(
 		$db,
 		qq[
 	SELECT apply_error_count > 0
 	FROM pg_stat_subscription_stats
 	WHERE subname = '$sub_name'
 	])
 	  or die
 	  qq(Timed out while waiting for apply error for subscription '$sub_name');

 	# Truncate test table so that apply worker can continue.
 	$node_subscriber->safe_psql($db, qq(TRUNCATE $table_name));

 	return ($pub_name, $sub_name);
 }

 my $db = 'postgres';

 # There shouldn't be any subscription errors before starting logical replication.
 my $result = $node_subscriber->safe_psql($db,
 	qq(SELECT count(1) FROM pg_stat_subscription_stats));
 is($result, qq(0),
 	'Check that there are no subscription errors before starting logical replication.'
 );

 # Create the publication and subscription with sync and apply errors
 my $table1_name = 'test_tab1';
 my ($pub1_name, $sub1_name) =
   create_sub_pub_w_errors($node_publisher, $node_subscriber, $db,
 	$table1_name);

 # Apply and Sync errors are > 0 and reset timestamp is NULL
 is( $node_subscriber->safe_psql(
 		$db,
 		qq(SELECT apply_error_count > 0,
 	sync_error_count > 0,
 	stats_reset IS NULL
 	FROM pg_stat_subscription_stats
 	WHERE subname = '$sub1_name')
 	),
 	qq(t|t|t),
 	qq(Check that apply errors and sync errors are both > 0 and stats_reset is NULL for subscription '$sub1_name'.)
 );

 # Reset a single subscription
 $node_subscriber->safe_psql($db,
 	qq(SELECT pg_stat_reset_subscription_stats((SELECT subid FROM pg_stat_subscription_stats WHERE subname = '$sub1_name')))
 );

 # Apply and Sync errors are 0 and stats reset is not NULL
 is( $node_subscriber->safe_psql(
 		$db,
 		qq(SELECT apply_error_count = 0,
 	sync_error_count = 0,
 	stats_reset IS NOT NULL
 	FROM pg_stat_subscription_stats
 	WHERE subname = '$sub1_name')
 	),
 	qq(t|t|t),
 	qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL after reset for subscription '$sub1_name'.)
 );

 # Get reset timestamp
 my $reset_time1 = $node_subscriber->safe_psql($db,
 	qq(SELECT stats_reset FROM pg_stat_subscription_stats WHERE subname = '$sub1_name')
 );

 # Reset single sub again
 $node_subscriber->safe_psql(
 	$db,
 	qq(SELECT pg_stat_reset_subscription_stats((SELECT subid FROM
 	pg_stat_subscription_stats WHERE subname = '$sub1_name')))
 );

 # check reset timestamp is newer after reset
 is( $node_subscriber->safe_psql(
 		$db,
 		qq(SELECT stats_reset > '$reset_time1'::timestamptz FROM
 	pg_stat_subscription_stats WHERE subname = '$sub1_name')
 	),
 	qq(t),
 	qq(Check reset timestamp for '$sub1_name' is newer after second reset.));

 # Make second subscription and publication
 my $table2_name = 'test_tab2';
 my ($pub2_name, $sub2_name) =
   create_sub_pub_w_errors($node_publisher, $node_subscriber, $db,
 	$table2_name);

 # Apply and Sync errors are > 0 and reset timestamp is NULL
 is( $node_subscriber->safe_psql(
 		$db,
 		qq(SELECT apply_error_count > 0,
 	sync_error_count > 0,
 	stats_reset IS NULL
 	FROM pg_stat_subscription_stats
 	WHERE subname = '$sub2_name')
 	),
 	qq(t|t|t),
 	qq(Confirm that apply errors and sync errors are both > 0 and stats_reset is NULL for sub '$sub2_name'.)
 );

 # Reset all subscriptions
 $node_subscriber->safe_psql($db,
 	qq(SELECT pg_stat_reset_subscription_stats(NULL)));

 # Apply and Sync errors are 0 and stats reset is not NULL
 is( $node_subscriber->safe_psql(
 		$db,
 		qq(SELECT apply_error_count = 0,
 	sync_error_count = 0,
 	stats_reset IS NOT NULL
 	FROM pg_stat_subscription_stats
 	WHERE subname = '$sub1_name')
 	),
 	qq(t|t|t),
 	qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL for sub '$sub1_name' after reset.)
 );

 is( $node_subscriber->safe_psql(
 		$db,
 		qq(SELECT apply_error_count = 0,
 	sync_error_count = 0,
 	stats_reset IS NOT NULL
 	FROM pg_stat_subscription_stats
 	WHERE subname = '$sub2_name')
 	),
 	qq(t|t|t),
 	qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL for sub '$sub2_name' after reset.)
 );

 $reset_time1 = $node_subscriber->safe_psql($db,
 	qq(SELECT stats_reset FROM pg_stat_subscription_stats WHERE subname = '$sub1_name')
 );
 my $reset_time2 = $node_subscriber->safe_psql($db,
 	qq(SELECT stats_reset FROM pg_stat_subscription_stats WHERE subname = '$sub2_name')
 );

 # Reset all subscriptions
 $node_subscriber->safe_psql($db,
 	qq(SELECT pg_stat_reset_subscription_stats(NULL)));

 # check reset timestamp for sub1 is newer after reset
 is( $node_subscriber->safe_psql(
 		$db,
 		qq(SELECT stats_reset > '$reset_time1'::timestamptz FROM
 	pg_stat_subscription_stats WHERE subname = '$sub1_name')
 	),
 	qq(t),
 	qq(Confirm that reset timestamp for '$sub1_name' is newer after second reset.)
 );

 # check reset timestamp for sub2 is newer after reset
 is( $node_subscriber->safe_psql(
 		$db,
 		qq(SELECT stats_reset > '$reset_time2'::timestamptz FROM
 	pg_stat_subscription_stats WHERE subname = '$sub2_name')
 	),
 	qq(t),
 	qq(Confirm that reset timestamp for '$sub2_name' is newer after second reset.)
 );

 # Get subscription 1 oid
 my $sub1_oid = $node_subscriber->safe_psql($db,
 	qq(SELECT oid FROM pg_subscription WHERE subname = '$sub1_name'));

 # Drop subscription 1
 $node_subscriber->safe_psql($db, qq(DROP SUBSCRIPTION $sub1_name));

 # Subscription stats for sub1 should be gone
 is( $node_subscriber->safe_psql(
 		$db, qq(SELECT pg_stat_have_stats('subscription', 0, $sub1_oid))),
 	qq(f),
 	qq(Subscription stats for subscription '$sub1_name' should be removed.));

 # Get subscription 2 oid
 my $sub2_oid = $node_subscriber->safe_psql($db,
 	qq(SELECT oid FROM pg_subscription WHERE subname = '$sub2_name'));

 # Diassociate the subscription 2 from its replication slot and drop it
 $node_subscriber->safe_psql(
 	$db,
 	qq(
 ALTER SUBSCRIPTION $sub2_name DISABLE;
 ALTER SUBSCRIPTION $sub2_name SET (slot_name = NONE);
 DROP SUBSCRIPTION $sub2_name;
 			    ));

 # Subscription stats for sub2 should be gone
 is( $node_subscriber->safe_psql(
 		$db, qq(SELECT pg_stat_have_stats('subscription', 0, $sub2_oid))),
 	qq(f),
 	qq(Subscription stats for subscription '$sub2_name' should be removed.));

 # Since disabling subscription doesn't wait for walsender to release the replication
 # slot and exit, wait for the slot to become inactive.
 $node_publisher->poll_query_until(
 	$db,
 	qq(SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = '$sub2_name' AND active_pid IS NULL))
 ) or die "slot never became inactive";

 $node_publisher->safe_psql($db,
 	qq(SELECT pg_drop_replication_slot('$sub2_name')));

 $node_subscriber->stop('fast');
 $node_publisher->stop('fast');

 done_testing();

	# Copyright (c) 2021-2023, PostgreSQL Global Development Group

	# Tests for subscription stats.
	use strict;
	use warnings;
	use PostgreSQL::Test::Cluster;
	use PostgreSQL::Test::Utils;
	use Test::More;

	# Create publisher node.
	my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
	$node_publisher->init(allows_streaming => 'logical');
	$node_publisher->start;

	# Create subscriber node.
	my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
	$node_subscriber->init(allows_streaming => 'logical');
	$node_subscriber->start;


	sub create_sub_pub_w_errors
	{
	my ($node_publisher, $node_subscriber, $db, $table_name) = @_;
	# Initial table setup on both publisher and subscriber. On subscriber we
	# create the same tables but with primary keys. Also, insert some data that
	# will conflict with the data replicated from publisher later.
	$node_publisher->safe_psql(
	$db,
	qq[
	BEGIN;
	CREATE TABLE $table_name(a int);
	INSERT INTO $table_name VALUES (1);
	COMMIT;
	]);
	$node_subscriber->safe_psql(
	$db,
	qq[
	BEGIN;
	CREATE TABLE $table_name(a int primary key);
	INSERT INTO $table_name VALUES (1);
	COMMIT;
	]);

	# Set up publication.
	my $pub_name = $table_name . '_pub';
	my $publisher_connstr = $node_publisher->connstr . qq( dbname=$db);

	$node_publisher->safe_psql($db,
	qq(CREATE PUBLICATION $pub_name FOR TABLE $table_name));

	# Create subscription. The tablesync for table on subscription will enter into
	# infinite error loop due to violating the unique constraint.
	my $sub_name = $table_name . '_sub';
	$node_subscriber->safe_psql($db,
	qq(CREATE SUBSCRIPTION $sub_name CONNECTION '$publisher_connstr' PUBLICATION $pub_name)
	);

	$node_publisher->wait_for_catchup($sub_name);

	# Wait for the tablesync error to be reported.
	$node_subscriber->poll_query_until(
	$db,
	qq[
	SELECT sync_error_count > 0
	FROM pg_stat_subscription_stats
	WHERE subname = '$sub_name'
	])
	or die
	qq(Timed out while waiting for tablesync errors for subscription '$sub_name');

	# Truncate test_tab1 so that tablesync worker can continue.
	$node_subscriber->safe_psql($db, qq(TRUNCATE $table_name));

	# Wait for initial tablesync to finish.
	$node_subscriber->poll_query_until(
	$db,
	qq[
	SELECT count(1) = 1 FROM pg_subscription_rel
	WHERE srrelid = '$table_name'::regclass AND srsubstate in ('r', 's')
	])
	or die
	qq(Timed out while waiting for subscriber to synchronize data for table '$table_name'.);

	# Check test table on the subscriber has one row.
	my $result =
	$node_subscriber->safe_psql($db, qq(SELECT a FROM $table_name));
	is($result, qq(1), qq(Check that table '$table_name' now has 1 row.));

	# Insert data to test table on the publisher, raising an error on the
	# subscriber due to violation of the unique constraint on test table.
	$node_publisher->safe_psql($db, qq(INSERT INTO $table_name VALUES (1)));

	# Wait for the apply error to be reported.
	$node_subscriber->poll_query_until(
	$db,
	qq[
	SELECT apply_error_count > 0
	FROM pg_stat_subscription_stats
	WHERE subname = '$sub_name'
	])
	or die
	qq(Timed out while waiting for apply error for subscription '$sub_name');

	# Truncate test table so that apply worker can continue.
	$node_subscriber->safe_psql($db, qq(TRUNCATE $table_name));

	return ($pub_name, $sub_name);
	}

	my $db = 'postgres';

	# There shouldn't be any subscription errors before starting logical replication.
	my $result = $node_subscriber->safe_psql($db,
	qq(SELECT count(1) FROM pg_stat_subscription_stats));
	is($result, qq(0),
	'Check that there are no subscription errors before starting logical replication.'
	);

	# Create the publication and subscription with sync and apply errors
	my $table1_name = 'test_tab1';
	my ($pub1_name, $sub1_name) =
	create_sub_pub_w_errors($node_publisher, $node_subscriber, $db,
	$table1_name);

	# Apply and Sync errors are > 0 and reset timestamp is NULL
	is( $node_subscriber->safe_psql(
	$db,
	qq(SELECT apply_error_count > 0,
	sync_error_count > 0,
	stats_reset IS NULL
	FROM pg_stat_subscription_stats
	WHERE subname = '$sub1_name')
	),
	qq(t\|t\|t),
	qq(Check that apply errors and sync errors are both > 0 and stats_reset is NULL for subscription '$sub1_name'.)
	);

	# Reset a single subscription
	$node_subscriber->safe_psql($db,
	qq(SELECT pg_stat_reset_subscription_stats((SELECT subid FROM pg_stat_subscription_stats WHERE subname = '$sub1_name')))
	);

	# Apply and Sync errors are 0 and stats reset is not NULL
	is( $node_subscriber->safe_psql(
	$db,
	qq(SELECT apply_error_count = 0,
	sync_error_count = 0,
	stats_reset IS NOT NULL
	FROM pg_stat_subscription_stats
	WHERE subname = '$sub1_name')
	),
	qq(t\|t\|t),
	qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL after reset for subscription '$sub1_name'.)
	);

	# Get reset timestamp
	my $reset_time1 = $node_subscriber->safe_psql($db,
	qq(SELECT stats_reset FROM pg_stat_subscription_stats WHERE subname = '$sub1_name')
	);

	# Reset single sub again
	$node_subscriber->safe_psql(
	$db,
	qq(SELECT pg_stat_reset_subscription_stats((SELECT subid FROM
	pg_stat_subscription_stats WHERE subname = '$sub1_name')))
	);

	# check reset timestamp is newer after reset
	is( $node_subscriber->safe_psql(
	$db,
	qq(SELECT stats_reset > '$reset_time1'::timestamptz FROM
	pg_stat_subscription_stats WHERE subname = '$sub1_name')
	),
	qq(t),
	qq(Check reset timestamp for '$sub1_name' is newer after second reset.));

	# Make second subscription and publication
	my $table2_name = 'test_tab2';
	my ($pub2_name, $sub2_name) =
	create_sub_pub_w_errors($node_publisher, $node_subscriber, $db,
	$table2_name);

	# Apply and Sync errors are > 0 and reset timestamp is NULL
	is( $node_subscriber->safe_psql(
	$db,
	qq(SELECT apply_error_count > 0,
	sync_error_count > 0,
	stats_reset IS NULL
	FROM pg_stat_subscription_stats
	WHERE subname = '$sub2_name')
	),
	qq(t\|t\|t),
	qq(Confirm that apply errors and sync errors are both > 0 and stats_reset is NULL for sub '$sub2_name'.)
	);

	# Reset all subscriptions
	$node_subscriber->safe_psql($db,
	qq(SELECT pg_stat_reset_subscription_stats(NULL)));

	# Apply and Sync errors are 0 and stats reset is not NULL
	is( $node_subscriber->safe_psql(
	$db,
	qq(SELECT apply_error_count = 0,
	sync_error_count = 0,
	stats_reset IS NOT NULL
	FROM pg_stat_subscription_stats
	WHERE subname = '$sub1_name')
	),
	qq(t\|t\|t),
	qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL for sub '$sub1_name' after reset.)
	);

	is( $node_subscriber->safe_psql(
	$db,
	qq(SELECT apply_error_count = 0,
	sync_error_count = 0,
	stats_reset IS NOT NULL
	FROM pg_stat_subscription_stats
	WHERE subname = '$sub2_name')
	),
	qq(t\|t\|t),
	qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL for sub '$sub2_name' after reset.)
	);

	$reset_time1 = $node_subscriber->safe_psql($db,
	qq(SELECT stats_reset FROM pg_stat_subscription_stats WHERE subname = '$sub1_name')
	);
	my $reset_time2 = $node_subscriber->safe_psql($db,
	qq(SELECT stats_reset FROM pg_stat_subscription_stats WHERE subname = '$sub2_name')
	);

	# Reset all subscriptions
	$node_subscriber->safe_psql($db,
	qq(SELECT pg_stat_reset_subscription_stats(NULL)));

	# check reset timestamp for sub1 is newer after reset
	is( $node_subscriber->safe_psql(
	$db,
	qq(SELECT stats_reset > '$reset_time1'::timestamptz FROM
	pg_stat_subscription_stats WHERE subname = '$sub1_name')
	),
	qq(t),
	qq(Confirm that reset timestamp for '$sub1_name' is newer after second reset.)
	);

	# check reset timestamp for sub2 is newer after reset
	is( $node_subscriber->safe_psql(
	$db,
	qq(SELECT stats_reset > '$reset_time2'::timestamptz FROM
	pg_stat_subscription_stats WHERE subname = '$sub2_name')
	),
	qq(t),
	qq(Confirm that reset timestamp for '$sub2_name' is newer after second reset.)
	);

	# Get subscription 1 oid
	my $sub1_oid = $node_subscriber->safe_psql($db,
	qq(SELECT oid FROM pg_subscription WHERE subname = '$sub1_name'));

	# Drop subscription 1
	$node_subscriber->safe_psql($db, qq(DROP SUBSCRIPTION $sub1_name));

	# Subscription stats for sub1 should be gone
	is( $node_subscriber->safe_psql(
	$db, qq(SELECT pg_stat_have_stats('subscription', 0, $sub1_oid))),
	qq(f),
	qq(Subscription stats for subscription '$sub1_name' should be removed.));

	# Get subscription 2 oid
	my $sub2_oid = $node_subscriber->safe_psql($db,
	qq(SELECT oid FROM pg_subscription WHERE subname = '$sub2_name'));

	# Diassociate the subscription 2 from its replication slot and drop it
	$node_subscriber->safe_psql(
	$db,
	qq(
	ALTER SUBSCRIPTION $sub2_name DISABLE;
	ALTER SUBSCRIPTION $sub2_name SET (slot_name = NONE);
	DROP SUBSCRIPTION $sub2_name;
	));

	# Subscription stats for sub2 should be gone
	is( $node_subscriber->safe_psql(
	$db, qq(SELECT pg_stat_have_stats('subscription', 0, $sub2_oid))),
	qq(f),
	qq(Subscription stats for subscription '$sub2_name' should be removed.));

	# Since disabling subscription doesn't wait for walsender to release the replication
	# slot and exit, wait for the slot to become inactive.
	$node_publisher->poll_query_until(
	$db,
	qq(SELECT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = '$sub2_name' AND active_pid IS NULL))
	) or die "slot never became inactive";

	$node_publisher->safe_psql($db,
	qq(SELECT pg_drop_replication_slot('$sub2_name')));

	$node_subscriber->stop('fast');
	$node_publisher->stop('fast');

	done_testing();