PostgreSQL 表引擎 - ClickHouse Documentation

PostgreSQL 引擎允许对存储在远程 PostgreSQL 服务器上的数据执行 SELECT 和 INSERT 查询。

目前，该表引擎仅支持 PostgreSQL 12 及以上版本。

了解我们的 Managed Postgres 服务。它采用与计算资源物理同址的 NVMe 存储，相比使用 EBS 等网络附加存储的替代方案，对于磁盘受限型工作负载，性能最高可提升 10 倍，并且可让你通过 ClickPipes 中的 Postgres CDC 连接器将 Postgres 数据复制到 ClickHouse。

创建表

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 type1 [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 type2 [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = PostgreSQL({host:port, database, table, user, password[, schema, [, on_conflict]] | named_collection[, option=value [,..]]})
SETTINGS
    [ postgresql_connection_pool_size=16, ]
    [ postgresql_connection_pool_wait_timeout=5000, ]
    [ postgresql_connection_pool_retries=2, ]
    [ postgresql_connection_pool_auto_close_connection=false, ]
    [ postgresql_connection_attempt_timeout=2 ]
;

请参阅 CREATE TABLE 查询的详细说明。该表的结构可以与原始 PostgreSQL 表结构不同：

列名应与原始 PostgreSQL 表中的列名一致，但你也可以只使用其中部分列，且顺序不限。
列类型可以与原始 PostgreSQL 表中的不同。ClickHouse 会尝试将值转换为 ClickHouse 数据类型。
external_table_functions_use_nulls 设置定义了如何处理 Nullable 列。默认值：1。如果为 0，表函数不会创建 Nullable 列，而会插入默认值来代替 null。此规则也适用于数组中的 NULL 值。

引擎参数

host:port — PostgreSQL 服务器地址。
database — 远程数据库名称。
table — 远程表名，或按原样传递给 PostgreSQL 的查询 (参见传递查询而不是表名) 。
user — PostgreSQL 用户。
password — 用户密码。
schema — 非默认表 schema。可选。
on_conflict — 冲突解决策略。示例：ON CONFLICT DO NOTHING。可选。注意：添加此选项会降低插入效率。

建议在生产环境中使用命名集合 (自 21.11 版本起可用) 。以下是一个示例：

<named_collections>
    <postgres_creds>
        <host>localhost</host>
        <port>5432</port>
        <user>postgres</user>
        <password>****</password>
        <schema>schema1</schema>
    </postgres_creds>
</named_collections>

某些参数可以通过键值参数覆盖：

SELECT * FROM postgresql(postgres_creds, table='table1');

设置

PostgreSQL 表引擎 (以及 postgresql 表函数) 使用的连接池可以通过 SETTINGS 子句按表配置。若未指定某项设置，则默认使用对应查询级别 postgresql_* 设置的值。

`postgresql_connection_pool_size`

连接池大小 (如果所有连接都在使用中，查询将等待，直到有连接被释放) 。必须为非零值。默认值：16。

`postgresql_connection_pool_wait_timeout`

连接池为空时，push/pop 操作的超时时间，单位为毫秒。0 表示在连接池为空时会阻塞等待。默认值：5000。

`postgresql_connection_pool_retries`

连接池 push/pop 操作的重试次数。默认值：2。

`postgresql_connection_pool_auto_close_connection`

在将连接返回到连接池之前，先关闭该连接。默认值：false。

`postgresql_connection_attempt_timeout`

单次连接到 PostgreSQL 端点时的连接超时时间，单位为秒。该值会作为连接 URL 的 connect_timeout 参数传递。默认值：2。示例：

CREATE TABLE pg_table
(
    `float_nullable` Nullable(Float32),
    `str` String,
    `int_id` Int32
)
ENGINE = PostgreSQL('localhost:5432', 'public', 'test', 'postgres_user', 'postgres_password')
SETTINGS postgresql_connection_pool_size = 32, postgresql_connection_pool_auto_close_connection = 1;

实现细节

PostgreSQL 端的 SELECT 查询会在只读的 PostgreSQL 事务中以 COPY (SELECT ...) TO STDOUT 的形式运行，并在每次 SELECT 查询后提交。简单的 WHERE 子句，如 =, !=, >, >=, <, <= 和 IN，会在 PostgreSQL 服务器上执行。所有 JOIN、聚合、排序、IN [ array ] 条件以及 LIMIT 采样约束，都只会在对 PostgreSQL 的查询结束后于 ClickHouse 中执行。

传递查询而不是表名

table 参数可以不是表名，而是一个按原样传递给 PostgreSQL 的 SELECT 查询。表的结构会从查询结果中推断出来。该查询既可以写成子查询，也可以包装在 query 函数中：

CREATE TABLE pg_table ENGINE = PostgreSQL('localhost:5432', 'test', (SELECT a, b FROM t1 JOIN t2 USING (id) WHERE a > 0), 'user', 'password');
CREATE TABLE pg_table ENGINE = PostgreSQL('localhost:5432', 'test', query('SELECT a, b FROM t1 JOIN t2 USING (id) WHERE a > 0'), 'user', 'password');

这对于将 JOIN、聚合或任何其他处理下推到 PostgreSQL 很有帮助。这样的表是只读的：不允许对其执行 INSERT。postgresql 表函数也支持相同的语法。

子查询形式 (SELECT ...) 会由 ClickHouse 解析，并在发送到服务器之前按 PostgreSQL 方言重新序列化 (包括 PostgreSQL 标识符引用和字符串字面量转义) 。因此，它必须是有效的 ClickHouse SQL。若要传递 ClickHouse 不会解析的 PostgreSQL 特有语法，请使用 query('...') 形式，其文本会原样发送到 PostgreSQL。周围 ClickHouse 查询中的任何外层 WHERE、LIMIT、聚合等，都不会下推到传入的查询中，而是在拉取完整查询结果后由 ClickHouse 应用。要限制从 PostgreSQL 读取的数据，请将过滤器放在传入的查询内部。使用 external_table_strict_query = 1 时，无法下推的外层过滤器会直接抛出异常，而不是在本地应用。

PostgreSQL 端的 INSERT 查询会在 PostgreSQL 事务中以 COPY "table_name" (field1, field2, ... fieldN) FROM STDIN 的形式运行，并在每条 INSERT 语句后自动提交。 PostgreSQL 的 Array 类型会转换为 ClickHouse 数组。

请注意：在 PostgreSQL 中，以 type_name[] 形式创建的数组数据，可能会在同一列的不同行中包含维度数量不同的多维数组。但在 ClickHouse 中，同一列的所有表行只允许使用维度数量相同的多维数组。

支持多个副本，必须使用 | 列出。例如：

CREATE TABLE test_replicas (id UInt32, name String) ENGINE = PostgreSQL(`postgres{2|3|4}:5432`, 'clickhouse', 'test_replicas', 'postgres', 'mysecretpassword');

支持为 PostgreSQL 字典源配置副本优先级。映射中的数值越大，优先级越低。最高优先级为 0。在下面的示例中，副本 example01-1 的优先级最高：

<postgresql>
    <port>5432</port>
    <user>clickhouse</user>
    <password>qwerty</password>
    <replica>
        <host>example01-1</host>
        <priority>1</priority>
    </replica>
    <replica>
        <host>example01-2</host>
        <priority>2</priority>
    </replica>
    <db>db_name</db>
    <table>table_name</table>
    <where>id=10</where>
    <invalidate_query>SQL_QUERY</invalidate_query>
</postgresql>
</source>

使用示例

PostgreSQL 中的表

postgres=# CREATE TABLE "public"."test" (
"int_id" SERIAL,
"int_nullable" INT NULL DEFAULT NULL,
"float" FLOAT NOT NULL,
"str" VARCHAR(100) NOT NULL DEFAULT '',
"float_nullable" FLOAT NULL DEFAULT NULL,
PRIMARY KEY (int_id));

CREATE TABLE

postgres=# INSERT INTO test (int_id, str, "float") VALUES (1,'test',2);
INSERT 0 1

postgresql> SELECT * FROM test;
int_id | int_nullable | float | str  | float_nullable
--------+--------------+-------+------+----------------
       1 |              |     2 | test |
(1 row)

在 ClickHouse 中创建表，并连接到上面创建的 PostgreSQL 表

本示例使用 PostgreSQL 表引擎将 ClickHouse 表连接到 PostgreSQL 表，并通过 PostgreSQL 数据库执行 SELECT 和 INSERT 语句：

CREATE TABLE default.postgresql_table
(
    `float_nullable` Nullable(Float32),
    `str` String,
    `int_id` Int32
)
ENGINE = PostgreSQL('localhost:5432', 'public', 'test', 'postgres_user', 'postgres_password');

使用 SELECT 查询将 PostgreSQL 表中的初始数据插入 ClickHouse 表

postgresql 表函数可将数据从 PostgreSQL 复制到 ClickHouse，通常用于把查询或分析工作放在 ClickHouse 中执行而不是在 PostgreSQL 中执行，从而提升数据的查询性能；也可用于将数据从 PostgreSQL 迁移到 ClickHouse。由于我们要将数据从 PostgreSQL 复制到 ClickHouse，因此会在 ClickHouse 中使用一个 MergeTree 表引擎，并将其命名为 postgresql_copy：

CREATE TABLE default.postgresql_copy
(
    `float_nullable` Nullable(Float32),
    `str` String,
    `int_id` Int32
)
ENGINE = MergeTree
ORDER BY (int_id);

INSERT INTO default.postgresql_copy
SELECT * FROM postgresql('localhost:5432', 'public', 'test', 'postgres_user', 'postgres_password');

将 PostgreSQL 表中的增量数据插入到 ClickHouse 表

如果要在初次插入后，继续在 PostgreSQL 表与 ClickHouse 表之间进行持续同步，可以在 ClickHouse 中使用 WHERE 子句，只插入基于时间戳或唯一序列 ID 在 PostgreSQL 中新增的数据。这需要跟踪此前已插入的最大 ID 或时间戳，例如：

SELECT max(`int_id`) AS maxIntID FROM default.postgresql_copy;

然后插入 PostgreSQL 表中大于该最大值的数据

INSERT INTO default.postgresql_copy
SELECT * FROM postgresql('localhost:5432', 'public', 'test', 'postgres_user', 'postgres_password')
WHERE int_id > (SELECT max(int_id) FROM default.postgresql_copy);

从生成的 ClickHouse 表中查询数据

SELECT * FROM postgresql_copy WHERE str IN ('test');

┌─float_nullable─┬─str──┬─int_id─┐
│           ᴺᵁᴸᴸ │ test │      1 │
└────────────────┴──────┴────────┘

使用非默认 schema

postgres=# CREATE SCHEMA "nice.schema";

postgres=# CREATE TABLE "nice.schema"."nice.table" (a integer);

postgres=# INSERT INTO "nice.schema"."nice.table" SELECT i FROM generate_series(0, 99) as t(i)

CREATE TABLE pg_table_schema_with_dots (a UInt32)
        ENGINE PostgreSQL('localhost:5432', 'clickhouse', 'nice.table', 'postgrsql_user', 'password', 'nice.schema');

另请参阅

​创建表

​设置

​postgresql_connection_pool_size

​postgresql_connection_pool_wait_timeout

​postgresql_connection_pool_retries

​postgresql_connection_pool_auto_close_connection

​postgresql_connection_attempt_timeout

​实现细节

​传递查询而不是表名

​使用示例

​PostgreSQL 中的表

​在 ClickHouse 中创建表，并连接到上面创建的 PostgreSQL 表

​使用 SELECT 查询将 PostgreSQL 表中的初始数据插入 ClickHouse 表

​将 PostgreSQL 表中的增量数据插入到 ClickHouse 表

​从生成的 ClickHouse 表中查询数据

​使用非默认 schema

​相关内容

创建表

设置

`postgresql_connection_pool_size`

`postgresql_connection_pool_wait_timeout`

`postgresql_connection_pool_retries`

`postgresql_connection_pool_auto_close_connection`

`postgresql_connection_attempt_timeout`

实现细节

传递查询而不是表名

使用示例

PostgreSQL 中的表

在 ClickHouse 中创建表，并连接到上面创建的 PostgreSQL 表

使用 SELECT 查询将 PostgreSQL 表中的初始数据插入 ClickHouse 表

将 PostgreSQL 表中的增量数据插入到 ClickHouse 表

从生成的 ClickHouse 表中查询数据

使用非默认 schema

相关内容