Коннектор Iceberg#

Apache Iceberg — открытый формат таблиц для огромных аналитических наборов данных. Коннектор Iceberg позволяет выполнять запросы к данным, которые хранятся в файлах формата Iceberg, как определено в спецификации таблиц Iceberg. Коннектор поддерживает версии 1 и 2 спецификации таблиц Apache Iceberg.

Состояние таблицы хранится в файлах метаданных. При любом изменении состояния таблицы создается новый файл метаданных, который атомарной заменой подменяет старые метаданные. Файл метаданных таблицы отслеживает схему таблицы, конфигурацию секционирования, пользовательские свойства и снимки содержимого таблицы.

Файлы данных Iceberg хранятся в формате Parquet, ORC или Avro, который задается свойством format в определении таблицы.

Iceberg разработан для устранения известных ограничений масштабируемости Hive, где метаданные таблиц хранятся в метахранилище на основе реляционной базы данных, например MySQL. Hive отслеживает в метахранилище расположения секций, но не отдельные файлы данных. Запросы Trino, использующие Коннектор Hive, должны сначала обратиться к метахранилищу, чтобы получить расположения секций, затем обратиться к базовой файловой системе, чтобы перечислить все файлы данных внутри каждой секции, а затем прочитать метаданные из каждого файла данных.

Поскольку Iceberg хранит пути к файлам данных в файлах метаданных, он обращается к базовой файловой системе только за файлами, которые необходимо прочитать.

Требования#

Для использования Iceberg требуется:

Общая конфигурация#

Чтобы настроить коннектор Iceberg, создайте файл свойств каталога etc/catalog/example.properties, который ссылается на коннектор iceberg.

Каталог метахранилища Hive используется по умолчанию.

Необходимо выбрать и настроить одну из поддерживаемых файловых систем.

connector.name=iceberg
hive.metastore.uri=thrift://example.net:9083
fs.x.enabled=true

Замените свойство конфигурации fs.x.enabled нужной файловой системой.

Доступны и другие типы каталогов метаданных, перечисленные в разделе требований этой темы. У каждого типа метахранилища есть собственные свойства конфигурации, а также общие свойства конфигурации метахранилища.

Следующие свойства конфигурации не зависят от используемой реализации каталога:

Общие свойства конфигурации Iceberg#

Имя свойства

Описание

Значение по умолчанию

iceberg.catalog.type

Определяет используемый тип метахранилища. Возможные значения:

  • hive_metastore

  • glue

  • jdbc

  • rest

  • nessie

  • snowflake

hive_metastore

iceberg.file-format

Определяет формат файлов для хранения данных таблиц Iceberg. Возможные значения:

  • PARQUET

  • ORC

  • AVRO

PARQUET

iceberg.compression-codec

Кодек сжатия, используемый при записи файлов. Возможные значения:

  • NONE

  • SNAPPY

  • LZ4

  • ZSTD

  • GZIP

ZSTD

iceberg.use-file-size-from-metadata

Читать размеры файлов из метаданных, а не из файловой системы. Это свойство следует использовать только как обходной путь для этой проблемы. Проблема исправлена в Iceberg версии 0.11.0.

true

iceberg.max-partitions-per-writer

Максимальное число секций, обрабатываемых одним писателем. Эквивалентное свойство сеанса каталога — max_partitions_per_writer.

100

iceberg.target-max-file-size

Целевой максимальный размер записываемых файлов; фактический размер может быть больше.

1GB

iceberg.unique-table-location

Использовать случайные уникальные расположения таблиц.

true

iceberg.dynamic-filtering.wait-timeout

Максимальное время ожидания завершения динамических фильтров при создании фрагментов.

1s

iceberg.delete-schema-locations-fallback

Удалять ли расположения схем, когда Trino не может определить, содержат ли они внешние файлы.

false

iceberg.minimum-assigned-split-weight

Десятичное значение в диапазоне (0, 1], используемое как минимум для весов, назначаемых каждому фрагменту. Низкое значение может повысить производительность для таблиц с небольшими файлами. Более высокое значение может повысить производительность запросов с сильно перекошенными агрегациями или соединениями.

0.05

iceberg.table-statistics-enabled

Включает Статистика таблиц. Эквивалентное свойство сеанса каталога для использования в отдельном сеансе — statistics_enabled. Установите false, чтобы запретить Оптимизации на основе стоимости использовать статистику для выбора более эффективного плана запроса и тем самым повышения производительности обработки. Устанавливать false не рекомендуется; это не отключает сбор статистики.

true

iceberg.extended-statistics.collect-on-write

Включает сбор расширенной статистики для операций записи. Эквивалентное свойство сеанса каталога — collect_extended_statistics_on_write.

true

iceberg.projection-pushdown-enabled

Включает проталкивание проекций.

true

iceberg.hive-catalog-name

Каталог, в который выполняется перенаправление при обращении к таблице Hive.

iceberg.register-table-procedure.enabled

Включите, чтобы разрешить пользователю вызывать процедуру register_table.

false

iceberg.add-files-procedure.enabled

Включите, чтобы разрешить пользователю вызывать процедуру add_files.

false

iceberg.query-partition-filter-required

Установите true, чтобы запрос был обязан использовать фильтр по секции для схем, указанных в iceberg.query-partition-filter-required-schemas. Эквивалентное свойство сеанса каталога — query_partition_filter_required.

false

iceberg.query-partition-filter-required-schemas

Задает список схем, для которых Trino может требовать, чтобы запросы к исходным таблицам использовали фильтр по ключам секционирования. Эквивалентное свойство сеанса — query_partition_filter_required_schemas. Список используется, если свойство конфигурации iceberg.query-partition-filter-required или свойство сеанса каталога query_partition_filter_required установлено в true.

[]

iceberg.incremental-refresh-enabled

Установите false, чтобы операция обновления материализованного представления всегда выполняла полное обновление. Для временного применения в конкретном каталоге можно использовать свойство сеанса каталога incremental_refresh_enabled. В большинстве случаев инкрементальное обновление выгоднее полного, поскольку требуется сканировать гораздо меньшую часть исходных таблиц. Хотя инкрементальное обновление может сканировать меньше данных, оно может создавать больше файлов данных, поскольку для вставки новых записей использует операцию добавления.

true

iceberg.materialized-views.refresh-max-snapshots-to-expire

Максимальное число снимков материализованного представления, срок действия которых истекает при обновлении. Значение 0 означает, что снимки не удаляются. Более высокие значения могут увеличить время обновления материализованного представления, если в нем много снимков. Все снимки в пределах периода, заданного свойством конфигурации iceberg.materialized-views.refresh-snapshot-retention-period, сохраняются.

200

iceberg.materialized-views.refresh-snapshot-retention-period

Период хранения снимков материализованного представления. Все снимки старше этого значения удаляются при обновлении материализованного представления, но не больше предела, заданного свойством конфигурации iceberg.materialized-views.refresh-max-snapshots-to-expire.

4h

iceberg.metadata-cache.enabled

Установите false, чтобы отключить кэширование файлов метаданных в памяти координатора. Этот кэш не используется, если fs.cache.enabled установлено в true.

true

iceberg.object-store-layout.enabled

Установите true, чтобы включить схему размещения файлов объектного хранилища Iceberg. При включении этой схемы детерминированный хэш добавляется сразу после пути записи данных.

false

iceberg.expire-snapshots.min-retention

Минимальный период хранения для команды expire_snapshot. Эквивалентное свойство сеанса — expire_snapshots_min_retention.

7d

iceberg.remove-orphan-files.min-retention

Минимальный период хранения для команды remove_orphan_files. Эквивалентное свойство сеанса — remove_orphan_files_min_retention.

7d

iceberg.idle-writer-min-file-size

Минимальный объем данных, записанных одним писателем секции, после которого он может считаться простаивающим и быть закрыт движком. Эквивалентное свойство сеанса — idle_writer_min_file_size.

16MB

iceberg.sorted-writing-enabled

Включает сортированную запись в таблицы с заданным порядком сортировки. Эквивалентное свойство сеанса — sorted_writing_enabled.

true

iceberg.sorted-writing.local-staging-path

Локальный каталог, который Trino может использовать для промежуточной записи в сортированные таблицы. Заполнитель ${USER} можно использовать, чтобы для каждого пользователя применялось отдельное расположение. Если свойство не настроено, при записи в сортированные таблицы для промежуточных данных используется целевое хранилище, что может быть неэффективно при записи в объектные хранилища, такие как S3. Если fs.hadoop.enabled не включено, для использования этой возможности требуется настроить локальную файловую систему.

iceberg.allowed-extra-properties

Список дополнительных свойств, которые разрешено задавать для таблиц Iceberg. Используйте *, чтобы разрешить все свойства.

[]

iceberg.split-manager-threads

Число потоков для создания фрагментов.

Удвоенное число процессоров на узле координатора.

iceberg.planning-threads

