-- SELECT * EXCLUDE — drop columns by name (not rename)
SELECT * EXCLUDE (_fivetran_deleted, _loaded_at, ssn, credit_card)
FROM customers;

-- SELECT * REPLACE — override specific columns inline
SELECT * REPLACE (
  amount / 100.0   AS amount,     -- cents → dollars in-place
  upper(status)   AS status      -- normalize casing
)
FROM orders;

-- COLUMNS(regex) — select by pattern
SELECT COLUMNS('.*_at$')          -- all timestamp columns
FROM events;

SELECT COLUMNS('revenue|amount|cost')  -- financial columns
FROM mart_revenue;

-- Apply a function to ALL matching columns at once:
SELECT
  id,
  round(COLUMNS('.*_amount$'), 2)  -- round every amount column
FROM orders;

-- COLUMNS with lambda (wildcard aggregation):
SELECT
  min(COLUMNS('.*')),   -- min of every column
  max(COLUMNS('.*')),   -- max of every column
  count(COLUMNS('.*')) -- non-null count of every column
FROM events;

-- GROUP BY ALL — infers non-aggregated columns automatically
SELECT
  date_trunc('day', event_ts) AS day,
  event_type,
  platform,
  count(*)                   AS n,
  avg(latency_ms)            AS avg_latency
FROM events
GROUP BY ALL;     -- auto: GROUP BY day, event_type, platform

-- ORDER BY ALL — sorts by every non-aggregated column
SELECT year, quarter, revenue
FROM mart_revenue
ORDER BY ALL;    -- ORDER BY year, quarter

-- ORDER BY alias — reference output column name
SELECT
  event_ts::date          AS day,
  count(*)               AS event_count
FROM events
GROUP BY day             -- reference alias directly!
ORDER BY event_count DESC;  -- same here

-- FROM first — anti-Cartesian safety + readability
FROM events
SELECT event_id, event_type
WHERE  event_ts > '2024-01-01'
LIMIT  100;  -- FROM before SELECT is valid DuckDB!

-- PIVOT: rows → columns (dynamic cross-tab)
PIVOT orders
ON     status          -- unique values become column headers
USING  count(*)        -- aggregation for each cell
GROUP BY customer_id;

-- Result: customer_id | PENDING | COMPLETED | CANCELLED

-- PIVOT with explicit values (for deterministic column order):
PIVOT revenue_data
ON    quarter IN (Q1, Q2, Q3, Q4)
USING sum(amount) AS rev
GROUP BY product_line;

-- UNPIVOT: columns → rows (melt wide to long)
UNPIVOT wide_metrics
ON     revenue, cost, profit    -- columns to collapse
INTO
  NAME  metric_name             -- new key column
  VALUE metric_value;            -- new value column

-- Dynamic PIVOT on expression (all months in data):
PIVOT (
  SELECT product, strftime(order_date, '%Y-%m') AS month, revenue
  FROM orders
)
ON    month
USING sum(revenue)
GROUP BY product;

-- QUALIFY filters on window function results.
-- Without QUALIFY, you need a subquery or CTE.

-- Latest order per customer (Top-N per group):
SELECT customer_id, order_id, created_at, amount
FROM   orders
QUALIFY row_number() OVER (
  PARTITION BY customer_id
  ORDER BY     created_at DESC
) = 1;

-- Deduplicate on composite key, keep latest:
SELECT * FROM raw_events
QUALIFY row_number() OVER (
  PARTITION BY event_id
  ORDER BY     _ingested_at DESC
) = 1;

─────────────────────────────────────────────────

-- ASOF JOIN — time-series "as of" matching.
-- For each event, find the exchange rate in effect AT THAT TIME.

SELECT
  o.order_id,
  o.amount_usd,
  o.amount_usd * r.rate_gbp AS amount_gbp
FROM       orders o
ASOF JOIN  exchange_rates r
           ON  o.currency    = r.currency
           AND o.created_at >= r.valid_from;

-- ASOF finds the largest r.valid_from <= o.created_at.
-- No self-joins, no correlated subqueries. Pure beauty.

-- DuckDB has first-class LIST type and lambda functions.

-- list_transform: map over a list
SELECT list_transform([1,2,3,4], x -> x * x);
-- → [1, 4, 9, 16]

-- list_filter: filter with a predicate
SELECT list_filter(tags, t -> t LIKE 'prod_%')
FROM events;

