Motor de tabela Iceberg

Este motor oferece uma integração de dados com tabelas Apache Iceberg no Amazon S3, Azure, HDFS e com tabelas armazenadas localmente.

Sem um esquema explícito, a tabela Iceberg já deve existir no armazenamento. Para criar uma nova tabela Iceberg standalone em um backend com permissão de escrita, especifique seu esquema na instrução CREATE TABLE.

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])

A descrição dos argumentos coincide com a descrição dos argumentos dos motores S3, AzureBlobStorage, HDFS e File, respectivamente. format indica o formato dos arquivos de dados na tabela Iceberg.

Para IcebergS3, é possível usar um parâmetro opcional extra_credentials para passar um role_arn para controle de acesso baseado em função no ClickHouse Cloud. Consulte S3 seguro para ver as etapas de configuração.

Os parâmetros do motor podem ser especificados usando Coleções nomeadas

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

Usando coleções nomeadas:

<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')

O motor de tabela Iceberg detecta automaticamente o backend de armazenamento com base na configuração disk e encaminha para IcebergS3, IcebergAzure ou IcebergLocal, conforme o caso. Quando nenhum disk é especificado, a implementação padrão é IcebergS3.

A tabela a seguir mostra como os tipos de dados do Iceberg são mapeados para os tipos de dados do ClickHouse durante a inferência de esquema (para leitura).

Os mapeamentos de tipos de dados acima se aplicam às leituras. As limitações a seguir se aplicam quando o ClickHouse cria ou altera esquemas Iceberg e grava dados:

• O ClickHouse não pode criar um esquema Iceberg que contenha Bool, Decimal, FixedString, Int8, UInt8, Int16 ou UInt16, nem adicionar ou modificar uma coluna para um desses tipos. A operação falha com uma exceção de tipo sem suporte. Isso não impede a inserção de valores Bool ou Decimal em colunas Iceberg existentes.
• Quando o ClickHouse cria ou altera um esquema Iceberg, ele mapeia todas as colunas DateTime e DateTime64 para timestamp do Iceberg, que tem precisão de microssegundos e não inclui timezone. O ClickHouse não pode gerar os tipos de esquema timestamptz, timestamp_ns ou timestamptz_ns; portanto, maior precisão e a semântica de timezone não são representadas no esquema Iceberg.
• O ClickHouse não pode gravar em uma tabela Iceberg que use uma coluna decimal, fixed, timestamp_ns ou timestamptz_ns como campo de partição direto. A operação falha com uma exceção de tipo sem suporte.
• Para arquivos de dados que contêm tipos cujos limites o ClickHouse não consegue serializar, incluindo Bool e Decimal, o ClickHouse omite todos os limites inferiores e superiores das colunas na entrada do manifesto Iceberg. Os tamanhos das colunas e as contagens de valores nulos ainda são incluídos, e os dados permanecem corretos, mas os leitores não podem usar a poda min-max no nível do manifesto para esses arquivos.

O ClickHouse oferece suporte à leitura de tabelas Iceberg cujo esquema evoluiu ao longo do tempo. Isso inclui tabelas em que colunas foram adicionadas, removidas ou reordenadas, bem como colunas que passaram de obrigatórias para Nullable. Além disso, há suporte para as seguintes conversões de tipo:

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

No momento, não é possível alterar estruturas aninhadas nem os tipos dos elementos dentro de arrays e maps.

Para ler uma tabela cujo esquema foi alterado após sua criação com inferência dinâmica de esquema, defina allow_dynamic_metadata_for_data_lakes = true ao criar a tabela.

O ClickHouse oferece suporte à poda de partições durante consultas SELECT em tabelas Iceberg, o que ajuda a otimizar o desempenho das consultas ao ignorar arquivos de dados irrelevantes. Para habilitar a poda de partições, defina use_iceberg_partition_pruning = 1. Para mais informações sobre a poda de partições do Iceberg, acesse https://iceberg.apache.org/spec/#partitioning

O ClickHouse oferece suporte a viagem no tempo em tabelas Iceberg, permitindo consultar dados históricos usando um timestamp específico ou um ID de snapshot.