Число потоков для чтения манифестов во время планирования.

Удвоенное число процессоров на узле координатора.

iceberg.metadata.parallelism

Число потоков для получения метаданных. Сейчас распараллелена только загрузка таблиц.

8

iceberg.file-delete-threads

Число потоков для удаления файлов при выполнении процедуры expire_snapshots или remove_orphan_files, а также при выполнении запросов DROP TABLE.

Удвоенное число процессоров на узле координатора.

iceberg.bucket-execution

Включает выполнение с учетом бакетов. Это позволяет движку использовать сведения о физическом бакетировании для оптимизации запросов за счет сокращения обмена данными.

true

Поддержка отказоустойчивого выполнения#

Коннектор поддерживает отказоустойчивую обработку запросов. Операции чтения и записи поддерживаются при любой политике повторных попыток.

Конфигурация доступа к файловым системам#

Коннектор поддерживает доступ к следующим файловым системам:

Необходимо включить и настроить доступ к конкретной файловой системе. Устаревшая поддержка не рекомендуется и будет удалена.

Сопоставление типов#

Коннектор читает и записывает данные в поддерживаемые файловые форматы Avro, ORC и Parquet согласно спецификации Iceberg.

Поскольку Trino и Iceberg поддерживают типы, которых нет у другой стороны, этот коннектор изменяет некоторые типы при чтении или записи данных. Типы данных могут сопоставляться по-разному в двух направлениях между Trino и источником данных. В следующих разделах описано сопоставление типов для каждого направления.

Спецификация Iceberg включает поддерживаемые типы данных и сопоставление с представлением в файлах Avro, ORC или Parquet:

Сопоставление типов Iceberg с типами Trino#

Коннектор сопоставляет типы Iceberg с соответствующими типами Trino по следующей таблице:

Сопоставление типов Iceberg с типами Trino#

Тип Iceberg

Тип Trino

BOOLEAN

BOOLEAN

INT

INTEGER

LONG

BIGINT

FLOAT

REAL

DOUBLE

DOUBLE

DECIMAL(p,s)

DECIMAL(p,s)

DATE

DATE

TIME

TIME(6)

TIMESTAMP

TIMESTAMP(6)

TIMESTAMPTZ

TIMESTAMP(6) WITH TIME ZONE

STRING

VARCHAR

UUID

UUID

BINARY

VARBINARY

FIXED (L)

VARBINARY

STRUCT(...)

ROW(...)

LIST(e)

ARRAY(e)

MAP(k,v)

MAP(k,v)

Другие типы не поддерживаются.

Сопоставление типов Trino с типами Iceberg#

Коннектор сопоставляет типы Trino с соответствующими типами Iceberg по следующей таблице:

Сопоставление типов Trino с типами Iceberg#

Тип Trino

Тип Iceberg

BOOLEAN

BOOLEAN

INTEGER

INT

BIGINT

LONG

REAL

FLOAT

DOUBLE

DOUBLE

DECIMAL(p,s)

DECIMAL(p,s)

DATE

DATE

TIME(6)

TIME

TIMESTAMP(6)

TIMESTAMP

TIMESTAMP(6) WITH TIME ZONE

TIMESTAMPTZ

VARCHAR

STRING

UUID

UUID

VARBINARY

BINARY

ROW(...)

STRUCT(...)

ARRAY(e)

LIST(e)

MAP(k,v)

MAP(k,v)

Другие типы не поддерживаются.

Безопасность#

Аутентификация Kerberos#

Коннектор Iceberg поддерживает аутентификацию Kerberos для метахранилища Hive и HDFS и настраивается теми же параметрами, что и коннектор Hive. Дополнительные сведения см. в разделе HDFS file system support.

Авторизация#

Коннектор Iceberg позволяет выбрать один из нескольких способов авторизации на уровне каталога.

Проверки авторизации для коннектора можно включить, задав свойство iceberg.security в файле свойств каталога. Это свойство должно иметь одно из следующих значений:

Значения безопасности Iceberg#

Значение свойства

Описание

ALLOW_ALL

Проверки авторизации не выполняются.

SYSTEM

Коннектор использует контроль доступа системного уровня.

READ_ONLY

Разрешены операции чтения данных или метаданных, например SELECT. Операции, записывающие данные или метаданные, например CREATE TABLE, INSERT или DELETE, не разрешены.

FILE

Проверки авторизации выполняются с помощью файла конфигурации контроля доступа уровня каталога, путь к которому задается свойством конфигурации каталога security.config-file. Сведения о файле конфигурации авторизации см. в Файлы контроля доступа на уровне каталога.

Поддержка SQL#

Этот коннектор предоставляет доступ на чтение и запись к данным и метаданным в Iceberg. Помимо глобально доступных операторов и операторов операций чтения, коннектор поддерживает следующие возможности:

Базовые примеры использования#

Коннектор поддерживает создание схем. Схему можно создать с указанным расположением или без него.

Схему можно создать оператором CREATE SCHEMA и свойством схемы location. Таблицы в этой схеме, для которых в операторе CREATE TABLE не задано явное свойство location, располагаются в подкаталоге внутри каталога, соответствующего расположению схемы.

Создание схемы в S3:

CREATE SCHEMA example.example_s3_schema
WITH (location = 's3://my-bucket/a/path/');

Создание схемы в S3-совместимом объектном хранилище, например MinIO:

CREATE SCHEMA example.example_s3a_schema
WITH (location = 's3a://my-bucket/a/path/');

Создание схемы в HDFS:

CREATE SCHEMA example.example_hdfs_schema
WITH (location='hdfs://hadoop-master:9000/user/hive/warehouse/a/path/');

В HDFS расположение при необходимости можно опустить:

CREATE SCHEMA example.example_hdfs_schema;

Коннектор Iceberg поддерживает создание таблиц с помощью синтаксиса CREATE TABLE. При необходимости укажите свойства таблицы, поддерживаемые этим коннектором:

CREATE TABLE example_table (
    c1 INTEGER,
    c2 DATE,
    c3 DOUBLE
)
WITH (
    format = 'PARQUET',
    partitioning = ARRAY['c1', 'c2'],
    sorted_by = ARRAY['c3'],
    location = 's3://my-bucket/a/path/'
);

Если свойство таблицы location не указано, содержимое таблицы хранится в подкаталоге внутри каталога, соответствующего расположению схемы.

Коннектор Iceberg поддерживает создание таблиц с помощью CREATE TABLE AS с синтаксисом SELECT:

CREATE TABLE tiny_nation
WITH (
    format = 'PARQUET'
)
AS
    SELECT *
    FROM nation
    WHERE nationkey < 10;

Другой вариант создания таблиц с помощью CREATE TABLE AS использует синтаксис VALUES:

CREATE TABLE yearly_clicks (
    year,
    clicks
)
WITH (
    partitioning = ARRAY['year']
)
AS VALUES
    (2021, 10000),
    (2022, 20000);

Процедуры#

Используйте оператор CALL для выполнения операций с данными или административных задач. Процедуры доступны в системной схеме каждого каталога. В следующем фрагменте показано, как вызвать example_procedure в каталоге examplecatalog:

CALL examplecatalog.system.example_procedure();

Регистрация таблицы#

Коннектор может регистрировать существующие таблицы Iceberg в метахранилище, если для каталога свойство iceberg.register-table-procedure.enabled установлено в true.

Процедура system.register_table позволяет вызывающей стороне зарегистрировать существующую таблицу Iceberg в метахранилище, используя ее существующие метаданные и файлы данных:

CALL example.system.register_table(
  schema_name => 'testdb', 
  table_name => 'customer_orders', 
  table_location => 'hdfs://hadoop-master:9000/user/hive/warehouse/customer_orders-581fad8517934af6be1857a903559d44');

Кроме того, можно указать имя файла, чтобы зарегистрировать таблицу с конкретными метаданными. Это можно использовать для регистрации таблицы с определенным состоянием либо когда коннектор не может автоматически определить, какую версию метаданных использовать:

CALL example.system.register_table(
  schema_name => 'testdb', 
  table_name => 'customer_orders', 
  table_location => 'hdfs://hadoop-master:9000/user/hive/warehouse/customer_orders-581fad8517934af6be1857a903559d44', 
  metadata_file_name => '00003-409702ba-4735-4645-8f14-09537cc0b2c8.metadata.json');

Чтобы предотвратить доступ неавторизованных пользователей к данным, эта процедура по умолчанию отключена. Она включается только если iceberg.register-table-procedure.enabled установлено в true.

Отмена регистрации таблицы#

Коннектор может удалять существующие таблицы Iceberg из метахранилища. После отмены регистрации таблицу больше нельзя запрашивать из Trino.

Процедура system.unregister_table позволяет вызывающей стороне отменить регистрацию существующей таблицы Iceberg в метахранилище без удаления данных:

CALL example.system.unregister_table(
  schema_name => 'testdb', 
  table_name => 'customer_orders');

