Moteur de table Iceberg

Ce moteur fournit une intégration de données avec des tables Apache Iceberg sur Amazon S3, Azure, HDFS et en stockage local.

Sans schéma explicite, la table Iceberg doit déjà exister dans le stockage. Pour créer une nouvelle table Iceberg standalone sur un backend accessible en écriture, spécifiez son schéma dans l’instruction 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 description de ces arguments est identique à celle des arguments des moteurs S3, AzureBlobStorage, HDFS et File. format désigne le format des fichiers de données de la table Iceberg.

Pour IcebergS3, vous pouvez utiliser le paramètre facultatif extra_credentials pour transmettre un role_arn afin de mettre en place un contrôle d'accès basé sur les rôles dans ClickHouse Cloud. Voir Sécuriser S3 pour les étapes de configuration.

Les paramètres du moteur peuvent être spécifiés à l'aide de collections nommées

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

Utilisation des collections nommées :

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

Le moteur de table Iceberg détecte automatiquement le backend de stockage en fonction du paramètre disk et redirige vers IcebergS3, IcebergAzure ou IcebergLocal selon le cas. Lorsqu'aucun disk n'est spécifié, l'implémentation IcebergS3 est utilisée par défaut.

Le tableau suivant montre comment les types de données Iceberg sont associés aux types de données ClickHouse lors de l’inférence de schéma (à des fins de lecture).

Les mappages de types de données ci-dessus s’appliquent aux lectures. Les limitations suivantes s’appliquent lorsque ClickHouse crée ou fait évoluer des schémas Iceberg et écrit des données :

• ClickHouse ne peut pas créer de schéma Iceberg contenant Bool, Decimal, FixedString, Int8, UInt8, Int16 ou UInt16, ni ajouter ou modifier une colonne pour lui attribuer l’un de ces types. L’opération échoue avec une exception de type non pris en charge. Cela n’empêche pas l’insertion de valeurs Bool ou Decimal dans des colonnes Iceberg existantes.
• Lorsque ClickHouse crée ou fait évoluer un schéma Iceberg, il mappe chaque colonne DateTime et DateTime64 vers le type timestamp d’Iceberg, qui a une précision à la microseconde et ne comporte pas de fuseau horaire. ClickHouse ne peut pas générer les types de schéma timestamptz, timestamp_ns ou timestamptz_ns ; une précision supérieure et la sémantique des fuseaux horaires ne sont donc pas représentées dans le schéma Iceberg.
• ClickHouse ne peut pas écrire dans une table Iceberg qui utilise une colonne decimal, fixed, timestamp_ns ou timestamptz_ns comme champ de partition direct. L’opération échoue avec une exception de type non pris en charge.
• Pour les fichiers de données contenant des types dont ClickHouse ne peut pas sérialiser les limites, notamment Bool et Decimal, ClickHouse omet toutes les limites inférieures et supérieures des colonnes dans l’entrée du manifeste Iceberg. Les tailles des colonnes et le nombre de valeurs nulles sont toujours inclus, et les données restent correctes, mais les lecteurs ne peuvent pas utiliser l’élagage min-max au niveau du manifeste pour ces fichiers.

ClickHouse prend en charge la lecture de tables Iceberg dont le schéma a évolué au fil du temps. Cela inclut les tables où des colonnes ont été ajoutées, supprimées ou réordonnées, ainsi que celles dont des colonnes sont passées de required à Nullable. De plus, les conversions de type suivantes sont prises en charge :

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

À ce jour, il n'est pas possible de modifier les structures imbriquées ni les types des éléments au sein des Array et des Map.

Pour lire une table dont le schéma a changé après sa création avec l'inférence dynamique de schéma, définissez allow_dynamic_metadata_for_data_lakes = true lors de la création de la table.

ClickHouse prend en charge l’élagage des partitions dans les requêtes SELECT sur les tables Iceberg, ce qui permet d’optimiser les performances des requêtes en évitant de lire les fichiers de données non pertinents. Pour activer l’élagage des partitions, définissez use_iceberg_partition_pruning = 1. Pour plus d’informations sur l’élagage des partitions dans Iceberg, consultez https://iceberg.apache.org/spec/#partitioning

ClickHouse prend en charge le time travel pour les tables Iceberg, ce qui vous permet d'interroger des données historiques à l'aide d'un horodatage spécifique ou d'un ID d'instantané.

