Using Impala with Iceberg Tables

• ACID compliance: DML operations are atomic, queries always read a consistent snapshot.
• Hidden partitioning: Iceberg produces partition values by taking a column value and optionally transforming it. Partition information is stored in the Iceberg metadata files. Iceberg is able to TRUNCATE column values or calculate a hash of them and use it for partitioning. Readers don't need to be aware of the partitioning of the table.
• Partition layout evolution: When the data volume or the query patterns change you can update the layout of a table. Since hidden partitioning is used, you don't need to rewrite the data files during partition layout evolution.
• Schema evolution: supports add, drop, update, or rename schema elements, and has no side-effects.
• Time travel: enables reproducible queries that use exactly the same table snapshot, or lets users easily examine changes.
• Cloning Iceberg tables: create an empty Iceberg table based on the definition of another Iceberg table.

When you have an existing Iceberg table that is not yet present in the Hive Metastore, you can use the CREATE EXTERNAL TABLE command in Impala to add the table to the Hive Metastore and make Impala able to interact with this table. Currently Impala supports HadoopTables, HadoopCatalog, and HiveCatalog. If you have an existing table in HiveCatalog, and you are using the same Hive Metastore, you need no further actions.

• HadoopTables. When the table already exists in a HadoopTable it means there is a location on the file system that contains your table. Use the following command to add this table to Impala's catalog:

  
  CREATE EXTERNAL TABLE ice_hadoop_tbl
  STORED AS ICEBERG
  LOCATION '/path/to/table'
  TBLPROPERTIES('iceberg.catalog'='hadoop.tables');
            

• HadoopCatalog. A table in HadoopCatalog means that there is a catalog location in the file system under which Iceberg tables are stored. Use the following command to add a table in a HadoopCatalog to Impala:

  
  CREATE EXTERNAL TABLE ice_hadoop_cat
  STORED AS ICEBERG
  TBLPROPERTIES('iceberg.catalog'='hadoop.catalog',
                'iceberg.catalog_location'='/path/to/catalog',
                'iceberg.table_identifier'='namespace.table');
            

• Alternatively, you can also use custom catalogs to use existing tables. It means you need to define your catalog in hive-site.xml. The advantage of this method is that other engines are more likely to be able to interact with this table. Please note that the automatic metadata update will not work for these tables, you will have to manually call REFRESH on the table when it changes outside Impala. To globally register different catalogs, set the following Hadoop configurations:

  Config Key	Description
  iceberg.catalog.<catalog_name>.type	type of catalog: hive, hadoop, or left unset if using a custom catalog
  iceberg.catalog.<catalog_name>.catalog-impl	catalog implementation, must not be null if type is empty
  iceberg.catalog.<catalog_name>.<key>	any config key and value pairs for the catalog

  For example, to register a HadoopCatalog called 'hadoop', set the following properties in hive-site.xml:

  
  iceberg.catalog.hadoop.type=hadoop;
  iceberg.catalog.hadoop.warehouse=hdfs://example.com:8020/warehouse;
              

  Then in the CREATE TABLE statement you can just refer to the catalog name:

  
  CREATE EXTERNAL TABLE ice_catalogs STORED AS ICEBERG TBLPROPERTIES('iceberg.catalog'='<CATALOG-NAME>');
              

• If the table already exists in HiveCatalog then Impala should be able to see it without any additional commands.

CREATE TABLE ice_t (i INT) STORED AS ICEBERG;
        

By default Impala assumes that the Iceberg table uses Parquet data files. ORC and AVRO are also supported, but we need to tell Impala via setting the table property 'write.format.default' to e.g. 'ORC'.

CREATE TABLE ice_ctas STORED AS ICEBERG AS SELECT i, b FROM value_tbl;

CREATE TABLE ice_ctas_part PARTITIONED BY(d) STORED AS ICEBERG AS SELECT s, ts, d FROM value_tbl;

CREATE TABLE ice_ctas_part_spec PARTITIONED BY SPEC (truncate(3, s)) STORED AS ICEBERG AS SELECT cast(t as INT), s, d FROM value_tbl;
        

          DROP TABLE ice_t;
        

When external.table.purge table property is set to true, then the DROP TABLE statement will also delete the data files. This property is set to true when Impala creates the Iceberg table via CREATE TABLE. When CREATE EXTERNAL TABLE is used (the table already exists in some catalog) then this external.table.purge is set to false, i.e. DROP TABLE doesn't remove any files, only the table definition in HMS.

You can get information about the supported Iceberg data types in the Iceberg spec.

Impala currently does not support schema evolution for tables with AVRO file format.

See schema evolution for more details.

The Iceberg spec has information about partitioning Iceberg tables. With Iceberg, we are not limited to value-based partitioning, we can also partition our tables via several partition transforms.

CREATE TABLE ice_p (i INT, d DATE, s STRING, t TIMESTAMP)
PARTITIONED BY SPEC (BUCKET(5, i), MONTH(d), TRUNCATE(3, s), HOUR(t))
STORED AS ICEBERG;
        

