import duckdb
import pandas as pd

# Create sample data
df = pd.DataFrame({
    'order_id': range(1, 100001),
    'customer_id': [i % 500 for i in range(100000)],
    'amount': [round(10 + (i * 7.3) % 990, 2) for i in range(100000)],
    'region': ['East China', 'North China', 'South China', 'Southwest', 'Northwest'][i % 5]
})

# Method 1: Use from_df (zero-copy, recommended)
con = duckdb.connect(':memory:')
con.execute("CREATE TABLE orders AS SELECT * FROM df")

# Method 2: Register as a view (fully zero-copy)
con.register('orders_view', df)
result = con.execute("SELECT region, SUM(amount) FROM orders_view GROUP BY region").fetchdf()
print(result)

# Complex analytical query
query = """
SELECT 
    customer_id,
    COUNT(*) as order_count,
    SUM(amount) as total_spent,
    AVG(amount) as avg_order_value,
    MAX(amount) as max_order
FROM orders
GROUP BY customer_id
HAVING COUNT(*) > 5
ORDER BY total_spent DESC
LIMIT 10
"""

top_customers = con.execute(query).fetchdf()
print(top_customers.to_string(index=False))

import time

# Generate larger dataset for benchmarking
large_df = pd.DataFrame({
    'id': range(1, 10_000_001),
    'value': [i * 1.5 for i in range(10_000_000)],
    'category': ['A' if i % 3 == 0 else 'B' if i % 3 == 1 else 'C' for i in range(10_000_000)]
})

# Zero-copy approach
start = time.time()
con.register('large_data', large_df)
result_zero = con.execute("SELECT category, COUNT(*), SUM(value) FROM large_data GROUP BY category").fetchdf()
zero_time = time.time() - start
print(f"Zero-copy approach: {zero_time:.3f}s")

# Traditional serialization approach
start = time.time()
result_ser = con.execute("SELECT category, COUNT(*), SUM(value) FROM large_data GROUP BY category").fetchdf()
ser_time = time.time() - start
print(f"Serialization approach: {ser_time:.3f}s")

import polars as pl
import duckdb

# Create Polars DataFrame
pl_df = pl.DataFrame({
    'product_id': range(1, 5001),
    'product_name': [f'Product{i}' for i in range(1, 5001)],
    'price': [round(10 + (i * 3.7) % 500, 2) for i in range(5000)],
    'sales': [int((i * 13) % 100) for i in range(5000)],
    'category': ['Electronics', 'Clothing', 'Food', 'Home', 'Books'][i % 5]
})

# Connect to DuckDB
con = duckdb.connect(':memory:')

# Register Polars DataFrame (zero-copy)
con.register('products', pl_df)

# Perform aggregate analysis in DuckDB
query = """
SELECT 
    category,
    COUNT(*) as product_count,
    ROUND(AVG(price), 2) as avg_price,
    SUM(sales) as total_sales
FROM products
GROUP BY category
ORDER BY total_sales DESC
"""

result = con.execute(query).fetch_arrow_table()
# Convert Arrow result back to Polars
result_pl = pl.from_arrow(result)
print(result_pl)

import polars as pl
import duckdb

# Step 1: Clean and preprocess with Polars
raw_data = pl.read_csv('/data/sales_raw.csv')
cleaned = (
    raw_data
    .filter(pl.col('amount') > 0)
    .with_columns([
        pl.col('date').str.strptime(pl.Date, '%Y-%m-%d'),
        (pl.col('amount') * pl.col('quantity')).alias('revenue'),
    ])
    .drop_nulls()
)

# Step 2: Register cleaned data in DuckDB
con = duckdb.connect(':memory:')
con.register('cleaned_sales', cleaned)

# Step 3: Complex analysis and reporting in DuckDB
report_query = """
SELECT 
    DATE_TRUNC('month', date) as month,
    category,
    COUNT(*) as order_count,
    SUM(revenue) as total_revenue,
    AVG(revenue) as avg_revenue_per_order,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY revenue) as p95_revenue
FROM cleaned_sales
WHERE date >= DATE '2025-01-01'
GROUP BY month, category
ORDER BY month, total_revenue DESC
"""

report = con.execute(report_query).fetchdf()
print(report.head(10).to_string())

import pandas as pd
import polars as pl
import duckdb
import time

# Generate 10 million row test dataset
n_rows = 10_000_000

# ---- Pandas approach ----
start = time.time()
pdf = pd.DataFrame({
    'value': [i * 1.1 for i in range(n_rows)],
    'group': [f'g{i % 100}' for i in range(n_rows)]
})
result_pd = pdf.groupby('group')['value'].agg(['sum', 'mean', 'count'])
pd_time = time.time() - start
print(f"Pandas aggregation: {pd_time:.3f}s")

