v3-0002-Support-WHERE-clause-and-COLUMN-list-in-table-arg.patch

From 61623a4d40bb7f477bd0db192fe08b727bf4e1b4 Mon Sep 17 00:00:00 2001
From: Khanna <Shubham.Khanna@fujitsu.com>
Date: Thu, 21 Aug 2025 13:44:43 +0530
Subject: [PATCH v3 2/2] Support WHERE clause and COLUMN list in --table
 argument

This patch support the specification of both a WHERE clause (row filter) and a
column list in the table specification via pg_createsubscriber, and modify the
utility's name (for example, to a more descriptive or aligned name).
For eg:-
CREATE PUBLICATION pub FOR TABLE schema.table (col1, col2) WHERE
(predicate);
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 49 +++++++++++++++++++++
 1 file changed, 49 insertions(+)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 0017a740b72..ecfb032593d 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -35,6 +35,8 @@ typedef struct TableSpec
 {
 	char	   *spec;
 	char	   *dbname;
+	char	   *att_names;
+	char	   *row_filter;
 	char	   *pattern_regex;
 	char	   *pattern_part1_regex;
 	char	   *pattern_part2_regex;
@@ -143,6 +145,7 @@ static void drop_existing_subscriptions(PGconn *conn, const char *subname,
 										const char *dbname);
 static void get_publisher_databases(struct CreateSubscriberOptions *opt,
 									bool dbnamespecified);
+static char *pg_strcasestr(const char *haystack, const char *needle);
 
 #define	USEC_PER_SEC	1000000
 #define	WAIT_INTERVAL	1		/* 1 second */
@@ -177,6 +180,24 @@ enum WaitPMResult
 	POSTMASTER_STILL_STARTING
 };
 
+char *
+pg_strcasestr(const char *haystack, const char *needle)
+{
+	if (!*needle)
+		return (char *) haystack;
+	for (; *haystack; haystack++)
+	{
+		const char *h = haystack;
+		const char *n = needle;
+
+		while (*h && *n && pg_tolower((unsigned char) *h) == pg_tolower((unsigned char) *n))
+			++h, ++n;
+		if (!*n)
+			return (char *) haystack;
+	}
+	return NULL;
+}
+
 /*
  * Cleanup objects that were created by pg_createsubscriber if there is an
  * error.
@@ -1746,6 +1767,11 @@ create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
 								  first_table ? "" : ", ",
 								  escaped_schema, escaped_table);
 
+				if (tbl->att_names && strlen(tbl->att_names) > 0)
+					appendPQExpBuffer(str, " ( %s )", tbl->att_names);
+
+				if (tbl->row_filter && strlen(tbl->row_filter) > 0)
+					appendPQExpBuffer(str, " WHERE %s", tbl->row_filter);
 				first_table = false;
 
 				PQfreemem(escaped_schema);
@@ -2334,6 +2360,8 @@ main(int argc, char **argv)
 					char	   *second_dot;
 					char	   *dbname_arg = NULL;
 					char	   *schema_table_part;
+					char	   *paren_start = NULL;
+					char	   *where_start = NULL;
 					TableSpec  *ts;
 					PQExpBuffer dbbuf;
 					PQExpBuffer schemabuf;
@@ -2346,6 +2374,27 @@ main(int argc, char **argv)
 
 					copy_arg = pg_strdup(optarg);
 
+					where_start = pg_strcasestr(copy_arg, " where ");
+					if (where_start != NULL)
+					{
+						*where_start = '\0';
+						where_start += 7;
+					}
+
+					paren_start = strchr(copy_arg, '(');
+					if (paren_start != NULL)
+					{
+						char	   *paren_end = strrchr(paren_start, ')');
+
+						if (!paren_end)
+							pg_fatal("unmatched '(' in --table argument \"%s\"", optarg);
+
+						*paren_start = '\0';
+						*paren_end = '\0';
+
+						paren_start++;
+					}
+
 					first_dot = strchr(copy_arg, '.');
 					if (first_dot != NULL)
 						second_dot = strchr(first_dot + 1, '.');
-- 
2.41.0.windows.3