Миграция таблицы#

Коннектор может читать из таблиц Hive, перенесенных в Iceberg, и записывать в них.

Используйте процедуру system.migrate, чтобы перенести таблицу из формата Hive в формат Iceberg с загрузкой файлов данных исходной таблицы. Схема таблицы, секционирование, свойства и расположение копируются из исходной таблицы. Бакетированная таблица Hive переносится как небакетированная таблица Iceberg. Файлы данных в таблице Hive должны использовать формат Parquet, ORC или Avro.

Процедуру нужно вызывать для конкретного каталога example, передавая соответствующие имена схемы и таблицы через обязательные параметры schema_name и table_name:

CALL example.system.migrate(
    schema_name => 'testdb',
    table_name => 'customer_orders');

Миграция завершается ошибкой, если какая-либо секция таблицы использует неподдерживаемый файловый формат.

Кроме того, можно передать аргумент recursive_directory, чтобы перенести таблицу Hive, содержащую подкаталоги:

CALL example.system.migrate(
    schema_name => 'testdb',
    table_name => 'customer_orders',
    recursive_directory => 'true');

Значение по умолчанию — fail; при нем процедура миграции выбрасывает исключение, если обнаружены подкаталоги. Установите значение true, чтобы переносить вложенные каталоги, или false, чтобы игнорировать их.

Добавление файлов#

Коннектор может добавлять файлы из таблиц или расположений в существующую таблицу Iceberg, если для каталога свойство iceberg.add-files-procedure.enabled установлено в true.

Используйте процедуру add_files_from_table, чтобы добавить в существующую таблицу Iceberg файлы из таблицы Hive в текущем каталоге, или add_files, чтобы добавить существующие файлы из указанного расположения.

Файлы данных должны быть в формате Parquet, ORC или Avro.

Процедура добавляет файлы в целевую таблицу, указанную после ALTER TABLE, и загружает их из исходной таблицы, заданной обязательными параметрами schema_name и table_name. Исходная таблица должна быть доступна в том же каталоге, что и целевая, и использовать формат Hive. Целевая таблица должна использовать формат Iceberg. Каталог должен использовать коннектор Iceberg.

Следующие примеры копируют данные из таблицы Hive hive_customer_orders в схеме legacy каталога example в таблицу Iceberg iceberg_customer_orders в схеме lakehouse каталога example:

ALTER TABLE example.lakehouse.iceberg_customer_orders 
EXECUTE add_files_from_table(
    schema_name => 'legacy',
    table_name => 'customer_orders');

Либо можно задать текущий каталог и схему оператором USE и опустить сведения о каталоге и схеме:

USE example.lakehouse;
ALTER TABLE iceberg_customer_orders 
EXECUTE add_files_from_table(
    schema_name => 'legacy',
    table_name => 'customer_orders');

Используйте аргумент partition_filter, чтобы добавить файлы из указанных секций. Следующий пример добавляет файлы из секции, где region равен ASIA, а country равен JAPAN:

ALTER TABLE example.lakehouse.iceberg_customer_orders 
EXECUTE add_files_from_table(
    schema_name => 'legacy',
    table_name => 'customer_orders',
    partition_filter => map(ARRAY['region', 'country'], ARRAY['ASIA', 'JAPAN']));

Кроме того, можно передать аргумент recursive_directory, чтобы перенести таблицу Hive, содержащую подкаталоги:

ALTER TABLE example.lakehouse.iceberg_customer_orders 
EXECUTE add_files_from_table(
    schema_name => 'legacy',
    table_name => 'customer_orders',
    recursive_directory => 'true');

Значение recursive_directory по умолчанию — fail; при нем процедура выбрасывает исключение, если обнаружены подкаталоги. Установите значение true, чтобы добавлять файлы из вложенных каталогов, или false, чтобы игнорировать их.

Процедура add_files поддерживает добавление файлов, а значит и содержащихся в них данных, в целевую таблицу, указанную после ALTER TABLE. Она загружает файлы из пути объектного хранилища, заданного обязательным параметром location. Файлы должны использовать указанный format; допустимые значения — ORC и PARQUET. Целевая таблица Iceberg должна использовать тот же формат, что и добавляемые файлы. Процедура не проверяет совместимость схем файлов с целевой таблицей Iceberg. Свойство location поддерживается для секционированных таблиц.

Следующие примеры копируют файлы формата ORC из расположения s3://my-bucket/a/path в таблицу Iceberg iceberg_customer_orders в схеме lakehouse каталога example:

ALTER TABLE example.lakehouse.iceberg_customer_orders 
EXECUTE add_files(
    location => 's3://my-bucket/a/path',
    format => 'ORC');

Функции#

Функции доступны в схеме system каждого каталога. Функции можно вызывать в SQL-операторе. Например, следующий фрагмент показывает, как выполнить функцию system.bucket в каталоге Iceberg:

SELECT system.bucket('trino', 16);

bucket#

Эта функция предоставляет бакетное преобразование Iceberg, чтобы пользователи могли определить, в какой бакет попадает конкретное значение. Функция принимает два аргумента: значение секции и число бакетов.

Поддерживаемые типы для первого аргумента этой функции:

  • TINYINT

  • SMALLINT

  • INTEGER

  • BIGINT

  • VARCHAR

  • VARBINARY

  • DATE

  • TIMESTAMP

  • TIMESTAMP WITH TIME ZONE

Например, чтобы узнать, какой номер бакета будет назначен конкретной строке, можно выполнить:

SELECT system.bucket('trino', 16);

Эту функцию можно использовать в предложении WHERE, чтобы работать только с определенным бакетом:

SELECT count(*)
FROM customer
WHERE system.bucket(custkey, 16) = 2;

Управление данными#

Функциональность Управление данными включает поддержку операторов INSERT, UPDATE, DELETE, TRUNCATE и MERGE.

Удаление по секции#

Для секционированных таблиц коннектор Iceberg поддерживает удаление целых секций, если предложение WHERE задает фильтры только по столбцам секционирования с identity-преобразованием, которые могут соответствовать целым секциям. Для определения таблицы из раздела Секционированные таблицы следующий SQL-оператор удаляет все секции, где country равен US:

DELETE FROM example.testdb.customer_orders
WHERE country = 'US';

Удаление секции выполняется, если предложение WHERE соответствует этим условиям.

Удаление на уровне строк#

Таблицы, использующие версию 2 спецификации Iceberg, поддерживают удаление отдельных строк путем записи файлов позиционного удаления.

Управление схемами и таблицами#

Функциональность Управление схемами и таблицами включает поддержку:

Эволюция схемы#

Iceberg поддерживает эволюцию схемы с безопасными операциями добавления, удаления и переименования столбцов, включая вложенные структуры.

Iceberg поддерживает изменение типов столбцов только для расширяющих операций:

  • INTEGER to BIGINT

  • REAL to DOUBLE

  • DECIMAL(p,s) to DECIMAL(p2,s) when p2 > p (scale cannot change)

Секционирование также можно изменить, и коннектор по-прежнему сможет запрашивать данные, созданные до изменения секционирования.

ALTER TABLE EXECUTE#

Коннектор поддерживает следующие команды для использования с ALTER TABLE EXECUTE.

optimize#

Команда optimize используется для перезаписи содержимого указанной таблицы, чтобы объединить его в меньшее число более крупных файлов. Если таблица секционирована, уплотнение данных выполняется отдельно для каждой секции, выбранной для оптимизации. Эта операция повышает производительность чтения.

Все файлы размером меньше необязательного параметра file_size_threshold (значение порога по умолчанию — 100MB) объединяются, если для секции выполнено любое из следующих условий:

  • присутствует больше одного файла данных для объединения

  • присутствует хотя бы один файл данных с присоединенными файлами удаления

ALTER TABLE test_table EXECUTE optimize

Следующий оператор объединяет файлы таблицы размером меньше 128 мегабайт:

ALTER TABLE test_table EXECUTE optimize(file_size_threshold => '128MB')

Можно использовать предложение WHERE со столбцами, по которым таблица секционирована, чтобы отфильтровать оптимизируемые секции:

ALTER TABLE test_partitioned_table EXECUTE optimize
WHERE partition_key = 1

Можно использовать более сложное предложение WHERE, чтобы сузить область действия процедуры optimize. В следующем примере значения временной метки приводятся к датам, а сравнение используется для оптимизации только секций с данными за 2022 год или позднее:

ALTER TABLE test_table EXECUTE optimize
WHERE CAST(timestamp_tz AS DATE) > DATE '2021-12-31'

Используйте предложение WHERE со столбцами метаданных, чтобы отфильтровать оптимизируемые файлы.

ALTER TABLE test_table EXECUTE optimize
WHERE "$file_modified_time" > date_trunc('day', CURRENT_TIMESTAMP);
optimize_manifests#

