Iceberg 表引擎

该引擎为 Amazon S3、Azure、HDFS 以及本地存储中的 Apache Iceberg 表提供数据集成。

如果未显式指定 schema，Iceberg 表必须已存在于存储中。要在可写后端上创建新的独立 Iceberg 表，请在 CREATE TABLE 语句中指定其 schema。

CREATE TABLE iceberg_table_s3
    ENGINE = IcebergS3(url, [, NOSIGN | access_key_id, secret_access_key, [session_token]], format, [,compression], [,extra_credentials])

CREATE TABLE iceberg_table_azure
    ENGINE = IcebergAzure(connection_string|storage_account_url, container_name, blobpath, [account_name, account_key, format, compression])

CREATE TABLE iceberg_table_hdfs
    ENGINE = IcebergHDFS(path_to_table, [,format] [,compression_method])

CREATE TABLE iceberg_table_local
    ENGINE = IcebergLocal(path_to_table, [,format] [,compression_method])

这些参数的说明分别与 S3、AzureBlobStorage、HDFS 和 File 表引擎中的参数说明一致。 format 表示 Iceberg 表中数据文件的格式。

对于 IcebergS3，可使用可选参数 extra_credentials 传递 role_arn，以便在 ClickHouse Cloud 中进行基于角色的访问。有关配置步骤，请参阅 安全访问 S3。

可以使用 命名集合 指定引擎参数。

CREATE TABLE iceberg_table ENGINE=IcebergS3('http://test.s3.amazonaws.com/clickhouse-bucket/test_table', 'test', 'test')

使用命名集合：

<clickhouse>
    <named_collections>
        <iceberg_conf>
            <url>http://test.s3.amazonaws.com/clickhouse-bucket/</url>
            <access_key_id>test</access_key_id>
            <secret_access_key>test</secret_access_key>
        </iceberg_conf>
    </named_collections>
</clickhouse>

CREATE TABLE iceberg_table ENGINE=IcebergS3(iceberg_conf, filename = 'test_table')

Iceberg 表引擎会根据 disk 设置自动检测存储后端，并相应路由到 IcebergS3、IcebergAzure 或 IcebergLocal。如果未指定 disk，则默认使用 IcebergS3 实现。

下表展示了在进行 schema 推断时 (用于读取) Iceberg 数据类型与 ClickHouse 数据类型之间的映射关系。

上述数据类型映射适用于读取操作。ClickHouse 创建或演进 Iceberg schema 并写入数据时，存在以下限制：

• ClickHouse 无法创建包含 Bool、Decimal、FixedString、Int8、UInt8、Int16 或 UInt16 的 Iceberg schema，也无法将列添加为或修改为上述类型之一。操作会因不支持的类型异常而失败。但这并不妨碍向现有 Iceberg 列中插入 Bool 或 Decimal 值。
• 当 ClickHouse 创建或演进 Iceberg schema 时，会将所有 DateTime 和 DateTime64 列映射为 Iceberg timestamp，后者具有微秒精度且不带时区。ClickHouse 无法生成 timestamptz、timestamp_ns 或 timestamptz_ns schema 类型，因此 Iceberg schema 无法表示更高精度和时区语义。
• ClickHouse 无法向使用 decimal、fixed、timestamp_ns 或 timestamptz_ns 列作为直接分区字段的 Iceberg 表写入数据。操作会因不支持的类型异常而失败。
• 对于包含 ClickHouse 无法序列化其边界值的类型 (包括 Bool 和 Decimal) 的数据文件，ClickHouse 会在 Iceberg 清单条目中省略所有列的下界和上界。仍会包含列大小和 NULL 计数，且数据保持正确，但读取器无法对这些文件使用清单级 min-max pruning。

ClickHouse 支持读取 schema 会随时间演进的 Iceberg 表。这包括列被添加、删除或重新排序的表，以及列从必填变为可为空的表。此外，还支持以下类型转换：