-- list_reduce: fold
SELECT list_reduce([10,20,30], (acc, x) -> acc + x);
-- → 60

-- Nested lambda: normalize prices in a list
SELECT
  product_id,
  list_transform(
    price_history,
    p -> round(p * exchange_rate, 2)
  ) AS price_gbp
FROM products;

-- Aggregate into list, then filter:
SELECT
  customer_id,
  list_filter(
    list(order_id ORDER BY created_at),
    (o, i) -> i < 5    -- first 5 orders only (index lambda)
  ) AS first_5_orders
FROM orders
GROUP BY 1;

-- Struct: composite type with named fields
SELECT
  struct_pack(name := 'Alice', age := 32) AS person,
  person.name,    -- dot access on struct
  person['age']; -- bracket access

-- Unnest array column into rows:
SELECT order_id, unnest(line_items) AS item
FROM orders;

-- Unnest with ordinality (keep array index):
SELECT order_id, item, idx
FROM   orders,
       unnest(line_items) WITH ORDINALITY t(item, idx);

-- SUMMARIZE — instant profiling of any table/query:
SUMMARIZE events;
-- Returns: count, nulls, min, max, avg, std,
--          q25, q50, q75 for EVERY column. Zero code.

-- generate_series — date spines, sequences:
SELECT generate_series(
  '2024-01-01'::date,
  '2024-12-31'::date,
  INTERVAL '1 day'
) AS day;

-- SAMPLE — fast approximate queries:
SELECT avg(amount), stddev(amount)
FROM events USING SAMPLE 10%;         -- reservoir sample

FROM events USING SAMPLE 100000 ROWS; -- fixed row count

import duckdb

# In-memory (default) — fastest, no persistence:
con = duckdb.connect()            # or duckdb.connect(':memory:')

# Persistent file — survives process restart:
con = duckdb.connect('analytics.duckdb')

# Read-only file — safe for concurrent reads:
con = duckdb.connect('analytics.duckdb', read_only=True)

# Module-level shortcut (implicit in-memory connection):
duckdb.sql("SELECT 42")  # no connect() needed

# Context manager — auto-close:
with duckdb.connect('analytics.duckdb') as con:
    result = con.sql("SELECT count(*) FROM events").fetchone()

# Configuration at connect-time:
con = duckdb.connect(config={
    'threads':      8,
    'memory_limit': '16GB',
    'temp_directory': '/tmp/duckdb_spill'
})

# A Relation is a lazy query plan — nothing executes until .execute()
rel = con.table('events')
rel = con.read_parquet('s3://bucket/events/*.parquet')
rel = con.sql("SELECT * FROM events")

# Chain operations (all lazy):
result = (
    con.read_parquet('events.parquet')
       .filter("event_type = 'purchase'")
       .project("user_id, amount, event_ts")
       .aggregate("user_id, sum(amount) AS total")
       .order("total DESC")
       .limit(100)
)

# Execute and collect:
df    = result.df()     # pandas
pl_df = result.pl()     # polars
arrow = result.arrow()  # pyarrow
rows  = result.fetchall()  # list of tuples

# Inspect the query plan:
print(result.explain())
print(result.query_df("SELECT * FROM query_relation"))

import duckdb
from duckdb.typing import VARCHAR, BIGINT, DOUBLE
import re, hashlib

con = duckdb.connect()

# --- SCALAR UDF ---
def mask_email(email: str) -> str:
    if not email: return None
    local, _, domain = email.partition('@')
    return f"{local[:2]}***@{domain}"

con.create_function(
    'mask_email',
    mask_email,
    [VARCHAR],   # input types
    VARCHAR,     # return type
    null_handling='special'  # handle NULLs in Python
)

con.sql("SELECT mask_email(email) FROM customers")

# --- VECTORIZED UDF (Arrow-native, batched) ---
import pyarrow as pa

def hash_ids(ids: pa.Array) -> pa.Array:
    # Runs on the full 1024-row vector at once — no Python loop!
    hashed = [
        hashlib.sha256(str(v).encode()).hexdigest()[:16]
        for v in ids.to_pylist()
    ]
    return pa.array(hashed, type=pa.string())

con.create_function(
    'hash_id', hash_ids,
    [BIGINT], VARCHAR,
    type=duckdb.functional.PythonUDFType.ARROW
)

