Filter CDC changes at transform time, not just apply time

src/bin/pgcopydb/cli_stream.c

@@ -1263,6 +1263,24 @@ stream_start_in_mode(LogicalStreamMode mode)
 		exit(EXIT_CODE_INTERNAL_ERROR);
 	}
 
+	/*
+	 * Open the source catalog and load setup information including filters, so
+	 * the transform step (prefetch) can skip changes for filtered-out tables
+	 * instead of materializing SQL that apply would only discard. Mirrors the
+	 * catchup path; followDB manages the source catalog from here on.
+	 */
+	if (!catalog_open_from_specs(&copySpecs))
+	{
+		/* errors have already been logged */
+		exit(EXIT_CODE_INTERNAL_ERROR);
+	}
+
+	if (!catalog_register_setup_from_specs(&copySpecs))
+	{
+		/* errors have already been logged */
+		exit(EXIT_CODE_INTERNAL_ERROR);
+	}
+
 	/*
 	 * Refrain from logging SQL statements in the apply module, because they
 	 * contain user data. That said, when --trace has been used, bypass that

src/bin/pgcopydb/filtering.c

@@ -1139,3 +1139,102 @@ filters_from_json(const char *jsonString, SourceFilters *filters)
 
 	return true;
 }
+
+
+/*
+ * shouldFilterOutTable checks if a given table should be filtered out based
+ * on the configured filters. It scans the in-memory filter lists only (no
+ * catalog or database access) and returns false when filters is NULL, so it
+ * fails open. Shared by CDC transform-time and apply-time filtering.
+ */
+bool
+shouldFilterOutTable(const char *nspname, const char *relname,
+					 SourceFilters *filters)
+{
+	if (filters == NULL)
+	{
+		return false;
+	}
+
+	/* Check exclude-schema filter */
+	for (int i = 0; i < filters->excludeSchemaList.count; i++)
+	{
+		if (strcmp(filters->excludeSchemaList.array[i].nspname, nspname) == 0)
+		{
+			log_trace("Filtering out table \"%s\".\"%s\" (schema in exclude-schema list)",
+					  nspname, relname);
+			return true;
+		}
+	}
+
+	/* Check include-only-schema filter */
+	if (filters->includeOnlySchemaList.count > 0)
+	{
+		bool found = false;
+		for (int i = 0; i < filters->includeOnlySchemaList.count; i++)
+		{
+			if (strcmp(filters->includeOnlySchemaList.array[i].nspname, nspname) == 0)
+			{
+				found = true;
+				break;
+			}
+		}
+		if (!found)
+		{
+			log_trace(
+				"Filtering out table \"%s\".\"%s\" (schema not in include-only-schema list)",
+				nspname, relname);
+			return true;
+		}
+	}
+
+	/* Check include-only-table filter */
+	if (filters->includeOnlyTableList.count > 0)
+	{
+		bool found = false;
+		for (int i = 0; i < filters->includeOnlyTableList.count; i++)
+		{
+			SourceFilterTable *table = &(filters->includeOnlyTableList.array[i]);
+			if (strcmp(table->nspname, nspname) == 0 &&
+				strcmp(table->relname, relname) == 0)
+			{
+				found = true;
+				break;
+			}
+		}
+		if (!found)
+		{
+			log_trace("Filtering out table \"%s\".\"%s\" (not in include-only list)",
+					  nspname, relname);
+			return true;
+		}
+	}
+
+	/* Check exclude-table filter */
+	for (int i = 0; i < filters->excludeTableList.count; i++)
+	{
+		SourceFilterTable *table = &(filters->excludeTableList.array[i]);
+		if (strcmp(table->nspname, nspname) == 0 &&
+			strcmp(table->relname, relname) == 0)
+		{
+			log_trace("Filtering out table \"%s\".\"%s\" (in exclude-table list)",
+					  nspname, relname);
+			return true;
+		}
+	}
+
+	/* Check exclude-table-data filter */
+	for (int i = 0; i < filters->excludeTableDataList.count; i++)
+	{
+		SourceFilterTable *table = &(filters->excludeTableDataList.array[i]);
+		if (strcmp(table->nspname, nspname) == 0 &&
+			strcmp(table->relname, relname) == 0)
+		{
+			log_trace("Filtering out table \"%s\".\"%s\" (in exclude-table-data list)",
+					  nspname, relname);
+			return true;
+		}
+	}
+
+	return false;
+}

src/bin/pgcopydb/filtering.h