ALTER TABLE ice_p SET PARTITION SPEC (VOID(i), VOID(d), TRUNCATE(3, s), HOUR(t), i);
        

CREATE TABLE ice_p (i INT, b INT) PARTITIONED BY (p1 INT, p2 STRING) STORED AS ICEBERG;
        

One can inspect a table's partition spec by the SHOW PARTITIONS or SHOW CREATE TABLE commands.

Impala is also able to insert new data to Iceberg tables. Currently the INSERT INTO and INSERT OVERWRITE DML statements are supported. One can also remove the contents of an Iceberg table via the TRUNCATE command.

CREATE TABLE ice_p (i INT, b INT) PARTITIONED BY SPEC (bucket(17, i)) STORED AS ICEBERG;
INSERT INTO ice_p VALUES (1, 2);
        

INSERT OVERWRITE statements can replace data in the table with the result of a query. For partitioned tables Impala does a dynamic overwrite, which means partitions that have rows produced by the SELECT query will be replaced. And partitions that have no rows produced by the SELECT query remain untouched. INSERT OVERWRITE is not allowed for tables that use the BUCKET partition transform because dynamic overwrite behavior would be too random in this case. If one needs to replace all contents of a table, they can still use TRUNCATE and INSERT INTO.

Impala can only write Iceberg tables with Parquet data files.

DELETE FROM ice_t where i = 3;
        

More information about the DELETE statement can be found at DELETE Statement (Impala 2.8 or higher only).

ALTER TABLE ice_t DROP PARTITION (i = 3);
ALTER TABLE ice_t DROP PARTITION (day(date_col) < '2024-10-01');
ALTER TABLE ice_t DROP PARTITION (year(timestamp_col) = '2024');
        

Any non-identity transforms must be included in the partition selector like (day(date_col)). Operands for filtering date and timestamp-based columns with transforms must be provided as strings, for example: (day(date_col) = '2024-10-01'). This is a metadata-only operation, the datafiles targeted by the deleted partitions do not get purged or removed from the filesystem, only a new snapshot is getting created with the remaining partitions.

More information about the ALTER TABLE DROP PARTITION statement can be found at ALTER TABLE Statement.

UPDATE ice_t SET val = val + 1;
UPDATE ice_t SET k = 4 WHERE i = 5;
UPDATE ice_t SET ice_t.k = o.k, ice_t.j = o.j, FROM ice_t, other_table o where ice_t.id = o.id;
        

The UPDATE FROM statement can be used to update a target Iceberg table based on a source table (or view) that doesn't need to be an Iceberg table. If there are multiple matches on the JOIN condition, Impala will raise an error.

More information about the UPDATE statement can be found at UPDATE Statement (Impala 2.8 or higher only).

MERGE INTO ice_t USING source ON ice_t.a = source.id WHEN NOT MATCHED THEN INSERT VALUES(id, source.column1);
MERGE INTO ice_t USING source ON ice_t.a = source.id WHEN MATCHED THEN DELETE;
MERGE INTO ice_t USING source ON ice_t.a = source.id WHEN MATCHED THEN UPDATE SET b = source.b;
MERGE INTO ice_t USING source ON ice_t.a = source.id
  WHEN MATCHED AND ice_t.a < 100 THEN UPDATE SET b = source.b
  WHEN MATCHED THEN DELETE
  WHEN NOT MATCHED THEN INSERT VALUES(id, source.column1);
        

More information about the MERGE statement can be found at MERGE Statement.

OPTIMIZE TABLE [db_name.]table_name [FILE_SIZE_THRESHOLD_MB=value];
        

When a table is optimized, a new snapshot is created. The old table state is still accessible by time travel to previous snapshots, because the rewritten data and delete files are not removed physically. Issue the ALTER TABLE ... EXECUTE expire_snapshots(...) command to remove the old files from the file system.

Note that OPTIMIZE TABLE without a specified FILE_SIZE_THRESHOLD_MB rewrites the entire table, therefore the operation can take a long time to complete depending on the size of the table. It is recommended to specify a file size threshold for recurring table maintenance jobs to save resources.

Iceberg stores the table states in a chain of snapshots. By default, Impala uses the current snapshot of the table. But for Iceberg tables, it is also possible to query an earlier state of the table.

SELECT * FROM ice_t FOR SYSTEM_TIME AS OF '2022-01-04 10:00:00';
SELECT * FROM ice_t FOR SYSTEM_TIME AS OF now() - interval 5 days;
SELECT * FROM ice_t FOR SYSTEM_VERSION AS OF 123456;
        

DESCRIBE HISTORY [db_name.]table_name
  [FROM timestamp];

DESCRIBE HISTORY [db_name.]table_name
  [BETWEEN timestamp AND timestamp]
        

