Repository: cloudnative-pg/cloudnative-pg

Issue:
[Bug]: Publication incorrectly compiles SQL when used with multiple tables
### Is there an existing issue already for this bug?

- [x] I have searched for an existing issue, and could not find anything. I believe this is a new bug.

### I have read the troubleshooting guide

- [x] I have read the troubleshooting guide and I think this is a new bug.

### I am running a supported version of CloudNativePG

- [x] I have read the troubleshooting guide and I think this is a new bug.

### Contact Details

ismailova.maryam@gmail.com

### Version

1.26 (latest patch)

### What version of Kubernetes are you using?

1.31

### What is your Kubernetes environment?

Other

### How did you install the operator?

Helm

### What happened?

When creating a publication for multiple tables cnpg incorrectly generates the SQL, and thus fails to create the publication.
In my case this is a publication for a database and 2 tables in the same scheme.
I have tried following example:
```
spec:
  cluster:
    name: clusterX
  dbname: log_crm
  name: dbz_publication
  publicationReclaimPolicy: delete
  target:
    objects:
    - table:
        name: log_crm.crm_employees
    - table:
        name: log_crm.crm_user_actions
```
And this generates following log and fails:
```
"message":"syntax error at or near \"TABLE\"","query":"CREATE PUBLICATION \"dbz_publication\" FOR TABLE \"log_crm.crm_employees\", TABLE \"log_crm.crm_user_actions\""
```

One can understand that for each table object in the list the operator tries to append a separate TABLE <table_name> , however seemingly postgres actually expect a single TABLE definition followed by a list of tables.

I have also tried following way to try and create all through a single table object:
```
spec:
  cluster:
    name: clusterX
  dbname: log_crm
  name: dbz_publication
  publicationReclaimPolicy: delete
  target:
    objects:
    - table:
        name: log_crm.crm_employees,log_crm.crm_user_actions
```

However, this as well fails:
```
"message":"relation \"log_crm.crm_employees,log_crm.crm_user_actions\" does not exist","query":"CREATE PUBLICATION \"dbz_publication\" FOR TABLE \"log_crm.crm_employees,log_crm.crm_user_actions\""
```

Let me note that unlike Publication CRD,  cmd using cnpg plugin works pretty well:
```
$ kubectl cnpg publication create log-crm-infra-2221-postgresql --dbname log_crm --publication dbz_publication --table log_crm.crm_employees,log_crm.crm_user_actions --dry-run
```
Comand results in following SQL command generated:
```
CREATE PUBLICATION "dbz_publication" FOR TABLE log_crm.crm_employees, log_crm.crm_user_actions
```

### Cluster resource

```shell
apiVersion: postgresql.cnpg.io/v1
kind: Cluster
metadata:
  name: clusterX
  namespace: default
  annotations:
    helm.sh/hook-weight: "-5"
spec:
  instances: 1
  
  imageName: ghcr.io/cloudnative-pg/postgresql:13.15
  imagePullPolicy: IfNotPresent
  postgresUID: 26
  postgresGID: 26
  storage:
    size: 20Gi
  resources:
    limits:
      cpu: "0.5"
      memory: 2Gi
    requests:
      cpu: "0.5"
      memory: 2Gi
  
  affinity:
    additionalPodAntiAffinity:
      preferredDuringSchedulingIgnoredDuringExecution:
      - podAffinityTerm:
          labelSelector:
            matchExpressions:
            - key: cnpg.io/cluster
              operator: In
              values:
              - clusterX
            - key: cnpg.io/podRole
              operator: In
              values:
              - instance
          topologyKey: node.cluster.x-k8s.io/esxi-host
        weight: 100
    enablePodAntiAffinity: true
    nodeSelector:
      postgres: "true"
    podAntiAffinityType: required
    tolerations:
    - effect: NoSchedule
      key: node-role.kubernetes.io/postgres
      operator: Exists
    topologyKey: kubernetes.io/hostname

  primaryUpdateMethod: switchover
  primaryUpdateStrategy: unsupervised
  logLevel: info
  enableSuperuserAccess: true
  enablePDB: true
  postgresql:
    parameters:
      archive_timeout: 1min
      autovacuum_analyze_scale_factor: "0.05"
      autovacuum_naptime: 15s
      autovacuum_vacuum_cost_limit: "200"
      autovacuum_vacuum_scale_factor: "0.1"
      effective_cache_size: 1024MB
      log_lock_waits: "on"
      log_min_duration_statement: 50ms
      max_connections: "400"
      max_slot_wal_keep_size: 2GB
      pg_stat_statements.max: "10000"
      pg_stat_statements.track: all
      shared_buffers: 512MB
      timezone: UTC
      track_activity_query_size: "4096"
      wal_compression: "on"
  managed:
    roles:
      - comment: Role placeholder for debezium grants. Further grants updated manually
        ensure: present
        name: debezium_grants
        replication: true
      - ensure: present
        inRoles:
        - debezium_grants
        inherit: true
        login: true
        name: debezium
        passwordSecret:
          name: clusterX-debezium-credentials-db
    

  monitoring:
    enablePodMonitor: false
    disableDefaultQueries: false
  
  
  bootstrap:
    initdb:
      database: log_crm
      secret:
        name: clusterX-credentials-db
```

