motor de tabla de Iceberg

Este motor proporciona una integración de datos con tablas Apache Iceberg en Amazon S3, Azure, HDFS y tablas almacenadas localmente.

Sin un esquema explícito, la tabla Iceberg ya debe existir en el almacenamiento. Para crear una nueva tabla Iceberg independiente en un backend con capacidad de escritura, especifica su esquema en la instrucción 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])

La descripción de estos argumentos coincide con la de los argumentos de los motores S3, AzureBlobStorage, HDFS y File, respectivamente. format indica el formato de los archivos de datos de la tabla Iceberg.

Para IcebergS3, se puede usar el parámetro opcional extra_credentials para pasar un role_arn para acceso basado en roles en ClickHouse Cloud. Consulta Secure S3 para ver los pasos de configuración.

Los parámetros del motor pueden especificarse mediante colecciones con nombre

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

Uso de colecciones con nombre:

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

El motor de tabla Iceberg detecta automáticamente el backend de almacenamiento según la configuración disk y deriva a IcebergS3, IcebergAzure o IcebergLocal, según corresponda. Cuando no se especifica ningún disk, usa de forma predeterminada la implementación IcebergS3.

La siguiente tabla muestra cómo se asignan los tipos de datos de Iceberg a los tipos de datos de ClickHouse durante la inferencia del esquema (para su lectura).

Las correspondencias de tipos de datos anteriores se aplican a las lecturas. Las siguientes limitaciones se aplican cuando ClickHouse crea o modifica esquemas Iceberg y escribe datos:

• ClickHouse no puede crear un esquema Iceberg que contenga Bool, Decimal, FixedString, Int8, UInt8, Int16 o UInt16, ni añadir o modificar una columna para que use uno de estos tipos. La operación falla con una excepción de tipo no admitido. Esto no impide insertar valores Bool o Decimal en columnas Iceberg existentes.
• Cuando ClickHouse crea o modifica un esquema Iceberg, asigna todas las columnas DateTime y DateTime64 al tipo timestamp de Iceberg, que tiene precisión de microsegundos y no incluye zona horaria. ClickHouse no puede generar tipos de esquema timestamptz, timestamp_ns ni timestamptz_ns, por lo que el esquema Iceberg no representa una precisión mayor ni la semántica de zona horaria.
• ClickHouse no puede escribir en una tabla Iceberg que use una columna decimal, fixed, timestamp_ns o timestamptz_ns como campo de partición directo. La operación falla con una excepción de tipo no admitido.
• En los archivos de datos que contienen tipos cuyos límites ClickHouse no puede serializar, incluidos Bool y Decimal, ClickHouse omite todos los límites inferiores y superiores de las columnas en la entrada del manifiesto Iceberg. Los tamaños de las columnas y los recuentos de valores nulos se siguen incluyendo, y los datos permanecen correctos, pero los lectores no pueden usar la poda mín-máx. a nivel de manifiesto para esos archivos.

ClickHouse admite leer tablas Iceberg cuyo esquema ha evolucionado con el tiempo. Esto incluye tablas en las que se han agregado, eliminado o reordenado columnas, así como columnas que han pasado de obligatorias a Nullable. Además, se admiten las siguientes conversiones de tipos:

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

Actualmente, no es posible cambiar estructuras anidadas ni los tipos de los elementos dentro de arrays y maps.

Para leer una tabla cuyo esquema haya cambiado después de su creación con inferencia dinámica de esquema, establezca allow_dynamic_metadata_for_data_lakes = true al crear la tabla.

ClickHouse admite la poda de particiones en las consultas SELECT sobre tablas Iceberg, lo que ayuda a optimizar el rendimiento de las consultas al omitir archivos de datos irrelevantes. Para habilitar la poda de particiones, establezca use_iceberg_partition_pruning = 1. Para obtener más información sobre la poda de particiones de Iceberg, consulte https://iceberg.apache.org/spec/#partitioning

ClickHouse admite el viaje en el tiempo en tablas Iceberg, lo que permite consultar datos históricos con una marca temporal específica o un ID de instantánea.