Com o tempo, operações frequentes de gravação em uma tabela Iceberg podem acumular um grande número de pequenos arquivos de manifesto na lista de manifestos do snapshot atual. Uma lista de manifestos extensa torna o planejamento das consultas mais lento, porque cada arquivo de manifesto precisa ser lido para localizar os arquivos de dados. O ClickHouse pode compactar esses arquivos de manifesto em menos arquivos, porém maiores, usando a instrução OPTIMIZE TABLE ... MANIFEST:

OPTIMIZE TABLE example_table MANIFEST SETTINGS allow_experimental_iceberg_compaction = 1;

Isso gera um novo snapshot (uma operação replace) que faz referência aos mesmos arquivos de dados por meio de um conjunto consolidado de arquivos de manifesto. Nenhum arquivo de dados é reescrito, e nenhuma linha é adicionada, excluída ou deduplicada — apenas a camada de manifesto é reorganizada.

• O recurso é experimental e é controlado pela configuração allow_experimental_iceberg_compaction. A instrução lança uma exceção se a configuração não estiver habilitada.
• A compactação só é tentada quando o número de arquivos de manifesto na lista de manifestos do snapshot atual excede o limite definido pela configuração iceberg_manifest_min_count_to_compact (padrão 30). Se a contagem atual for menor ou igual ao limite, a compactação será ignorada e nenhum novo snapshot será criado. Defina um limite menor para compactar de forma mais imediata.
• OPTIMIZE TABLE ... MANIFEST é suportado apenas para tabelas Iceberg. Executá-lo em qualquer outro motor de tabela lança uma exceção.
• OPTIMIZE TABLE ... MANIFEST é suportado apenas para tabelas Iceberg com format-version 2. Executá-lo em uma tabela com format-version 1 lança uma exceção, assim como executá-lo em uma tabela com format-version 3, porque os metadados first_row_id de row-lineage da v3 ainda não passam por round-trip na reescrita de manifest.
• OPTIMIZE TABLE ... MANIFEST não é suportado para tabelas Iceberg criptografadas cujos arquivos de dados contêm key_metadata por arquivo. Preservar esses metadados de criptografia em uma reescrita de manifest ainda não foi implementado, portanto a instrução lança uma exceção NOT_IMPLEMENTED.

O ClickHouse oferece suporte à leitura de tabelas Iceberg que usam os seguintes métodos de exclusão:

• Exclusões por posição
• Exclusões por igualdade (suportadas a partir da versão 25.8+)

O método de exclusão a seguir não é suportado:

