Trino connector: Iceberg catalog

Introduction

Apache Iceberg is an open table format for huge analytic datasets. The Iceberg catalog allows Trino querying data stored in files written in Iceberg format, as defined in the Iceberg Table Spec. The catalog supports Apache Iceberg table spec versions 1 and 2.

Requirements

To use Iceberg, you need:

• Network access from the Trino coordinator and workers to the distributed object storage.
• Access to a Hive metastore service (HMS), an AWS Glue catalog, a JDBC catalog, a REST catalog, or a Nessie server.
• Data files stored in a supported file format. These can be configured using file format configuration properties per catalog:
  • ORC
  • Parquet (default)

Schema operations

Create a schema

Create a schema

CREATE SCHEMA catalog.schema_name

Table operations

Create table

The Apache Gravitino Trino connector supports basic Iceberg table creation statements, such as defining fields, allowing null values, and adding comments. The Apache Gravitino Trino connector supports CREATE TABLE AS SELECT.

CREATE OR REPLACE TABLE AS SELECT is not supported. The Iceberg connector caches the table's UUID at query-plan time; dropping and recreating the table inside the same transaction causes the subsequent insert phase to detect a UUID mismatch and fail. Use DROP TABLE followed by CREATE TABLE AS SELECT as an alternative.

Create an Iceberg table

CREATE TABLE catalog.schema_name.table_name
(
  name varchar,
  salary int
)

Alter table

Support for the following alter table operations:

• Rename table
• Add a column
• Drop a column
• Rename a column
• Change a column type
• Set a table property

Select

The Apache Gravitino Trino connector supports most SELECT statements, allowing the execution of queries successfully. It doesn't support certain query optimizations, such as pushdown and pruning functionalities.

Update

UPDATE is only supported for table using v2 or higher of the Iceberg specification.

Delete

Support the deletion of entire partitions and deletion of individual rows for table using v2 or higher of the Iceberg specification. See also Delete limitation.

Merge

MERGE is only supported for table using v2 or higher of the Iceberg specification.

Table procedures

The Apache Gravitino Trino connector delegates Iceberg table maintenance procedures to the underlying Iceberg connector, so they can be invoked via ALTER TABLE ... EXECUTE on Iceberg tables managed by Gravitino. The following procedures are supported:

Procedure	Description
expire_snapshots	Remove old snapshots and their associated metadata/data files to reclaim storage.
remove_orphan_files	Remove files in the table's data directory that are not referenced by any snapshot.
optimize	Rewrite small data files into fewer, larger files to improve read performance (a.k.a. rewrite_data_files).
rewrite_manifests	Rewrite the table's manifest files to optimize metadata scans.

Run Iceberg maintenance procedures

-- Expire snapshots older than the default retention threshold
ALTER TABLE iceberg_test.database_01.table_01 EXECUTE expire_snapshots;

-- Expire snapshots older than 7 days (requires the minimum retention override
-- to be less than or equal to the requested threshold)
ALTER TABLE iceberg_test.database_01.table_01
  EXECUTE expire_snapshots(retention_threshold => '7d');

-- Remove orphan files
ALTER TABLE iceberg_test.database_01.table_01 EXECUTE remove_orphan_files;

-- Compact small data files
ALTER TABLE iceberg_test.database_01.table_01 EXECUTE optimize;

-- Compact small data files whose size is under a threshold
ALTER TABLE iceberg_test.database_01.table_01
  EXECUTE optimize(file_size_threshold => '128MB');

-- Rewrite manifests for faster metadata scans
ALTER TABLE iceberg_test.database_01.table_01 EXECUTE rewrite_manifests;

For the full list of parameters accepted by each procedure, see the Trino Iceberg connector documentation.

Table and schema properties

Create a schema with properties

Iceberg schema does not support properties.

Create a table with properties

Create a table with properties

CREATE TABLE catalog.dbname.tablename
(
  name varchar,
  salary int
) WITH (
  KEY = 'VALUE',
  ...      
);

The following tables are the properties supported by the Iceberg table:

Property	Description	Default Value	Required	Reserved
partitioning	Partition columns for the table	(none)	No	No
sorted_by	Sorted columns for the table	(none)	No	No

Reserved properties: A reserved property is one can't be set by users but can be read by users.

Examples

Complete the following steps before you can use the Iceberg catalog in Trino through Apache Gravitino:

Create a metalake named test and a catalog named iceberg_test in Apache Gravitino.

Create a metalake and an Iceberg catalog

curl -X POST -H "Content-Type: application/json" \
-d '{
  "name": "test",
  "comment": "comment",
  "properties": {}
}' http://{gravitino_host}:8090/api/metalakes

curl -X POST -H "Content-Type: application/json" \
-d '{
  "name": "iceberg_test",
  "type": "RELATIONAL",
  "comment": "comment",
  "provider": "lakehouse-iceberg",
  "properties": {
    "uri": "thrift://{hive_host}:9083",
    "catalog-backend": "hive",
    "warehouse": "hdfs://{hdfs_host}:9000/user/iceberg/warehouse"
  }
}' http://{gravitino_host}:8090/api/metalakes/test/catalogs

