Ignore dropped and generated columns from the column list.

We don't allow different column lists for the same table in the different
publications of the single subscription. A publication with a column list
except for dropped and generated columns should be considered the same as
a publication with no column list (which implicitly includes all columns
as part of the columns list). However, as we were not excluding the
dropped and generated columns from the column list combining such
publications leads to an error "cannot use different column lists for
table ...".

We decided not to backpatch this fix as there is a risk of users seeing
this as a behavior change and also we didn't see any field report of this
case.

Author: Shi yu
Reviewed-by: Amit Kapila
Discussion: https://postgr.es/m/OSZPR01MB631091CCBC56F195B1B9ACB0FDFE9@OSZPR01MB6310.jpnprd01.prod.outlook.com

src/backend/catalog/pg_publication.c

@@ -1153,6 +1153,36 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
 			nulls[2] = true;
 		}
 
+		/* Show all columns when the column list is not specified. */
+		if (nulls[1] == true)
+		{
+			Relation	rel = table_open(relid, AccessShareLock);
+			int			nattnums = 0;
+			int16	   *attnums;
+			TupleDesc	desc = RelationGetDescr(rel);
+			int			i;
+
+			attnums = (int16 *) palloc(desc->natts * sizeof(int16));
+
+			for (i = 0; i < desc->natts; i++)
+			{
+				Form_pg_attribute att = TupleDescAttr(desc, i);
+
+				if (att->attisdropped || att->attgenerated)
+					continue;
+
+				attnums[nattnums++] = att->attnum;
+			}
+
+			if (nattnums > 0)
+			{
+				values[1] = PointerGetDatum(buildint2vector(attnums, nattnums));
+				nulls[1] = false;
+			}
+
+			table_close(rel, AccessShareLock);
+		}
+
 		rettuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
 
 		SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(rettuple));

src/backend/catalog/system_views.sql

@@ -371,9 +371,8 @@ CREATE VIEW pg_publication_tables AS
         C.relname AS tablename,
         ( SELECT array_agg(a.attname ORDER BY a.attnum)
           FROM pg_attribute a
-          WHERE a.attrelid = GPT.relid AND a.attnum > 0 AND
-                NOT a.attisdropped AND
-                (a.attnum = ANY(GPT.attrs) OR GPT.attrs IS NULL)
+          WHERE a.attrelid = GPT.relid AND
+                a.attnum = ANY(GPT.attrs)
         ) AS attnames,
         pg_get_expr(GPT.qual, GPT.relid) AS rowfilter
     FROM pg_publication P,

src/backend/replication/pgoutput/pgoutput.c

@@ -1058,16 +1058,31 @@ pgoutput_column_list_init(PGOutputData *data, List *publications,
 				/* Build the column list bitmap in the per-entry context. */
 				if (!pub_no_list)	/* when not null */
 				{
+					int			i;
+					int			nliveatts = 0;
+					TupleDesc	desc = RelationGetDescr(relation);
+
 					pgoutput_ensure_entry_cxt(data, entry);
 
 					cols = pub_collist_to_bitmapset(cols, cfdatum,
 													entry->entry_cxt);
 
+					/* Get the number of live attributes. */
+					for (i = 0; i < desc->natts; i++)
+					{
+						Form_pg_attribute att = TupleDescAttr(desc, i);
+
+						if (att->attisdropped || att->attgenerated)
+							continue;
+
+						nliveatts++;
+					}
+
 					/*
 					 * If column list includes all the columns of the table,
 					 * set it to NULL.
 					 */
-					if (bms_num_members(cols) == RelationGetNumberOfAttributes(relation))
+					if (bms_num_members(cols) == nliveatts)
 					{
 						bms_free(cols);
 						cols = NULL;

src/include/catalog/catversion.h

@@ -57,6 +57,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	202301092
+#define CATALOG_VERSION_NO	202301131
 
 #endif

src/test/regress/expected/rules.out

@@ -1446,7 +1446,7 @@ pg_publication_tables| SELECT p.pubname,
     c.relname AS tablename,
     ( SELECT array_agg(a.attname ORDER BY a.attnum) AS array_agg
            FROM pg_attribute a
-          WHERE ((a.attrelid = gpt.relid) AND (a.attnum > 0) AND (NOT a.attisdropped) AND ((a.attnum = ANY ((gpt.attrs)::smallint[])) OR (gpt.attrs IS NULL)))) AS attnames,
+          WHERE ((a.attrelid = gpt.relid) AND (a.attnum = ANY ((gpt.attrs)::smallint[])))) AS attnames,
     pg_get_expr(gpt.qual, gpt.relid) AS rowfilter
    FROM pg_publication p,
     LATERAL pg_get_publication_tables((p.pubname)::text) gpt(relid, attrs, qual),

src/test/subscription/t/031_column_list.pl

@@ -1184,6 +1184,53 @@ $result = $node_publisher->safe_psql(
 is( $result, qq(t
 t), 'check the number of columns in the old tuple');
 
+# TEST: Generated and dropped columns are not considered for the column list.
+# So, the publication having a column list except for those columns and a
+# publication without any column (aka all columns as part of the columns
+# list) are considered to have the same column list.
+$node_publisher->safe_psql(
+	'postgres', qq(
+	CREATE TABLE test_mix_4 (a int PRIMARY KEY, b int, c int, d int GENERATED ALWAYS AS (a + 1) STORED);
+	ALTER TABLE test_mix_4 DROP COLUMN c;
+
+	CREATE PUBLICATION pub_mix_7 FOR TABLE test_mix_4 (a, b);
+	CREATE PUBLICATION pub_mix_8 FOR TABLE test_mix_4;
+
+	-- initial data
+	INSERT INTO test_mix_4 VALUES (1, 2);
+));
+
+$node_subscriber->safe_psql(
+	'postgres', qq(
+	DROP SUBSCRIPTION sub1;
+	CREATE TABLE test_mix_4 (a int PRIMARY KEY, b int, c int, d int);
+));
+
+$node_subscriber->safe_psql(
+	'postgres', qq(
+	CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub_mix_7, pub_mix_8;
+));
+
+$node_subscriber->wait_for_subscription_sync;
+
+is( $node_subscriber->safe_psql(
+		'postgres', "SELECT * FROM test_mix_4 ORDER BY a"),
+	qq(1|2||),
+	'initial synchronization with multiple publications with the same column list'
+);
+
+$node_publisher->safe_psql(
+	'postgres', qq(
+	INSERT INTO test_mix_4 VALUES (3, 4);
+));
+
+$node_publisher->wait_for_catchup('sub1');
+
+is( $node_subscriber->safe_psql(
+		'postgres', "SELECT * FROM test_mix_4 ORDER BY a"),
+	qq(1|2||
+3|4||),
+	'replication with multiple publications with the same column list');
 
 # TEST: With a table included in multiple publications with different column
 # lists, we should catch the error when creating the subscription.