[BUG] RDS PostgreSQL source: CREATE PUBLICATION uses unquoted identifiers, failing on hyphenated or mixed-case database/schema/table names

Describe the bug

The RDS/Aurora PostgreSQL source builds its CREATE PUBLICATION … FOR TABLE … statement by concatenating schema-discovered identifiers undelimited. PostgreSQL folds undelimited identifiers to lower case and rejects some characters outright, so any database, schema, or table whose name requires a delimited (double-quoted) identifier — a hyphen, an uppercase letter, or a reserved word — makes publication creation fail. That aborts the pipeline during leader/stream initialization, so the pipeline never reaches a running state and no data is ingested.

There is no configuration workaround. The identifiers come from schema discovery rather than from tables.include, and the publication and slot names are generated per run with a random suffix, so the publication cannot be pre-created out of band either.

To Reproduce

The failure is entirely in the generated DDL, so it reproduces against plain PostgreSQL — no RDS or Aurora required:

docker run -d --name pgdemo -e POSTGRES_PASSWORD=demo postgres:17 -c wal_level=logical

-- Case 1: a database name containing a hyphen
CREATE DATABASE "My-Db-1";
\c "My-Db-1"
CREATE SCHEMA dbo;
CREATE TABLE dbo."MyTable" (id int PRIMARY KEY);

CREATE PUBLICATION dp_unquoted FOR TABLE My-Db-1.dbo.MyTable;        -- what is generated today
-- ERROR:  syntax error at or near "-"
-- LINE 1: CREATE PUBLICATION dp_unquoted FOR TABLE My-Db-1.dbo.MyTable...

CREATE PUBLICATION dp_quoted FOR TABLE "My-Db-1"."dbo"."MyTable";    -- delimited
-- CREATE PUBLICATION

-- Case 2: a lower-case database with a mixed-case table (a different error, same root cause)
\c postgres
CREATE DATABASE mydb;
\c mydb
CREATE SCHEMA dbo;
CREATE TABLE dbo."MyTable" (id int PRIMARY KEY);

CREATE PUBLICATION dp_unquoted FOR TABLE mydb.dbo.MyTable;           -- what is generated today
-- ERROR:  relation "dbo.mytable" does not exist

CREATE PUBLICATION dp_quoted FOR TABLE "mydb"."dbo"."MyTable";       -- delimited
-- CREATE PUBLICATION

Verified on PostgreSQL 17.10.

Through a pipeline: point an rds source with engine: aurora-postgresql and stream: true at a database named My-Db-1 containing dbo."MyTable". Initialization fails and the leader loop logs:

ERROR org.opensearch.dataprepper.plugins.source.rds.leader.LeaderScheduler - Exception occurred in primary leader scheduling loop
java.lang.RuntimeException: org.postgresql.util.PSQLException: ERROR: syntax error at or near "-"
  Position: 79
    at ...rds.schema.PostgresSchemaManager.createLogicalReplicationSlot(PostgresSchemaManager.java:102)
    at ...rds.leader.LeaderScheduler.createStreamPartition(LeaderScheduler.java:189)

(That trace is from the Amazon OpenSearch Ingestion managed runtime, so its line numbers reflect that build rather than current main; the code path is unchanged on main.)

Expected behavior

Generated DDL should delimit each part of the database/schema/table identifier, so that names containing hyphens, uppercase letters, or reserved words work — consistent with how the slot name is already bound safely via setString.

Environment

• Data Prepper: reproduced on the Amazon OpenSearch Ingestion managed runtime; code path confirmed unchanged on main (and in 2.16.0). PostgresSchemaManager.java has not been modified since #5699 (May 2025).
• Source: Aurora PostgreSQL 17 (engine: aurora-postgresql), stream: true
• Standalone repro above: PostgreSQL 17.10 on Debian

Additional context

Root cause. LeaderScheduler.createStreamPartition takes the table list from schema discovery:

tableNames = new ArrayList<>(dbTableMetadata.getTableColumnDataTypeMap().keySet());

Those keys are built by PostgresSchemaManager.getTableNames as raw database.schema.table strings:

tableNames.add(databaseName + "." + schemaName + "." + tableName);

and createLogicalReplicationSlot concatenates them into the statement without delimiting anything:

StringBuilder createPublicationStatementBuilder = new StringBuilder("CREATE PUBLICATION ")
        .append(publicationName)
        .append(" FOR TABLE ");
for (int i = 0; i < tableNames.size(); i++) {
    createPublicationStatementBuilder.append(tableNames.get(i));
    ...

Because the names come from JDBC metadata (TABLE_SCHEM / TABLE_NAME), they carry the true stored case — so delimiting them is not just safe, it is what makes the lookup correct.

I have a fix ready and will open a PR referencing this issue.

Possible follow-up, out of scope here: the same class splits a fully qualified name on "\\." in several places (getPrimaryKeysForTable, getColumnDataTypesForTable, getEnumColumnsForTable), which also misparses any identifier containing a literal dot. That is a pre-existing, separate issue from the quoting bug.

Start in PostgresSchemaManager.createLogicalReplicationSlot and review how LeaderScheduler.createStreamPartition supplies the discovered table names. Reproduce the generated DDL with the PostgreSQL examples in the issue, then verify that database, schema, and table identifiers are delimited and publication creation succeeds for hyphenated and mixed-case names.