import duckdb

con = duckdb.connect("analytics.duckdb")

# Method 1: Direct write from DataFrame (most common)
import pandas as pd
df = pd.read_csv("large_dataset.csv")
con.execute("CREATE TABLE events AS SELECT * FROM df")
con.execute("COPY events TO 'events.parquet' (FORMAT PARQUET)")

# Method 2: One-line CSV to Parquet conversion
con.execute("""
    COPY (SELECT * FROM read_csv_auto('large_dataset.csv'))
    TO 'events.parquet' (FORMAT PARQUET)
""")

# Method 3: Parallel write for multiple files (utilizes multi-threading)
con.execute("""
    COPY (SELECT * FROM read_csv_auto('data/*.csv'))
    TO 'events_output.parquet' (FORMAT PARQUET, COMPRESSION ZSTD, CHUNK_SIZE 32768)
""")
print("✅ Parquet file generated")

-- Performance and size comparison of different compression algorithms
-- ZSTD: Best compression ratio, ideal for storage
-- SNAPPY: Balance between speed and compression, recommended default
-- UNCOMPRESSED: Maximum speed, ideal for temporary files

-- ZSTD compression (highest compression ratio, for long-term storage)
COPY events TO 'events_zstd.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);

-- SNAPPY compression (balanced approach, recommended for production)
COPY events TO 'events_snappy.parquet' (FORMAT PARQUET, COMPRESSION SNAPPY);

-- BROTLI compression (compression ratio close to ZSTD, faster decompression)
COPY events TO 'events_brotli.parquet' (FORMAT PARQUET, COMPRESSION BROTLI);

-- Check file sizes with different compression methods
SELECT 
    filename,
    round(file_size / 1024 / 1024, 2) AS size_mb,
    round(file_size / 1024 / 1024 / 1024, 2) AS size_gb
FROM (VALUES
    ('events_uncompressed.parquet', system.parquet_metadata('events_uncompressed.parquet').row_groups[1].total_compressed_size),
    ('events_snappy.parquet', system.parquet_metadata('events_snappy.parquet').row_groups[1].total_compressed_size),
    ('events_zstd.parquet', system.parquet_metadata('events_zstd.parquet').row_groups[1].total_compressed_size)
) AS t(filename, file_size);

import duckdb

con = duckdb.connect("analytics.duckdb")

# ❌ Wrong: Read all columns (wastes IO and memory)
df_bad = con.execute("SELECT * FROM events.parquet").df()

# ✅ Correct: Only read needed columns (reduces 80%+ IO)
df_good = con.execute("""
    SELECT event_type, user_id, timestamp 
    FROM read_parquet('events.parquet')
    WHERE timestamp >= '2026-01-01'
""").df()

# ✅ Advanced: Leverage predicate pushdown
# DuckDB automatically pushes WHERE conditions to the Parquet read layer
df_optimized = con.execute("""
    SELECT user_id, event_type, COUNT(*) as cnt
    FROM read_parquet('events.parquet')
    WHERE event_type IN ('purchase', 'signup')
      AND timestamp >= '2026-06-01'
    GROUP BY user_id, event_type
""").df()

-- DuckDB automatically optimizes to:
-- 1. Read Parquet file statistics (min/max of each column)
-- 2. Skip Row Groups that don't meet conditions
-- 3. Only read data blocks that satisfy conditions

-- Example: 10GB Parquet file, reading only 1% of data
EXPLAIN ANALYZE
SELECT *
FROM read_parquet('events_large.parquet')
WHERE event_date = '2026-09-25'
  AND user_id > 1000000;

-- Output example (key information):
-- Output: []
--   Parallel Project: []
--     Parallel TableFunction: read_parquet
--       Filters: event_date = '2026-09-25' AND user_id > 1000000
--       RowGroupsSkipped: 48/50  ← Skipped 96% of data!
--       RowsRead: 2,345,678 / 234,567,890  ← Only read 1%