• Vetores de exclusão (introduzido na 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

Observação: não é possível especificar os parâmetros iceberg_timestamp_ms e iceberg_snapshot_id na mesma consulta.

• Snapshots normalmente são criados quando:

  • Novos dados são gravados na tabela
  • É realizada algum tipo de compactação de dados

• Alterações de esquema normalmente não criam snapshots - Isso resulta em comportamentos importantes ao usar viagem no tempo com tabelas que passaram por evolução de esquema.

Estes cenários usam o Spark para ilustrar alterações de esquema feitas por um gravador Iceberg externo.

Considere esta sequência de operações:

-- 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|
+------------+------------+-----+

Resultados da consulta em diferentes timestamps:

• Em ts1 & ts2: aparecem apenas as duas colunas originais
• Em ts3: aparecem as três colunas, com NULL no preço da primeira linha

Uma consulta de viagem no tempo no momento atual pode mostrar um esquema diferente do esquema atual da tabela:

-- 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|
    +------------+------------+-----+

Isso acontece porque ALTER TABLE não cria um novo snapshot; para a tabela atual, o Spark obtém o valor de schema_id do arquivo de metadados mais recente, e não de um snapshot.

A segunda é que, ao usar a viagem no tempo, não é possível obter o estado da tabela antes de qualquer dado ter sido gravado nela:

-- 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.

No ClickHouse, o comportamento é consistente com o do Spark. Você pode mentalmente substituir as consultas Select do Spark pelas consultas Select do ClickHouse, e isso funcionará da mesma forma.

Ao usar o motor de tabela Iceberg no ClickHouse, o sistema precisa localizar o arquivo metadata.json correto que descreve a estrutura da tabela Iceberg. Veja como esse processo de resolução funciona:

1. Especificação direta do caminho:

• Se você definir iceberg_metadata_file_path, o sistema usará esse caminho exato, combinando-o com o caminho do diretório da tabela Iceberg.
• Quando essa configuração é fornecida, todas as outras configurações de resolução são ignoradas.

2. Correspondência do UUID da tabela:

• Se iceberg_metadata_table_uuid for especificado, o sistema:
  • Examinará apenas os arquivos .metadata.json no diretório metadata
  • Filtrará os arquivos que contêm um campo table-uuid correspondente ao UUID especificado (sem diferenciar maiúsculas de minúsculas)

3. Busca padrão:

• Se nenhuma das configurações acima for fornecida, todos os arquivos .metadata.json no diretório metadata passam a ser candidatos

Depois de identificar os arquivos candidatos usando as regras acima, o sistema determina qual deles é o mais recente:

• Se iceberg_recent_metadata_file_by_last_updated_ms_field estiver habilitado:

  • O arquivo com o maior valor de last-updated-ms será selecionado

• Caso contrário:

  • O arquivo com o maior número de versão será selecionado
  • (A versão aparece como V em nomes de arquivo no formato V.metadata.json ou V-uuid.metadata.json)

Observação: Todas as configurações mencionadas (salvo indicação explícita em contrário) são configurações no nível da engine e devem ser especificadas durante a criação da tabela, como mostrado abaixo:

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

Observação: Embora os catálogos Iceberg normalmente façam a resolução de metadados, o motor de tabela Iceberg do ClickHouse interpreta diretamente arquivos armazenados no S3 como tabelas Iceberg, por isso é importante entender essas regras de resolução.

O motor de tabela e a função de tabela Iceberg oferecem suporte a cache de dados, assim como S3, AzureBlobStorage e HDFS. Veja aqui.

O motor de tabela Iceberg e a função de tabela oferecem suporte a um cache de metadados que armazena informações dos arquivos de manifesto, da lista de manifestos e do JSON de metadados. O cache é armazenado em memória. Esse recurso é controlado pela configuração use_iceberg_metadata_files_cache, que vem habilitada por padrão.

A pré-busca assíncrona de metadados pode ser habilitada na criação da tabela Iceberg definindo iceberg_metadata_async_prefetch_period_ms. Se for definido como 0 (padrão) ou se o cache de metadados não estiver habilitado, a pré-busca assíncrona será desabilitada. Para habilitar esse recurso, deve ser informado um valor diferente de zero, em milissegundos. Esse valor representa o intervalo entre os ciclos de pré-busca.

Se estiver habilitado, o servidor executará uma operação recorrente em segundo plano para listar o catálogo remoto e detectar uma nova versão dos metadados. Em seguida, ele fará a análise desses metadados e percorrerá recursivamente o snapshot, buscando arquivos ativos de lista de manifestos e arquivos de manifesto. Os arquivos já disponíveis no cache de metadados não serão baixados novamente. Ao final de cada ciclo de pré-busca, o snapshot de metadados mais recente estará disponível no cache de metadados.

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

Para aproveitar ao máximo a pré-busca assíncrona de metadados em operações de leitura, o parâmetro iceberg_metadata_staleness_ms deve ser especificado como parâmetro de consulta ou de sessão. Por padrão (0 - não especificado), no contexto de cada consulta, o servidor buscará os metadados mais recentes no catálogo remoto. Ao especificar uma tolerância à defasagem dos metadados, o servidor pode usar a versão em cache do snapshot de metadados sem consultar o catálogo remoto. Se houver uma versão dos metadados no cache e ela tiver sido baixada dentro do intervalo de defasagem informado, ela será usada para processar a consulta. Caso contrário, a versão mais recente será buscada no catálogo remoto.

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

Observação: A pré-busca assíncrona de metadados é executada no ICEBERG_SCEDULE_POOL, um threadpool do servidor para operações em segundo plano em tabelas Iceberg ativas. O tamanho desse threadpool é controlado pelo parâmetro de configuração do servidor iceberg_background_schedule_pool_size (o padrão é 10).

Observação: Atualmente, espera-se que o tamanho do cache de metadados seja suficiente para armazenar integralmente o snapshot de metadados mais recente de todas as tabelas ativas, se a pré-busca assíncrona estiver habilitada.

• função de tabela iceberg