Avec le temps, des écritures fréquentes dans une table Iceberg peuvent entraîner l’accumulation d’un grand nombre de petits fichiers manifestes dans la liste de manifestes de l’instantané actuel. Une liste de manifestes trop longue ralentit la planification des requêtes, car chaque fichier manifeste doit être lu pour localiser les fichiers de données. ClickHouse peut compacter ces fichiers manifestes en un nombre plus réduit de fichiers plus volumineux à l’aide de l’instruction OPTIMIZE TABLE ... MANIFEST :

OPTIMIZE TABLE example_table MANIFEST SETTINGS allow_experimental_iceberg_compaction = 1;

Cela produit un nouvel instantané (une opération replace) qui pointe vers les mêmes fichiers de données au moyen d’un ensemble consolidé de fichiers manifestes. Aucun fichier de données n’est réécrit et aucune ligne n’est ajoutée, supprimée ou dédupliquée — seule la couche des manifestes est réorganisée.

• Cette fonctionnalité est expérimentale et protégée par le paramètre allow_experimental_iceberg_compaction. L'instruction lève une exception si ce paramètre n'est pas activé.
• La compaction n'est tentée que lorsque le nombre de fichiers manifest dans la liste de manifestes de l'instantané actuel dépasse le seuil défini par le paramètre iceberg_manifest_min_count_to_compact (valeur par défaut : 30). Si ce nombre est inférieur ou égal au seuil, la compaction est ignorée et aucun nouveau instantané n'est créé. Abaissez ce seuil pour déclencher la compaction plus tôt.
• OPTIMIZE TABLE ... MANIFEST est pris en charge uniquement pour les tables Iceberg. Son exécution sur tout autre moteur de table lève une exception.
• OPTIMIZE TABLE ... MANIFEST est pris en charge uniquement pour les tables Iceberg au format version 2. Son exécution sur une table au format version 1 lève une exception, de même que son exécution sur une table au format version 3, car les métadonnées v3 first_row_id de lignée de lignes ne sont pas encore restituées à l'identique lors de la réécriture du manifeste.
• OPTIMIZE TABLE ... MANIFEST n'est pas pris en charge pour les tables Iceberg chiffrées dont les fichiers de données contiennent un key_metadata par fichier. La préservation de ces métadonnées de chiffrement lors d'une réécriture du manifeste n'est pas encore implémentée ; l'instruction lève donc une exception NOT_IMPLEMENTED.

ClickHouse prend en charge la lecture des tables Iceberg utilisant les méthodes de suppression suivantes :

• Suppressions par position
• Suppressions par égalité (prises en charge à partir de la version 25.8+)

La méthode de suppression suivante n’est pas prise en charge :

• Vecteurs de suppression (introduits dans la version 3)

 SELECT * FROM example_table ORDER BY 1
SETTINGS iceberg_timestamp_ms = 1714636800000

 SELECT * FROM example_table ORDER BY 1
SETTINGS iceberg_snapshot_id = 3547395809148285433

Remarque : vous ne pouvez pas spécifier les paramètres iceberg_timestamp_ms et iceberg_snapshot_id dans une même requête.

• Les instantanés sont généralement créés lorsque :

  • De nouvelles données sont écrites dans la table
  • Une compaction des données est effectuée

• Les changements de schéma ne créent généralement pas d’instantanés - Cela entraîne des comportements importants lors de l’utilisation du time travel avec des tables ayant fait l’objet d’une évolution du schéma.

Ces scénarios utilisent Spark pour illustrer les modifications de schéma apportées par un writer Iceberg externe.

Considérez la séquence d’opérations suivante :

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

Résultats de la requête à différents horodatages :

• À ts1 & ts2 : seules les deux colonnes d’origine apparaissent
• À ts3 : les trois colonnes apparaissent, avec NULL comme prix pour la première ligne

Une requête time travel exécutée à l’instant présent peut afficher un schéma différent de celui de la table actuelle :

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

Cela se produit parce que ALTER TABLE ne crée pas de nouvel instantané ; pour la table actuelle, Spark récupère la valeur de schema_id à partir du fichier de métadonnées le plus récent, et non d’un instantané.

Le deuxième point est que, lors d’une requête en time travel, vous ne pouvez pas obtenir l’état de la table avant qu’aucune donnée n’y ait été écrite :

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

Dans ClickHouse, le comportement est identique à celui de Spark. Vous pouvez considérer qu’il suffit de remplacer mentalement les requêtes Select de Spark par des requêtes Select de ClickHouse, et cela fonctionnera de la même manière.

Lors de l’utilisation du moteur de table Iceberg dans ClickHouse, le système doit localiser le bon fichier metadata.json, qui décrit la structure de la table Iceberg. Voici comment ce processus de résolution fonctionne :

