# Install DuckDB CLI (v1.5.2 or later)
curl https://install.duckdb.org | sh

# Or via Python
pip install duckdb

-- Install the Iceberg extension (community maintained)
INSTALL iceberg FROM community;
LOAD iceberg;

-- Verify installation
SELECT version();

-- Create mock sales data for Iceberg write testing
CREATE TABLE raw_sales AS
SELECT 
 range AS order_id,
 '2026-0' || (range % 9 + 1)::VARCHAR AS month,
 (random() * 1000)::INTEGER AS customer_id,
 CASE WHEN random() < 0.3 THEN 'Electronics'
 WHEN random() < 0.6 THEN 'Clothing'
 ELSE 'Household'
 END AS category,
 (random() * 5000 + 10)::DECIMAL(10,2) AS amount,
 DATE '2026-01-01' + INTERVAL (range % 365) DAY AS order_date
FROM range(1, 100000);

-- Quick stats
SELECT count(*) AS total_orders,
 round(sum(amount)) AS total_revenue
FROM raw_sales;

-- Attach a local Iceberg database (file-based)
ATTACH 'sales_iceberg' AS sales_db (TYPE iceberg);

USE sales_db;

-- Create a partitioned table (by month)
CREATE TABLE orders (
 order_id INTEGER,
 month VARCHAR,
 customer_id INTEGER,
 category VARCHAR,
 amount DECIMAL(10,2),
 order_date DATE
) PARTITION_BY (month);

-- Bulk insert
INSERT INTO orders 
SELECT * FROM raw_sales;

-- Verify
SELECT month, count(*) AS orders, round(sum(amount)) AS revenue
FROM orders
GROUP BY month
ORDER BY month;

-- View Iceberg snapshot history
SELECT snapshot_id, parent_id, timestamp, manifest_list
FROM iceberg_snapshots('sales_iceberg/orders');

-- Start a transaction: update electronics pricing
BEGIN TRANSACTION;

UPDATE orders 
SET amount = amount * 1.1
WHERE category = 'Electronics';

-- Preview changes
SELECT category, round(sum(amount)) AS revenue
FROM orders
WHERE category = 'Electronics'
GROUP BY category;

COMMIT;

-- ⏱ Time travel: query the snapshot before the update
-- Get historical snapshot IDs
SELECT snapshot_id, timestamp 
FROM iceberg_snapshots('sales_iceberg/orders')
ORDER BY timestamp DESC;

-- Query a specific version
SELECT category, round(sum(amount)) AS revenue
FROM orders
FOR SYSTEM_VERSION AS OF 1234567890
WHERE category = 'Electronics'
GROUP BY category;

-- Rollback to a specific snapshot
ALTER TABLE orders ROLLBACK TO 1234567890;

-- Create incremental update data
CREATE TABLE daily_updates AS
SELECT order_id, '2026-05' AS month, 
 customer_id, 'Electronics' AS category,
 amount * 1.2 AS amount, order_date
FROM raw_sales
WHERE order_id <= 100;

-- Execute MERGE
MERGE INTO orders t
USING daily_updates s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET 
 amount = s.amount,
 category = s.category
WHEN NOT MATCHED THEN INSERT 
 (order_id, month, customer_id, category, amount, order_date)
VALUES 
 (s.order_id, s.month, s.customer_id, s.category, s.amount, s.order_date);

-- Connect to Unity Catalog (requires UC endpoint)
ATTACH 'uc:my_catalog.my_schema' AS uc_db 
 (TYPE uc, 
 endpoint 'https://your-uc-instance/api/2.1/unity-catalog',
 token 'your_token_here');

-- Query Iceberg tables in Unity Catalog
SELECT * FROM uc_db.sales_region_iceberg
WHERE region = 'APAC'
LIMIT 100;

-- Connect to AWS Glue Catalog
ATTACH 'glue:my_database' AS glue_db 
 (TYPE glue, 
 region 'us-east-1');

-- Query Iceberg tables in Glue
SELECT year, count(*) AS flights
FROM glue_db.flights_iceberg
WHERE year >= 2024
GROUP BY year
ORDER BY year DESC;

-- Good partitioning: low-cardinality fields (month, region)
CREATE TABLE orders_partitioned (
 order_id INTEGER,
 month VARCHAR,
 region VARCHAR,
 amount DECIMAL(10,2)
) PARTITION_BY (month, region);

-- Avoid: high-cardinality partitions (order_id, customer_id)
-- These create thousands of tiny files and kill performance

-- Check current file count
SELECT count(*) AS file_count,
 round(sum(file_size_in_bytes) / 1024 / 1024) AS total_mb
FROM iceberg_files('sales_iceberg/orders');

-- Rewrite small files into larger ones
CALL iceberg_rewrite_data_files(
 'sales_iceberg/orders',
 strategy => 'binpack',
 target_bytes_per_file => '134217728' -- 128MB
);

-- DuckDB leverages Iceberg's partition pruning automatically
EXPLAIN ANALYZE
SELECT month, round(sum(amount)) AS revenue
FROM orders
WHERE month IN ('2026-01', '2026-02', '2026-03')
GROUP BY month;
-- DuckDB skips irrelevant partitions

import duckdb
import pandas as pd
from datetime import datetime

def ingest_to_iceberg(csv_path: str, table_path: str, partition_col: str):
 """
 Auto-ingest CSV data into Iceberg with compaction
 """
 con = duckdb.connect()
 
 # Load extensions
 con.execute("INSTALL iceberg FROM community; LOAD iceberg;")
 
 # Attach Iceberg database
 con.execute(f"ATTACH '{table_path}' AS db (TYPE iceberg)")
 
 # Read CSV, add ingest date, write to Iceberg
 con.execute(f"""
 CREATE OR REPLACE TABLE db.raw_data AS 
 SELECT *, '{datetime.now().strftime('%Y-%m-%d')}' AS ingest_date
 FROM read_csv_auto('{csv_path}')
 """)
 
 # Create partitioned table
 con.execute(f"""
 CREATE OR REPLACE TABLE db.partitioned_data 
 PARTITION_BY ({partition_col}) AS
 SELECT * FROM db.raw_data
 """)
 
 # Compact small files
 con.execute(f"CALL iceberg_rewrite_data_files('{table_path}/partitioned_data', 'binpack', 134217728)")
 
 # Stats
 rows = con.execute(f"SELECT count(*) FROM db.partitioned_data").fetchone()[0]
 print(f"✅ Successfully ingested {rows} rows into {table_path}")
 
 con.close()

# Usage
ingest_to_iceberg(
 csv_path='/data/sales_2026.csv',
 table_path='s3://my-bucket/iceberg/sales',
 partition_col='month'
)