Repository: xataio/pgstream (go)

Issue:
Syntax error when building create table query
In `postgres_wal_ddl_adapter.go` we are doing a syntax error when building `create table` query:

```go
func (a *ddlAdapter) buildCreateTableQuery(schemaName string, table schemalog.Table) *query {
	createQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (", quotedTableName(schemaName, table.Name))
	uniqueConstraints := make([]string, 0, len(table.Columns))
	for _, col := range table.Columns {
		createQuery = fmt.Sprintf("%s %s,\n", createQuery, a.buildColumnDefinition(&col)) // THE PROBLEMATIC LINE
		// if there's a unique constraint associated to the column, and it's not
		// the primary key, explicitly add it
		if uniqueConstraint := a.buildUniqueConstraint(col); uniqueConstraint != "" && !slices.Contains(table.PrimaryKeyColumns, col.Name) {
			uniqueConstraints = append(uniqueConstraints, uniqueConstraint)
		}
	}
```

Here we add a comma after each column. This causes a syntax error if there's no unique constraint or primary key on the table. Resulting in a query like: `CREATE TABLE IF NOT EXISTS "test_sc"."test_table" ("a" integer,)`

Pull request description:
Fix syntax error with ddl adapter
Fixes #305 

Patch:
diff --git a/pkg/wal/processor/postgres/postgres_wal_ddl_adapter.go b/pkg/wal/processor/postgres/postgres_wal_ddl_adapter.go
index 9d36139..7a5c690 100644
--- a/pkg/wal/processor/postgres/postgres_wal_ddl_adapter.go
+++ b/pkg/wal/processor/postgres/postgres_wal_ddl_adapter.go
@@ -78,8 +78,9 @@ func (a *ddlAdapter) schemaDiffToQueries(schemaName string, diff *schemalog.Diff
 func (a *ddlAdapter) buildCreateTableQuery(schemaName string, table schemalog.Table) *query {
 	createQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (", quotedTableName(schemaName, table.Name))
 	uniqueConstraints := make([]string, 0, len(table.Columns))
+	columnDefinitions := make([]string, 0, len(table.Columns))
 	for _, col := range table.Columns {
-		createQuery = fmt.Sprintf("%s %s,\n", createQuery, a.buildColumnDefinition(&col))
+		columnDefinitions = append(columnDefinitions, a.buildColumnDefinition(&col))
 		// if there's a unique constraint associated to the column, and it's not
 		// the primary key, explicitly add it
 		if uniqueConstraint := a.buildUniqueConstraint(col); uniqueConstraint != "" && !slices.Contains(table.PrimaryKeyColumns, col.Name) {
@@ -87,8 +88,9 @@ func (a *ddlAdapter) buildCreateTableQuery(schemaName string, table schemalog.Ta
 		}
 	}
 
-	for _, constraint := range uniqueConstraints {
-		createQuery = fmt.Sprintf("%s %s,\n", createQuery, constraint)
+	createQuery = fmt.Sprintf("%s\n%s", createQuery, strings.Join(columnDefinitions, ",\n"))
+	if len(uniqueConstraints) > 0 {
+		createQuery = fmt.Sprintf("%s,\n%s", createQuery, strings.Join(uniqueConstraints, ",\n"))
 	}
 
 	primaryKeys := make([]string, 0, len(table.PrimaryKeyColumns))
@@ -97,10 +99,10 @@ func (a *ddlAdapter) buildCreateTableQuery(schemaName string, table schemalog.Ta
 	}
 
 	if len(primaryKeys) > 0 {
-		createQuery = fmt.Sprintf("%s PRIMARY KEY (%s)\n", createQuery, strings.Join(primaryKeys, ","))
+		createQuery = fmt.Sprintf("%s,\nPRIMARY KEY (%s)\n", createQuery, strings.Join(primaryKeys, ","))
 	}
 
-	createQuery = fmt.Sprintf("%s)", createQuery)
+	createQuery += ")"
 
 	return a.newDDLQuery(createQuery)
 }


Test patch:
diff --git a/pkg/wal/processor/postgres/postgres_wal_ddl_adapter_test.go b/pkg/wal/processor/postgres/postgres_wal_ddl_adapter_test.go
index febbdd3..fe7a329 100644
--- a/pkg/wal/processor/postgres/postgres_wal_ddl_adapter_test.go
+++ b/pkg/wal/processor/postgres/postgres_wal_ddl_adapter_test.go
@@ -178,7 +178,30 @@ func TestDDLAdapter_schemaDiffToQueries(t *testing.T) {
 
 			wantQueries: []*query{
 				{
-					sql:   fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s ( \"id\" uuid NOT NULL,\n \"name\" text,\n \"age\" int NOT NULL DEFAULT 0,\n UNIQUE (\"name\"),\n PRIMARY KEY (\"id\")\n)", quotedTableName(testSchema, table1)),
+					sql:   fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (\n\"id\" uuid NOT NULL,\n\"name\" text,\n\"age\" int NOT NULL DEFAULT 0,\nUNIQUE (\"name\"),\nPRIMARY KEY (\"id\")\n)", quotedTableName(testSchema, table1)),
+					isDDL: true,
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "ok - table added no primary key or unique constraint",
+			diff: &schemalog.Diff{
+				TablesAdded: []schemalog.Table{
+					{
+						Name: table1,
+						Columns: []schemalog.Column{
+							{Name: "id", DataType: "uuid", Nullable: false, Unique: false},
+							{Name: "name", DataType: "text", Nullable: true, Unique: false},
+							{Name: "age", DataType: "int", Nullable: false, DefaultValue: &defaultAge},
+						},
+					},
+				},
+			},
+
+			wantQueries: []*query{
+				{
+					sql:   fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (\n\"id\" uuid NOT NULL,\n\"name\" text,\n\"age\" int NOT NULL DEFAULT 0)", quotedTableName(testSchema, table1)),
 					isDDL: true,
 				},
 			},