1. Spécification directe du chemin :

• Si vous définissez iceberg_metadata_file_path, le système utilisera ce chemin exact en le combinant avec le chemin du répertoire de la table Iceberg.
• Lorsque ce paramètre est défini, tous les autres paramètres de résolution sont ignorés.

2. Correspondance de l’UUID de la table :

• Si iceberg_metadata_table_uuid est spécifié, le système :
  • examinera uniquement les fichiers .metadata.json du répertoire metadata
  • filtrera les fichiers contenant un champ table-uuid correspondant à l’UUID spécifié (sans distinction entre majuscules et minuscules)

3. Recherche par défaut :

• Si aucun des paramètres ci-dessus n’est défini, tous les fichiers .metadata.json du répertoire metadata sont considérés comme des candidats

Après avoir identifié les fichiers candidats à l’aide des règles ci-dessus, le système détermine lequel est le plus récent :

• Si iceberg_recent_metadata_file_by_last_updated_ms_field est activé :

  • Le fichier ayant la valeur last-updated-ms la plus élevée est sélectionné

• Sinon :

  • Le fichier ayant le numéro de version le plus élevé est sélectionné
  • (La version apparaît sous la forme V dans les noms de fichier au format V.metadata.json ou V-uuid.metadata.json)

Remarque : Tous les paramètres mentionnés (sauf indication contraire explicite) sont des paramètres au niveau du moteur et doivent être spécifiés lors de la création de la table, comme indiqué ci-dessous :

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

Remarque : Bien que les catalogues Iceberg gèrent généralement la résolution des métadonnées, le moteur de table Iceberg de ClickHouse interprète directement les fichiers stockés dans S3 comme des tables Iceberg, d’où l’importance de comprendre ces règles de résolution.

Le moteur de table Iceberg et la fonction de table correspondante prennent en charge le cache de données, comme les stockages S3, AzureBlobStorage et HDFS. Voir ici.

Le moteur de table et la fonction de table Iceberg prennent en charge un cache de métadonnées qui stocke des informations sur les fichiers manifeste, la liste de manifestes et le JSON de métadonnées. Le cache est stocké en mémoire. Cette fonctionnalité est contrôlée par le paramètre use_iceberg_metadata_files_cache, qui est activé par défaut.

Le préchargement asynchrone des métadonnées peut être activé lors de la création d’une table Iceberg en définissant iceberg_metadata_async_prefetch_period_ms. Si cette valeur est définie sur 0 (valeur par défaut), ou si le cache des métadonnées n’est pas activé, le préchargement asynchrone est désactivé. Pour activer cette fonctionnalité, vous devez fournir une valeur non nulle en millisecondes. Elle représente l’intervalle entre les cycles de préchargement.

S’il est activé, le serveur exécutera une opération récurrente en arrière-plan pour lister le catalogue distant et détecter une nouvelle version des métadonnées. Il l’analysera ensuite et parcourra récursivement l’instantané, en récupérant les fichiers de liste de manifests actifs ainsi que les fichiers manifeste. Les fichiers déjà présents dans le cache des métadonnées ne seront pas téléchargés à nouveau. À la fin de chaque cycle de préchargement, le dernier instantané des métadonnées est disponible dans le cache des métadonnées.

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

Afin de tirer pleinement parti du préchargement asynchrone des métadonnées lors des opérations de lecture, le paramètre iceberg_metadata_staleness_ms doit être défini comme paramètre de requête ou de session. Par défaut (0 - non spécifié), pour chaque requête, le serveur récupère les métadonnées les plus récentes depuis le catalogue distant. En spécifiant une tolérance à l'ancienneté des métadonnées, le serveur est autorisé à utiliser la version en cache de l'instantané des métadonnées sans interroger le catalogue distant. S'il existe une version des métadonnées en cache et qu'elle a été téléchargée dans le délai d'ancienneté indiqué, elle sera utilisée pour traiter la requête. Sinon, la version la plus récente sera récupérée depuis le catalogue distant.

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

Remarque : Le préchargement asynchrone des métadonnées s'exécute dans ICEBERG_SCEDULE_POOL, le threadpool côté serveur dédié aux opérations en arrière-plan sur les tables Iceberg actives. La taille de ce threadpool est contrôlée par le paramètre de configuration du serveur iceberg_background_schedule_pool_size (la valeur par défaut est 10).

Remarque : On considère actuellement que la taille du cache de métadonnées est suffisante pour contenir intégralement le dernier instantané de métadonnées de toutes les tables actives, si le préchargement asynchrone est activé.

• fonction de table Iceberg