Перезаписывает файлы манифестов, группируя их по столбцам секционирования. Это можно использовать для оптимизации планирования сканирования, когда есть много небольших файлов манифестов или когда в запросах чтения есть фильтры по секциям, но файлы манифестов не сгруппированы по секциям. Свойство таблицы Iceberg commit.manifest.target-size-bytes управляет максимальным размером файлов манифестов, создаваемых этой процедурой.

optimize_manifests можно запустить следующим образом:

ALTER TABLE test_table EXECUTE optimize_manifests;

metric_name                      | metric_value
---------------------------------+--------------
rewritten_manifests_count        |            2
added_manifests_count            |            1
kept_manifests_count             |            1
processed_manifest_entries_count |            2
expire_snapshots#

Команда expire_snapshots удаляет все снимки и все связанные с ними метаданные и файлы данных. Рекомендуется регулярно завершать срок действия снимков, чтобы удалять файлы данных, которые больше не нужны, и сохранять небольшой размер метаданных таблицы. Процедура затрагивает все снимки старше периода времени, настроенного параметром retention_threshold.

expire_snapshots можно запустить следующим образом:

ALTER TABLE test_table EXECUTE expire_snapshots(retention_threshold => '7d');

Значение retention_threshold должно быть больше или равно iceberg.expire-snapshots.min-retention в каталоге, иначе процедура завершится ошибкой с похожим сообщением: Retention specified (1.00d) is shorter than the minimum retention configured in the system (7.00d). Значение этого свойства по умолчанию — 7d.

Команда принимает необязательный параметр retain_last, задающий минимальное число снимков-предков, которые нужно сохранить (по умолчанию 1), независимо от значения retention_threshold.

Команда принимает необязательный параметр clean_expired_metadata (по умолчанию false). Если он равен true, очищаются метаданные, такие как спецификации секций и схемы, на которые больше не ссылаются снимки.

remove_orphan_files#

Команда remove_orphan_files удаляет из каталога данных таблицы все файлы, которые не связаны из файлов метаданных и старше значения параметра retention_threshold. Рекомендуется время от времени удалять потерянные файлы, чтобы контролировать размер каталога данных таблицы.

remove_orphan_files можно запустить следующим образом:

ALTER TABLE test_table EXECUTE remove_orphan_files(retention_threshold => '7d');

        metric_name         | metric_value
----------------------------+--------------
 processed_manifests_count  |            2
 active_files_count         |           98
 scanned_files_count        |           97
 deleted_files_count        |            0

Значение retention_threshold должно быть больше или равно iceberg.remove-orphan-files.min-retention в каталоге, иначе процедура завершится ошибкой с похожим сообщением: Retention specified (1.00d) is shorter than the minimum retention configured in the system (7.00d). Значение этого свойства по умолчанию — 7d.

Вывод запроса содержит следующие метрики:

Вывод#

Имя свойства

Описание

processed_manifests_count

Количество файлов манифестов, прочитанных remove_orphan_files.

active_files_count

Количество файлов, относящихся к снимкам, срок действия которых не истек.

scanned_files_count

Количество файлов, просканированных из файловой системы.

deleted_files_count

Количество файлов, удаленных remove_orphan_files.
drop_extended_stats#

Команда drop_extended_stats удаляет из таблицы все сведения расширенной статистики.

drop_extended_stats можно запустить следующим образом:

ALTER TABLE test_table EXECUTE drop_extended_stats;

ALTER TABLE SET PROPERTIES#

Коннектор поддерживает изменение свойств существующих таблиц с помощью ALTER TABLE SET PROPERTIES.

После создания таблицы можно изменить следующие свойства таблицы:

  • format

  • format_version

  • partitioning

  • sorted_by

  • max_commit_retry

  • delete_after_commit_enabled

  • max_previous_versions

  • object_store_layout_enabled

  • data_location

Например, чтобы обновить таблицу с версии 1 спецификации Iceberg до версии 2:

ALTER TABLE table_name SET PROPERTIES format_version = 2;

Или чтобы задать столбец my_new_partition_column как столбец секционирования таблицы:

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

Текущие значения свойств таблицы можно вывести с помощью SHOW CREATE TABLE.

Свойства таблицы#

Свойства таблицы предоставляют или задают метаданные для базовых таблиц. Это важно для операторов CREATE TABLE AS. Свойства таблицы передаются коннектору с помощью предложения WITH.

Свойства таблицы Iceberg#

Имя свойства

Описание

format

Необязательно задает формат файлов данных таблицы: PARQUET, ORC или AVRO. По умолчанию используется значение свойства конфигурации каталога iceberg.file-format, у которого значение по умолчанию — PARQUET.

compression_codec

Необязательно задает кодек сжатия, используемый для записи таблицы: NONE, ZSTD, SNAPPY, LZ4 или GZIP. По умолчанию используется значение свойства конфигурации каталога iceberg.compression-codec, у которого значение по умолчанию — ZSTD.

partitioning

Необязательно задает секционирование таблицы. Если таблица секционирована по столбцам c1 и c2, свойство секционирования выглядит как partitioning = ARRAY['c1', 'c2'].

sorted_by

Порядок сортировки, применяемый при записи содержимого каждого файла, записываемого в таблицу. Если файлы таблицы сортируются по столбцам c1 и c2, свойство порядка сортировки выглядит как sorted_by = ARRAY['c1', 'c2']. Порядок сортировки применяется независимо к содержимому каждого выходного файла, а не ко всему набору данных.

location

Необязательно задает URI расположения таблицы в файловой системе.

format_version

Необязательно задает версию формата спецификации Iceberg для новых таблиц: 1, 2 или 3. По умолчанию — 2. Версия 2 требуется для удалений на уровне строк. Поддержка версии 3 экспериментальная; обновления и удаления на уровне строк, а также OPTIMIZE, не поддерживаются. Таблицы с возможностями v3, такими как значения столбцов по умолчанию и шифрование, не поддерживаются.

max_commit_retry

Число повторных попыток фиксации перед ошибкой. По умолчанию используется значение свойства конфигурации каталога iceberg.max-commit-retry, у которого значение по умолчанию — 4.

delete_after_commit_enabled

Удалять ли файлы метаданных самой старой отслеживаемой версии после каждой фиксации таблицы. По умолчанию используется значение свойства конфигурации каталога iceberg.delete-after-commit-enabled, у которого значение по умолчанию — false.

max_previous_versions

Максимальное число файлов метаданных предыдущих версий, которые нужно отслеживать. По умолчанию используется значение свойства конфигурации каталога iceberg.max-previous-versions, у которого значение по умолчанию — 100.

orc_bloom_filter_columns

Список столбцов через запятую для фильтра Блума ORC. Он повышает производительность запросов с предикатами равенства и IN при чтении файлов ORC. Требуется формат ORC. По умолчанию — [].

orc_bloom_filter_fpp

Вероятность ложноположительного срабатывания фильтров Блума ORC. Требуется формат ORC. По умолчанию — 0.05.

parquet_bloom_filter_columns

Список столбцов через запятую для фильтра Блума Parquet. Он повышает производительность запросов с предикатами равенства и IN при чтении файлов Parquet. Требуется формат Parquet. По умолчанию — [].

object_store_layout_enabled

Включена ли схема размещения файлов объектного хранилища Iceberg. По умолчанию — false.

data_location

Необязательно задает URI расположения файлов данных таблицы в файловой системе.

extra_properties

Дополнительные свойства, добавляемые к таблице Iceberg. Trino не использует эти свойства; они доступны в таблице метаданных $properties. Свойства не включаются в вывод операторов SHOW CREATE TABLE.

Определение таблицы ниже задает использование файлов Parquet, секционирование по столбцам c1 и c2 и расположение в файловой системе /var/example_tables/test_table:

CREATE TABLE test_table (
    c1 INTEGER,
    c2 DATE,
    c3 DOUBLE)
WITH (
    format = 'PARQUET',
    partitioning = ARRAY['c1', 'c2'],
    location = '/var/example_tables/test_table');

Определение таблицы ниже задает использование файлов ORC с compression_codec SNAPPY, индексом фильтра Блума по столбцам c1 и c2, вероятностью ложноположительного срабатывания 0.05 и расположением в файловой системе /var/example_tables/test_table:

CREATE TABLE test_table (
    c1 INTEGER,
    c2 DATE,
    c3 DOUBLE)
WITH (
    format = 'ORC',
    compression_codec = 'SNAPPY',
    location = '/var/example_tables/test_table',
    orc_bloom_filter_columns = ARRAY['c1', 'c2'],
    orc_bloom_filter_fpp = 0.05);

Определение таблицы ниже задает использование файлов Avro с секционированием по полю child1 в столбце parent:

CREATE TABLE test_table (
    data INTEGER,
    parent ROW(child1 DOUBLE, child2 INTEGER))
WITH (
    format = 'AVRO',
    partitioning = ARRAY['"parent.child1"']);

Metadata-таблицы#

