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'")