-- Create partitioned directory structure
-- data/
-- ├── year=2024/month=01/events.parquet
-- ├── year=2024/month=02/events.parquet
-- ├── ...
-- └── year=2026/month=09/events.parquet

-- Auto-detect Hive-style partitions
SELECT *
FROM read_parquet('data/year=*/month=*/events.parquet')
WHERE year = 2026 AND month = 9;

-- DuckDB will only read the 2026-09 partition file
-- Instead of scanning all 33 months of data!

import duckdb

con = duckdb.connect("analytics.duckdb")

# Scenario: One Parquet file per month, need cross-month aggregation
# Auto-detect all Parquet files in the directory
result = con.execute("""
    SELECT 
        strftime(timestamp, '%Y-%m') AS month,
        event_type,
        COUNT(*) AS event_count,
        SUM(amount) AS total_amount
    FROM read_parquet('data/monthly_*.parquet', union_by_name=true)
    WHERE timestamp >= '2026-01-01'
    GROUP BY month, event_type
    ORDER BY month DESC
""").df()

print(result.to_markdown())

-- View column encoding info of Parquet files
SELECT 
    column_name,
    compression,
    num_values,
    num_rows,
    null_count,
    distinct_count,
    round(compressed_size / 1024 / 1024, 2) AS compressed_mb,
    round(uncompressed_size / 1024 / 1024, 2) AS uncompressed_mb
FROM parquet_schema('events.parquet')
ORDER BY column_name;

import duckdb

con = duckdb.connect("analytics.duckdb")

# Process large files in chunks to avoid memory overflow
chunk_size = 100_000  # Process 100k rows at a time

for i, batch in enumerate(con.execute("""
    SELECT * FROM read_parquet('huge_dataset.parquet')
""").fetchmany(chunk_size)):
    process_batch(batch)
    print(f"Processed batch {i+1}")

import duckdb
import pyarrow as pa

con = duckdb.connect("analytics.duckdb")

-- DuckDB can output Arrow format directly, zero data copy!
arrow_table = con.execute("SELECT * FROM read_parquet('events.parquet')").arrow()

-- Directly pass to Polars/Pandas with zero copy!
import polars as pl
df = pl.from_arrow(arrow_table)

-- Or pass to ML models
from sklearn.ensemble import RandomForestClassifier
X = arrow_table.to_pandas()[['feature1', 'feature2', 'feature3']]
y = arrow_table.to_pandas()['label']

-- DuckDB automatically parallelizes reading multiple Parquet files
-- Utilizing all CPU cores for acceleration
SELECT 
    region,
    SUM(sales) AS total_sales,
    AVG(price) AS avg_price
FROM read_parquet('/data/sales/*.parquet')
GROUP BY region;

-- You can control parallelism
SET parallel_threads = 8;  -- Uses all cores by default

-- 1. Use appropriate compression algorithms
-- Frequent reads → SNAPPY (balanced)
-- Long-term storage → ZSTD (high compression)
-- Temporary data → UNCOMPRESSED (fastest)

-- 2. Set proper chunk size
COPY large_table TO 'output.parquet' 
(FORMAT PARQUET, COMPRESSION ZSTD, CHUNK_SIZE 65536);

-- 3. Partition on write
COPY (SELECT * FROM source WHERE date >= '2026-01-01')
TO 'output_by_month/' (FORMAT PARQUET, PARTITION_BY (date_trunc('month', date)));

-- 1. Always specify needed columns, never use SELECT *
SELECT user_id, event_type, timestamp FROM events.parquet;

-- 2. Use WHERE conditions to trigger predicate pushdown
SELECT user_id FROM events.parquet 
WHERE event_type = 'purchase' AND timestamp > '2026-09-01';

-- 3. Use LIMIT to preview large tables
SELECT * FROM read_parquet('huge.parquet') LIMIT 100;

-- 4. Check execution plan to confirm optimizations
EXPLAIN ANALYZE 
SELECT user_id, COUNT(*) FROM read_parquet('events.parquet')
WHERE event_date = '2026-09-25' GROUP BY user_id;