### Relevant log output

```shell
"message":"syntax error at or near \"TABLE\"","query":"CREATE PUBLICATION \"dbz_publication\" FOR TABLE \"log_crm.crm_employees\", TABLE \"log_crm.crm_user_actions\""

"message":"relation \"log_crm.crm_employees,log_crm.crm_user_actions\" does not exist","query":"CREATE PUBLICATION \"dbz_publication\" FOR TABLE \"log_crm.crm_employees,log_crm.crm_user_actions\""
```

### Code of Conduct

- [x] I agree to follow this project's Code of Conduct

Hints from issue comments:
@dosubot: <!-- Answer -->
This is a confirmed bug in the Publication CRD's SQL generation logic. When you specify multiple tables, the operator incorrectly generates SQL like `CREATE PUBLICATION ... FOR TABLE table1, TABLE table2`, which is not valid in PostgreSQL. The correct syntax is to use a single `TABLE` keyword followed by a comma-separated list of tables, like `CREATE PUBLICATION ... FOR TABLE table1, table2` ([PostgreSQL docs](https://www.postgresql.org/docs/current/sql-createpublication.html)).

The cnpg kubectl plugin generates the correct SQL, but the CRD/controller implementation does not. There are currently no existing issues or pull requests tracking or fixing this bug in the main repository. As a workaround, you can use the cnpg plugin to create publications with multiple tables until the CRD logic is fixed. The bug is in how the CRD aggregates table names for the SQL statement, not in your YAML or usage.