• int -> long
• float -> double
• decimal(P, S) -> decimal(P', S) where P' > P.

目前，尚不支持更改嵌套结构，也不支持更改数组和 Map 中元素的类型。

要读取使用动态 schema 推断创建后 schema 又发生变化的表，请在创建表时设置 allow_dynamic_metadata_for_data_lakes = true。

ClickHouse 支持在针对 Iceberg 表的 SELECT 查询中进行分区裁剪，这有助于通过跳过无关的数据文件来优化查询性能。要启用分区裁剪，请设置 use_iceberg_partition_pruning = 1。有关 Iceberg 分区裁剪的更多信息，请参阅 https://iceberg.apache.org/spec/#partitioning

ClickHouse 支持 Iceberg 表的时间旅行，让你能够通过特定的时间戳或快照 ID 查询历史数据。

随着时间推移，对 Iceberg 表的频繁写入会在当前快照的 manifest 列表中累积大量小型 manifest 文件。manifest 列表过长会拖慢查询计划，因为必须读取每个 manifest 文件才能找出数据文件。ClickHouse 可以使用 OPTIMIZE TABLE ... MANIFEST 语句，将这些 manifest 文件合并整理为数量更少、体积更大的文件：

OPTIMIZE TABLE example_table MANIFEST SETTINGS allow_experimental_iceberg_compaction = 1;

这会生成一个新的快照 (一次 replace 操作) ，通过一组合并后的 manifest 文件引用相同的数据文件。不会重写任何数据文件，也不会添加、删除或去重任何行——仅重新整理 manifest 这一层。

• 该功能为 Experimental，并受 allow_experimental_iceberg_compaction 设置控制。如果该设置未启用，该语句会抛出异常。
• 仅当当前快照的 manifest 列表中的 manifest 文件数量超过 iceberg_manifest_min_count_to_compact 设置指定的阈值 (默认值为 30) 时，才会尝试执行合并整理。如果当前数量小于或等于该阈值，则会跳过合并整理，且不会创建新的快照。将该阈值调低，可更积极地执行合并整理。
• OPTIMIZE TABLE ... MANIFEST 仅支持 Iceberg 表。对任何其他表引擎运行它都会抛出异常。
• OPTIMIZE TABLE ... MANIFEST 仅支持 Iceberg format-version 2 表。对 format-version 1 表运行它会抛出异常，对 format-version 3 表运行也会抛出异常，因为 v3 的行血缘 first_row_id 元数据尚未能在 manifest 重写过程中完整保留。
• OPTIMIZE TABLE ... MANIFEST 不支持已加密的 Iceberg 表，这类表的数据文件包含按文件存储的 key_metadata。目前尚未实现跨 manifest 重写保留此加密元数据，因此该语句会抛出 NOT_IMPLEMENTED 异常。

ClickHouse 支持读取使用以下删除方法的 Iceberg 表：

• 位置删除
• 等值删除 (自 25.8+ 版本起支持)

以下删除方法不支持：

• 删除向量 (在 v3 中引入)

 SELECT * FROM example_table ORDER BY 1
SETTINGS iceberg_timestamp_ms = 1714636800000

 SELECT * FROM example_table ORDER BY 1
SETTINGS iceberg_snapshot_id = 3547395809148285433

注意：在同一个查询中，不能同时指定 iceberg_timestamp_ms 和 iceberg_snapshot_id 参数。

• 快照通常会在以下情况下创建：

  • 向表中写入新数据时
  • 执行某种数据合并整理时

• schema 变更通常不会创建快照——这会在对经历过 schema 演进的表使用时间旅行时带来一些重要特性。

这些场景使用 Spark 说明外部 Iceberg 写入器所做的 schema 变更。

考虑以下操作顺序：

-- Create a table with two columns
CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example (
order_number int,
product_code string
  )
USING iceberg
OPTIONS ('format-version'='2')

-- Insert data into the table
INSERT INTO spark_catalog.db.time_travel_example VALUES
    (1, 'Mars')

  ts1 = now() // A piece of pseudo code

-- Alter table to add a new column
ALTER TABLE spark_catalog.db.time_travel_example ADD COLUMN (price double)

ts2 = now()

-- Insert data into the table
INSERT INTO spark_catalog.db.time_travel_example VALUES (2, 'Venus', 100)

ts3 = now()

-- Query the table at each timestamp
  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts1;

+------------+------------+
|order_number|product_code|
+------------+------------+
|           1|        Mars|
+------------+------------+
  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts2;

+------------+------------+
|order_number|product_code|
+------------+------------+
|           1|        Mars|
+------------+------------+

  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts3;

+------------+------------+-----+
|order_number|product_code|price|
+------------+------------+-----+
|           1|        Mars| NULL|
|           2|       Venus|100.0|
+------------+------------+-----+

不同时间戳下的查询结果：

• 在 ts1 和 ts2 时：仅显示最初的两列
• 在 ts3 时：显示全部三列，其中第一行的 price 为 NULL

在当前时刻执行时间旅行查询时，显示出的 schema 可能与当前表的 schema 不同：

-- Create a table
CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example_2 (
order_number int,
product_code string
  )
USING iceberg
OPTIONS ('format-version'='2')

-- Insert initial data into the table
INSERT INTO spark_catalog.db.time_travel_example_2 VALUES (2, 'Venus');

-- Alter table to add a new column
ALTER TABLE spark_catalog.db.time_travel_example_2 ADD COLUMN (price double);

ts = now();

-- Query the table at a current moment but using timestamp syntax

  SELECT * FROM spark_catalog.db.time_travel_example_2 TIMESTAMP AS OF ts;

    +------------+------------+
    |order_number|product_code|
    +------------+------------+
    |           2|       Venus|
    +------------+------------+

-- Query the table at a current moment
  SELECT * FROM spark_catalog.db.time_travel_example_2;
    +------------+------------+-----+
    |order_number|product_code|price|
    +------------+------------+-----+
    |           2|       Venus| NULL|
    +------------+------------+-----+

之所以会出现这种情况，是因为 ALTER TABLE 不会创建新的 快照；对于当前表，Spark 获取 schema_id 的值时，取自最新的元数据文件，而不是某个 快照。

第二点是，进行时间旅行时，你无法获取该表在尚未写入任何数据之前的状态：

-- Create a table
CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example_3 (
order_number int,
product_code string
  )
USING iceberg
OPTIONS ('format-version'='2');

ts = now();

-- Query the table at a specific timestamp
  SELECT * FROM spark_catalog.db.time_travel_example_3 TIMESTAMP AS OF ts; -- Finises with error: Cannot find a snapshot older than ts.

在 ClickHouse 中，其行为与 Spark 保持一致。你可以直接把 Spark 的 Select 查询理解为 ClickHouse 的 Select 查询，两者的工作方式是一样的。

在 ClickHouse 中使用 Iceberg 表引擎时，系统需要找到描述 Iceberg 表结构的正确 metadata.json 文件。下面介绍这一解析过程的工作原理：

1. 直接指定路径：

• 如果设置了 iceberg_metadata_file_path，系统会将其与 Iceberg 表目录路径拼接，使用这个精确路径。
• 提供此设置后，其他所有解析设置都会被忽略。

2. 表 UUID 匹配：

• 如果指定了 iceberg_metadata_table_uuid，系统将：
  • 只检查 metadata 目录中的 .metadata.json 文件
  • 筛选出包含 table-uuid 字段且与指定 UUID 匹配 (不区分大小写) 的文件

3. 默认搜索：

• 如果上述两个设置都未提供，则 metadata 目录中的所有 .metadata.json 文件都会作为候选文件

根据上述规则识别出候选文件后，系统会进一步确定其中哪个文件最新：

• 如果启用了 iceberg_recent_metadata_file_by_last_updated_ms_field：

  • 选择 last-updated-ms 值最大的文件

• 否则：

  • 选择版本号最大的文件
  • (在格式为 V.metadata.json 或 V-uuid.metadata.json 的文件名中，版本号显示为 V)

注意：上述提到的所有设置 (除非另有明确说明) 均为引擎级设置，必须在创建表时按如下所示指定：

CREATE TABLE example_table ENGINE = Iceberg(
    's3://bucket/path/to/iceberg_table'
) SETTINGS iceberg_metadata_table_uuid = '6f6f6407-c6a5-465f-a808-ea8900e35a38';

注意：虽然 Iceberg 目录通常负责元数据解析，但 ClickHouse 的 Iceberg 表引擎会直接将存储在 S3 中的文件识别为 Iceberg 表，因此理解这些解析规则非常重要。

Iceberg 表引擎和表函数支持数据缓存，与 S3、AzureBlobStorage 和 HDFS 存储相同。请参见此处。

Iceberg 表引擎和表函数支持元数据缓存，可缓存 manifest 文件、manifest 列表和 metadata json 的相关信息。缓存存储在内存中。此功能由设置 use_iceberg_metadata_files_cache 控制，默认启用。

在创建 Iceberg 表时，可通过设置 iceberg_metadata_async_prefetch_period_ms 启用异步元数据预取。如果将其设为 0 (默认值) ，或未启用元数据缓存，则会禁用异步预取。 要启用此功能，需要指定一个非零的毫秒值，表示两次预取周期之间的时间间隔。

启用后，服务器会在后台周期性执行一项操作，列出远程 目录 并检测新的元数据版本。随后会对其进行解析，并递归遍历快照，拉取活动的 manifest 列表和 manifest 文件。 已存在于元数据缓存中的文件不会被重复下载。每个预取周期结束时，最新的元数据快照都会存放在元数据缓存中。

CREATE TABLE example_table ENGINE = Iceberg(
    's3://bucket/path/to/iceberg_table'
) SETTINGS
    iceberg_metadata_async_prefetch_period_ms = 60000;

为了在读取操作中尽可能利用异步元数据预取，应将 iceberg_metadata_staleness_ms 指定为查询参数或会话参数。默认情况下 (0，即未指定) ，在每个查询的上下文中，服务器都会从远程 目录 拉取最新元数据。 通过指定可容忍的元数据过期程度，服务器便可在不调用远程 目录 的情况下使用已缓存的元数据 快照 版本。如果缓存中存在元数据版本，且其下载时间处于给定的过期窗口内，则会使用该版本来处理查询。 否则，将从远程 目录 拉取最新版本。

SELECT count() FROM icebench_table WHERE ...
SETTINGS iceberg_metadata_staleness_ms=120000

注意：异步元数据预取在 ICEBERG_SCEDULE_POOL 中运行，这是一个服务器端线程池，用于对活动 Iceberg 表执行后台操作。该线程池的大小由服务器配置参数 iceberg_background_schedule_pool_size 控制 (默认值为 10) 。

注意：目前预期是，如果启用了异步预取，元数据缓存的大小应足以完整容纳所有活动表的最新元数据快照。

• Iceberg 表函数