@@ -139,4 +139,7 @@ bool parse_filters(const char *filebname, SourceFilters *filters);
 bool filters_as_json(SourceFilters *filters, JSON_Value *jsFilter);
 bool filters_from_json(const char *jsonString, SourceFilters *filters);
 
+bool shouldFilterOutTable(const char *nspname, const char *relname,
+						  SourceFilters *filters);
+
 #endif  /* FILTERING_H */

src/bin/pgcopydb/ld_apply.c

@@ -1815,107 +1815,6 @@ extractTableNameFromPrepare(const char *stmt,
 }
 
 
-/*
- * shouldFilterOutTable checks if a given table should be filtered out based
- * on the configured filters.
- *
- * Note: This function checks in-memory filter lists. For extension filtering
- * during CDC, we also need to check the catalog filter table (see usage in
- * parseSQLAction where we pass the catalog context).
- */
-static bool
-shouldFilterOutTable(const char *nspname, const char *relname,
-					 SourceFilters *filters)
-{
-	if (filters == NULL)
-	{
-		return false;
-	}
-
-	/* Check exclude-schema filter */
-	for (int i = 0; i < filters->excludeSchemaList.count; i++)
-	{
-		if (strcmp(filters->excludeSchemaList.array[i].nspname, nspname) == 0)
-		{
-			log_trace("Filtering out table \"%s\".\"%s\" (schema in exclude-schema list)",
-					  nspname, relname);
-			return true;
-		}
-	}
-
-	/* Check include-only-schema filter */
-	if (filters->includeOnlySchemaList.count > 0)
-	{
-		bool found = false;
-		for (int i = 0; i < filters->includeOnlySchemaList.count; i++)
-		{
-			if (strcmp(filters->includeOnlySchemaList.array[i].nspname, nspname) == 0)
-			{
-				found = true;
-				break;
-			}
-		}
-		if (!found)
-		{
-			log_trace(
-				"Filtering out table \"%s\".\"%s\" (schema not in include-only-schema list)",
-				nspname, relname);
-			return true;
-		}
-	}
-
-	/* Check include-only-table filter */
-	if (filters->includeOnlyTableList.count > 0)
-	{
-		bool found = false;
-		for (int i = 0; i < filters->includeOnlyTableList.count; i++)
-		{
-			SourceFilterTable *table = &(filters->includeOnlyTableList.array[i]);
-			if (strcmp(table->nspname, nspname) == 0 &&
-				strcmp(table->relname, relname) == 0)
-			{
-				found = true;
-				break;
-			}
-		}
-		if (!found)
-		{
-			log_trace("Filtering out table \"%s\".\"%s\" (not in include-only list)",
-					  nspname, relname);
-			return true;
-		}
-	}
-
-	/* Check exclude-table filter */
-	for (int i = 0; i < filters->excludeTableList.count; i++)
-	{
-		SourceFilterTable *table = &(filters->excludeTableList.array[i]);
-		if (strcmp(table->nspname, nspname) == 0 &&
-			strcmp(table->relname, relname) == 0)
-		{
-			log_trace("Filtering out table \"%s\".\"%s\" (in exclude-table list)",
-					  nspname, relname);
-			return true;
-		}
-	}
-
-	/* Check exclude-table-data filter */
-	for (int i = 0; i < filters->excludeTableDataList.count; i++)
-	{
-		SourceFilterTable *table = &(filters->excludeTableDataList.array[i]);
-		if (strcmp(table->nspname, nspname) == 0 &&
-			strcmp(table->relname, relname) == 0)
-		{
-			log_trace("Filtering out table \"%s\".\"%s\" (in exclude-table-data list)",
-					  nspname, relname);
-			return true;
-		}
-	}
-
-	return false;
-}
-
-
 /*
  * parseSQLAction returns the action that is implemented in the given SQL
  * query.

src/bin/pgcopydb/ld_stream.c

@@ -426,6 +426,9 @@ stream_init_context(StreamSpecs *specs)
 	/* transform needs some catalog lookups (pkey, type oid) */
 	privateContext->sourceDB = specs->sourceDB;
 
+	/* table filtering configuration, used by parseMessage() at transform time */
+	privateContext->filters = specs->filters;
+
 	return true;
 }
 

src/bin/pgcopydb/ld_transform.c

@@ -1154,6 +1154,55 @@ stream_transform_file(StreamSpecs *specs, char *jsonfilename, char *sqlfilename)
 }
 
 