Коннектор предоставляет несколько metadata-таблиц для каждой таблицы Iceberg. Эти metadata-таблицы содержат сведения о внутренней структуре таблицы Iceberg. К каждой metadata-таблице можно обратиться, добавив имя metadata-таблицы к имени таблицы:

SELECT * FROM "test_table$properties";
$properties table#

Таблица $properties предоставляет доступ к общим сведениям о конфигурации таблицы Iceberg и любым дополнительным парам ключ-значение метаданных, которыми помечена таблица.

Свойства текущего снимка таблицы Iceberg test_table можно получить с помощью следующего запроса:

SELECT * FROM "test_table$properties";

 key                   | value    |
-----------------------+----------+
write.format.default   | PARQUET  |
$history table#

Таблица $history предоставляет журнал изменений метаданных, выполненных для таблицы Iceberg.

Журнал изменений таблицы Iceberg test_table можно получить с помощью следующего запроса:

SELECT * FROM "test_table$history";

 made_current_at                  | snapshot_id          | parent_id            | is_current_ancestor
----------------------------------+----------------------+----------------------+--------------------
2022-01-10 08:11:20 Europe/Vienna | 8667764846443717831  |  <null>              |  true
2022-01-10 08:11:34 Europe/Vienna | 7860805980949777961  | 8667764846443717831  |  true

Вывод запроса содержит следующие столбцы:

Столбцы истории#

Имя

Тип

Описание

made_current_at

TIMESTAMP(3) WITH TIME ZONE

Время, когда снимок стал активным.

snapshot_id

BIGINT

Идентификатор снимка.

parent_id

BIGINT

Идентификатор родительского снимка.

is_current_ancestor

BOOLEAN

Является ли этот снимок предком текущего снимка.
$metadata_log_entries table#

Таблица $metadata_log_entries предоставляет представление записей журнала метаданных таблицы Iceberg.

Сведения о записях журнала метаданных таблицы Iceberg test_table можно получить с помощью следующего запроса:

SELECT * FROM "test_table$metadata_log_entries";

             timestamp                 |                                                              file                                                          | latest_snapshot_id  | latest_schema_id | latest_sequence_number
---------------------------------------+----------------------------------------------------------------------------------------------------------------------------+---------------------+------------------+------------------------
 2024-01-16 15:55:31.172 Europe/Vienna | hdfs://hadoop-master:9000/user/hive/warehouse/test_table/metadata/00000-39174715-be2a-48fa-9949-35413b8b736e.metadata.json | 1221802298419195590 |                0 |                      1
 2024-01-16 17:19:56.118 Europe/Vienna | hdfs://hadoop-master:9000/user/hive/warehouse/test_table/metadata/00001-e40178c9-271f-4a96-ad29-eed5e7aef9b0.metadata.json | 7124386610209126943 |                0 |                      2

Вывод запроса содержит следующие столбцы:

Столбцы записей журнала метаданных#

Имя

Тип

Описание

timestamp

TIMESTAMP(3) WITH TIME ZONE

Время создания метаданных.

file

VARCHAR

Расположение файла метаданных.

latest_snapshot_id

BIGINT

Идентификатор последнего снимка на момент обновления метаданных.

latest_schema_id

INTEGER

Идентификатор последней схемы на момент обновления метаданных.

latest_sequence_number

BIGINT

Порядковый номер данных файла метаданных.
$snapshots table#

Таблица $snapshots предоставляет подробное представление снимков таблицы Iceberg. Снимок состоит из одного или нескольких файлов манифестов, а полное содержимое таблицы представлено объединением всех файлов данных в этих манифестах.

Сведения о снимках таблицы Iceberg test_table можно получить с помощью следующего запроса:

SELECT * FROM "test_table$snapshots";

 committed_at                      | snapshot_id          | parent_id            | operation          |  manifest_list                                                                                                                           |   summary
----------------------------------+----------------------+----------------------+--------------------+------------------------------------------------------------------------------------------------------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
2022-01-10 08:11:20 Europe/Vienna | 8667764846443717831  |  <null>              |  append            |   hdfs://hadoop-master:9000/user/hive/warehouse/test_table/metadata/snap-8667764846443717831-1-100cf97e-6d56-446e-8961-afdaded63bc4.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-01-10 08:11:34 Europe/Vienna | 7860805980949777961  | 8667764846443717831  |  append            |   hdfs://hadoop-master:9000/user/hive/warehouse/test_table/metadata/snap-7860805980949777961-1-faa19903-1455-4bb8-855a-61a1bbafbaa7.avro | {changed-partition-count=1, added-data-files=1, total-equality-deletes=0, added-records=1, total-position-deletes=0, added-files-size=442, total-delete-files=0, total-files-size=442, total-records=1, total-data-files=1}

Вывод запроса содержит следующие столбцы:

Столбцы снимков#

Имя

Тип

Описание

committed_at

TIMESTAMP(3) WITH TIME ZONE

Время, когда снимок стал активным.

snapshot_id

BIGINT

Идентификатор снимка.

parent_id

BIGINT

Идентификатор родительского снимка.

operation

VARCHAR

Тип операции, выполненной для таблицы Iceberg. Поддерживаемые типы операций в Iceberg:

  • append, когда добавляются новые данные.

  • replace, когда файлы удаляются и заменяются без изменения данных в таблице.

  • overwrite, когда новые данные добавляются с перезаписью существующих.

  • delete, когда данные удаляются из таблицы и новые данные не добавляются.

manifest_list

VARCHAR

Список файлов манифестов Avro, содержащих подробные сведения об изменениях снимка.

summary

map(VARCHAR, VARCHAR)

Сводка изменений, внесенных между предыдущим и текущим снимками.
$manifests and $all_manifests tables#

Таблицы $manifests и $all_manifests предоставляют подробный обзор манифестов, соответствующих снимкам, выполненным в журнале таблицы Iceberg. Таблица $manifests содержит данные текущего снимка. Таблица $all_manifests содержит данные всех снимков.

Сведения о манифестах таблицы Iceberg test_table можно получить с помощью следующего запроса:

SELECT * FROM "test_table$manifests";

 content | path                                                                                                           | length          | partition_spec_id    | added_snapshot_id     | added_data_files_count  | added_rows_count | existing_data_files_count   | existing_rows_count | deleted_data_files_count    | deleted_rows_count | partition_summaries
---------+----------------------------------------------------------------------------------------------------------------+-----------------+----------------------+-----------------------+-------------------------+------------------+-----------------------------+---------------------+-----------------------------+--------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------
 0       | hdfs://hadoop-master:9000/user/hive/warehouse/test_table/metadata/faa19903-1455-4bb8-855a-61a1bbafbaa7-m0.avro |  6277           |   0                  | 7860805980949777961   | 1                       | 100              | 0                           | 0                   | 0                           | 0                  | {{contains_null=false, contains_nan= false, lower_bound=1, upper_bound=1},{contains_null=false, contains_nan= false, lower_bound=2021-01-12, upper_bound=2021-01-12}}

Вывод запроса содержит следующие столбцы:

Столбцы манифестов#

Имя

Тип

Описание

content

INTEGER

Тип содержимого, хранящегося в манифесте. Поддерживаемые типы содержимого в Iceberg:

  • DATA(0) — манифесты, отслеживающие файлы данных.

  • DELETES(1) — манифесты, отслеживающие файлы удаления.

path

VARCHAR

Расположение файла манифеста.

length

BIGINT

Длина файла манифеста.

partition_spec_id

INTEGER

Идентификатор спецификации секционирования, использованной для записи файла манифеста.

added_snapshot_id

BIGINT

Идентификатор снимка, во время которого была добавлена эта запись манифеста.

added_data_files_count

INTEGER

Число файлов данных со статусом ADDED в файле манифеста.

added_rows_count

BIGINT

Общее число строк во всех файлах данных со статусом ADDED в файле манифеста.

existing_data_files_count

INTEGER

Число файлов данных со статусом EXISTING в файле манифеста.

existing_rows_count

BIGINT

Общее число строк во всех файлах данных со статусом EXISTING в файле манифеста.

deleted_data_files_count

INTEGER

Число файлов данных со статусом DELETED в файле манифеста.

deleted_rows_count

BIGINT

Общее число строк во всех файлах данных со статусом DELETED в файле манифеста.

partition_summaries

ARRAY(row(contains_null BOOLEAN, contains_nan BOOLEAN, lower_bound VARCHAR, upper_bound VARCHAR))

Метаданные диапазона секции.
Таблица $partitions#

Таблица $partitions предоставляет подробный обзор секций таблицы Iceberg.

Сведения о секциях таблицы Iceberg test_table можно получить с помощью следующего запроса:

SELECT * FROM "test_table$partitions";

 partition             | record_count  | file_count    | total_size    |  data
