@@ -0,0 +1,2434 @@
+package org.apache.flink.api.java.io.jdbc.catalog;%0A%0Aimport org.apache.flink.table.api.EnvironmentSettings;%0Aimport org.apache.flink.table.api.TableEnvironment;%0Aimport org.apache.flink.table.api.TableUtils;%0Aimport org.apache.flink.types.Row;%0A%0Aimport org.junit.Test;%0A%0Aimport java.util.List;%0A%0Aimport static org.apache.flink.api.java.io.jdbc.catalog.PostgresCatalog.DEFAULT_DATABASE;%0Aimport static org.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM;%0Aimport static org.junit.Assert.assertEquals;%0A%0A/**%0A * E2E test for %7B@link PostgresCatalog%7D.%0A */%0Apublic class PostgresCatalogITCase extends PostgresCatalogTestBase %7B%0A%0A%09@Test%0A%09public void test_withoutSchema() throws Exception %7B%0A%09%09TableEnvironment tEnv = getTableEnvWithPgCatalog();%0A%0A%09%09List%3CRow%3E results = TableUtils.collectToList(%0A%09%09%09tEnv.sqlQuery(String.format(%22select * from %25s%22, TABLE1)));%0A%09%09assertEquals(%22%5B1%5D%22, results.toString());%0A%09%7D%0A%0A%09@Test%0A%09public void test_withSchema() throws Exception %7B%0A%09%09TableEnvironment tEnv = getTableEnvWithPgCatalog();%0A%0A%09%09List%3CRow%3E results = TableUtils.collectToList(%0A%09%09%09tEnv.sqlQuery(String.format(%22select * from %60%25s%60%22, PostgresTablePath.fromFlinkTableName(TABLE1))));%0A%09%09assertEquals(%22%5B1%5D%22, results.toString());%0A%09%7D%0A%0A%09@Test%0A%09public void test_fullPath() throws Exception %7B%0A%09%09TableEnvironment tEnv = getTableEnvWithPgCatalog();%0A%0A%09%09List%3CRow%3E results = TableUtils.collectToList(%0A%09%09%09tEnv.sqlQuery(String.format(%22select * from %25s.%25s.%60%25s%60%22,%0A%09%09%09%09TEST_CATALOG_NAME,%0A%09%09%09%09DEFAULT_DATABASE,%0A%09%09%09%09PostgresTablePath.fromFlinkTableName(TABLE1))));%0A%09%09assertEquals(%22%5B1%5D%22, results.toString());%0A%09%7D%0A%0A%09@Test%0A%09public void test_insert() throws Exception %7B%0A%09%09TableEnvironment tEnv = getTableEnvWithPgCatalog();%0A%0A%09%09tEnv.sqlUpdate(String.format(%22insert into %25s select * from %60%25s%60%22, TABLE4, TABLE1));%0A%09%09tEnv.execute(%22test%22);%0A%0A%09%09List%3CRow%3E results = TableUtils.collectToList(%0A%09%09%09tEnv.sqlQuery(String.format(%22select * from %25s%22, TABLE1)));%0A%09%09assertEquals(%22%5B1%5D%22, results.toString());%0A%09%7D%0A%0A%09private TableEnvironment getTableEnvWithPgCatalog() %7B%0A%09%09EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();%0A%09%09TableEnvironment tableEnv = TableEnvironment.create(settings);%0A%09%09tableEnv.getConfig().getConfiguration().setInteger(TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM.key(), 1);%0A%0A%09%09tableEnv.registerCatalog(TEST_CATALOG_NAME, catalog);%0A%09%09tableEnv.useCatalog(TEST_CATALOG_NAME);%0A%09%09return tableEnv;%0A%09%7D%0A%7D%0A

add back PostgresCatalogITCase