+/*
+ * dequoteIdentifier writes the bare form of a possibly double-quoted SQL
+ * identifier into out. wal2json stores identifiers escaped by
+ * pgsql_escape_identifier (e.g. "public", or "Weird""Name"), whereas filter
+ * lists hold bare names; this strips the surrounding quotes and collapses
+ * doubled "" back to a single ". Identifiers without surrounding quotes (e.g.
+ * from test_decoding) are copied verbatim.
+ */
+static void
+dequoteIdentifier(const char *ident, char *out, size_t outsize)
+{
+	if (ident == NULL || outsize == 0)
+	{
+		if (outsize > 0)
+		{
+			out[0] = '\0';
+		}
+		return;
+	}
+
+	size_t len = strlen(ident);
+
+	if (len >= 2 && ident[0] == '"' && ident[len - 1] == '"')
+	{
+		size_t j = 0;
+
+		/* copy the content between the surrounding quotes, un-doubling "" */
+		for (size_t i = 1; i + 1 < len && j + 1 < outsize; i++)
+		{
+			if (ident[i] == '"' && i + 2 < len && ident[i + 1] == '"')
+			{
+				out[j++] = '"';
+				i++;    /* skip the second quote of the pair */
+			}
+			else
+			{
+				out[j++] = ident[i];
+			}
+		}
+
+		out[j] = '\0';
+	}
+	else
+	{
+		strlcpy(out, ident, outsize);
+	}
+}
+
+
 /*
  * parseMessage parses a JSON message as emitted by the logical decoding output
  * plugin (either test_decoding or wal2json) into our own internal
@@ -1510,6 +1559,38 @@ parseMessage(StreamContext *privateContext, char *message, JSON_Value *json)
 				}
 			}
 
+			/*
+			 * Skip DML/TRUNCATE destined for tables that are filtered out, so
+			 * we never render PREPARE/EXECUTE SQL for changes that apply would
+			 * only discard. Detection here mirrors the apply-time check; the
+			 * apply-time skip stays as defense-in-depth.
+			 */
+
+			/*
+			 * Table names on the statement are SQL-escaped (wal2json quotes
+			 * them via pgsql_escape_identifier), but the filter lists hold bare
+			 * identifiers. Dequote before comparing so the check matches the
+			 * apply-time behavior (which unquotes when parsing PREPARE text).
+			 */
+			if (stmtNspname != NULL)
+			{
+				char nsp[PG_NAMEDATALEN] = { 0 };
+				char rel[PG_NAMEDATALEN] = { 0 };
+
+				dequoteIdentifier(stmtNspname, nsp, sizeof(nsp));
+				dequoteIdentifier(stmtRelname, rel, sizeof(rel));
+
+				if (shouldFilterOutTable(nsp, rel, privateContext->filters))
+				{
+					log_trace("Skipping %c on filtered-out table %s.%s",
+							  metadata->action, nsp, rel);
+					free(stmt);
+					privateContext->stmt = NULL;
+					break;
+				}
+			}
+
+			/* skip DML targeting materialized views */
 			if (stmtNspname != NULL &&
 				lookupMatViewCache(privateContext->matViewCache,
 								   stmtNspname, stmtRelname))

tests/Makefile

@@ -12,6 +12,7 @@ BUILD_ARGS = --build-arg PGVERSION=$(PGVERSION)
 
 all: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions \
 	 cdc-wal2json cdc-test-decoding cdc-endpos-between-transaction cdc-low-level \
+	 cdc-filtering \
 	 cdc-prune \
 	 follow-wal2json follow-standby follow-9.6 follow-data-only follow-target-reconnect \
 	 endpos-in-multi-wal-txn exclude-extension \
@@ -57,6 +58,9 @@ cdc-endpos-between-transaction: build
 cdc-low-level: build
 	$(MAKE) -C $@
 
+cdc-filtering: build
+	$(MAKE) -C $@
+
 follow-wal2json: build
 	$(MAKE) -C $@
 
@@ -108,7 +112,7 @@ build:
 
 .PHONY: all build
 .PHONY: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions
-.PHONY: cdc-wal2json cdc-test-decoding cdc-low-level cdc-prune
+.PHONY: cdc-wal2json cdc-test-decoding cdc-low-level cdc-filtering cdc-prune
 .PHONY: follow-wal2json follow-standby follow-9.6 follow-target-reconnect
 .PHONY: endpos-in-multi-wal-txn exclude-extension
 .PHONY: blob-snapshot-release follow-defer-indexes fk-not-valid defer-validate-fks