import duckdb

# Simplest approach: one-line import
con = duckdb.connect("big_data.duckdb")
con.execute("CREATE TABLE events AS SELECT * FROM read_csv_auto('events_2026.csv')")

# Read all CSV files in directory (automatic parallelism)
con.execute("""
    CREATE TABLE events AS
    SELECT * FROM read_csv_auto('events_2026/*.csv')
""")

# Filter by time range with predicate pushdown (only reads needed data)
con.execute("""
    CREATE TABLE events_2026_q3 AS
    SELECT * FROM read_csv_auto('events_2026/*.csv')
    WHERE event_date >= '2026-07-01'
      AND event_date < '2026-10-01'
""")

# Probe schema from first 1000 rows
schema = con.execute("""
    SELECT column_name, column_type
    FROM (SELECT * FROM read_csv_auto('events_2026/*.csv', SAMPLE_SIZE=1000))
    LIMIT 0
""").fetchdf()

print(schema)
#   column_name    column_type
# 0  event_id       BIGINT
# 1  user_id        BIGINT
# 2  event_type     VARCHAR
# 3  event_date     DATE
# 4  amount         DOUBLE
# ...

# Import with detected schema for precision
con.execute("""
    CREATE TABLE events (
        event_id BIGINT,
        user_id BIGINT,
        event_type VARCHAR,
        event_date DATE,
        amount DOUBLE
    )
""")
con.execute("""
    INSERT INTO events
    SELECT * FROM read_csv_auto('events_2026/*.csv', TYPES(...))
""")

# Directly read gzip-compressed CSV (auto-decompress + parallel parse)
con.execute("CREATE TABLE events AS SELECT * FROM read_csv_auto('events_2026.csv.gz')")

# Directly read gzip-compressed JSON
con.execute("CREATE TABLE events AS SELECT * FROM read_json_auto('events_2026.json.gz')")

con = duckdb.connect("big_data.duckdb", config={
    'threads': '8',              # Use all 8 cores
    'max_memory': '24GB',        # Cap memory usage
    'temp_directory': '/tmp/duckdb_temp',  # Spill-to-disk directory
    'parallel': '8'              # Parallel read threads
})

# Parallel reading of multiple gzip files
con.execute("""
    CREATE TABLE events AS
    SELECT * FROM read_csv_auto('events_2026/*.csv.gz', parallel=8)
""")

import duckdb

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

# Create table structure
con.execute("""
    CREATE TABLE events (
        event_id BIGINT,
        user_id BIGINT,
        event_type VARCHAR,
        event_date DATE,
        amount DOUBLE,
        metadata JSON
    )
""")

# Batch write with Appender (20x faster than INSERT)
appender = con.appender("events")

# Write in batches (1 million rows per batch)
batch_size = 1_000_000
for i in range(1000):  # 1000 batches = 1 billion rows
    batch = generate_batch(batch_size)  # Replace with your data source
    appender.append_df(batch)
    if i % 50 == 0:
        print(f"Progress: {i*batch_size:,} / 1,000,000,000")

appender.flush()
con.close()
print("Load complete!")

# Step 1: Read and write Parquet (leveraging columnar compression)
con1 = duckdb.connect(":memory:")
con1.execute("""
    CREATE TABLE events AS
    SELECT * FROM read_csv_auto('events_2026/*.csv.gz')
""")
con1.execute("COPY events TO 'events_2026.parquet' (FORMAT PARQUET)")
con1.close()

# Step 2: Load from Parquet into DuckDB (extremely fast)
con2 = duckdb.connect("big_data.duckdb")
con2.execute("CREATE TABLE events AS SELECT * FROM 'events_2026.parquet'")
con2.close()

#!/usr/bin/env python3
"""DuckDB Billion-Record Loading Script"""
import duckdb
import os
import time

# Configuration
DATABASE = "/data/warehouse/events.duckdb"
INPUT_PATTERN = "/data/raw/events_2026/*.csv.gz"
TEMP_DIR = "/tmp/duckdb_temp"
THREADS = 8
MAX_MEMORY = "24GB"

os.makedirs(TEMP_DIR, exist_ok=True)

# Connection config
config = {
    'threads': str(THREADS),
    'max_memory': MAX_MEMORY,
    'temp_directory': TEMP_DIR,
    'preserve_insert_order': 'false',  # Don't guarantee order, faster writes
    'allow_unsigned_extensions': 'true',
}

print(f"[*] Starting DuckDB load at {time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"[*] Config: threads={THREADS}, max_memory={MAX_MEMORY}")

con = duckdb.connect(DATABASE, config=config)

# Create table structure
con.execute("""
    CREATE TABLE IF NOT EXISTS events (
        event_id BIGINT,
        user_id BIGINT,
        event_type VARCHAR,
        event_date DATE,
        amount DOUBLE,
        country VARCHAR,
        device VARCHAR,
        browser VARCHAR,
        os VARCHAR,
        session_id VARCHAR,
        created_at TIMESTAMP,
        metadata JSON
    )
""")

# Parallel read gzip CSV
start = time.time()
con.execute(f"""
    CREATE TABLE events_raw AS
    SELECT * FROM read_csv_auto('{INPUT_PATTERN}',
        parallel={THREADS},
        all_varchar=false,
        auto_detect=true
    )
""")
elapsed = time.time() - start
print(f"[*] Load completed in {elapsed:.1f}s")

# Clean up temp data
con.execute("DROP TABLE events_raw")
con.execute("VACUUM")

# Verify
row_count = con.execute("SELECT COUNT(*) FROM events").fetchone()[0]
print(f"[*] Total rows: {row_count:,}")
print(f"[*] Database size: {os.path.getsize(DATABASE) / 1024**3:.2f} GB")
print(f"[*] Completed at {time.strftime('%Y-%m-%d %H:%M:%S')}")

con.close()

# Monitor loading progress
con = duckdb.connect("big_data.duckdb")

# Check current query status
result = con.execute("SELECT * FROM duckdb_transactions()").fetchall()
for row in result:
    print(row)

# Monitor memory usage
memory_info = con.execute("""
    SELECT * FROM duckdb_memory()
""").fetchall()
for row in memory_info:
    print(row)

# View execution plan (confirm predicate pushdown)
con.execute("EXPLAIN ANALYZE SELECT * FROM events WHERE event_date >= '2026-07-01'")