!1380 Support PostgreSQL and openGauss Update/Delete
Merge pull request !1380 from Anllick/openGaussPgUpdate
This commit is contained in:
Gitee
commit
1512e1adaf
|
|
@ -45,6 +45,95 @@ Finally, you can access the `hetutb` table in the `public` schema:
|
|||
|
||||
If you used a different name for your catalog properties file, use that catalog name instead of `opengauss` in the above examples.
|
||||
|
||||
## openGauss Update/Delete Support
|
||||
|
||||
### Create openGauss Table
|
||||
|
||||
Example：
|
||||
|
||||
```sql
|
||||
CREATE TABLE opengauss_table (
|
||||
id int,
|
||||
name varchar(255));
|
||||
```
|
||||
|
||||
### INSERT on openGauss tables
|
||||
|
||||
Example：
|
||||
|
||||
```sql
|
||||
INSERT INTO opengauss_table
|
||||
VALUES
|
||||
(1, 'Jack'),
|
||||
(2, 'Bob');
|
||||
```
|
||||
|
||||
### UPDATE on openGauss tables
|
||||
|
||||
Example：
|
||||
|
||||
```sql
|
||||
UPDATE opengauss_table
|
||||
SET name='Tim'
|
||||
WHERE id=1;
|
||||
```
|
||||
|
||||
Above example updates the column `name`'s value to `Tim` of rows with column `id` having value `1`.
|
||||
|
||||
SELECT result before UPDATE:
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM opengauss_table;
|
||||
id | name
|
||||
----+------
|
||||
1 | Jack
|
||||
2 | Bob
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
SELECT result after UPDATE
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM opengauss_table;
|
||||
id | name
|
||||
----+------
|
||||
2 | Bob
|
||||
1 | Tim
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
### DELETE on openGauss tables
|
||||
|
||||
Example：
|
||||
|
||||
```sql
|
||||
DELETE FROM opengauss_table
|
||||
WHERE id=2;
|
||||
```
|
||||
|
||||
Above example delete the rows with column `id` having value `2`.
|
||||
|
||||
SELECT result before DELETE:
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM opengauss_table;
|
||||
id | name
|
||||
----+------
|
||||
2 | Bob
|
||||
1 | Tim
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
SELECT result after DELETE:
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM opengauss_table;
|
||||
id | name
|
||||
----+------
|
||||
1 | Tim
|
||||
(1 row)
|
||||
```
|
||||
|
||||
****Note:****
|
||||
|
||||
> - When the compatibility type of the openGuass database is O (DBCOMPATIBILITY = A), the `Date` data type is not supported.
|
||||
|
|
@ -64,4 +153,4 @@ openGauss Connector Limitations
|
|||
|
||||
The following SQL statements are not yet supported:
|
||||
|
||||
[DELETE](../sql/delete.md), [GRANT](../sql/grant.md), [REVOKE](../sql/revoke.md), [SHOW GRANTS](../sql/show-grants.md), [SHOW ROLES](../sql/show-roles.md), [SHOW ROLE GRANTS](../sql/show-role-grants.md)
|
||||
[GRANT](../sql/grant.md), [REVOKE](../sql/revoke.md), [SHOW GRANTS](../sql/show-grants.md), [SHOW ROLES](../sql/show-roles.md), [SHOW ROLE GRANTS](../sql/show-role-grants.md)
|
||||
|
|
|
|||
|
|
@ -46,9 +46,98 @@ Finally, you can access the `clicks` table in the `web` schema:
|
|||
|
||||
If you used a different name for your catalog properties file, use that catalog name instead of `postgresql` in the above examples.
|
||||
|
||||
## PostgreSQL Update/Delete Support
|
||||
|
||||
### Create PostgreSQL Table
|
||||
|
||||
Example：
|
||||
|
||||
```sql
|
||||
CREATE TABLE postgresql_table (
|
||||
id int,
|
||||
name varchar(255));
|
||||
```
|
||||
|
||||
### INSERT on PostgreSQL tables
|
||||
|
||||
Example：
|
||||
|
||||
```sql
|
||||
INSERT INTO postgresql_table
|
||||
VALUES
|
||||
(1, 'Jack'),
|
||||
(2, 'Bob');
|
||||
```
|
||||
|
||||
### UPDATE on PostgreSQL tables
|
||||
|
||||
Example：
|
||||
|
||||
```sql
|
||||
UPDATE postgresql_table
|
||||
SET name='Tim'
|
||||
WHERE id=1;
|
||||
```
|
||||
|
||||
Above example updates the column `name`'s value to `Tim` of rows with column `id` having value `1`.
|
||||
|
||||
SELECT result before UPDATE:
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM postgresql_table;
|
||||
id | name
|
||||
----+------
|
||||
1 | Jack
|
||||
2 | Bob
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
SELECT result after UPDATE
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM postgresql_table;
|
||||
id | name
|
||||
----+------
|
||||
2 | Bob
|
||||
1 | Tim
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
### DELETE on PostgreSQL tables
|
||||
|
||||
Example：
|
||||
|
||||
```sql
|
||||
DELETE FROM postgresql_table
|
||||
WHERE id=2;
|
||||
```
|
||||
|
||||
Above example delete the rows with column `id` having value `2`.
|
||||
|
||||
SELECT result before DELETE:
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM postgresql_table;
|
||||
id | name
|
||||
----+------
|
||||
2 | Bob
|
||||
1 | Tim
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
SELECT result after DELETE:
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM postgresql_table;
|
||||
id | name
|
||||
----+------
|
||||
1 | Tim
|
||||
(1 row)
|
||||
```
|
||||
|
||||
PostgreSQL Connector Limitations
|
||||
--------------------------------
|
||||
|
||||
The following SQL statements are not yet supported:
|
||||
|
||||
[DELETE](../sql/delete.md), [GRANT](../sql/grant.md), [REVOKE](../sql/revoke.md), [SHOW GRANTS](../sql/show-grants.md), [SHOW ROLES](../sql/show-roles.md), [SHOW ROLE GRANTS](../sql/show-role-grants.md)
|
||||
[GRANT](../sql/grant.md), [REVOKE](../sql/revoke.md), [SHOW GRANTS](../sql/show-grants.md), [SHOW ROLES](../sql/show-roles.md), [SHOW ROLE GRANTS](../sql/show-role-grants.md)
|
||||
|
|
@ -41,6 +41,95 @@ openGauss连接器为每个openGauss模式提供一个模式。可通过执行`S
|
|||
|
||||
如果对目录属性文件使用不同的名称，请使用该目录名称，而不要使用上述示例中的`opengauss`。
|
||||
|
||||
## openGauss Update/Delete 支持
|
||||
|
||||
### 使用openGauss连接器创建表
|
||||
|
||||
示例：
|
||||
|
||||
```sql
|
||||
CREATE TABLE opengauss_table (
|
||||
id int,
|
||||
name varchar(255));
|
||||
```
|
||||
|
||||
### 对表执行INSERT
|
||||
|
||||
示例：
|
||||
|
||||
```sql
|
||||
INSERT INTO opengauss_table
|
||||
VALUES
|
||||
(1, 'Jack'),
|
||||
(2, 'Bob');
|
||||
```
|
||||
|
||||
### 对表执行UPDATE
|
||||
|
||||
示例：
|
||||
|
||||
```sql
|
||||
UPDATE opengauss_table
|
||||
SET name='Tim'
|
||||
WHERE id=1;
|
||||
```
|
||||
|
||||
上述示例将列`id`中值为`1`所在行的列`name`的值更新为`Tim`。
|
||||
|
||||
UPDATE前的SELECT结果：
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM opengauss_table;
|
||||
id | name
|
||||
----+------
|
||||
1 | Jack
|
||||
2 | Bob
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
UPDATE后的SELECT结果
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM opengauss_table;
|
||||
id | name
|
||||
----+------
|
||||
2 | Bob
|
||||
1 | Tim
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
### 对表执行DELETE
|
||||
|
||||
示例：
|
||||
|
||||
```sql
|
||||
DELETE FROM opengauss_table
|
||||
WHERE id=2;
|
||||
```
|
||||
|
||||
以上示例删除了值为`2`的列`id`的行。
|
||||
|
||||
DELETE前的SELECT结果：
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM opengauss_table;
|
||||
id | name
|
||||
----+------
|
||||
2 | Bob
|
||||
1 | Tim
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
DELETE后的SELECT结果：
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM opengauss_table;
|
||||
id | name
|
||||
----+------
|
||||
1 | Tim
|
||||
(1 row)
|
||||
```
|
||||
|
||||
**注意**
|
||||
|
||||
> - openGuass数据库兼容类型为O（即DBCOMPATIBILITY = A）时不支持`Date`数据类型。
|
||||
|
|
@ -59,4 +148,4 @@ openGauss连接器为每个openGauss模式提供一个模式。可通过执行`S
|
|||
|
||||
暂不支持以下SQL语句：
|
||||
|
||||
[DELETE](../sql/delete.md)、[GRANT](../sql/grant.md)、[REVOKE](../sql/revoke.md)、[SHOW GRANTS](../sql/show-grants.md)、[SHOW ROLES](../sql/show-roles.md)、[SHOW ROLE GRANTS](../sql/show-role-grants.md)
|
||||
[GRANT](../sql/grant.md)、[REVOKE](../sql/revoke.md)、[SHOW GRANTS](../sql/show-grants.md)、[SHOW ROLES](../sql/show-roles.md)、[SHOW ROLE GRANTS](../sql/show-role-grants.md)
|
||||
|
|
@ -41,8 +41,97 @@ PostgreSQL连接器为每个PostgreSQL模式提供一个模式。可通过执行
|
|||
|
||||
如果对目录属性文件使用不同的名称，请使用该目录名称，而不要使用上述示例中的`postgresql`。
|
||||
|
||||
## PostgreSQL Update/Delete 支持
|
||||
|
||||
### 使用PostgreSQL连接器创建表
|
||||
|
||||
示例：
|
||||
|
||||
```sql
|
||||
CREATE TABLE postgresql_table (
|
||||
id int,
|
||||
name varchar(255));
|
||||
```
|
||||
|
||||
### 对表执行INSERT
|
||||
|
||||
示例：
|
||||
|
||||
```sql
|
||||
INSERT INTO postgresql_table
|
||||
VALUES
|
||||
(1, 'Jack'),
|
||||
(2, 'Bob');
|
||||
```
|
||||
|
||||
### 对表执行UPDATE
|
||||
|
||||
示例：
|
||||
|
||||
```sql
|
||||
UPDATE postgresql_table
|
||||
SET name='Tim'
|
||||
WHERE id=1;
|
||||
```
|
||||
|
||||
上述示例将列`id`中值为`1`所在行的列`name`的值更新为`Tim`。
|
||||
|
||||
UPDATE前的SELECT结果：
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM postgresql_table;
|
||||
id | name
|
||||
----+------
|
||||
1 | Jack
|
||||
2 | Bob
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
UPDATE后的SELECT结果
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM postgresql_table;
|
||||
id | name
|
||||
----+------
|
||||
2 | Bob
|
||||
1 | Tim
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
### 对表执行DELETE
|
||||
|
||||
示例：
|
||||
|
||||
```sql
|
||||
DELETE FROM postgresql_table
|
||||
WHERE id=2;
|
||||
```
|
||||
|
||||
以上示例删除了值为`2`的列`id`的行。
|
||||
|
||||
DELETE前的SELECT结果：
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM postgresql_table;
|
||||
id | name
|
||||
----+------
|
||||
2 | Bob
|
||||
1 | Tim
|
||||
(2 rows)
|
||||
```
|
||||
|
||||
DELETE后的SELECT结果：
|
||||
|
||||
```sql
|
||||
lk:default> SELECT * FROM postgresql_table;
|
||||
id | name
|
||||
----+------
|
||||
1 | Tim
|
||||
(1 row)
|
||||
```
|
||||
|
||||
## PostgreSQL连接器限制
|
||||
|
||||
暂不支持以下SQL语句：
|
||||
|
||||
[DELETE](../sql/delete.md)、[GRANT](../sql/grant.md)、[REVOKE](../sql/revoke.md)、[SHOW GRANTS](../sql/show-grants.md)、[SHOW ROLES](../sql/show-roles.md)、[SHOW ROLE GRANTS](../sql/show-role-grants.md)
|
||||
[GRANT](../sql/grant.md)、[REVOKE](../sql/revoke.md)、[SHOW GRANTS](../sql/show-grants.md)、[SHOW ROLES](../sql/show-roles.md)、[SHOW ROLE GRANTS](../sql/show-role-grants.md)
|
||||
|
|
@ -18,6 +18,8 @@ import io.prestosql.plugin.postgresql.TestPostgreSqlIntegrationSmokeTest;
|
|||
import org.intellij.lang.annotations.Language;
|
||||
import org.testng.annotations.Test;
|
||||
|
||||
import java.sql.SQLException;
|
||||
|
||||
import static io.airlift.tpch.TpchTable.ORDERS;
|
||||
import static java.lang.String.format;
|
||||
import static org.testng.Assert.assertEquals;
|
||||
|
|
@ -39,6 +41,154 @@ public class TestOpenGaussIntegrationSmokeTest
|
|||
openGaussServer);
|
||||
}
|
||||
|
||||
@Test
|
||||
@Override
|
||||
public void testUpdateByOneField()
|
||||
throws SQLException
|
||||
{
|
||||
@Language("SQL") String createTableSql = format("" +
|
||||
"CREATE TABLE %s.tpch.test_update (\n" +
|
||||
"id int,\n" +
|
||||
"name varchar,\n" +
|
||||
"sex char,\n" +
|
||||
"age int,\n" +
|
||||
"score varchar,\n" +
|
||||
"birthday date,\n" +
|
||||
"salary double precision\n" +
|
||||
")",
|
||||
getSession().getCatalog().get());
|
||||
|
||||
assertUpdate(createTableSql);
|
||||
assertUpdate("INSERT INTO test_update VALUES(1, 'Bob', '1', 24, 'excellent', date'1997-09-28', 40000)", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES(2, 'Jack', '1', 25, 'good', date'1996-08-14', 35000)", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES(3, 'Rose', '2', 22, 'excellent', date'1999-07-11', 10000)", 1);
|
||||
|
||||
assertQuery("SELECT * FROM test_update where id = 1", "VALUES(1, 'Bob', '1', 24, 'excellent', date'1997-09-28', 40000)");
|
||||
assertUpdate("UPDATE test_update SET name = 'Kitty', age = 26, birthday = date'1995-08-16' where id = 1", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 1", "VALUES(1, 'Kitty', '1', 26, 'excellent', date'1995-08-16', 40000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE score = 'good'", "VALUES(2, 'Jack', '1', 25, 'good', date'1996-08-14', 35000)");
|
||||
assertUpdate("UPDATE test_update SET name = 'Jane', sex = '2' where score = 'good'", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE score = 'good'", "VALUES(2, 'Jane', '2', 25, 'good', date'1996-08-14', 35000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE salary = 10000", "VALUES(3, 'Rose', '2', 22, 'excellent', date'1999-07-11', 10000)");
|
||||
assertUpdate("UPDATE test_update SET score = 'qualified' where salary = 10000.00 ", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE salary = 10000", "VALUES(3, 'Rose', '2', 22, 'qualified', date'1999-07-11', 10000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE age = 22", "VALUES(3, 'Rose', '2', 22, 'qualified', date'1999-07-11', 10000)");
|
||||
assertUpdate("UPDATE test_update SET age = age + 1, birthday = date'1998-08-16' where age = 22 ", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE age = 23", "VALUES(3, 'Rose', '2', 23, 'qualified', date'1998-08-16', 10000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE birthday = date'1996-08-14'", "VALUES(2, 'Jane', '2', 25, 'good', date'1996-08-14', 35000)");
|
||||
assertUpdate("UPDATE test_update SET salary = 36000 where birthday = date'1996-08-14'", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE birthday = date'1996-08-14'", "VALUES(2, 'Jane', '2', 25, 'good', date'1996-08-14', 36000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE name = 'Jane'", "VALUES(2, 'Jane', '2', 25, 'good', date'1996-08-14', 36000)");
|
||||
assertUpdate("UPDATE test_update SET score = 'bad', birthday = date'1995-10-16' where name = 'Jane'", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE name = 'Jane'", "VALUES(2, 'Jane', '2', 25, 'bad', date'1995-10-16', 36000)");
|
||||
|
||||
assertUpdate("DROP TABLE test_update");
|
||||
}
|
||||
|
||||
@Test
|
||||
@Override
|
||||
public void testUpdateByMutiField()
|
||||
throws SQLException
|
||||
{
|
||||
@Language("SQL") String createTableSql = format("" +
|
||||
"CREATE TABLE %s.tpch.test_update (\n" +
|
||||
"name varchar,\n" +
|
||||
"id int,\n" +
|
||||
"score varchar\n" +
|
||||
")",
|
||||
getSession().getCatalog().get());
|
||||
|
||||
assertUpdate(createTableSql);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Bob', 1, 'excellent')", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Tim', 2, 'good')", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Jane', 3, 'good')", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Rose', 4, 'bad')", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Petty', 5, 'good')", 1);
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 4", "VALUES('Rose', 4, 'bad')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 5", "VALUES('Petty', 5, 'good')");
|
||||
assertUpdate("UPDATE test_update SET name = 'Kitt' WHERE id IN (4, 5) AND NAME LIKE 'P%'", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 4", "VALUES('Rose', 4, 'bad')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 5", "VALUES('Kitt', 5, 'good')");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 2", "VALUES('Tim', 2, 'good')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 3", "VALUES('Jane', 3, 'good')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 4", "VALUES('Rose', 4, 'bad')");
|
||||
assertUpdate("UPDATE test_update SET score = 'excellent' WHERE id BETWEEN 2 And 4 AND score != 'bad'", 2);
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 2", "VALUES('Tim', 2, 'excellent')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 3", "VALUES('Jane', 3, 'excellent')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 4", "VALUES('Rose', 4, 'bad')");
|
||||
|
||||
assertUpdate("DROP TABLE test_update");
|
||||
}
|
||||
|
||||
@Test
|
||||
@Override
|
||||
public void testDeleteByOneField()
|
||||
throws SQLException
|
||||
{
|
||||
@Language("SQL") String createTableSql = format("" +
|
||||
"CREATE TABLE %s.tpch.test_delete (\n" +
|
||||
"id int,\n" +
|
||||
"name varchar,\n" +
|
||||
"sex char,\n" +
|
||||
"age int,\n" +
|
||||
"score varchar,\n" +
|
||||
"birthday date,\n" +
|
||||
"salary double precision\n" +
|
||||
")",
|
||||
getSession().getCatalog().get());
|
||||
assertUpdate(createTableSql);
|
||||
assertUpdate("INSERT INTO test_delete VALUES(1, 'Bob', '1', 24, 'excellent', date'1997-09-28', 40000)", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES(2, 'Jack', '1', 25, 'good', date'1996-08-14', 35000)", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES(3, 'Jane', '2', 23, 'bad', date'1998-07-25', 15000)", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES(4, 'Rose', '2', 22, 'excellent', date'1999-07-11', 10000)", 1);
|
||||
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(4)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE name = 'Bob'", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(3)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE birthday = date'1996-08-14'", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(2)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE salary = 15000", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(1)");
|
||||
assertUpdate("DROP TABLE test_delete");
|
||||
}
|
||||
|
||||
@Test
|
||||
@Override
|
||||
public void testDeleteByMutiField()
|
||||
throws SQLException
|
||||
{
|
||||
@Language("SQL") String createTableSql = format("" +
|
||||
"CREATE TABLE %s.tpch.test_delete (\n" +
|
||||
"name varchar,\n" +
|
||||
"id int,\n" +
|
||||
"score varchar\n" +
|
||||
")",
|
||||
getSession().getCatalog().get());
|
||||
|
||||
assertUpdate(createTableSql);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Bob', 1, 'excellent')", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Tim', 2, 'good')", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Jane', 3, 'good')", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Rose', 4, 'bad')", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Petty', 5, 'good')", 1);
|
||||
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(5)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE id BETWEEN 2 AND 4 AND name IN ('Rose', 'Petty')", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(4)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE name = 'Petty' AND score = 'good'", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(3)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE name LIKE '%o%' OR score = 'good'", 3);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(0)");
|
||||
assertUpdate("DROP TABLE test_delete");
|
||||
}
|
||||
|
||||
@Test
|
||||
@Override
|
||||
public void testMaterializedView()
|
||||
|
|
|
|||
|
|
@ -157,5 +157,10 @@
|
|||
<artifactId>annotations</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.airlift</groupId>
|
||||
<artifactId>log</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
|
|
|
|||
|
|
@ -16,9 +16,11 @@ package io.prestosql.plugin.postgresql;
|
|||
import com.fasterxml.jackson.core.JsonFactory;
|
||||
import com.fasterxml.jackson.core.JsonParser;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.google.common.base.VerifyException;
|
||||
import com.google.common.collect.ImmutableList;
|
||||
import com.google.common.collect.ImmutableMap;
|
||||
import io.airlift.json.ObjectMapperProvider;
|
||||
import io.airlift.log.Logger;
|
||||
import io.airlift.slice.DynamicSliceOutput;
|
||||
import io.airlift.slice.Slice;
|
||||
import io.airlift.slice.SliceOutput;
|
||||
|
|
@ -26,8 +28,10 @@ import io.prestosql.plugin.jdbc.BaseJdbcClient;
|
|||
import io.prestosql.plugin.jdbc.BaseJdbcConfig;
|
||||
import io.prestosql.plugin.jdbc.BlockReadFunction;
|
||||
import io.prestosql.plugin.jdbc.BlockWriteFunction;
|
||||
import io.prestosql.plugin.jdbc.BooleanWriteFunction;
|
||||
import io.prestosql.plugin.jdbc.ColumnMapping;
|
||||
import io.prestosql.plugin.jdbc.ConnectionFactory;
|
||||
import io.prestosql.plugin.jdbc.DoubleWriteFunction;
|
||||
import io.prestosql.plugin.jdbc.JdbcColumnHandle;
|
||||
import io.prestosql.plugin.jdbc.JdbcIdentity;
|
||||
import io.prestosql.plugin.jdbc.JdbcTableHandle;
|
||||
|
|
@ -35,8 +39,15 @@ import io.prestosql.plugin.jdbc.JdbcTypeHandle;
|
|||
import io.prestosql.plugin.jdbc.LongWriteFunction;
|
||||
import io.prestosql.plugin.jdbc.SliceWriteFunction;
|
||||
import io.prestosql.plugin.jdbc.StatsCollecting;
|
||||
import io.prestosql.plugin.jdbc.WriteFunction;
|
||||
import io.prestosql.plugin.jdbc.WriteMapping;
|
||||
import io.prestosql.plugin.jdbc.WriteNullFunction;
|
||||
import io.prestosql.plugin.jdbc.optimization.JdbcPushDownModule;
|
||||
import io.prestosql.spi.PrestoException;
|
||||
import io.prestosql.spi.block.Block;
|
||||
import io.prestosql.spi.connector.ColumnHandle;
|
||||
import io.prestosql.spi.connector.ConnectorSession;
|
||||
import io.prestosql.spi.connector.ConnectorTableHandle;
|
||||
import io.prestosql.spi.connector.ConnectorTableMetadata;
|
||||
import io.prestosql.spi.connector.SchemaTableName;
|
||||
import io.prestosql.spi.connector.TableNotFoundException;
|
||||
|
|
@ -59,19 +70,23 @@ import java.sql.ResultSet;
|
|||
import java.sql.SQLException;
|
||||
import java.sql.Types;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.OptionalLong;
|
||||
import java.util.UUID;
|
||||
import java.util.function.BiFunction;
|
||||
|
||||
import static com.fasterxml.jackson.core.JsonFactory.Feature.CANONICALIZE_FIELD_NAMES;
|
||||
import static com.fasterxml.jackson.databind.SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS;
|
||||
import static com.google.common.base.Strings.isNullOrEmpty;
|
||||
import static com.google.common.base.Verify.verify;
|
||||
import static com.google.common.collect.ImmutableList.toImmutableList;
|
||||
import static io.airlift.slice.SizeOf.SIZE_OF_LONG;
|
||||
import static io.airlift.slice.Slices.wrappedLongArray;
|
||||
import static io.prestosql.plugin.jdbc.JdbcErrorCode.JDBC_ERROR;
|
||||
import static io.prestosql.plugin.jdbc.StandardColumnMappings.timestampColumnMapping;
|
||||
import static io.prestosql.plugin.postgresql.TypeUtils.getJdbcObjectArray;
|
||||
import static io.prestosql.plugin.postgresql.TypeUtils.jdbcObjectArrayToBlock;
|
||||
import static io.prestosql.plugin.postgresql.TypeUtils.toBoxedArray;
|
||||
|
|
@ -82,13 +97,21 @@ import static io.prestosql.spi.type.DateTimeEncoding.packDateTimeWithZone;
|
|||
import static io.prestosql.spi.type.DateTimeEncoding.unpackMillisUtc;
|
||||
import static io.prestosql.spi.type.TimeZoneKey.UTC_KEY;
|
||||
import static io.prestosql.spi.type.TimestampWithTimeZoneType.TIMESTAMP_WITH_TIME_ZONE;
|
||||
import static io.prestosql.spi.type.VarcharType.VARCHAR;
|
||||
import static java.lang.String.format;
|
||||
import static java.nio.charset.StandardCharsets.UTF_8;
|
||||
import static java.sql.DatabaseMetaData.columnNoNulls;
|
||||
import static java.util.Locale.ENGLISH;
|
||||
|
||||
public abstract class BasePostgreSqlClient
|
||||
extends BaseJdbcClient
|
||||
{
|
||||
private static final Logger log = Logger.get(BasePostgreSqlClient.class);
|
||||
|
||||
/**
|
||||
* If disabled, do not accept sub-query push down.
|
||||
*/
|
||||
private final JdbcPushDownModule pushDownModule;
|
||||
protected static final String DUPLICATE_TABLE_SQLSTATE = "42P07";
|
||||
|
||||
protected final Type jsonType;
|
||||
|
|
@ -104,6 +127,7 @@ public abstract class BasePostgreSqlClient
|
|||
super(config, "\"", connectionFactory);
|
||||
this.jsonType = typeManager.getType(new TypeSignature(StandardTypes.JSON));
|
||||
this.uuidType = typeManager.getType(new TypeSignature(StandardTypes.UUID));
|
||||
this.pushDownModule = config.getPushDownModule();
|
||||
|
||||
switch (postgresqlConfig.getArrayMapping()) {
|
||||
case DISABLED:
|
||||
|
|
@ -241,56 +265,6 @@ public abstract class BasePostgreSqlClient
|
|||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<ColumnMapping> toPrestoType(ConnectorSession session, Connection connection, JdbcTypeHandle typeHandle)
|
||||
{
|
||||
String jdbcTypeName = typeHandle.getJdbcTypeName()
|
||||
.orElseThrow(() -> new PrestoException(JDBC_ERROR, "Type name is missing: " + typeHandle));
|
||||
|
||||
switch (jdbcTypeName) {
|
||||
case "uuid":
|
||||
return Optional.of(uuidColumnMapping());
|
||||
case "jsonb":
|
||||
case "json":
|
||||
return Optional.of(jsonColumnMapping());
|
||||
case "timestamptz":
|
||||
// PostgreSQL's "timestamp with time zone" is reported as Types.TIMESTAMP rather than Types.TIMESTAMP_WITH_TIMEZONE
|
||||
return Optional.of(timestampWithTimeZoneColumnMapping());
|
||||
default:
|
||||
break;
|
||||
}
|
||||
if (typeHandle.getJdbcType() == Types.VARCHAR && !jdbcTypeName.equals("varchar")) {
|
||||
// This can be e.g. an ENUM
|
||||
return Optional.of(typedVarcharColumnMapping(jdbcTypeName));
|
||||
}
|
||||
if (typeHandle.getJdbcType() == Types.TIMESTAMP) {
|
||||
return Optional.of(timestampColumnMapping());
|
||||
}
|
||||
if (typeHandle.getJdbcType() == Types.ARRAY && supportArrays) {
|
||||
if (!typeHandle.getArrayDimensions().isPresent()) {
|
||||
return Optional.empty();
|
||||
}
|
||||
JdbcTypeHandle elementTypeHandle = getArrayElementTypeHandle(connection, typeHandle);
|
||||
String elementTypeName = typeHandle.getJdbcTypeName()
|
||||
.orElseThrow(() -> new PrestoException(JDBC_ERROR, "Element type name is missing: " + elementTypeHandle));
|
||||
if (elementTypeHandle.getJdbcType() == Types.VARBINARY) {
|
||||
// PostgreSQL jdbc driver doesn't currently support array of varbinary (bytea[])
|
||||
// https://github.com/pgjdbc/pgjdbc/pull/1184
|
||||
return Optional.empty();
|
||||
}
|
||||
return toPrestoType(session, connection, elementTypeHandle)
|
||||
.map(elementMapping -> {
|
||||
ArrayType prestoArrayType = new ArrayType(elementMapping.getType());
|
||||
int arrayDimensions = typeHandle.getArrayDimensions().get();
|
||||
for (int i = 1; i < arrayDimensions; i++) {
|
||||
prestoArrayType = new ArrayType(prestoArrayType);
|
||||
}
|
||||
return arrayColumnMapping(session, prestoArrayType, elementTypeName);
|
||||
});
|
||||
}
|
||||
return super.toPrestoType(session, connection, typeHandle);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Optional<BiFunction<String, Long, String>> limitFunction()
|
||||
{
|
||||
|
|
@ -420,4 +394,255 @@ public abstract class BasePostgreSqlClient
|
|||
// so we pass an InputStreamReader instead.
|
||||
return JSON_FACTORY.createParser(new InputStreamReader(json.getInput(), UTF_8));
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConnectorTableHandle beginUpdate(ConnectorSession session, ConnectorTableHandle tableHandle, List<Type> updatedColumnTypes)
|
||||
{
|
||||
JdbcTableHandle jdbcTableHandle = (JdbcTableHandle) tableHandle;
|
||||
jdbcTableHandle.setUpdatedColumnTypes(updatedColumnTypes);
|
||||
jdbcTableHandle.setDeleteOrUpdate(true);
|
||||
return jdbcTableHandle;
|
||||
}
|
||||
|
||||
private void setStatement(ConnectorSession session, ConnectorTableHandle tableHandle, PreparedStatement statement, Block block, int position, int channel)
|
||||
throws SQLException
|
||||
{
|
||||
JdbcTableHandle jdbcTableHandle = (JdbcTableHandle) tableHandle;
|
||||
List<Type> updatedColumnTypes = jdbcTableHandle.getUpdatedColumnTypes();
|
||||
|
||||
List<WriteMapping> writeMappings = updatedColumnTypes.stream()
|
||||
.map(type ->
|
||||
{
|
||||
WriteMapping writeMapping = toWriteMapping(session, type);
|
||||
WriteFunction writeFunction = writeMapping.getWriteFunction();
|
||||
verify(
|
||||
type.getJavaType() == writeFunction.getJavaType(),
|
||||
"openLooKeng type %s is not compatible with write function %s accepting %s",
|
||||
type,
|
||||
writeFunction,
|
||||
writeFunction.getJavaType());
|
||||
return writeMapping;
|
||||
})
|
||||
.collect(toImmutableList());
|
||||
|
||||
List<WriteFunction> columnWriters = writeMappings.stream()
|
||||
.map(WriteMapping::getWriteFunction)
|
||||
.collect(toImmutableList());
|
||||
|
||||
List<WriteNullFunction> nullWriters = writeMappings.stream()
|
||||
.map(WriteMapping::getWriteNullFunction)
|
||||
.collect(toImmutableList());
|
||||
|
||||
int parameterIndex = channel + 1;
|
||||
|
||||
if (block.isNull(position)) {
|
||||
nullWriters.get(channel).setNull(statement, parameterIndex);
|
||||
return;
|
||||
}
|
||||
|
||||
Type type = jdbcTableHandle.getUpdatedColumnTypes().get(channel);
|
||||
Class<?> javaType = type.getJavaType();
|
||||
WriteFunction writeFunction = columnWriters.get(channel);
|
||||
if (javaType == boolean.class) {
|
||||
((BooleanWriteFunction) writeFunction).set(statement, parameterIndex, type.getBoolean(block, position));
|
||||
}
|
||||
else if (javaType == long.class) {
|
||||
((LongWriteFunction) writeFunction).set(statement, parameterIndex, type.getLong(block, position));
|
||||
}
|
||||
else if (javaType == double.class) {
|
||||
((DoubleWriteFunction) writeFunction).set(statement, parameterIndex, type.getDouble(block, position));
|
||||
}
|
||||
else if (javaType == Slice.class) {
|
||||
((SliceWriteFunction) writeFunction).set(statement, parameterIndex, type.getSlice(block, position));
|
||||
}
|
||||
else if (javaType == Block.class) {
|
||||
((BlockWriteFunction) writeFunction).set(statement, parameterIndex, (Block) type.getObject(block, position));
|
||||
}
|
||||
else {
|
||||
throw new VerifyException(format("Unexpected type %s with java type %s", type, javaType.getName()));
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void finishUpdate(ConnectorSession session, ConnectorTableHandle tableHandle, Collection<Slice> fragments)
|
||||
{
|
||||
}
|
||||
|
||||
private Map<String, String> getColumnNameMap(ConnectorSession session, JdbcTableHandle tableHandle)
|
||||
{
|
||||
HashMap<String, String> columnNameMap = new HashMap<>(); //<columnName in lower case, columnName in datasource>
|
||||
List<JdbcColumnHandle> columnList = getColumns(session, tableHandle);
|
||||
|
||||
for (JdbcColumnHandle columnHandle : columnList) {
|
||||
String columnName = columnHandle.getColumnName();
|
||||
columnNameMap.put(columnName.toLowerCase(ENGLISH), columnName);
|
||||
}
|
||||
|
||||
return columnNameMap;
|
||||
}
|
||||
|
||||
private List<String> getColumnNameFromDataSource(ConnectorSession session, JdbcTableHandle tableHandle, List<String> columns)
|
||||
{
|
||||
Map<String, String> columnNameMap = getColumnNameMap(session, tableHandle);
|
||||
List<String> updatedColumns = new ArrayList<>();
|
||||
|
||||
for (String columnName : columns) {
|
||||
String originName = columnNameMap.get(columnName.toLowerCase(ENGLISH));
|
||||
updatedColumns.add((originName != null) ? originName : columnName);
|
||||
}
|
||||
|
||||
return updatedColumns;
|
||||
}
|
||||
|
||||
private String buildRemoteSchemaTableName(JdbcTableHandle tableHandle)
|
||||
{
|
||||
StringBuilder remoteSchemaTable = new StringBuilder();
|
||||
|
||||
if (!isNullOrEmpty(tableHandle.getSchemaName())) {
|
||||
remoteSchemaTable.append(quoted(tableHandle.getSchemaName())).append(".");
|
||||
}
|
||||
|
||||
remoteSchemaTable.append(quoted(tableHandle.getTableName()));
|
||||
|
||||
return remoteSchemaTable.toString();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String buildUpdateSql(ConnectorSession session, ConnectorTableHandle handle, int setNum,
|
||||
List<String> updatedColumns)
|
||||
{
|
||||
JdbcTableHandle tableHandle = (JdbcTableHandle) handle;
|
||||
StringBuilder sqlBuilder = new StringBuilder();
|
||||
sqlBuilder.append(format("UPDATE %s SET ", buildRemoteSchemaTableName(tableHandle)));
|
||||
List<String> columnList = getColumnNameFromDataSource(session, tableHandle, updatedColumns);
|
||||
for (int i = 0; i < setNum; i++) {
|
||||
sqlBuilder.append(quoted(columnList.get(i)));
|
||||
sqlBuilder.append(" = ? ");
|
||||
if (i != setNum - 1) {
|
||||
sqlBuilder.append(", ");
|
||||
}
|
||||
}
|
||||
|
||||
sqlBuilder.append("WHERE " + "ctid" + "=?::tid");
|
||||
return sqlBuilder.toString();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setUpdateSql(ConnectorSession session, ConnectorTableHandle tableHandle, PreparedStatement statement, List<Block> columnValueAndCtidBlock, int position, List<String> updatedColumns)
|
||||
{
|
||||
Block ctids = columnValueAndCtidBlock.get(columnValueAndCtidBlock.size() - 1);
|
||||
|
||||
try {
|
||||
for (int i = 0; i < updatedColumns.size(); i++) {
|
||||
setStatement(session, tableHandle, statement, columnValueAndCtidBlock.get(i), position, i);
|
||||
}
|
||||
String ctid = ctids.getString(position, position, 10);
|
||||
statement.setString(updatedColumns.size() + 1, ctid);
|
||||
}
|
||||
catch (SQLException e) {
|
||||
throw new PrestoException(JDBC_ERROR, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public ColumnHandle getUpdateRowIdColumnHandle(ConnectorSession session, ConnectorTableHandle tableHandle, List<ColumnHandle> updatedColumns)
|
||||
{
|
||||
JdbcTypeHandle jdbcTypeHandle = new JdbcTypeHandle(Types.CHAR, Optional.of("char"), 10, 0, Optional.empty());
|
||||
return new JdbcColumnHandle("ctid", jdbcTypeHandle, VARCHAR, true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ColumnHandle getDeleteRowIdColumnHandle(ConnectorSession session, ConnectorTableHandle tableHandle)
|
||||
{
|
||||
JdbcTypeHandle jdbcTypeHandle = new JdbcTypeHandle(Types.CHAR, Optional.of("char"), 10, 0, Optional.empty());
|
||||
return new JdbcColumnHandle("ctid", jdbcTypeHandle, VARCHAR, true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<ConnectorTableHandle> applyDelete(ConnectorSession session, ConnectorTableHandle handle)
|
||||
{
|
||||
if (pushDownModule.equals(JdbcPushDownModule.DEFAULT)) {
|
||||
return Optional.empty();
|
||||
}
|
||||
return Optional.of(handle);
|
||||
}
|
||||
|
||||
private String extractSubQuery(String subQuery)
|
||||
{
|
||||
String query = subQuery.substring(subQuery.indexOf("WHERE"));
|
||||
int count = 1;
|
||||
int lastIndex = 0;
|
||||
for (int i = query.indexOf("(") + 1; i < query.length(); i++) {
|
||||
if (String.valueOf(query.charAt(i)).equals("(")) {
|
||||
count++;
|
||||
}
|
||||
if (String.valueOf(query.charAt(i)).equals(")")) {
|
||||
count--;
|
||||
}
|
||||
if (count == 0) {
|
||||
lastIndex = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
return query.substring(0, lastIndex + 1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public OptionalLong executeDelete(ConnectorSession session, ConnectorTableHandle handle)
|
||||
{
|
||||
JdbcIdentity identity = JdbcIdentity.from(session);
|
||||
JdbcTableHandle basePostgresqlHandle = (JdbcTableHandle) handle;
|
||||
try (Connection connection = connectionFactory.openConnection(identity)) {
|
||||
String sql = "DELETE FROM " + basePostgresqlHandle.getSchemaPrefixedTableName();
|
||||
if (basePostgresqlHandle.getGeneratedSql().isPresent()) {
|
||||
String subQuery = basePostgresqlHandle.getGeneratedSql().get().getSql();
|
||||
sql = String.format("%s %s", sql, extractSubQuery(subQuery));
|
||||
try (PreparedStatement statement = connection.prepareStatement(sql)) {
|
||||
return OptionalLong.of(statement.executeUpdate());
|
||||
}
|
||||
}
|
||||
else {
|
||||
try (PreparedStatement statement = connection.prepareStatement(sql)) {
|
||||
log.debug("Execute: %s", sql);
|
||||
return OptionalLong.of(statement.executeUpdate());
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (SQLException e) {
|
||||
throw new PrestoException(JDBC_ERROR, e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConnectorTableHandle beginDelete(ConnectorSession session, ConnectorTableHandle tableHandle)
|
||||
{
|
||||
JdbcTableHandle jdbcTableHandle = (JdbcTableHandle) tableHandle;
|
||||
jdbcTableHandle.setDeleteOrUpdate(true);
|
||||
return jdbcTableHandle;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void finishDelete(ConnectorSession session, ConnectorTableHandle tableHandle, Collection<Slice> fragments)
|
||||
{
|
||||
}
|
||||
|
||||
@Override
|
||||
public String buildDeleteSql(ConnectorTableHandle handle)
|
||||
{
|
||||
JdbcTableHandle tableHandle = (JdbcTableHandle) handle;
|
||||
return format(
|
||||
"DELETE FROM %s WHERE ctid=%s", buildRemoteSchemaTableName(tableHandle), "?::tid");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setDeleteSql(PreparedStatement statement, Block ctids, int position)
|
||||
{
|
||||
String ctid = ctids.getString(position, position, 10);
|
||||
try {
|
||||
statement.setString(1, ctid);
|
||||
}
|
||||
catch (SQLException e) {
|
||||
throw new PrestoException(JDBC_ERROR, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,8 +16,12 @@ package io.prestosql.plugin.postgresql;
|
|||
import io.prestosql.plugin.jdbc.BaseJdbcConfig;
|
||||
import io.prestosql.plugin.jdbc.ColumnMapping;
|
||||
import io.prestosql.plugin.jdbc.ConnectionFactory;
|
||||
import io.prestosql.plugin.jdbc.JdbcIdentity;
|
||||
import io.prestosql.plugin.jdbc.JdbcSplit;
|
||||
import io.prestosql.plugin.jdbc.JdbcTypeHandle;
|
||||
import io.prestosql.plugin.jdbc.SliceWriteFunction;
|
||||
import io.prestosql.plugin.jdbc.WriteMapping;
|
||||
import io.prestosql.spi.PrestoException;
|
||||
import io.prestosql.spi.connector.ConnectorSession;
|
||||
import io.prestosql.spi.type.ArrayType;
|
||||
import io.prestosql.spi.type.StandardTypes;
|
||||
|
|
@ -29,8 +33,15 @@ import org.postgresql.util.PGobject;
|
|||
|
||||
import javax.inject.Inject;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Types;
|
||||
import java.util.Optional;
|
||||
|
||||
import static io.airlift.slice.Slices.utf8Slice;
|
||||
import static io.prestosql.plugin.jdbc.ColumnMapping.DISABLE_PUSHDOWN;
|
||||
import static io.prestosql.plugin.jdbc.JdbcErrorCode.JDBC_ERROR;
|
||||
import static io.prestosql.plugin.jdbc.StandardColumnMappings.timestampColumnMapping;
|
||||
import static io.prestosql.plugin.jdbc.StandardColumnMappings.timestampWriteFunction;
|
||||
import static io.prestosql.plugin.jdbc.StandardColumnMappings.tinyintWriteFunction;
|
||||
import static io.prestosql.plugin.jdbc.StandardColumnMappings.varbinaryWriteFunction;
|
||||
|
|
@ -49,6 +60,64 @@ public class PostgreSqlClient
|
|||
super(config, postgresqlConfig, connectionFactory, typeManager);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Connection getConnection(JdbcIdentity identity, JdbcSplit split)
|
||||
throws SQLException
|
||||
{
|
||||
Connection connection = connectionFactory.openConnection(identity);
|
||||
return connection;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<ColumnMapping> toPrestoType(ConnectorSession session, Connection connection, JdbcTypeHandle typeHandle)
|
||||
{
|
||||
String jdbcTypeName = typeHandle.getJdbcTypeName()
|
||||
.orElseThrow(() -> new PrestoException(JDBC_ERROR, "Type name is missing: " + typeHandle));
|
||||
|
||||
switch (jdbcTypeName) {
|
||||
case "uuid":
|
||||
return Optional.of(uuidColumnMapping());
|
||||
case "jsonb":
|
||||
case "json":
|
||||
return Optional.of(jsonColumnMapping());
|
||||
case "timestamptz":
|
||||
// PostgreSQL's "timestamp with time zone" is reported as Types.TIMESTAMP rather than Types.TIMESTAMP_WITH_TIMEZONE
|
||||
return Optional.of(timestampWithTimeZoneColumnMapping());
|
||||
default:
|
||||
break;
|
||||
}
|
||||
if (typeHandle.getJdbcType() == Types.VARCHAR && !jdbcTypeName.equals("varchar")) {
|
||||
// This can be e.g. an ENUM
|
||||
return Optional.of(typedVarcharColumnMapping(jdbcTypeName));
|
||||
}
|
||||
if (typeHandle.getJdbcType() == Types.TIMESTAMP) {
|
||||
return Optional.of(timestampColumnMapping());
|
||||
}
|
||||
if (typeHandle.getJdbcType() == Types.ARRAY && supportArrays) {
|
||||
if (!typeHandle.getArrayDimensions().isPresent()) {
|
||||
return Optional.empty();
|
||||
}
|
||||
JdbcTypeHandle elementTypeHandle = getArrayElementTypeHandle(connection, typeHandle);
|
||||
String elementTypeName = typeHandle.getJdbcTypeName()
|
||||
.orElseThrow(() -> new PrestoException(JDBC_ERROR, "Element type name is missing: " + elementTypeHandle));
|
||||
if (elementTypeHandle.getJdbcType() == Types.VARBINARY) {
|
||||
// PostgreSQL jdbc driver doesn't currently support array of varbinary (bytea[])
|
||||
// https://github.com/pgjdbc/pgjdbc/pull/1184
|
||||
return Optional.empty();
|
||||
}
|
||||
return toPrestoType(session, connection, elementTypeHandle)
|
||||
.map(elementMapping -> {
|
||||
ArrayType prestoArrayType = new ArrayType(elementMapping.getType());
|
||||
int arrayDimensions = typeHandle.getArrayDimensions().get();
|
||||
for (int i = 1; i < arrayDimensions; i++) {
|
||||
prestoArrayType = new ArrayType(prestoArrayType);
|
||||
}
|
||||
return arrayColumnMapping(session, prestoArrayType, elementTypeName);
|
||||
});
|
||||
}
|
||||
return super.toPrestoType(session, connection, typeHandle);
|
||||
}
|
||||
|
||||
@Override
|
||||
public WriteMapping toWriteMapping(ConnectorSession session, Type type)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -84,6 +84,111 @@ public class TestPostgreSqlIntegrationSmokeTest
|
|||
assertFalse(getQueryRunner().tableExists(getSession(), "test_drop"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testUpdateByOneField()
|
||||
throws SQLException
|
||||
{
|
||||
execute("CREATE TABLE tpch.test_update (id serial primary key, name varchar, sex char, age int, score varchar, birthday date, salary double precision)");
|
||||
assertUpdate("INSERT INTO test_update VALUES(1, 'Bob', '1', 24, 'excellent', date'1997-09-28', 40000)", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES(2, 'Jack', '1', 25, 'good', date'1996-08-14', 35000)", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES(3, 'Rose', '2', 22, 'excellent', date'1999-07-11', 10000)", 1);
|
||||
|
||||
assertQuery("SELECT * FROM test_update where id = 1", "VALUES(1, 'Bob', '1', 24, 'excellent', date'1997-09-28', 40000)");
|
||||
assertUpdate("UPDATE test_update SET name = 'Kitty', age = 26, birthday = date'1995-08-16' where id = 1", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 1", "VALUES(1, 'Kitty', '1', 26, 'excellent', date'1995-08-16', 40000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE score = 'good'", "VALUES(2, 'Jack', '1', 25, 'good', date'1996-08-14', 35000)");
|
||||
assertUpdate("UPDATE test_update SET name = 'Jane', sex = '2' where score = 'good'", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE score = 'good'", "VALUES(2, 'Jane', '2', 25, 'good', date'1996-08-14', 35000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE salary = 10000", "VALUES(3, 'Rose', '2', 22, 'excellent', date'1999-07-11', 10000)");
|
||||
assertUpdate("UPDATE test_update SET score = 'qualified' where salary = 10000.00 ", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE salary = 10000", "VALUES(3, 'Rose', '2', 22, 'qualified', date'1999-07-11', 10000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE age = 22", "VALUES(3, 'Rose', '2', 22, 'qualified', date'1999-07-11', 10000)");
|
||||
assertUpdate("UPDATE test_update SET age = age + 1, birthday = date'1998-08-16' where age = 22 ", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE age = 23", "VALUES(3, 'Rose', '2', 23, 'qualified', date'1998-08-16', 10000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE birthday = date'1996-08-14'", "VALUES(2, 'Jane', '2', 25, 'good', date'1996-08-14', 35000)");
|
||||
assertUpdate("UPDATE test_update SET salary = 36000 where birthday = date'1996-08-14'", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE birthday = date'1996-08-14'", "VALUES(2, 'Jane', '2', 25, 'good', date'1996-08-14', 36000)");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE name = 'Jane'", "VALUES(2, 'Jane', '2', 25, 'good', date'1996-08-14', 36000)");
|
||||
assertUpdate("UPDATE test_update SET score = 'bad', birthday = date'1995-10-16' where name = 'Jane'", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE name = 'Jane'", "VALUES(2, 'Jane', '2', 25, 'bad', date'1995-10-16', 36000)");
|
||||
|
||||
assertUpdate("DROP TABLE test_update");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testUpdateByMutiField()
|
||||
throws SQLException
|
||||
{
|
||||
execute("CREATE TABLE tpch.test_update (name varchar, id int, score varchar)");
|
||||
assertUpdate("INSERT INTO test_update VALUES('Bob', 1, 'excellent')", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Tim', 2, 'good')", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Jane', 3, 'good')", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Rose', 4, 'bad')", 1);
|
||||
assertUpdate("INSERT INTO test_update VALUES('Petty', 5, 'good')", 1);
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 4", "VALUES('Rose', 4, 'bad')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 5", "VALUES('Petty', 5, 'good')");
|
||||
assertUpdate("UPDATE test_update SET name = 'Kitt' WHERE id IN (4, 5) AND NAME LIKE 'P%'", 1);
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 4", "VALUES('Rose', 4, 'bad')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 5", "VALUES('Kitt', 5, 'good')");
|
||||
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 2", "VALUES('Tim', 2, 'good')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 3", "VALUES('Jane', 3, 'good')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 4", "VALUES('Rose', 4, 'bad')");
|
||||
assertUpdate("UPDATE test_update SET score = 'excellent' WHERE id BETWEEN 2 And 4 AND score != 'bad'", 2);
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 2", "VALUES('Tim', 2, 'excellent')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 3", "VALUES('Jane', 3, 'excellent')");
|
||||
assertQuery("SELECT * FROM test_update WHERE id = 4", "VALUES('Rose', 4, 'bad')");
|
||||
|
||||
assertUpdate("DROP TABLE test_update");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDeleteByOneField()
|
||||
throws SQLException
|
||||
{
|
||||
execute("CREATE TABLE tpch.test_delete (id serial primary key, name varchar, sex char, age int, score varchar, birthday date, salary double precision)");
|
||||
assertUpdate("INSERT INTO test_delete VALUES(1, 'Bob', '1', 24, 'excellent', date'1997-09-28', 40000)", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES(2, 'Jack', '1', 25, 'good', date'1996-08-14', 35000)", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES(3, 'Jane', '2', 23, 'bad', date'1998-07-25', 15000)", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES(4, 'Rose', '2', 22, 'excellent', date'1999-07-11', 10000)", 1);
|
||||
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(4)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE name = 'Bob'", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(3)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE birthday = date'1996-08-14'", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(2)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE salary = 15000", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(1)");
|
||||
assertUpdate("DROP TABLE test_delete");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDeleteByMutiField()
|
||||
throws SQLException
|
||||
{
|
||||
execute("CREATE TABLE tpch.test_delete (name varchar, id int, score varchar)");
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Bob', 1, 'excellent')", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Tim', 2, 'good')", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Jane', 3, 'good')", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Rose', 4, 'bad')", 1);
|
||||
assertUpdate("INSERT INTO test_delete VALUES('Petty', 5, 'good')", 1);
|
||||
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(5)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE id BETWEEN 2 AND 4 AND name IN ('Rose', 'Petty')", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(4)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE name = 'Petty' AND score = 'good'", 1);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(3)");
|
||||
assertUpdate("DELETE FROM test_delete WHERE name LIKE '%o%' OR score = 'good'", 3);
|
||||
assertQuery("SELECT COUNT() FROM test_delete", "VALUES(0)");
|
||||
assertUpdate("DROP TABLE test_delete");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testInsert()
|
||||
throws Exception
|
||||
|
|
|
|||
Loading…
Reference in New Issue