Con el tiempo, las escrituras frecuentes en una tabla Iceberg pueden acumular una gran cantidad de archivos de manifiesto pequeños en la lista de manifiestos de la instantánea actual. Una lista de manifiestos extensa ralentiza la planificación de consultas, ya que es necesario leer cada archivo de manifiesto para localizar los archivos de datos. ClickHouse puede compactar estos archivos de manifiesto en un menor número de archivos más grandes mediante la sentencia OPTIMIZE TABLE ... MANIFEST:

OPTIMIZE TABLE example_table MANIFEST SETTINGS allow_experimental_iceberg_compaction = 1;

Esto genera una nueva instantánea (una operación replace) que hace referencia a los mismos archivos de datos mediante un conjunto consolidado de archivos de manifiesto. No se reescribe ningún archivo de datos ni se añade, elimina o deduplica ninguna fila; solo se reorganiza la capa de manifiestos.

• La funcionalidad es experimental y está controlada por la configuración allow_experimental_iceberg_compaction. La sentencia lanza una excepción si la configuración no está habilitada.
• La compactación solo se intenta cuando el número de archivos de manifiesto en la lista de manifiestos de la instantánea actual supera el umbral indicado por la configuración iceberg_manifest_min_count_to_compact (valor predeterminado: 30). Si el recuento actual es menor o igual que el umbral, se omite la compactación y no se crea ninguna instantánea nueva. Establezca un umbral más bajo para que la compactación se intente antes.
• OPTIMIZE TABLE ... MANIFEST solo es compatible con tablas Iceberg. Ejecutarlo sobre cualquier otro motor de tabla lanza una excepción.
• OPTIMIZE TABLE ... MANIFEST solo es compatible con tablas Iceberg de format-version 2. Ejecutarlo sobre una tabla de format-version 1 lanza una excepción, y lo mismo ocurre si se ejecuta sobre una tabla de format-version 3, porque los metadatos first_row_id de row-lineage de v3 aún no se preservan al reescribir el manifiesto.
• OPTIMIZE TABLE ... MANIFEST no es compatible con tablas Iceberg cifradas cuyos data files contienen key_metadata por archivo. Aún no está implementado preservar estos metadatos de cifrado al reescribir el manifiesto, por lo que la sentencia lanza una excepción NOT_IMPLEMENTED.

ClickHouse admite la lectura de tablas Iceberg que utilizan los siguientes métodos de eliminación:

• Eliminaciones por posición
• Eliminaciones por igualdad (compatibles a partir de la versión 25.8+)

El siguiente método de eliminación no es compatible:

