Iceberg Connector

Overview

openLooKeng Iceberg is an open table format for huge analytic datasets. The Iceberg connector allows querying data stored in files written in Iceberg format. Iceberg data files can be stored in either Parquet, ORC format, as determined by the format property in the table definition. The table format defaults to ORC.

Requirements

To use Iceberg, you need:

• Network access from the openLooKeng coordinator and workers to the distributed object storage.
• Access to a Hive metastore service (HMS) .
• Network access from the openLooKeng coordinator to the HMS. Hive metastore access with the Thrift protocol defaults to using port 9083.

Hive metastore catalog

The Hive metastore catalog is the default implementation. When using it, the Iceberg connector supports the same metastore configuration properties as the Hive connector. At a minimum, hive.metastore.uri must be configured.

connector.name=iceberg
hive.metastore.uri=thrift://localhost:9083

General configuration

These configuration properties are independent of which catalog implementation is used.

Iceberg general configuration properties With the following content creationetc/catalog/iceberg.properties，please repalcelocalhost:9083 with the right ip and port：

SQL support

This connector provides read access and write access to data and metadata in Iceberg. In addition to the globally available and read operation statements, the connector supports the following features:

Table Related Statements

• create

CREATE TABLE ordersg (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (partitioning = ARRAY['month(order_date)', 'bucket(account_number, 10)', 'country']);

Column related statements

• INSERT

alter table ordersg add COLUMN zip varchar;

• RENAME

ALTER TABLE ordersg RENAME COLUMN zip TO zip_r;

• DELETE

ALTER TABLE ordersg DROP COLUMN zip_r;

Data related Statements

• INSERT

insert into ordersg values(1,date'1988-11-09',666888,'tim','US');

• DELETE

Delete from ordersg where order_id = 1;

• UPDATE

update ordersg set customer = 'Alice' where order_id = 2;

• SELECT

select * from ordersg; 

Partitioned tables

Iceberg supports partitioning by specifying transforms over the table columns. A partition is created for each unique tuple value produced by the transforms. Identity transforms are simply the column name. Other transforms are:

In this example, the table is partitioned by the month of order_date, a hash of account_number (with 10 buckets), and country:

CREATE TABLE ordersg (
                         order_id BIGINT,
                         order_date DATE,
                         account_number BIGINT,
                         customer VARCHAR,
                         country VARCHAR)
    WITH (partitioning = ARRAY['month(order_date)', 'bucket(account_number, 10)', 'country']);

Manually Modifying Partitions

ALTER TABLE ordersg SET PROPERTIES partitioning = ARRAY['month(order_date)'];

For partitioned tables, the Iceberg connector supports the deletion of entire partitions if the WHERE clause specifies filters only on the identity-transformed partitioning columns, that can match entire partitions. Given the table definition above, this SQL will delete all partitions for which country is US:

DELETE FROM iceberg.testdb.ordersg WHERE country = 'US';

Rolling back to a previous snapshot

Iceberg supports a “snapshot” model of data, where table snapshots are identified by an snapshot IDs.

The connector provides a system snapshots table for each Iceberg table. Snapshots are identified by BIGINT snapshot IDs. You can find the latest snapshot ID for table customer_orders by running the following command:

SELECT snapshot_id FROM "ordersg $snapshots" ORDER BY committed_at DESC LIMIT 1;

A SQL procedure system.rollback_to_snapshot allows the caller to roll back the state of the table to a previous snapshot id:

CALL iceberg.system.rollback_to_snapshot('testdb', 'ordersg', 8954597067493422955);

Metadata tables

The connector exposes several metadata tables for each Iceberg table. These metadata tables contain information about the internal structure of the Iceberg table. You can query each metadata table by appending the metadata table name to the table name:

SELECT * FROM "ordersg$data";

$data table#

The $data table is an alias for the Iceberg table itself.

The statement:

SELECT * FROM "ordersg$data";

is equivalent to:

SELECT * FROM ordersg;

$properties table

The $properties table provides access to general information about Iceberg table configuration and any additional metadata key/value pairs that the table is tagged with.

You can retrieve the properties of the current snapshot of the Iceberg table by using the following query:

SELECT * FROM "ordersg$properties";

key | value | ———————–+———-+ write.format.default | PARQUET |

$history table#

The $history table provides a log of the metadata changes performed on the Iceberg table.

You can retrieve the changelog of the Iceberg table by using the following query:

SELECT * FROM "ordersg$history";

made_current_at | snapshot_id | parent_id | is_current_ancestor ———————————-+———————-+———————-+——————– 2022-08-19 05:42:37.854 UTC | 7464177163099901858 | 7924870943332311497 | true 2022-08-19 05:44:35.212 UTC | 2228056193383591891 | 7464177163099901858 | true

The output of the query has the following columns:

$snapshots table

The $snapshots table provides a detailed view of snapshots of the Iceberg table. A snapshot consists of one or more file manifests, and the complete table contents is represented by the union of all the data files in those manifests.

You can retrieve the information about the snapshots of the Iceberg table by using the following query:

SELECT * FROM "ordersg$snapshots";

| committed_at | snapshot_id | parent_id | operation | manifest_list | summary | | 2022-08-08 08:20:04.126 UTC | 7026585913702073835 | | append | hdfs://hadoop1:9000/home/gitama/hadoop/hive/user/hive/warehouse/test_100.db/orders08/metadata/snap-7026585913702073835-1-d0b5ba3d-6363-4f32-974e-79bb68d19423.avro | {changed-partition-count=0, total-equality-deletes=0, total-position-deletes=0, total-delete-files=0, total-files-size=0, total-records=0, total-data-files=0} | | 2022-08-08 08:21:58.343 UTC | 629134202395791160 | 7026585913702073835 | append | hdfs://hadoop1:9000/home/gitama/hadoop/hive/user/hive/warehouse/test_100.db/orders08/metadata/snap-629134202395791160-1-b6e9c1c3-0532-4bf8-a814-a159494e272d.avro | {changed-partition-count=1, added-data-files=1, total-equality-deletes=0, added-records=1, total-position-deletes=0, added-files-size=289, total-delete-files=0, total-files-size=289, total-records=1, total-data-files=1} |

The output of the query has the following columns:

$manifests table#

The $manifests table provides a detailed overview of the manifests corresponding to the snapshots performed in the log of the Iceberg table. You can retrieve the information about the manifests of the Iceberg table by using the following query:

SELECT * FROM "ordersg$manifests";

Path | length | partition_spec_id | added_snapshot_id | added_data_files_count | existing_data_files_count | deleted_data_files_count | partitions
—————————————+———————+———————+———————+——————-+——————–+——————–+——————–+——————–+——————- hdfs://hadoop1:9000/home/gitama/hadoop/hive/user/hive/warehouse/test_100.db/orders08/metadata/b6e9c1c3-0532-4bf8-a814-a159494e272d-m0.avro | 6534 | 0 | 629134202395791160 | 1 | 0 | 0 | [ ]

The output of the query has the following columns:

$partitions table

The $partitions table provides a detailed overview of the partitions of the Iceberg table. You can retrieve the information about the partitions of the Iceberg table by using the following query:

SELECT * FROM "ordersg$partitions";

| record_count | file_count | total_size | data | ————–+————-+———————+———————+————————————| 1 | 1 | 289 | {id={min=null, max=null, null_count=0, nan_count=null}, name={min=null, max=null, null_count=0, nan_count=null}} |

The output of the query has the following columns: Partitions columns

$files table#

The $files table provides a detailed overview of the data files in current snapshot of the Iceberg table.

To retrieve the information about the data files of the Iceberg table use the following query:

SELECT * FROM "ordersg$files";

| content | file_path | file_format | record_count | file_size_in_bytes | column_sizes | value_counts | null_value_counts | nan_value_counts | lower_bounds | upper_bounds | key_metadata | split_offsets | equality_ids |
| 0 | hdfs://192.168.31.120:9000/user/hive/warehouse/orders19/data/20220819_034313_39152_vdmku-1709db2a-dc6f-4ef9-bb77-23f4c150801f.orc | ORC | 1 | 354 | | {1=1, 2=1, 3=1}","{1=0, 2=0, 3=0} | | | | | | |
| 0 | hdfs://192.168.31.120:9000/user/hive/warehouse/orders19/data/20220819_054009_11365_xq568-1803130c-6b7b-4da6-b460-dfb44f176ef4.orc | ORC | 1 | 413 | | {1=1, 2=1, 3=1, 4=1} | {1=0, 2=0, 3=0, 4=1} | | | | | | |

The output of the query has the following columns: Files columns

ALTER TABLE EXECUTE

The connector supports the following commands for use with ALTER TABLE EXECUTE(For details, see Merging Files).

File merging

The optimize command is used for rewriting the active content of the specified table so that it is merged into fewer but larger files. In case that the table is partitioned, the data compaction acts separately on each partition selected for optimization. This operation improves read performance.

All files with a size below the optional file_size_threshold parameter (default value for the threshold is 100MB) are merged:

ALTER TABLE ordersg EXECUTE optimize;

The following statement merges the files in a table that are under 10 megabytes in size:

ALTER TABLE ordersg EXECUTE optimize(file_size_threshold => '10MB');

ALTER TABLE SET PROPERTIES

The connector supports modifying the properties on existing tables using ALTER TABLE SET PROPERTIES.

The following table properties can be updated after a table is created:

• format
• partitioning

ALTER TABLE ordersg SET PROPERTIES format ='PARQUET';

Or to set the column country as a partition column on a table:

ALTER TABLE ordersg SET PROPERTIES partitioning = ARRAY[<existing partition columns>, 'country'];

You can use SHOW CREATE TABLE Ordersg to display the current values of the TABLE properties.

openLooKeng to Iceberg type mapping