-----------------------+---------------+---------------+---------------+------------------------------------------------------
{c1=1, c2=2021-01-12}  |  2            | 2             |  884          | {c3={min=1.0, max=2.0, null_count=0, nan_count=NULL}}
{c1=1, c2=2021-01-13}  |  1            | 1             |  442          | {c3={min=1.0, max=1.0, null_count=0, nan_count=NULL}}

Вывод запроса содержит следующие столбцы:

Столбцы секций#

Имя

Тип

Описание

partition

ROW(...)

Строка, содержащая сопоставление имен столбцов секции со значениями столбцов секции.

record_count

BIGINT

Число записей в секции.

file_count

BIGINT

Число файлов, сопоставленных с секцией.

total_size

BIGINT

Размер всех файлов в секции.

data

ROW(... ROW (min ..., max ... , null_count BIGINT, nan_count BIGINT))

Метаданные диапазона секции.
Таблица $files#

Таблица $files предоставляет подробный обзор файлов данных в текущем снимке таблицы Iceberg.

Чтобы получить сведения о файлах данных таблицы Iceberg test_table, используйте следующий запрос:

SELECT * FROM "test_table$files";

 content  | file_path                                                                                                                     | record_count    | file_format   | 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://hadoop-master:9000/user/hive/warehouse/test_table/data/c1=3/c2=2021-01-14/af9872b2-40f3-428f-9c87-186d2750d84e.parquet |  1              |  PARQUET      |  442                 | {1=40, 2=40, 3=44}   |  {1=1, 2=1, 3=1}  |  {1=0, 2=0, 3=0}   | <null>            |  {1=3, 2=2021-01-14, 3=1.3} |  {1=3, 2=2021-01-14, 3=1.3} |  <null>        | <null>         |   <null>

Вывод запроса содержит следующие столбцы:

Столбцы файлов#

Имя

Тип

Описание

content

INTEGER

Тип содержимого, хранящегося в файле. Поддерживаемые типы содержимого в Iceberg:

  • DATA(0)

  • POSITION_DELETES(1)

  • EQUALITY_DELETES(2)

file_path

VARCHAR

Расположение файла данных.

file_format

VARCHAR

Формат файла данных.

spec_id

INTEGER

Идентификатор спецификации, используемый для отслеживания файла, содержащего строку.

partition

ROW(...)

Строка, содержащая сопоставление имен столбцов секции со значениями столбцов секции.

record_count

BIGINT

Число записей, содержащихся в файле данных.

file_size_in_bytes

BIGINT

Размер файла данных.

column_sizes

map(INTEGER, BIGINT)

Сопоставление идентификатора столбца Iceberg с соответствующим размером в файле.

value_counts

map(INTEGER, BIGINT)

Сопоставление идентификатора столбца Iceberg с соответствующим числом записей в файле.

null_value_counts

map(INTEGER, BIGINT)

Сопоставление идентификатора столбца Iceberg с соответствующим числом значений NULL в файле.

nan_value_counts

map(INTEGER, BIGINT)

Сопоставление идентификатора столбца Iceberg с соответствующим числом нечисловых значений в файле.

lower_bounds

map(INTEGER, BIGINT)

Сопоставление идентификатора столбца Iceberg с соответствующей нижней границей в файле.

upper_bounds

map(INTEGER, BIGINT)

Сопоставление идентификатора столбца Iceberg с соответствующей верхней границей в файле.

key_metadata

VARBINARY

Метаданные о ключе шифрования, использованном для шифрования этого файла, если применимо.

split_offsets

array(BIGINT)

Список рекомендуемых расположений фрагментов.

equality_ids

array(INTEGER)

Набор идентификаторов полей, используемых для сравнения на равенство в файлах удаления по равенству.

sort_order_id

INTEGER

Идентификатор, представляющий порядок сортировки для этого файла.

readable_metrics

JSON

Метрики файла в удобочитаемой форме.
Таблицы $entries и $all_entries#

Таблицы $entries и $all_entries предоставляют записи манифестов таблицы как для файлов данных, так и для файлов удаления. Таблица $entries содержит данные текущего снимка. Таблица $all_entries содержит данные всех снимков.

Чтобы получить сведения о записях таблицы Iceberg test_table, используйте следующий запрос:

SELECT * FROM "test_table$entries";

Сокращенный пример вывода:

 status |   snapshot_id  | sequence_number | file_sequence_number |              data_file              |                readable_metrics                |
--------+----------------+-----------------+----------------------+-------------------------------------+------------------------------------------------+
      2 | 57897183625154 |              0  |                   0  | {"content":0,...,"sort_order_id":0} | {"c1":{"column_size":103,...,"upper_bound":3}} |

Metadata-таблицы содержат следующие столбцы:

Столбцы файлов#

Имя

Тип

Описание

status

INTEGER

Числовой признак статуса для отслеживания добавлений и удалений. Удаления носят только информационный характер и не используются при сканировании:

  • EXISTING(0)

  • ADDED(1)

  • DELETED(2)

snapshot_id

BIGINT

Идентификатор снимка ссылки.

sequence_number

BIGINT

Порядковый номер данных файла. Наследуется, когда значение равно null, а статус равен 1.

file_sequence_number

BIGINT

Порядковый номер файла, указывающий, когда файл был добавлен. Наследуется, когда значение равно null, а статус равен 1.

data_file

ROW

Метаданные, включая путь к файлу, формат файла, размер файла и другие сведения.

readable_metrics

JSON

Метрики файла в формате JSON, такие как размер столбца, количество значений и другие.
Таблица $refs#

Таблица $refs предоставляет сведения о ссылках Iceberg, включая ветви и теги.

Ссылки таблицы Iceberg test_table можно получить с помощью следующего запроса:

SELECT * FROM "test_table$refs";

name            | type   | snapshot_id | max_reference_age_in_ms | min_snapshots_to_keep | max_snapshot_age_in_ms |
----------------+--------+-------------+-------------------------+-----------------------+------------------------+
example_tag     | TAG    | 10000000000 | 10000                   | null                  | null                   |
example_branch  | BRANCH | 20000000000 | 20000                   | 2                     | 30000                  |

Вывод запроса содержит следующие столбцы:

Столбцы ссылок#

Имя

Тип

Описание

name

VARCHAR

Имя ссылки.

type

VARCHAR

Тип ссылки: BRANCH или TAG.

snapshot_id

BIGINT

Идентификатор снимка ссылки.

max_reference_age_in_ms

BIGINT

Максимальный возраст ссылки до возможного истечения срока ее действия.

min_snapshots_to_keep

INTEGER

Только для ветви: минимальное число снимков, которые нужно сохранить в ветви.

max_snapshot_age_in_ms

BIGINT

Только для ветви: максимальный допустимый возраст снимка в ветви. Более старые снимки в ветви будут удалены по истечении срока действия.

Metadata-столбцы#

Помимо определенных столбцов, коннектор Iceberg автоматически предоставляет метаданные пути в виде скрытого столбца в каждой таблице:

  • $partition: путь секции для этой строки

  • $path: полный путь в файловой системе к файлу для этой строки

  • $file_modified_time: временная метка последнего изменения файла для этой строки

Эти столбцы можно использовать в SQL-операторах как любые другие столбцы. Их можно выбирать напрямую или использовать в условных выражениях. Например, можно просмотреть путь к файлу для каждой записи:

SELECT *, "$partition", "$path", "$file_modified_time"
FROM example.web.page_views;

Получение всех записей, относящихся к конкретному файлу, с использованием фильтра "$path":

SELECT *
FROM example.web.page_views
WHERE "$path" = '/usr/iceberg/table/web.page_views/data/file_01.parquet'

Получение всех записей, относящихся к конкретному файлу, с использованием фильтра "$file_modified_time":

SELECT *
FROM example.web.page_views
WHERE "$file_modified_time" = CAST('2022-07-01 01:02:03.456 UTC' AS TIMESTAMP WITH TIME ZONE)

Системные таблицы#

Коннектор предоставляет metadata-таблицы в системной схеме.

Таблица iceberg_tables#

Таблица iceberg_tables позволяет перечислять только таблицы Iceberg из заданного каталога. Оператор SHOW TABLES, information_schema.tables и jdbc.tables возвращают все таблицы, существующие в базовом метахранилище, даже если коннектор Iceberg никак не может их обработать. Это может произойти, если другие коннекторы, например Hive или Delta Lake, используют то же метахранилище, каталог и схему для хранения своих таблиц.

Таблица содержит следующие столбцы:

Столбцы iceberg_tables#

Имя

Тип

Описание

table_schema

VARCHAR

Имя схемы, в которой находится таблица.

table_name

VARCHAR

Имя таблицы.

Следующий запрос выводит таблицы Iceberg из всех схем каталога example.

SELECT * FROM example.system.iceberg_tables;

 table_schema | table_name  |
--------------+-------------+
 tpcds        | store_sales |
 tpch         | nation      |
 tpch         | region      |
 tpch         | orders      |

DROP TABLE#