For More information about the Iceberg catalog, refer to Iceberg catalog.

Set the value of configuration gravitino.metalake to the metalake you have created, named 'test', and start the Trino container.

Use the Trino CLI to connect to the Trino container and run a query.

List catalogs

SHOW CATALOGS;

Expected output

    Catalog
----------------
 gravitino
 jmx
 system
 iceberg_test
(4 rows)

Query 20231017_082503_00018_6nt3n, FINISHED, 1 node

The gravitino catalog is a catalog defined By Trino catalog configuration. The iceberg_test catalog is the catalog created by you in Apache Gravitino. Other catalogs are regular user-configured Trino catalogs.

Create tables and schemas

Create the database_01 schema

CREATE SCHEMA iceberg_test.database_01;

Create the table_01 table

CREATE TABLE iceberg_test.database_01.table_01
(
name varchar,
salary int
) with (
  partitioning = ARRAY['salary'],
  sorted_by = ARRAY['name']
);

Write data

Insert data into table_01

INSERT INTO iceberg_test.database_01.table_01 (name, salary) VALUES ('ice', 12);

Insert data into table_01 from a select

INSERT INTO iceberg_test.database_01.table_01 (name, salary) SELECT * FROM iceberg_test.database_01.table_01;

Update data in table_01

UPDATE iceberg_test.database_01.table_01 SET name='ice_update' WHERE salary=12;

Delete data from table_01

DELETE FROM iceberg_test.database_01.table_01 WHERE name='ice';

Merge data into table_01

MERGE INTO iceberg_test.database_01.table_01 t USING iceberg_test.database_01.table_02 s
    ON (t.name = s.name)
    WHEN MATCHED AND s.name = 'bob'
        THEN DELETE
    WHEN MATCHED
        THEN UPDATE
            SET salary = s.salary + t.salary
    WHEN NOT MATCHED
        THEN INSERT (name, salary)
              VALUES (s.name, s.salary);

Query data

Query table_01

SELECT * FROM iceberg_test.database_01.table_01;

Modify a table

Add the age column

ALTER TABLE iceberg_test.database_01.table_01 ADD COLUMN age int;

Drop the age column

ALTER TABLE iceberg_test.database_01.table_01 DROP COLUMN age;

Rename table_01 to table_02

ALTER TABLE iceberg_test.database_01.table_01 RENAME TO iceberg_test.database_01.table_02;

Drop

Drop a schema

DROP SCHEMA iceberg_test.database_01;

Drop a table

DROP TABLE iceberg_test.database_01.table_01;

HDFS username and permissions

Before running any Insert statements for Iceberg tables in Trino, you must check that the user Trino is using to access HDFS has access to the warehouse directory. Override this username by setting the HADOOP_USER_NAME system property in the Trino JVM config, replacing hdfs_user with the appropriate username:

Set the HDFS user in the Trino JVM config

-DHADOOP_USER_NAME=hdfs_user

S3

When using AWS S3 within the Iceberg catalog, users need to configure the Trino Iceberg connector's AWS S3-related properties in the catalog's properties. Refer to the documentation of Hive connector with Amazon S3. These configurations must use the trino.bypass. prefix in the Iceberg catalog's attributes to be effective.

Run the following in the Trino CLI.

Create an Iceberg catalog on S3

call gravitino.system.create_catalog(
    'gt_iceberg',
    'lakehouse-iceberg',
    map(
        array['uri', 'catalog-backend', 'warehouse',
          'trino.bypass.hive.s3.aws-access-key', 'trino.bypass.hive.s3.aws-secret-key', 'trino.bypass.hive.s3.region',
          's3-access-key-id', 's3-secret-access-key', 's3-region', 'io-impl'
        ],
        array['thrift://hive:9083', 'hive', 's3a://trino-test-ice/dw2',
        '{aws_access_key}', '{aws_secret_key}', '{region}',
        '{aws_access_key}', '{aws_secret_key}', '{region}', 'org.apache.iceberg.aws.s3.S3FileIO']
    )
);

• The configurations of trino.bypass.hive.s3.aws-access-key, trino.bypass.hive.s3.aws-secret-key, trino.bypass.hive.s3.region are the required the configurations for the Apache Gravitino Trino connector.
• The configurations of s3-access-key-id, s3-secret-access-key, io-impl and s3-region. are the required the configurations for the Apache Gravitino Iceberg catalog.
• The location specifies the storage path on AWS S3. Ensure that the specified directory exists on AWS S3 before proceeding.

Once the Iceberg catalog is successfully created, users can create schemas and tables as follows:

Create a schema and table on S3

CREATE SCHEMA gt_iceberg.gt_db03;

CREATE TABLE gt_iceberg.gt_db03.tb01 (
    name varchar,
    salary int
);

After running the command, the tables are ready for data reading and writing operations on AWS S3.

TThe Iceberg catalog module in the Apache Gravitino server should add AWS S3 support. Refer to Apache Gravitino Iceberg catalog.