Fix postgres_cdc input

Allow quoted identifiers for the table names

Signed-off-by: Mihai Todor <[email protected]>

CHANGELOG.md

@@ -8,6 +8,7 @@ All notable changes to this project will be documented in this file.
 ### Fixed
 
 - `gcp_bigquery` output with parquet format no longer returns errors incorrectly. (@rockwotj)
+- `postgres_cdc` input now allows quoted identifiers for the table names. (@mihaitodor)
 
 ## 4.43.1 - 2024-12-09
 

internal/impl/postgresql/pglogicalstream/pglogrepl.go

@@ -430,7 +430,7 @@ func CreatePublication(ctx context.Context, conn *pgconn.PgConn, publicationName
 
 	// remove tables from publication
 	for _, dropTable := range tablesToRemoveFromPublication {
-		sq, err := sanitize.SQLQuery(fmt.Sprintf("ALTER PUBLICATION %s DROP TABLE %s;", publicationName, dropTable))
+		sq, err := sanitize.SQLQuery(fmt.Sprintf(`ALTER PUBLICATION %s DROP TABLE "%s";`, publicationName, dropTable))
 		if err != nil {
 			return fmt.Errorf("failed to sanitize drop table query: %w", err)
 		}

internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go

@@ -366,6 +366,18 @@ func SQLQuery(sql string, args ...any) (string, error) {
 // ValidatePostgresIdentifier checks if a string is a valid PostgreSQL identifier
 // This follows PostgreSQL's standard naming rules
 func ValidatePostgresIdentifier(name string) error {
+	if parts := strings.Split(name, "."); len(parts) == 2 {
+		if err := ValidatePostgresIdentifier(parts[0]); err != nil {
+			return fmt.Errorf("invalid schema identifier: %s", err)
+		}
+		name = parts[1]
+	}
+
+	// Strip quotes if they are present
+	if strings.HasPrefix(name, "\"") && strings.HasSuffix(name, "\"") {
+		name = strings.Trim(name, "\"")
+	}
+
 	if len(name) == 0 {
 		return errors.New("empty identifier is not allowed")
 	}