fix: postgres pgoutput part publication with partition root table

Author: jarad0628

connectors-common/debezium-bucket/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java

@@ -171,7 +171,15 @@ protected void initPublication() {
                                 Set<TableId> tablesToCapture = determineCapturedTables();
                                 Set<TableId> existsTables = new HashSet<>();
                                 String catalog = connection().getCatalog();
-                                try (ResultSet resultSet = stmt.executeQuery(String.format("select * from pg_publication_tables where pubname='%s'", publicationName))) {
+                                try (ResultSet resultSet = stmt.executeQuery(String.format("select schemaname, tablename from pg_publication_tables pt join pg_publication pp\n" +
+                                        "    on pt.pubname = pp.pubname\n" +
+                                        "where pt.pubname='%s' and pp.pubviaroot='false'\n" +
+                                        "union all\n" +
+                                        "select schemaname, pc.relname from pg_publication_tables pt join pg_publication pp\n" +
+                                        "    on pt.pubname = pp.pubname join pg_class pc on pc.oid in (SELECT inhrelid\n" +
+                                        "    FROM pg_inherits\n" +
+                                        "    WHERE inhparent = pt.tablename::regclass)\n" +
+                                        "where pt.pubname='%s' and pp.pubviaroot='true'", publicationName, publicationName))) {
                                     while (resultSet.next()) {
                                         existsTables.add(new TableId(catalog, resultSet.getString("schemaname"), resultSet.getString("tablename")));
                                     }

connectors/postgres-connector/src/main/java/io/tapdata/connector/postgres/PostgresConnector.java

@@ -861,7 +861,7 @@ private void createAllPublicationIfNotExist() throws SQLException {
     }
 
     private void createCustomPublicationIfNotExist(List<String> tableList) {
-        String sql = String.format("CREATE PUBLICATION %s FOR TABLE %s", slotName, tableList.stream().map(this::getSchemaAndTable).collect(Collectors.joining(", ")));
+        String sql = String.format("CREATE PUBLICATION %s FOR TABLE %s %s", slotName, tableList.stream().map(this::getSchemaAndTable).collect(Collectors.joining(", ")), postgresConfig.getPartitionRoot() ? "WITH (publish_via_partition_root = true)" : "");
         try {
             tapLogger.info("Create publication sql: {}", sql);
             postgresJdbcContext.execute(sql);