DESCRIBE HISTORY ice_t FROM '2022-01-04 10:00:00';
DESCRIBE HISTORY ice_t FROM now() - interval 5 days;
DESCRIBE HISTORY ice_t BETWEEN '2022-01-04 10:00:00' AND '2022-01-05 10:00:00';

Please note that time travel queries are executed using the old schema of the table from the point specified by the time travel parameters. Prior to Impala 4.3.0 the current table schema is used to query an older snapshot of the table, which might have had a different schema in the past.

Iceberg table modifications cause new table snapshots to be created; these snapshots represent an earlier version of the table. The ALTER TABLE [db_name.]table_name EXECUTE ROLLBACK statement can be used to roll back the table to a previous snapshot.

ALTER TABLE ice_tbl EXECUTE ROLLBACK(123456);
        

ALTER TABLE ice_tbl EXECUTE ROLLBACK('2022-01-04 10:00:00');
        

It is only possible to roll back to a snapshot that is a current ancestor of the table.

When a table is rolled back to a snapshot, a new snapshot is created with the same snapshot id, but with a new creation timestamp.

ALTER TABLE ice_tbl EXECUTE expire_snapshots('2022-01-04 10:00:00');
ALTER TABLE ice_tbl EXECUTE expire_snapshots(now() - interval 5 days);
        

Old metadata file clean up can be configured with write.metadata.delete-after-commit.enabled=true and write.metadata.previous-versions-max table properties. This allows automatic metadata file removal after operations that modify metadata such as expiring snapshots or inserting data.

Iceberg stores extensive metadata for each table (e.g. snapshots, manifests, data and delete files etc.), which is accessible in Impala in the form of virtual tables called metadata tables.

Metadata tables can be queried just like regular tables, including filtering, aggregation and joining with other metadata and regular tables. On the other hand, they are read-only, so it is not possible to change, add or remove records from them, they cannot be dropped and new metadata tables cannot be created. Metadata changes made in other ways (not through metadata tables) are reflected in the tables.

SHOW METADATA TABLES IN [db.]tbl [[LIKE] “pattern”]
        

Just like regular tables, metadata tables have schemas that can be queried with the DESCRIBE command. Note, however, that DESCRIBE FORMATTED|EXTENDED are not available for metadata tables.

DESCRIBE functional_parquet.iceberg_alltypes_part.history;
        

To retrieve information from metadata tables, use the usual SELECT statement. You can select any subset of the columns or all of them using ‘*’. Note that in contrast to regular tables, SELECT * on metadata tables always includes complex-typed columns in the result. Therefore, the query option EXPAND_COMPLEX_TYPES only applies to regular tables. This holds also in queries that mix metadata tables and regular tables: for SELECT * expressions from metadata tables, complex types will always be included, and for SELECT * expressions from regular tables, complex types will be included if and only if EXPAND_COMPLEX_TYPES is true.

Note that unnesting collections from metadata tables is not supported.

SELECT
    s.operation,
    h.is_current_ancestor,
    s.summary
FROM functional_parquet.iceberg_alltypes_part.history h
JOIN functional_parquet.iceberg_alltypes_part.snapshots s
  ON h.snapshot_id = s.snapshot_id
WHERE s.operation = 'append'
ORDER BY made_current_at;
        

Impala supports reading NDV (Number of Distinct Values) statistics from Puffin files. For the Puffin specification, see the Apache Iceberg Puffin site.

If there are Puffin stats for multiple snapshots, Impala chooses the most recent one for each column. Note that this means that the stats for different columns may come from different snapshots.

In case there are both HMS and Puffin stats for a column, the more recent one will be used - for HMS stats we use the 'impala.lastComputeStatsTime' table property, and for Puffin stats we use the snapshot timestamp to determine which one is more recent.

Reading Puffin stats is disabled by default; set the "--enable_reading_puffin_stats" startup flag to true to enable it.

Some engines, e.g. Trino, also write the NDV as a property (with key "ndv") in the "statistics" section of the metadata.json file for each blob, in addition to the Puffin file. If such a property is present for a blob, Impala will read the value from the metadata.json file instead of the Puffin file to reduce file I/O.

Note that Impala does not yet support writing Puffin statistics files.

          CREATE TABLE new_ice_tbl LIKE orig_ice_tbl;
        

Because of the Data Types of Iceberg and Impala do not correspond one by one, Impala can only clone between Iceberg tables.

iceberg.io-impl=org.apache.iceberg.hadoop.HadoopFileIO;
iceberg.io.manifest.cache-enabled=true;
iceberg.io.manifest.cache.max-total-bytes=104857600;
iceberg.io.manifest.cache.expiration-interval-ms=3600000;
iceberg.io.manifest.cache.max-content-length=8388608;
        

Manifest caching only works for tables that are loaded with either of HadoopCatalogs or HiveCatalogs. Individual HadoopCatalog and HiveCatalog will have separate manifest caches with the same configuration. By default, only 8 catalogs can have their manifest cache active in memory. This number can be raised by setting a higher value in the java system property iceberg.io.manifest.cache.fileio-max.