References:
- [CloudNativePG PublicationSpec and Target docs](https://cloudnative-pg.io/documentation/1.26/cloudnative-pg.v1/)
- [kubectl plugin documentation](https://cloudnative-pg.io/documentation/1.26/kubectl-plugin/)
- [Relevant code struct definitions](https://github.com/cloudnative-pg/cloudnative-pg/blob/54a8d1a8ec8ee45c9d7b39ab180c12355a61035a/api/v1/publication_types.go)


<!-- Dosu Comment Footer -->
*To reply, just mention [@dosu](https://go.dosu.dev/dosubot).*

---
<sup>How did I do? [Good](https://app.dosu.dev/response-feedback/5994074e-a522-46d1-8b6f-7d91e7872398?feedback_type=great_response) | [Irrelevant](https://app.dosu.dev/response-feedback/5994074e-a522-46d1-8b6f-7d91e7872398?feedback_type=irrelevant_answer) | [Incorrect](https://app.dosu.dev/response-feedback/5994074e-a522-46d1-8b6f-7d91e7872398?feedback_type=incorrect_sources) | [Verbose](https://app.dosu.dev/response-feedback/5994074e-a522-46d1-8b6f-7d91e7872398?feedback_type=too_verbose) | [Hallucination](https://app.dosu.dev/response-feedback/5994074e-a522-46d1-8b6f-7d91e7872398?feedback_type=hallucination) | [Report 🐛](https://app.dosu.dev/response-feedback/5994074e-a522-46d1-8b6f-7d91e7872398?feedback_type=bug_report) | [Other](https://app.dosu.dev/response-feedback/5994074e-a522-46d1-8b6f-7d91e7872398?feedback_type=other)</sup>&nbsp;&nbsp;[![Chat with Dosu](https://dosu.dev/dosu-chat-badge.svg)](https://app.dosu.dev/89b818f3-8b5e-449b-bfb3-64a9360d9665/ask?utm_source=github)&nbsp;[![Join Discord](https://img.shields.io/badge/join-5865F2?logo=discord&logoColor=white&label=)](https://go.dosu.dev/discord-bot)&nbsp;[![Share on X](https://img.shields.io/badge/X-share-black)](https://twitter.com/intent/tweet?text=%40dosu_ai%20helped%20me%20solve%20this%20issue!&url=https%3A//github.com/cloudnative-pg/cloudnative-pg/issues/8588)@maryam-ismayilova-m10: @dosu so is there any existing issue regarding this bug in the project? i couldn't find one first of all@maryam-ismayilova-m10: UPD* after a few tests i have noticed that this bug is relevant for postgres version 13 . I could not reproduce it on postgres clusters of versions 16 and 17. @maryam-ismayilova-m10: but given that postgres version 13 is still considered as supported by cloudnative-pg (according to the latest doc at https://cloudnative-pg.io/documentation/1.26/supported_releases/ ) , i believe this is still to be considered a bug. @jsilvela: I think you're right. PG 14 also seems not to allow the multiple-table-object syntax.
And even on PG 17, the examples on the Postgres documentation express multiple tables as `CREATE PUBLICATION .... TABLE a, b`.
In CloudNativePG, I believe the kubectl-plugin way would handle this syntax. See https://cloudnative-pg.io/documentation/1.27/kubectl-plugin/#logical-replication-publications
In any case, worth fixing IMO@leonardoce: The allowed grammar in PG14 is:

https://github.com/postgres/postgres/blob/bc476f8b8b64c3f6f4e1c77f4baceda842fe88dc/src/backend/parser/gram.y#L9591-L9624

Starting from PG15 the grammar is really different:

https://github.com/postgres/postgres/blob/fa42213d4af6bc0dce5b8e9bffeef9afa20b8dd8/src/backend/parser/gram.y#L10302-L10345

The new syntax has been introduced by https://github.com/postgres/postgres/commit/5a2832465fd8 and supports specifying both tables and schema.


Solution patch:
diff --git a/docs/src/logical_replication.md b/docs/src/logical_replication.md
index 02dd5fbe..1848c083 100644
--- a/docs/src/logical_replication.md
+++ b/docs/src/logical_replication.md
@@ -84,12 +84,46 @@ In the above example:
 - It includes all tables (`spec.target.allTables: true`) from the `app`
   database (`spec.dbname`).
 
+### Fine-grained control over publication tables
+
+While the `allTables` option provides a convenient way to replicate all tables
+in a database, PostgreSQL version 15 and later introduce enhanced flexibility
+through the [`CREATE PUBLICATION`](https://www.postgresql.org/docs/current/sql-createpublication.html)
+command. This allows you to precisely define which tables, or even which types
+of data changes, should be included in a publication.
+
 !!! Important
-    While `allTables` simplifies configuration, PostgreSQL offers fine-grained
-    control for replicating specific tables or targeted data changes. For advanced
-    configurations, consult the [PostgreSQL documentation](https://www.postgresql.org/docs/current/logical-replication.html).
-    Additionally, refer to the [CloudNativePG API reference](cloudnative-pg.v1.md#postgresql-cnpg-io-v1-PublicationTarget)
-    for details on declaratively customizing replication targets.
+    If you are using PostgreSQL versions earlier than 15, review the syntax and
+    options available for `CREATE PUBLICATION` in your specific release. Some
+    parameters and features may not be supported.
+
+For complex or tailored replication setups, refer to the
+[PostgreSQL logical replication documentation](https://www.postgresql.org/docs/current/logical-replication.html).
+
+Additionally, refer to the [CloudNativePG API reference](cloudnative-pg.v1.md#postgresql-cnpg-io-v1-PublicationTarget)
+for details on declaratively customizing replication targets.
+
+The following example defines a publication that replicates all tables in the
+`portal` schema of the `app` database, along with the `users` table from the
+`access` schema:
+
+```yaml
+apiVersion: postgresql.cnpg.io/v1
+kind: Publication
+metadata:
+  name: publisher
+spec:
+  cluster:
+    name: freddie
+  dbname: app
+  name: publisher
+  target:
+    objects:
+      - tablesInSchema: portal
+      - table:
+          name: users
+          schema: access
+```
 
 ### Required Fields in the `Publication` Manifest
 
@@ -347,7 +381,7 @@ metadata:
 spec:
   instances: 1
 
-  imageName: ghcr.io/cloudnative-pg/postgresql:16
+  imageName: ghcr.io/cloudnative-pg/postgresql:16-standard-trixie
 
   storage:
     size: 1Gi
@@ -396,6 +430,8 @@ metadata:
 spec:
   instances: 1
 
+  imageName: ghcr.io/cloudnative-pg/postgresql:18-standard-trixie
+
   storage:
     size: 1Gi
 
diff --git a/internal/management/controller/publication_controller_sql.go b/internal/management/controller/publication_controller_sql.go
index c9d2eee1..b9cd8a4f 100644
--- a/internal/management/controller/publication_controller_sql.go
+++ b/internal/management/controller/publication_controller_sql.go
@@ -151,38 +151,52 @@ func toPublicationTargetSQL(obj *apiv1.PublicationTarget) string {
 }
 
 func toPublicationTargetObjectsSQL(obj *apiv1.PublicationTarget) string {
-	result := ""
+	var parts []string
+	var currentTables []string
+
+	flushTables := func() {
+		if len(currentTables) > 0 {
+			parts = append(parts, fmt.Sprintf("TABLE %s", strings.Join(currentTables, ", ")))
+			currentTables = nil
+		}
+	}
+
 	for _, object := range obj.Objects {
-		if len(result) > 0 {
-			result += ", "
+		if len(object.TablesInSchema) > 0 {
+			// Flush any accumulated tables before adding schema
+			flushTables()
+			parts = append(parts, fmt.Sprintf("TABLES IN SCHEMA %s", pgx.Identifier{object.TablesInSchema}.Sanitize()))
+		} else if object.Table != nil {
+			// Accumulate table definitions
+			// Note: Grouping consecutive tables under a single TABLE keyword ensures
+			// compatibility with PostgreSQL 13/14. Mixed TABLE and TABLES IN SCHEMA
+			// in the same publication requires PostgreSQL 15+.
+			currentTables = append(currentTables, toTableDefinitionSQL(object.Table))
 		}
-		result += toPublicationObjectSQL(&object)
 	}
 
-	return result
-}
+	// Flush any remaining tables
+	flushTables()
 
-func toPublicationObjectSQL(obj *apiv1.PublicationTargetObject) string {
-	if len(obj.TablesInSchema) > 0 {
-		return fmt.Sprintf("TABLES IN SCHEMA %s", pgx.Identifier{obj.TablesInSchema}.Sanitize())
-	}
+	return strings.Join(parts, ", ")
+}
 
+func toTableDefinitionSQL(table *apiv1.PublicationTargetTable) string {
 	result := strings.Builder{}
-	result.WriteString("TABLE ")
 
-	if obj.Table.Only {
+	if table.Only {
 		result.WriteString("ONLY ")
 	}
 
-	if len(obj.Table.Schema) > 0 {
-		result.WriteString(fmt.Sprintf("%s.", pgx.Identifier{obj.Table.Schema}.Sanitize()))
+	if len(table.Schema) > 0 {
+		result.WriteString(fmt.Sprintf("%s.", pgx.Identifier{table.Schema}.Sanitize()))
 	}
 
-	result.WriteString(pgx.Identifier{obj.Table.Name}.Sanitize())
+	result.WriteString(pgx.Identifier{table.Name}.Sanitize())
 
-	if len(obj.Table.Columns) > 0 {
-		sanitizedColumns := make([]string, 0, len(obj.Table.Columns))
-		for _, column := range obj.Table.Columns {
+	if len(table.Columns) > 0 {
+		sanitizedColumns := make([]string, 0, len(table.Columns))
+		for _, column := range table.Columns {
 			sanitizedColumns = append(sanitizedColumns, pgx.Identifier{column}.Sanitize())
 		}
 		result.WriteString(fmt.Sprintf(" (%s)", strings.Join(sanitizedColumns, ", ")))



Test patch:
diff --git a/internal/management/controller/publication_controller_sql_test.go b/internal/management/controller/publication_controller_sql_test.go
index d7635e90..b65ed54a 100644
--- a/internal/management/controller/publication_controller_sql_test.go
+++ b/internal/management/controller/publication_controller_sql_test.go
@@ -173,56 +173,59 @@ var _ = Describe("publication sql", func() {
 	})
 })
 
-var _ = Describe("toPublicationObjectSQL", func() {
-	It("returns correct SQL for tables in schema", func() {
-		obj := &apiv1.PublicationTargetObject{
-			TablesInSchema: "public",
-		}
-		result := toPublicationObjectSQL(obj)
-		Expect(result).To(Equal(`TABLES IN SCHEMA "public"`))
-	})
-
-	It("returns correct SQL for table with schema and columns", func() {
-		obj := &apiv1.PublicationTargetObject{
-			Table: &apiv1.PublicationTargetTable{
-				Name:    "table",
-				Schema:  "test",
-				Columns: []string{"a", "b"},
+var _ = Describe("toPublicationCreateSQL", func() {
+	It("returns correct SQL for multiple tables", func() {
+		obj := &apiv1.Publication{
+			Spec: apiv1.PublicationSpec{
+				Name: "test_pub",
+				Target: apiv1.PublicationTarget{
+					Objects: []apiv1.PublicationTargetObject{
+						{Table: &apiv1.PublicationTargetTable{Name: "crm_employees", Schema: "log_crm"}},
+						{Table: &apiv1.PublicationTargetTable{Name: "crm_user_actions", Schema: "log_crm"}},
+					},
+				},
 			},
 		}
-		result := toPublicationObjectSQL(obj)
-		Expect(result).To(Equal(`TABLE "test"."table" ("a", "b")`))
-	})
 
-	It("returns correct SQL for table with only clause", func() {
-		obj := &apiv1.PublicationTargetObject{
-			Table: &apiv1.PublicationTargetTable{
-				Name: "table",
-				Only: true,
-			},
-		}
-		result := toPublicationObjectSQL(obj)
-		Expect(result).To(Equal(`TABLE ONLY "table"`))
+		sql := toPublicationCreateSQL(obj)
+		Expect(sql).To(Equal(
+			`CREATE PUBLICATION "test_pub" FOR TABLE "log_crm"."crm_employees", "log_crm"."crm_user_actions"`))
 	})
 
-	It("returns correct SQL for table without schema and columns", func() {
-		obj := &apiv1.PublicationTargetObject{
-			Table: &apiv1.PublicationTargetTable{
-				Name: "table",
+	It("returns correct SQL for mixed tables and schemas (requires pg15+)", func() {
+		obj := &apiv1.Publication{
+			Spec: apiv1.PublicationSpec{
+				Name: "test_pub",
+				Target: apiv1.PublicationTarget{
+					Objects: []apiv1.PublicationTargetObject{
+						{Table: &apiv1.PublicationTargetTable{Name: "table1"}},
+						{Table: &apiv1.PublicationTargetTable{Name: "table2"}},
+						{TablesInSchema: "public"},
+						{Table: &apiv1.PublicationTargetTable{Name: "table3"}},
+					},
+				},
 			},
 		}
-		result := toPublicationObjectSQL(obj)
-		Expect(result).To(Equal(`TABLE "table"`))
+
+		sql := toPublicationCreateSQL(obj)
+		Expect(sql).To(Equal(
+			`CREATE PUBLICATION "test_pub" FOR TABLE "table1", "table2", TABLES IN SCHEMA "public", TABLE "table3"`))
 	})
 
-	It("returns correct SQL for table with schema but without columns", func() {
-		obj := &apiv1.PublicationTargetObject{
-			Table: &apiv1.PublicationTargetTable{
-				Name:   "table",
-				Schema: "test",
+	It("returns correct SQL for multiple tables with columns", func() {
+		obj := &apiv1.Publication{
+			Spec: apiv1.PublicationSpec{
+				Name: "test_pub",
+				Target: apiv1.PublicationTarget{
+					Objects: []apiv1.PublicationTargetObject{
+						{Table: &apiv1.PublicationTargetTable{Name: "table1", Columns: []string{"a", "b"}}},
+						{Table: &apiv1.PublicationTargetTable{Name: "table2", Columns: []string{"c"}}},
+					},
+				},
 			},
 		}
-		result := toPublicationObjectSQL(obj)
-		Expect(result).To(Equal(`TABLE "test"."table"`))
+
+		sql := toPublicationCreateSQL(obj)
+		Expect(sql).To(Equal(`CREATE PUBLICATION "test_pub" FOR TABLE "table1" ("a", "b"), "table2" ("c")`))
 	})
 })