# --- AGGREGATE UDF ---
class GeometricMean:
    def __init__(self): self.values = []
    def step(self, x):
        if x and x > 0: self.values.append(x)
    def finalize(self):
        if not self.values: return None
        return (reduce(lambda a,b: a*b, self.values)
                ) ** (1/len(self.values))

con.create_aggregate_function('geo_mean', GeometricMean, [DOUBLE], DOUBLE)

# Jupyter magic — write SQL in cells directly:
%load_ext duckdb_magic

%%duckdb
SELECT year, sum(revenue) total
FROM read_parquet('data/*.parquet')
GROUP BY ALL
ORDER BY year;

# Query profiling in Python:
con.execute("PRAGMA enable_profiling")
con.execute("PRAGMA profiling_output='/tmp/profile.json'")
con.sql("SELECT * FROM big_table WHERE x > 1000")
# Open profile.json in chrome://tracing

# EXPLAIN ANALYZE:
print(con.execute("""
  EXPLAIN ANALYZE
  SELECT customer_id, sum(amount)
  FROM events
  WHERE event_ts > '2024-01-01'
  GROUP BY 1
""").fetchdf().to_string())

import ibis

# Ibis = portable dataframe API that compiles to SQL.
# DuckDB is the best Ibis backend (fastest, most complete).

con = ibis.duckdb.connect('analytics.duckdb')

events = con.table('events')

result = (
    events
    .filter([
        events.event_ts > ibis.timestamp('2024-01-01'),
        events.event_type.isin(['purchase', 'refund'])
    ])
    .group_by(['user_id', 'event_type'])
    .aggregate(
        n       = events.event_id.count(),
        total   = events.amount.sum(),
        avg_lat = events.latency_ms.mean()
    )
    .order_by(ibis.desc('total'))
    .limit(100)
)

# See the compiled SQL:
ibis.to_sql(result)

# Execute:
df = result.to_pandas()

# Switch backend to BigQuery with zero code changes:
# con = ibis.bigquery.connect(project='my-proj', dataset='my_ds')

-- Parallelism (default: all CPU cores)
SET threads = 16;
SET threads = 1;    -- deterministic debugging

-- Memory (default: 80% of RAM)
SET memory_limit = '32GB';
SET memory_limit = '-1';  -- unlimited

-- External spilling — enables larger-than-RAM queries
SET temp_directory = '/fast-nvme/duckdb_tmp';
SET max_temp_directory_size = '500GB';

-- Worker threads for Parquet I/O
SET worker_threads_for_csv_writing = 8;

-- Preserve insertion order (OFF = faster GROUP BY)
SET preserve_insertion_order = false;

-- Enable object cache (S3 metadata caching)
SET enable_object_cache = true;

-- HTTP metadata cache for remote Parquet
SET http_metadata_cache_enable = true;

-- Profile every query:
PRAGMA enable_progress_bar;
PRAGMA enable_profiling = 'json';
PRAGMA profiling_output = '/tmp/duck_profile.json';

-- Write partitioned Parquet — the ideal ingress landing zone:
COPY (
  SELECT *, year(event_ts) AS yr, month(event_ts) AS mo
  FROM events
) TO 's3://output/events' (
  FORMAT         parquet,
  PARTITION_BY   (yr, mo),           -- hive partitioning
  COMPRESSION    zstd,               -- best ratio/speed tradeoff
  ROW_GROUP_SIZE 122880,            -- 128K rows/group (sweet spot)
  OVERWRITE_OR_IGNORE true
);

-- Write sorted Parquet (dramatically improves filter pushdown):
COPY (
  SELECT * FROM events ORDER BY customer_id, event_ts
) TO 'events_sorted.parquet' (
  FORMAT parquet,
  COMPRESSION zstd
);
-- Sorted Parquet + row group stats = skip entire row groups on filter.
-- 100x scan speedup on customer_id range queries.

-- DuckDB native format (fastest for DuckDB-only workloads):
CHECKPOINT;  -- flush WAL to .duckdb file
-- The .duckdb file IS the storage — already column-grouped,
-- zone-mapped, FSST-compressed. No conversion needed.

import duckdb, boto3, orjson
from datetime import datetime, timezone

