fix(cdc): fix uppercase-identifier of pgcdc

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/common/PostgresValidator.java

@@ -180,7 +180,8 @@ private void validateTableSchema() throws SQLException {
         // check primary key
         // reference: https://wiki.postgresql.org/wiki/Retrieve_primary_key_columns
         try (var stmt = jdbcConnection.prepareStatement(ValidatorUtils.getSql("postgres.pk"))) {
-            stmt.setString(1, this.schemaName + "." + this.tableName);
+            stmt.setString(1, schemaName);
+            stmt.setString(2, tableName);
             var res = stmt.executeQuery();
             var pkFields = new HashSet<String>();
             while (res.next()) {

java/connector-node/risingwave-connector-service/src/main/resources/validate_sql.properties

@@ -6,7 +6,7 @@ mysql.table_schema=SELECT COLUMN_NAME, DATA_TYPE, COLUMN_KEY FROM information_sc
 mysql.grants=SHOW GRANTS FOR CURRENT_USER()
 postgres.wal=show wal_level
 postgres.table=SELECT EXISTS ( SELECT FROM pg_tables WHERE schemaname = ? AND tablename = ?)
-postgres.pk=SELECT a.attname, format_type(a.atttypid, a.atttypmod) AS data_type FROM pg_index i JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) WHERE i.indrelid = ?::regclass AND i.indisprimary
+postgres.pk=SELECT a.attname, format_type(a.atttypid, a.atttypmod) AS data_type FROM pg_index i JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) JOIN pg_class c ON c.oid = i.indrelid JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = ? AND c.relname = ? AND i.indisprimary
 postgres.table_schema=SELECT column_name, data_type FROM information_schema.columns WHERE table_schema = ? AND table_name = ? ORDER BY ordinal_position
 postgres.slot.check=SELECT slot_name FROM pg_replication_slots WHERE slot_name = ? and database = ?
 postgres.slot_limit.check=SELECT CASE WHEN (SELECT count(*) FROM pg_replication_slots) = (SELECT setting FROM pg_settings WHERE name='max_replication_slots')::int THEN 'true' ELSE 'false' END AS result;