Apache Iceberg + Parquet: Руководство для Data Engineers¶

-- Пользователь должен знать о партициях
SELECT * FROM orders 
WHERE year = 2026 AND month = 1 AND day = 23 
  AND customer_id = 'cust_123';

-- Iceberg автоматически использует партиции
SELECT * FROM orders 
WHERE created_at = '2026-01-23' 
  AND customer_id = 'cust_123';

-- Изначально: monthly partitioning
ALTER TABLE sales.orders 
SET PARTITION SPEC (month(created_at));

-- Трафик вырос, переходим на daily
ALTER TABLE sales.orders 
REPLACE PARTITION FIELD month(created_at) WITH day(created_at);

-- Старые данные остаются в monthly партициях
-- Новые данные пишутся в daily партиции
-- Запросы работают корректно с обоими

-- Чтение данных как было вчера
SELECT * FROM sales.orders 
TIMESTAMP AS OF '2026-01-22 10:00:00';

-- Чтение конкретного snapshot
SELECT * FROM sales.orders 
VERSION AS OF 123456789;

-- Список всех snapshots
SELECT * FROM sales.orders.snapshots 
ORDER BY committed_at DESC;

-- Rollback к предыдущему snapshot
CALL system.rollback_to_snapshot('sales.orders', 123456789);

-- Добавление колонки (non-breaking)
ALTER TABLE sales.orders 
ADD COLUMN discount_percent DOUBLE;

-- Удаление колонки
ALTER TABLE sales.orders 
DROP COLUMN old_field;

-- Переименование колонки
ALTER TABLE sales.orders 
RENAME COLUMN old_name TO new_name;

-- Изменение типа (widening only)
ALTER TABLE sales.orders 
ALTER COLUMN price TYPE DECIMAL(20,2);  -- было DECIMAL(10,2)

-- Query engine читает statistics из footer
SELECT * FROM orders WHERE total_amount > 100000;

-- Если max(total_amount) в row group < 100000, 
-- весь row group пропускается (не читается с диска)!

SELECT * FROM orders WHERE order_id = 'ord_abc123';

-- 1. Query engine проверяет bloom filter для каждого Parquet файла
-- 2. Если bloom filter говорит "NO" - файл пропускается (100% уверенность)
-- 3. Если bloom filter говорит "MAYBE" - файл читается (1% ложных срабатываний)

-- Удалить snapshots старше 30 дней, оставить минимум 10
CALL system.expire_snapshots(
  table => 'sales.orders',
  older_than => TIMESTAMP '2025-12-24',
  retain_last => 10
);

-- Удалить orphan files старше 3 дней
CALL system.remove_orphan_files(
  table => 'sales.orders',
  older_than => TIMESTAMP '2026-01-20'
);

-- Объединить small files в последних 7 днях
CALL system.rewrite_data_files(
  table => 'sales.orders',
  strategy => 'binpack',
  where => 'created_at >= current_date() - INTERVAL ''7'' DAY'
);

-- Сколько small files?
SELECT COUNT(*) as small_files_count
FROM sales.orders.files
WHERE file_size_in_bytes < 128 * 1024 * 1024;  -- < 128 MB

-- Z-order rewrite для последних 30 дней
CALL system.rewrite_data_files(
  table => 'sales.orders',
  strategy => 'sort',
  sort_order => 'zorder(customer_id, created_at)',
  where => 'created_at >= current_date() - INTERVAL ''30'' DAY'
);

-- Z-order полезен для таких запросов:
SELECT * FROM orders
WHERE customer_id = 'cust_123'   -- Filter 1
  AND created_at >= '2026-01-01'  -- Filter 2
  AND total_amount > 1000;        -- Filter 3

-- История snapshots
SELECT 
  snapshot_id,
  committed_at,
  operation,
  summary
FROM sales.orders.snapshots
ORDER BY committed_at DESC
LIMIT 10;

-- Список data files
SELECT 
  file_path,
  file_size_in_bytes / 1024^2 as size_mb,
  record_count,
  partition
FROM sales.orders.files
ORDER BY file_size_in_bytes DESC
LIMIT 20;

-- Статистика по партициям
SELECT 
  partition,
  COUNT(*) as file_count,
  SUM(record_count) as total_records,
  SUM(file_size_in_bytes) / 1024^3 as size_gb
FROM sales.orders.files
GROUP BY partition
ORDER BY partition DESC;

