Flinksql read / write PgSQL

One . Code

 StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, EnvironmentSettings.inStreamingMode());

        AbstractJdbcCatalog catalog = JdbcCatalogUtils.createCatalog(
                "myPgsql",
                "staging",
                "username",
                "password",
                "jdbc:postgresql://xxx:port"
        );

        tableEnv.registerCatalog("myPgsql", catalog);
        tableEnv.useCatalog("myPgsql");
        tableEnv.useDatabase("staging");
        boolean staging = catalog.tableExists(new ObjectPath("staging", "medical.hospital_department"));
        System.out.println(staging);
        System.out.println(Arrays.toString(tableEnv.listCatalogs()));
        System.out.println(Arrays.toString(tableEnv.listDatabases()));
        //System.out.println(Arrays.toString(tableEnv.listTables()));
        tableEnv.executeSql("select id,name from `medical.hospital_department`").print();

//sql api
tableEnv.executeSql(
                "CREATE TABLE hospital_position (\n" +
                        " id string,\n" +
                        " department_id string,\n" +
                        " waypoint_id STRING,\n" +
                        " zone_id STRING,\n" +
                        " PRIMARY KEY (id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        " 'connector' = 'jdbc',\n" +
                        " 'url' = 'jdbc:postgresql://xxxxx:port/ database ',\n" +
                        " 'username' = 'xx', " +
                        " 'password' = 'xxxx', " +
                        " 'table-name' = 'schema name . Table name '\n" +
                        ")"
        );

2 Step on the pit

If pgsql Used schema,name At this time, the table name needs `` escape , Otherwise, we will not find medical object

// `medical.hospital_department`  Must escape 
        tableEnv.executeSql("select id,name from `medical.hospital_department`").print();