• Vectores de eliminación (introducido en la 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

Nota: No puede especificar a la vez los parámetros iceberg_timestamp_ms e iceberg_snapshot_id en la misma consulta.

• Normalmente, las instantáneas se crean cuando:

  • Se escriben datos nuevos en la tabla
  • Se realiza algún tipo de compactación de datos

• Los cambios de esquema normalmente no crean instantáneas - Esto da lugar a comportamientos importantes al usar viaje en el tiempo con tablas que han pasado por una evolución del esquema.

Estos escenarios usan Spark para ilustrar los cambios de esquema realizados por un writer de Iceberg externo.

Considere la siguiente secuencia de operaciones:

-- 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 de la consulta en distintas marcas de tiempo:

• En ts1 & ts2: Solo aparecen las dos columnas originales
• En ts3: Aparecen las tres columnas, con NULL en el precio de la primera fila

Una consulta de viaje en el tiempo en el momento actual puede mostrar un esquema distinto del de la tabla actual:

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

Esto sucede porque ALTER TABLE no crea una nueva instantánea, sino que, para la tabla actual, Spark toma el valor de schema_id del archivo de metadatos más reciente, no de una instantánea.

La segunda es que, al usar viaje en el tiempo, no puedes obtener el estado de la tabla antes de que se haya escrito ningún dato en ella:

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

En ClickHouse, el comportamiento coincide con el de Spark. Puedes sustituir mentalmente las consultas Select de Spark por las de ClickHouse y funcionará igual.

Al usar el table engine Iceberg en ClickHouse, el sistema necesita localizar el archivo metadata.json correcto que describe la estructura de la tabla Iceberg. Así funciona este proceso de resolución:

1. Especificación directa de la ruta:

• Si estableces iceberg_metadata_file_path, el sistema usará esta ruta exacta combinándola con la ruta del directorio de la tabla Iceberg.
• Cuando se proporciona este ajuste, se ignoran todos los demás ajustes de resolución.

2. Coincidencia del UUID de la tabla:

• Si se especifica iceberg_metadata_table_uuid, el sistema:
  • Solo examinará los archivos .metadata.json del directorio metadata
  • Filtrará los archivos que contengan un campo table-uuid que coincida con el UUID especificado (sin distinguir entre mayúsculas y minúsculas)

3. Búsqueda predeterminada:

• Si no se proporciona ninguno de los ajustes anteriores, todos los archivos .metadata.json del directorio metadata pasan a ser candidatos

Después de identificar los archivos candidatos según las reglas anteriores, el sistema determina cuál es el más reciente:

• Si iceberg_recent_metadata_file_by_last_updated_ms_field está habilitado:

  • Se selecciona el archivo con el valor last-updated-ms más alto

• En caso contrario:

  • Se selecciona el archivo con el número de versión más alto
  • (La versión aparece como V en nombres de archivo con el formato V.metadata.json o V-uuid.metadata.json)

Nota: Todos los ajustes mencionados (salvo que se especifique explícitamente lo contrario) son ajustes a nivel de motor y deben especificarse durante la creación de la tabla, como se muestra a continuación:

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

Nota: Aunque los catálogos Iceberg normalmente se encargan de la resolución de metadatos, el motor de tabla Iceberg de ClickHouse interpreta directamente los archivos almacenados en S3 como tablas Iceberg, por lo que es importante comprender estas reglas de resolución.

El motor de tabla y la función de tabla Iceberg admiten el caché de datos, al igual que los almacenamientos S3, AzureBlobStorage y HDFS. Consulte aquí.

El motor de tabla y la función de tabla Iceberg admiten una caché de metadatos que almacena información de los archivos de manifiesto, la lista de manifiestos y el JSON de metadatos. La caché reside en memoria. Esta funcionalidad se controla mediante la configuración use_iceberg_metadata_files_cache, habilitada de forma predeterminada.

La precarga asíncrona de metadatos puede habilitarse al crear una tabla Iceberg configurando iceberg_metadata_async_prefetch_period_ms. Si se establece en 0 (valor predeterminado) o si la caché de metadatos no está habilitada, la precarga asíncrona se desactiva. Para habilitar esta función, debe proporcionarse un valor distinto de cero en milisegundos. Este valor representa el intervalo entre los ciclos de precarga.

Si está habilitada, el servidor ejecutará una operación recurrente en segundo plano para listar el catálogo remoto y detectar una nueva versión de los metadatos. A continuación, los analizará y recorrerá recursivamente la instantánea, obteniendo los archivos activos de lista de manifiestos y los archivos de manifiesto. Los archivos que ya estén disponibles en la caché de metadatos no se volverán a descargar. Al final de cada ciclo de precarga, la instantánea de metadatos más reciente estará disponible en la caché de metadatos.

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

Para aprovechar al máximo la precarga asíncrona de metadatos durante las operaciones de lectura, el parámetro iceberg_metadata_staleness_ms debe especificarse como parámetro de consulta o de sesión. De forma predeterminada (0, no especificado), en el contexto de cada consulta, el servidor obtendrá los metadatos más recientes del catálogo remoto. Al especificar una tolerancia a la desactualización de los metadatos, se permite que el servidor use la versión en caché de la instantánea de metadatos sin consultar el catálogo remoto. Si hay una versión de metadatos en caché y se descargó dentro de la ventana de desactualización indicada, se usará para procesar la consulta. De lo contrario, se obtendrá la versión más reciente del catálogo remoto.

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

Nota: La precarga asíncrona de metadatos se ejecuta en ICEBERG_SCEDULE_POOL, que es el threadpool del servidor para operaciones en segundo plano en tablas Iceberg activas. El tamaño de este threadpool se controla mediante el parámetro de configuración del servidor iceberg_background_schedule_pool_size (el valor predeterminado es 10).

Nota: Actualmente se espera que el tamaño de la caché de metadatos sea suficiente para alojar por completo la instantánea de metadatos más reciente de todas las tablas activas, si la precarga asíncrona está habilitada.

• función de tabla de Iceberg