-- Список manifests
SELECT 
  path,
  added_data_files_count,
  existing_data_files_count,
  deleted_data_files_count
FROM sales.orders.manifests
ORDER BY added_snapshot_id DESC;

-- История изменений таблицы
SELECT 
  made_current_at,
  snapshot_id,
  operation,
  is_current_ancestor
FROM sales.orders.history
ORDER BY made_current_at DESC;

-- 1. Table size
SELECT SUM(file_size_in_bytes) / 1024^3 AS table_size_gb
FROM sales.orders.files;

-- 2. Small files ratio
SELECT 
  COUNT(CASE WHEN file_size_in_bytes < 128*1024^2 THEN 1 END) AS small_files,
  COUNT(*) AS total_files,
  COUNT(CASE WHEN file_size_in_bytes < 128*1024^2 THEN 1 END) * 100.0 / COUNT(*) AS small_files_pct
FROM sales.orders.files;

-- 3. Snapshot age
SELECT 
  NOW() - MAX(committed_at) AS oldest_snapshot_age
FROM sales.orders.snapshots;

-- 4. Query performance (Trino)
SELECT 
  query_id,
  query,
  state,
  elapsed_time_ms,
  queued_time_ms,
  planning_time_ms,
  execution_time_ms,
  peak_memory_bytes / 1024^3 AS peak_memory_gb
FROM system.runtime.queries
WHERE query LIKE '%sales.orders%'
  AND state = 'FINISHED'
ORDER BY elapsed_time_ms DESC
LIMIT 10;

-- 1. Проверить predicate pushdown
EXPLAIN SELECT * FROM orders WHERE created_at >= '2026-01-23';

-- 2. Проверить partition pruning
EXPLAIN ANALYZE SELECT * FROM orders WHERE created_at >= '2026-01-23';

-- 3. Проверить file count
SELECT COUNT(*) FROM sales.orders.files 
WHERE partition LIKE '%created_at_day=2026-01-23%';

SELECT 
  DATE_TRUNC('day', committed_at) as date,
  COUNT(*) as files_created
FROM sales.orders.files
GROUP BY DATE_TRUNC('day', committed_at)
ORDER BY date DESC;

-- Daily compaction
CALL system.rewrite_data_files(
  table => 'sales.orders',
  strategy => 'binpack',
  where => 'created_at >= current_date() - 1'
);

-- Проверить количество snapshots
SELECT COUNT(*) FROM sales.orders.snapshots;

-- Проверить количество manifests
SELECT COUNT(*) FROM sales.orders.manifests;

-- Expire старые snapshots
CALL system.expire_snapshots(
  table => 'sales.orders',
  older_than => CURRENT_TIMESTAMP - INTERVAL '7' DAY,
  retain_last => 10
);

-- Оптимизировано с bloom filter
SELECT * FROM sales.orders
WHERE order_id = 'ord_abc123';

-- План запроса:
-- 1. Bloom filter проверка → пропустить 99% файлов
-- 2. Прочитать 1-2 файла
-- Latency: ~50ms p50, ~200ms p99

-- Оптимизировано с partition pruning
SELECT * FROM sales.orders
WHERE created_at >= CURRENT_DATE - INTERVAL '7' DAY
ORDER BY created_at DESC;

-- План запроса:
-- 1. Partition pruning → только 7 партиций
-- 2. Sort order → sequential read, no sorting needed
-- Latency: ~500ms p50, ~2s p99

-- Оптимизировано с sort order + z-order
SELECT 
  customer_id,
  COUNT(*) as total_orders,
  SUM(total_amount) as total_spent,
  MAX(created_at) as last_order
FROM sales.orders
WHERE customer_id = 'cust_123'
GROUP BY customer_id;

-- План запроса:
-- 1. Bloom filter на customer_id → skip большинство файлов
-- 2. Sort order → data локализована, sequential read
-- 3. Columnar format → читаем только нужные колонки
-- Latency: ~100ms

-- Columnar format shine
SELECT 
  DATE(created_at) as date,
  COUNT(*) as orders,
  SUM(total_amount) as revenue,
  AVG(total_amount) as avg_order_value
FROM sales.orders
WHERE created_at >= CURRENT_DATE - INTERVAL '30' DAY
GROUP BY DATE(created_at)
ORDER BY date DESC;

-- План запроса:
-- 1. Partition pruning → 30 партиций
-- 2. Columnar read → только created_at и total_amount колонки
-- 3. Vectorized aggregation
-- Latency: ~2s p50, ~10s p99