Allow access to Postgres partitioned tables

docs/src/main/sphinx/connector/postgresql.rst

@@ -12,7 +12,7 @@ Requirements
 
 To connect to PostgreSQL, you need:
 
-* PostgreSQL 9.6 or higher.
+* PostgreSQL 10.x or higher.
 * Network access from the Trino coordinator and workers to PostgreSQL.
   Port 5432 is the default port.
 

plugin/trino-postgresql/src/main/java/io/trino/plugin/postgresql/PostgreSqlClient.java

@@ -277,7 +277,7 @@ public PostgreSqlClient(
         this.varcharMapType = (MapType) typeManager.getType(mapType(VARCHAR.getTypeSignature(), VARCHAR.getTypeSignature()));
 
         ImmutableList.Builder<String> tableTypes = ImmutableList.builder();
-        tableTypes.add("TABLE", "VIEW", "MATERIALIZED VIEW", "FOREIGN TABLE");
+        tableTypes.add("TABLE", "PARTITIONED TABLE", "VIEW", "MATERIALIZED VIEW", "FOREIGN TABLE");
         if (postgreSqlConfig.isIncludeSystemTables()) {
             tableTypes.add("SYSTEM TABLE", "SYSTEM VIEW");
         }

plugin/trino-postgresql/src/test/java/io/trino/plugin/postgresql/TestPostgreSqlConnectorTest.java

@@ -204,6 +204,41 @@ public void testSystemTable()
                 .contains("orders");
     }
 
+    @Test
+    public void testPartitionedTables()
+            throws Exception
+    {
+        try (TestTable testTable = new TestTable(
+                postgreSqlServer::execute,
+                "test_part_tbl",
+                "(id int NOT NULL, payload varchar, logdate date NOT NULL) PARTITION BY RANGE (logdate)")) {
+            String values202111 = "(1, 'A', '2021-11-01'), (2, 'B', '2021-11-25')";
+            String values202112 = "(3, 'C', '2021-12-01')";
+            execute(format("CREATE TABLE %s_2021_11 PARTITION OF %s FOR VALUES FROM ('2021-11-01') TO ('2021-12-01')", testTable.getName(), testTable.getName()));
+            execute(format("CREATE TABLE %s_2021_12 PARTITION OF %s FOR VALUES FROM ('2021-12-01') TO ('2022-01-01')", testTable.getName(), testTable.getName()));
+            execute(format("INSERT INTO %s VALUES %s ,%s", testTable.getName(), values202111, values202112));
+            assertThat(computeActual("SHOW TABLES").getOnlyColumnAsSet())
+                    .contains(testTable.getName(), testTable.getName() + "_2021_11", testTable.getName() + "_2021_12");
+            assertQuery(format("SELECT * FROM %s", testTable.getName()), format("VALUES %s, %s", values202111, values202112));
+            assertQuery(format("SELECT * FROM %s_2021_12", testTable.getName()), "VALUES " + values202112);
+        }
+
+        try (TestTable testTable = new TestTable(
+                postgreSqlServer::execute,
+                "test_part_tbl",
+                "(id int NOT NULL, type varchar, logdate varchar) PARTITION BY LIST (type)")) {
+            String valuesA = "(1, 'A', '2021-11-11'), (4, 'A', '2021-12-25')";
+            String valuesB = "(3, 'B', '2021-12-12'), (2, 'B', '2021-12-28')";
+            execute(format("CREATE TABLE %s_a PARTITION OF %s FOR VALUES IN ('A')", testTable.getName(), testTable.getName()));
+            execute(format("CREATE TABLE %s_b PARTITION OF %s FOR VALUES IN ('B')", testTable.getName(), testTable.getName()));
+            assertUpdate(format("INSERT INTO %s VALUES %s ,%s", testTable.getName(), valuesA, valuesB), 4);
+            assertThat(computeActual("SHOW TABLES").getOnlyColumnAsSet())
+                    .contains(testTable.getName(), testTable.getName() + "_a", testTable.getName() + "_b");
+            assertQuery(format("SELECT * FROM %s", testTable.getName()), format("VALUES %s, %s", valuesA, valuesB));
+            assertQuery(format("SELECT * FROM %s_a", testTable.getName()), "VALUES " + valuesA);
+        }
+    }
+
     @Test
     public void testTableWithNoSupportedColumns()
             throws Exception

plugin/trino-postgresql/src/test/java/io/trino/plugin/postgresql/TestingPostgreSqlServer.java

@@ -41,6 +41,7 @@ public class TestingPostgreSqlServer
     private static final String PASSWORD = "test";
     private static final String DATABASE = "tpch";
 
+    private static final String LOG_PREFIX_REGEXP = "^([-:0-9. ]+UTC \\[[0-9]+\\] )";
     private static final String LOG_RUNNING_STATEMENT_PREFIX = "LOG:  execute <unnamed>: ";
     private static final String LOG_CANCELLATION_EVENT = "ERROR:  canceling statement due to user request";
     private static final String LOG_CANCELLED_STATEMENT_PREFIX = "STATEMENT:  ";
@@ -50,7 +51,7 @@ public class TestingPostgreSqlServer
     public TestingPostgreSqlServer()
     {
         // Use the oldest supported PostgreSQL version
-        dockerContainer = new PostgreSQLContainer<>("postgres:9.6")
+        dockerContainer = new PostgreSQLContainer<>("postgres:10")
                 .withDatabaseName(DATABASE)
                 .withUsername(USER)
                 .withPassword(PASSWORD)
@@ -82,13 +83,13 @@ protected List<RemoteDatabaseEvent> getRemoteDatabaseEvents()
         Iterator<String> logsIterator = logs.iterator();
         ImmutableList.Builder<RemoteDatabaseEvent> events = ImmutableList.builder();
         while (logsIterator.hasNext()) {
-            String logLine = logsIterator.next();
+            String logLine = logsIterator.next().replaceAll(LOG_PREFIX_REGEXP, "");
             if (logLine.startsWith(LOG_RUNNING_STATEMENT_PREFIX)) {
                 events.add(new RemoteDatabaseEvent(logLine.substring(LOG_RUNNING_STATEMENT_PREFIX.length()), RUNNING));
             }
             if (logLine.equals(LOG_CANCELLATION_EVENT)) {
                 // next line must be present
-                String cancelledStatementLogLine = logsIterator.next();
+                String cancelledStatementLogLine = logsIterator.next().replaceAll(LOG_PREFIX_REGEXP, "");
                 if (cancelledStatementLogLine.startsWith(LOG_CANCELLED_STATEMENT_PREFIX)) {
                     events.add(new RemoteDatabaseEvent(cancelledStatementLogLine.substring(LOG_CANCELLED_STATEMENT_PREFIX.length()), CANCELLED));
                 }