Коннектор Iceberg поддерживает удаление таблицы с помощью синтаксиса DROP TABLE. При успешном выполнении команды удаляются и данные таблицы Iceberg, и сведения о таблице в службе метахранилища. Удаление таблиц, у которых данные или метаданные хранятся в расположении, отличном от соответствующего базового каталога таблицы в объектном хранилище, не поддерживается.

COMMENT#

Коннектор Iceberg поддерживает установку комментариев для следующих объектов:

  • таблицы

  • представления

  • столбцы таблиц

  • столбцы материализованных представлений

Параметр COMMENT поддерживается как для таблицы, так и для столбцов таблицы в операции CREATE TABLE.

Параметр COMMENT поддерживается при добавлении столбцов таблицы через операции ALTER TABLE.

Коннектор поддерживает команду COMMENT для установки комментариев к существующим сущностям.

Секционированные таблицы#

Iceberg поддерживает секционирование путем указания преобразований над столбцами таблицы. Секция создается для каждого уникального значения кортежа, полученного преобразованиями. Identity-преобразование — это просто имя столбца. Другие преобразования:

Преобразования столбцов Iceberg#

Преобразование

Описание

year(ts)

Секция создается для каждого года. Значение секции — целочисленная разница в годах между ts и 1 января 1970 года.

month(ts)

Секция создается для каждого месяца каждого года. Значение секции — целочисленная разница в месяцах между ts и 1 января 1970 года.

day(ts)

Секция создается для каждого дня каждого года. Значение секции — целочисленная разница в днях между ts и 1 января 1970 года.

hour(ts)

Секция создается для каждого часа каждого дня. Значение секции — временная метка, в которой минуты и секунды установлены в ноль.

bucket(x, nbuckets)

Данные хэшируются в указанное число бакетов. Значение секции — целочисленный хэш x со значением от 0 до nbuckets - 1 включительно.

truncate(s, nchars)

Значение секции — первые nchars символов s.

В этом примере таблица секционирована по месяцу order_date, хэшу account_number (с 10 бакетами) и country:

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

Сортированные таблицы#

Коннектор поддерживает сортированные файлы как способ повышения производительности. При записи данные сортируются внутри каждого файла на основе указанного массива из одного или нескольких столбцов.

Сортировка особенно полезна, когда сортируемые столбцы имеют высокую кардинальность и используются как фильтр для выборочного чтения.

Порядок сортировки настраивается свойством таблицы sorted_by. При создании таблицы укажите массив из одного или нескольких столбцов, используемых для сортировки. Следующий пример настраивает столбец order_date таблицы orders в схеме customers каталога example:

