Fix postgres_cdc input

[mihaitodor]

Allow quoted identifiers for the table names.

Note: This is a very naive implementation and we might want to instead just quote the identifier like @rockwotj suggested (and implemented here for Snowflake).

If we always quote, then it's a bit unclear how to validate the identifiers because we can't tell when it genuinely needs to be quoted to apply the alternate validation rules: https://www.postgresql.org/docs/current/sql-syntax-lexical.html

Quoted identifiers can contain any character, except the character with code zero. (To include a double quote, write two double quotes.) This allows constructing table or column names that would otherwise not be possible, such as ones containing spaces or ampersands. The length limitation still applies.

Also, users might get confused if they specify the table name as upper case in the config for a non-quoted table:

> CREATE TABLE MIHAI(ID INT NOT NULL, CATEGORY VARCHAR);
> INSERT INTO "MIHAI"(ID, CATEGORY) VALUES(13, 'FOOBAR');
relation "MIHAI" does not exist
LINE 1: INSERT INTO "MIHAI"(ID, CATEGORY) VALUES(13, 'FOOBAR')

How to test

Run Postgres and connect to it via pgcli:

$ docker run --rm -it -e POSTGRES_USER=testuser -e "POSTGRES_PASSWORD=testpass" -e POSTGRES_DB=testdb -p5432:5432 postgres -c wal_level=logical
$ pgcli postgres://testuser:testpass@localhost:5432/testdb

Create a quoted table:

> CREATE SCHEMA "TestABC";
> CREATE TABLE "TestABC"."FooBar"(ID INT NOT NULL, CATEGORY VARCHAR);
> CREATE TABLE "TestABC"."BarFoo"(ID INT NOT NULL, CATEGORY VARCHAR);

Run Connect:

input:
  postgres_cdc:
    dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable
    include_transaction_markers: true
    stream_snapshot: false
    snapshot_memory_safety_factor: 1
    snapshot_batch_size: 0
    schema: public # No default (required)
    tables:
      - '"FooBar"'
      - '"BarFoo"'
    checkpoint_limit: 1024
    temporary_slot: false
    slot_name: "mihaifoo" # No default (optional)
    pg_standby_timeout: 10s
    pg_wal_monitor_interval: 3s
    max_parallel_snapshot_tables: 1
    auto_replay_nacks: true

output:
  stdout: {}

Insert rows:

> INSERT INTO "TestABC"."FooBar"(ID, CATEGORY) VALUES(13, 'FOOBAR');
> INSERT INTO "TestABC"."BarFoo"(ID, CATEGORY) VALUES(13, 'FOOBAR');

Comment out the "BarFoo" table in the Connect config, restart it and insert another row in "FooBar".

[rockwotj]

Okay updates here:

• Added a new type to clarify we always are working with table FQN (with schema)
• We only validate identifiers once - at the package entrypoint
• We validate quoted identifiers

[rockwotj]

I also updated one of our integration tests to use a PascalCase name to codify the test that Mihai wrote in the PR.

thank you @mihaitodor for this! If you have time today can you double check my commits look OK? If so please merge.

[rockwotj]

Technically this also opens us up to allowing a single input with different schemas - not sure if you see that as needed @ligfx

[ligfx]

Looks good to me and works for me locally. Love the normalization of passed in identifiers.

[ligfx]

return `"` + name.Replace(`"`, `""`) + `"`

?

[rockwotj]

Yeah that's simpler. This happens to be faster, but this isn't on a critical path. I'll fix it next time I hack on this input