"""
ARCHITECTURE:
  Kafka → Consumer → DuckDB (transform) → S3 Parquet (partitioned)
  Queries: DuckDB directly on S3 Parquet
  No cluster. No warehouse. No ETL server.
"""

con = duckdb.connect()
con.execute("INSTALL httpfs; LOAD httpfs")

def land_batch(events: list[dict], dt: datetime):
    # Ingest into DuckDB in-memory, transform, write Parquet
    con.execute("CREATE OR REPLACE TABLE staging AS SELECT * FROM events")

    path = (
        f"s3://my-lake/events/"
        f"year={dt.year}/month={dt.month:02d}/day={dt.day:02d}/"
        f"batch_{dt.timestamp():.0f}.parquet"
    )

    con.execute(f"""
        COPY (
            SELECT
                event_id,
                user_id,
                event_type,
                -- DuckDB parsing JSON inline:
                (payload->>'$.session_id')::VARCHAR AS session_id,
                (payload->>'$.amount')::DECIMAL(18,4) AS amount,
                event_ts::TIMESTAMP AS event_ts
            FROM staging
            WHERE event_id IS NOT NULL
            QUALIFY row_number() OVER (
                PARTITION BY event_id ORDER BY _ingested_at DESC
            ) = 1   -- deduplicate in the same query
        ) TO '{path}'
        (FORMAT parquet, COMPRESSION zstd, ROW_GROUP_SIZE 65536)
    """)
    return path

def query_lakehouse(start_date: str, end_date: str):
    return con.sql(f"""
        SELECT
            date_trunc('hour', event_ts) AS hour,
            event_type,
            count(*) AS n,
            sum(amount) AS revenue
        FROM read_parquet(
            's3://my-lake/events/year=*/month=*/day=*/*.parquet',
            hive_partitioning = true
        )
        WHERE event_ts BETWEEN '{start_date}' AND '{end_date}'
        GROUP BY ALL
        ORDER BY hour, revenue DESC
    """).df()

"""
Pattern: API Gateway → Lambda (DuckDB) → S3 Parquet
A serverless analytical query engine with zero infra.
Lambda has 10GB memory and 15min timeout.
DuckDB + 64-core Lambda = serious computation.
"""
import duckdb, json, os

def handler(event, context):
    con = duckdb.connect()
    con.execute("INSTALL httpfs; LOAD httpfs")

    # IAM role auto-resolved via instance metadata
    con.execute("""
        CREATE SECRET aws (
            TYPE s3,
            PROVIDER CREDENTIAL_CHAIN
        )
    """)

    query      = event['queryStringParameters']['q']
    start_date = event['queryStringParameters']['start']
    end_date   = event['queryStringParameters']['end']

    # Parameterized to avoid injection:
    result = con.execute("""
        SELECT event_type, count(*) n, sum(amount) rev
        FROM read_parquet(
            's3://my-lake/events/year=*/month=*/day=*/*.parquet',
            hive_partitioning = true
        )
        WHERE event_ts BETWEEN $1 AND $2
        GROUP BY ALL
        ORDER BY rev DESC
        LIMIT 1000
    """, [start_date, end_date]).df()

    return {
        'statusCode': 200,
        'body': result.to_json(orient='records'),
        'headers': {'Content-Type': 'application/json'}
    }

// DuckDB-WASM: full DuckDB compiled to WebAssembly.
// Query Parquet files in the browser. No server needed.

import * as duckdb from '@duckdb/duckdb-wasm';

const BUNDLE = duckdb.selectBundle({
  mvp: { mainModule: '.../duckdb-mvp.wasm',
         mainWorker: '.../duckdb-browser-mvp.worker.js' },
  eh:  { mainModule: '.../duckdb-eh.wasm',
         mainWorker:  '.../duckdb-browser-eh.worker.js' },
});

const worker = new Worker(BUNDLE.mainWorker);
const db     = new duckdb.AsyncDuckDB(logger, worker);
await db.instantiate(BUNDLE.mainModule);
const con = await db.connect();

// Register a Parquet file from a URL:
await db.registerFileURL('events.parquet',
  'https://example.com/events.parquet',
  duckdb.DuckDBDataProtocol.HTTP, false
);

// Query it — all in the browser, no round-trip to server:
const result = await con.query(`
  SELECT event_type, count(*) AS n
  FROM read_parquet('events.parquet')
  GROUP BY ALL ORDER BY n DESC
`);

console.log(result.toArray());