CREATE TABLE example.customers.orders (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (sorted_by = ARRAY['order_date']);

Направления сортировки или порядок значений null можно явно настроить следующим образом:

CREATE TABLE example.customers.orders (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (sorted_by = ARRAY['order_date DESC NULLS FIRST', 'order_id ASC NULLS LAST']);

Сортировку можно объединить с секционированием по тому же столбцу. Например:

CREATE TABLE example.customers.orders (
    order_id BIGINT,
    order_date DATE,
    account_number BIGINT,
    customer VARCHAR,
    country VARCHAR)
WITH (
    partitioning = ARRAY['month(order_date)'],
    sorted_by = ARRAY['order_date']
);

Сортированную запись можно отключить, установив свойство сеанса sorted_writing_enabled в false.

Использование снимков#

Iceberg поддерживает модель данных на основе снимков, где снимки таблицы идентифицируются ID снимка.

Коннектор предоставляет системную таблицу со сведениями о снимках для каждой таблицы Iceberg. Снимки идентифицируются ID снимков типа BIGINT. Например, ID снимков таблицы customer_orders можно найти, выполнив следующий запрос:

SELECT snapshot_id
FROM example.testdb."customer_orders$snapshots"
ORDER BY committed_at DESC;

Замена таблиц#

Коннектор поддерживает замену существующей таблицы как атомарную операцию. Атомарная замена таблицы создает новый снимок с новым определением таблицы как часть истории таблицы.

Для замены таблицы используйте CREATE OR REPLACE TABLE или CREATE OR REPLACE TABLE AS.

К более ранним снимкам таблицы можно обращаться через Запросы time travel.

В следующем примере таблица example_table заменяется полностью новым определением и данными из исходной таблицы:

CREATE OR REPLACE TABLE example_table
WITH (sorted_by = ARRAY['a'])
AS SELECT * FROM another_table;
Запросы time travel#

Коннектор позволяет запрашивать исторические данные. Это дает возможность запросить таблицу в том состоянии, в котором она была на момент создания предыдущего снимка, даже если с тех пор данные были изменены или удалены.

Исторические данные таблицы можно получить, указав идентификатор снимка, соответствующий нужной версии таблицы:

SELECT *
FROM example.testdb.customer_orders FOR VERSION AS OF 8954597067493422955;

Другой подход к получению исторических данных — указать момент времени в прошлом, например день или неделю назад. Для предоставления предыдущего состояния таблицы внутренне используется последний снимок таблицы, созданный до указанной в запросе временной метки или в этот момент:

SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF TIMESTAMP '2022-03-23 09:59:29.803 Europe/Vienna';

Коннектор позволяет создать новый снимок через замену таблицы Iceberg.

CREATE OR REPLACE TABLE example.testdb.customer_orders AS
SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF TIMESTAMP '2022-03-23 09:59:29.803 Europe/Vienna';

Чтобы использовать снимок таблицы в запросе, можно указать дату как момент времени в прошлом. Если часовой пояс сеанса — Europe/Vienna, следующие запросы эквивалентны:

SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF DATE '2022-03-23';

SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF TIMESTAMP '2022-03-23 00:00:00';

SELECT *
FROM example.testdb.customer_orders FOR TIMESTAMP AS OF TIMESTAMP '2022-03-23 00:00:00.000 Europe/Vienna';

Iceberg поддерживает именованные ссылки на снимки через ветви и теги. Time travel можно выполнять к ветвям и тегам таблицы.

SELECT *
FROM example.testdb.customer_orders FOR VERSION AS OF 'historical-tag';

SELECT *
FROM example.testdb.customer_orders FOR VERSION AS OF 'test-branch';
Откат к предыдущему снимку#

Используйте metadata-таблицу $snapshots, чтобы определить ID последнего снимка таблицы, как в следующем запросе:

SELECT snapshot_id
FROM example.testdb."customer_orders$snapshots"
ORDER BY committed_at DESC LIMIT 1;

Табличная процедура rollback_to_snapshot позволяет вызывающей стороне откатить состояние таблицы к предыдущему ID снимка:

ALTER TABLE testdb.customer_orders EXECUTE rollback_to_snapshot(8954597067493422955);

Ограничение столбца NOT NULL#

Коннектор Iceberg поддерживает установку ограничений NOT NULL для столбцов таблицы.

Ограничение NOT NULL можно задать для столбцов при создании таблиц с помощью синтаксиса CREATE TABLE:

CREATE TABLE example_table (
    year INTEGER NOT NULL,
    name VARCHAR NOT NULL,
    age INTEGER,
    address VARCHAR
);

При попытке вставить или обновить данные в таблице запрос завершается ошибкой, если он пытается установить значение NULL в столбец с ограничением NOT NULL.

Материализованные представления#

Коннектор Iceberg поддерживает Управление материализованными представлениями. В базовой системе каждое материализованное представление состоит из определения представления и таблицы хранения Iceberg. Имя таблицы хранения сохраняется как свойство материализованного представления. Данные хранятся в этой таблице.

Для управления созданной таблицей хранения, а значит размещением и производительностью, можно использовать Свойства таблицы. Например, следующее предложение с CREATE MATERIALIZED VIEW задает формат ORC для файлов данных и секционирует хранилище по дням с использованием столбца event_date:

WITH ( format = 'ORC', partitioning = ARRAY['event_date'] )

По умолчанию таблица хранения создается в той же схеме, что и определение материализованного представления. Свойство материализованного представления storage_schema можно использовать, чтобы указать схему, в которой создается таблица хранения.

Создание материализованного представления не заполняет его данными автоматически. Чтобы заполнить материализованное представление данными, необходимо выполнить REFRESH MATERIALIZED VIEW.

Данные в материализованном представлении можно обновить командой REFRESH MATERIALIZED VIEW. Эта операция может выполнять инкрементальное или полное обновление в зависимости от сложности определения материализованного представления и истории снимков исходных таблиц. При полном обновлении операция удаляет данные из таблицы хранения и вставляет в существующую таблицу данные, полученные в результате выполнения запроса материализованного представления. При инкрементальном обновлении существующие данные не удаляются из таблицы хранения; из исходных таблиц обрабатываются только дельта-записи, которые при необходимости добавляются в таблицу хранения. В обоих случаях данные заменяются или добавляются атомарно, поэтому пользователи могут продолжать запрашивать материализованное представление во время его обновления. Обновление материализованного представления также сохраняет ID снимков всех таблиц Iceberg, которые входят в запрос материализованного представления, в метаданных материализованного представления. При запросе материализованного представления ID снимков используются для проверки актуальности данных в таблице хранения.

Материализованные представления, запросы которых используют не-Iceberg таблицы, демонстрируют поведение по умолчанию для льготных периодов. Если все таблицы являются таблицами Iceberg, коннектор может определить, что данные не изменились, и продолжить использовать данные из таблиц хранения даже после истечения льготного периода.

Коннектор Iceberg поддерживает предложение WHEN STALE в CREATE MATERIALIZED VIEW для управления поведением, когда материализованное представление устарело.

Удаление материализованного представления с помощью DROP MATERIALIZED VIEW удаляет определение и таблицу хранения.

Табличные функции#

Коннектор поддерживает табличные функции, описанные в следующих разделах.

table_changes#

Позволяет читать изменения на уровне строк между двумя версиями таблицы Iceberg. Следующий запрос показывает пример отображения изменений таблицы t1 в схеме default текущего каталога. Возвращаются все изменения между начальным и конечным снимками.

SELECT
  *
FROM
  TABLE(
    system.table_changes(
      schema_name => 'default',
      table_name => 't1',
      start_snapshot_id => 6541165659943306573,
      end_snapshot_id => 6745790645714043599
    )
  );

Функция принимает следующие обязательные параметры:

  • schema_name

    Имя схемы, для которой вызывается функция.

  • table_name

    Имя таблицы, для которой вызывается функция.

  • start_snapshot_id

    Идентификатор исключаемого начального снимка.

  • end_snapshot_id

    Идентификатор включаемого конечного снимка.

Используйте metadata-таблицу $snapshots, чтобы определить ID снимков таблицы.

Функция возвращает столбцы, присутствующие в таблице, а также следующие значения для каждого изменения:

  • _change_type

    Тип произошедшего изменения. Возможные значения: insert и delete.

  • _change_version_id

    Идентификатор снимка, в котором произошло изменение.

  • _change_timestamp

    Временная метка, когда снимок стал активным.

  • _change_ordinal

    Порядковый номер изменения, полезный для сортировки результатов.

Пример:

Создайте таблицу:

CREATE TABLE test_schema.pages (page_url VARCHAR, domain VARCHAR, views INTEGER);

Вставьте данные:

INSERT INTO test_schema.pages
    VALUES
        ('url1', 'domain1', 1),
        ('url2', 'domain2', 2),
        ('url3', 'domain1', 3);
INSERT INTO test_schema.pages
    VALUES
        ('url4', 'domain1', 400),
        ('url5', 'domain2', 500),
        ('url6', 'domain3', 2);

Получите идентификаторы снимков изменений, выполненных в таблице:

SELECT
    snapshot_id,
    parent_id,
    operation
FROM test_schema."pages$snapshots";

     snapshot_id     |      parent_id      | operation
---------------------+---------------------+-----------
 2009020668682716382 |                NULL | append
 2135434251890923160 | 2009020668682716382 | append
 3108755571950643966 | 2135434251890923160 | append
(3 rows)

Выберите изменения, выполненные в упомянутых ранее операторах INSERT:

SELECT
    *
FROM
    TABLE(
            system.table_changes(
                    schema_name => 'test_schema',
                    table_name => 'pages',
                    start_snapshot_id => 2009020668682716382,
                    end_snapshot_id => 3108755571950643966
            )
    )
ORDER BY _change_ordinal ASC;

 page_url | domain  | views | _change_type | _change_version_id  |      _change_timestamp      | _change_ordinal
----------+---------+-------+--------------+---------------------+-----------------------------+-----------------
 url1     | domain1 |     1 | insert       | 2135434251890923160 | 2024-04-04 21:24:26.105 UTC |               0
 url2     | domain2 |     2 | insert       | 2135434251890923160 | 2024-04-04 21:24:26.105 UTC |               0
 url3     | domain1 |     3 | insert       | 2135434251890923160 | 2024-04-04 21:24:26.105 UTC |               0
 url4     | domain1 |   400 | insert       | 3108755571950643966 | 2024-04-04 21:24:28.318 UTC |               1
 url5     | domain2 |   500 | insert       | 3108755571950643966 | 2024-04-04 21:24:28.318 UTC |               1
 url6     | domain3 |     2 | insert       | 3108755571950643966 | 2024-04-04 21:24:28.318 UTC |               1
(6 rows)
Ограничения#

  • Таблицы с файлами удаления не поддерживаются. Табличная функция table_changes не поддерживает снимки, включающие файлы удаления. Такие файлы удаления обычно создаются операциями на уровне строк.

  • Функция table_changes сообщает об изменениях по снимкам в указанном диапазоне. Она не вычисляет итоговый эффект по нескольким снимкам. Например, если строка удалена в одном снимке и повторно вставлена в более позднем снимке в пределах диапазона, функция возвращает две записи (одно удаление и одну вставку), а не пропускает строку как не имеющую итогового изменения.

Производительность#

Коннектор включает ряд улучшений производительности, подробно описанных в следующих разделах.

Статистика таблицы#

Коннектор Iceberg может собирать статистику столбцов с помощью оператора ANALYZE.

Обновление статистики таблицы#

Если ваши запросы сложны и включают соединение больших наборов данных, выполнение ANALYZE для таблиц может повысить производительность запросов за счет сбора статистической информации о данных:

ANALYZE table_name;

Этот запрос собирает статистику для всех столбцов.

Для широких таблиц сбор статистики по всем столбцам может быть дорогим. Обычно это также не требуется: статистика полезна только для отдельных столбцов, таких как ключи соединения, предикаты или ключи группировки. Можно указать подмножество столбцов для анализа с помощью необязательного свойства columns:

ANALYZE table_name WITH (columns = ARRAY['col_1', 'col_2']);

Этот запрос собирает статистику для столбцов col_1 и col_2.

Обратите внимание: если ранее статистика была собрана для всех столбцов, перед повторным анализом ее нужно удалить командой drop_extended_stats.

Перенаправление таблиц#

Trino позволяет прозрачно перенаправлять операции над существующей таблицей в подходящий каталог на основе формата таблицы и конфигурации каталога.

В контексте коннекторов, зависящих от службы metastore, например Коннектор Hive, Коннектор Iceberg и Коннектор Delta Lake, metastore (служба Hive metastore, AWS Glue Data Catalog) может использоваться для работы с таблицами разных форматов. Поэтому база данных metastore может содержать разные таблицы с разными форматами.

В качестве конкретного примера используем следующий простой сценарий, в котором задействовано перенаправление таблицы:

USE example.example_schema;

EXPLAIN SELECT * FROM example_table;

                               Query Plan
-------------------------------------------------------------------------
Fragment 0 [SOURCE]
     ...
     Output[columnNames = [...]]
     │   ...
     └─ TableScan[table = another_catalog:example_schema:example_table]
            ...

Вывод оператора EXPLAIN указывает фактический каталог, который обрабатывает запрос SELECT к таблице example_table.

Функциональность перенаправления таблиц также работает при использовании полностью квалифицированных имен таблиц:

EXPLAIN SELECT * FROM example.example_schema.example_table;

                               Query Plan
-------------------------------------------------------------------------
Fragment 0 [SOURCE]
     ...
     Output[columnNames = [...]]
     │   ...
     └─ TableScan[table = another_catalog:example_schema:example_table]
            ...

Trino поддерживает перенаправление таблиц для следующих операций:

Trino не поддерживает перенаправление представлений.

Коннектор поддерживает перенаправление из таблиц Iceberg в таблицы Hive с помощью свойства конфигурации каталога iceberg.hive-catalog-name.

Кэш файловой системы#

Коннектор поддерживает настройку и использование кэширования файловой системы.

Кэширование метаданных Iceberg#

Коннектор Iceberg поддерживает кэширование метаданных в памяти координатора. Это кэширование метаданных включено по умолчанию и может быть отключено установкой свойства конфигурации iceberg.metadata-cache.enabled в false. Когда fs.cache.enabled установлено в true, метаданные кэшируются на локальных дисках с использованием реализации кэширования файловой системы. Если fs.cache.enabled включено, кэширование метаданных в памяти координатора отключается.

Кроме того, можно использовать следующие свойства конфигурации каталога:

Свойства конфигурации кэширования метаданных в памяти#

Свойство

Описание

fs.memory-cache.ttl

Максимальная продолжительность хранения файлов в кэше перед вытеснением. Минимальное значение 0s означает, что кэширование фактически отключено. По умолчанию — 1h.

fs.memory-cache.max-size

Максимальный общий размер данных в кэше. При увеличении этого значения учитывайте, что используется память координатора. По умолчанию — 2% от максимального размера кучи на узле.

fs.memory-cache.max-content-length

Максимальный размер файла, который можно кэшировать. По умолчанию — 15MB.