# ---- Polars approach ----
start = time.time()
pldf = pl.DataFrame({
    'value': [i * 1.1 for i in range(n_rows)],
    'group': [f'g{i % 100}' for i in range(n_rows)]
})
result_pl = pldf.group_by('group').agg([
    pl.col('value').sum(),
    pl.col('value').mean(),
    pl.count()
])
pl_time = time.time() - start
print(f"Polars aggregation: {pl_time:.3f}s")

# ---- DuckDB approach ----
start = time.time()
con = duckdb.connect(':memory:')
con.register('test_data', pldf)
result_db = con.execute("""
    SELECT group, 
           SUM(value) as sum_val, 
           AVG(value) as avg_val, 
           COUNT(*) as cnt
    FROM test_data 
    GROUP BY group
""").fetchdf()
db_time = time.time() - start
print(f"DuckDB aggregation: {db_time:.3f}s")

import polars as pl
import duckdb
import pandas as pd
from datetime import datetime

def build_etl_pipeline():
    """Build an e-commerce data analysis ETL pipeline"""
    
    # === Phase 1: Polars Data Cleaning ===
    print("Phase 1: Polars data cleaning...")
    raw = pl.scan_csv('/data/orders/*.csv')
    
    cleaned = (
        raw
        .filter(
            (pl.col('order_date') >= '2025-01-01') &
            (pl.col('status') == 'completed')
        )
        .with_columns([
            (pl.col('unit_price') * pl.col('quantity')).alias('line_total'),
            pl.col('order_date').str.to_date('%Y-%m-%d').alias('order_date_parsed'),
            pl.col('city').str.to_uppercase().alias('city_upper'),
        ])
        .select([
            'order_id', 'customer_id', 'product_id',
            'order_date_parsed', 'city_upper',
            'unit_price', 'quantity', 'line_total', 'status'
        ])
    )
    
    # === Phase 2: DuckDB Storage and Analysis ===
    print("Phase 2: DuckDB analysis...")
    con = duckdb.connect('analytics.duckdb')
    
    # Write Polars LazyFrame directly to DuckDB
    con.execute("CREATE TABLE orders AS SELECT * FROM cleaned")
    
    # Multi-dimensional aggregate analysis
    daily_summary = con.execute("""
        SELECT 
            DATE_TRUNC('day', order_date_parsed) as sale_date,
            city_upper,
            COUNT(DISTINCT customer_id) as unique_customers,
            COUNT(*) as total_orders,
            SUM(line_total) as daily_revenue,
            AVG(line_total) as avg_order_value
        FROM orders
        GROUP BY sale_date, city_upper
        ORDER BY sale_date, daily_revenue DESC
    """).fetchdf()
    
    # Customer value analysis
    customer_analysis = con.execute("""
        SELECT 
            customer_id,
            COUNT(DISTINCT order_id) as total_orders,
            SUM(line_total) as lifetime_value,
            MIN(order_date_parsed) as first_order,
            MAX(order_date_parsed) as last_order,
            AVG(line_total) as avg_order_value
        FROM orders
        GROUP BY customer_id
        HAVING COUNT(DISTINCT order_id) >= 3
        ORDER BY lifetime_value DESC
        LIMIT 100
    """).fetchdf()
    
    # === Phase 3: Pandas Post-processing ===
    print("Phase 3: Pandas post-processing...")
    
    # Trend analysis
    trend = daily_summary.groupby('sale_date')['daily_revenue'].sum()
    trend.index = pd.to_datetime(trend.index)
    trend_7d = trend.rolling('7D').mean()
    
    # Export reports
    with pd.ExcelWriter('/reports/daily_report.xlsx') as writer:
        daily_summary.to_excel(writer, sheet_name='Daily Summary', index=False)
        customer_analysis.to_excel(writer, sheet_name='Customer Analysis', index=False)
    
    con.close()
    print("ETL pipeline completed!")

if __name__ == '__main__':
    build_etl_pipeline()

# ❌ Wrong: Multiple serializations/deserializations
df_pandas = pd.read_csv('big_file.csv')  # Load into memory
con.register('temp', df_pandas)           # First conversion
result = con.execute("SELECT * FROM temp").fetchdf()  # Second conversion

# ✅ Correct: Use Arrow as intermediate format
import pyarrow.parquet as pq
table = pq.read_table('data.parquet')    # Arrow format
con.execute("CREATE TABLE data AS SELECT * FROM table")  # Zero-copy
result = con.execute("SELECT ...").fetch_arrow_table()   # Zero-copy