import duckdb

# Traditional approach: DuckDB → CSV/JSON → Pandas
conn = duckdb.connect()
result = conn.execute("SELECT * FROM large_table")
df = result.fetchdf() # Internal: DuckDB → Python tuple → Pandas DataFrame

import duckdb
import pyarrow as pa

# Create in-memory database
conn = duckdb.connect()

# Load data
conn.execute("""
 CREATE TABLE sales AS SELECT * FROM read_csv_auto('sales_large.csv')
""")

# Get query results in Arrow format
result = conn.execute("""
 SELECT 
 region,
 date_trunc('month', sale_date) AS month,
 SUM(amount) AS total_sales,
 COUNT(*) AS transaction_count
 FROM sales
 WHERE sale_date >= '2025-01-01'
 GROUP BY region, month
 ORDER BY region, month
""")

# Zero-copy: return as Arrow Table
arrow_table = result.fetch_arrow_table()
print(f"Rows: {arrow_table.num_rows}, Columns: {arrow_table.num_columns}")
print(f"Schema: {arrow_table.schema}")

import pyarrow as pa
import pyarrow.dataset as ds
import duckdb

# Create a PyArrow table
table = pa.table({
 'id': pa.array([1, 2, 3, 4, 5]),
 'name': pa.array(['Alice', 'Bob', 'Charlie', 'Diana', 'Eve']),
 'score': pa.array([95.5, 87.3, 92.1, 78.9, 88.4])
})

# DuckDB queries PyArrow table directly (zero-copy!)
conn = duckdb.connect()
result = conn.execute("""
 SELECT 
 name,
 score,
 RANK() OVER (ORDER BY score DESC) AS rank
 FROM table
 WHERE score > 85
 ORDER BY score DESC
""").fetch_arrow_table()

print(result)

import duckdb
import pyarrow as pa
import pyarrow.ipc as ipc
import tempfile

# Create sample data and write to Arrow IPC file
table = pa.table({
 'timestamp': pa.array([1000, 2000, 3000, 4000]),
 'temperature': pa.array([22.5, 23.1, 21.8, 24.2])
})

with tempfile.NamedTemporaryFile(suffix='.arrow', delete=False) as f:
 writer = ipc.new_file(f, table.schema)
 writer.write_table(table)
 writer.close()
 ipc_path = f.name

# DuckDB queries Arrow IPC file directly
conn = duckdb.connect()
conn.execute(f"CREATE TABLE temps AS SELECT * FROM read_arrow_ipc('{ipc_path}')")
result = conn.execute("SELECT AVG(temperature) as avg_temp FROM temps").fetchone()
print(f"Average temperature: {result[0]}°C")

import duckdb
import pyarrow as pa
import pyarrow.csv as csv

# Stream CSV → Arrow stream → DuckDB instant query
conn = duckdb.connect()

read_options = csv.ReadOptions(block_size=1024 * 1024 * 10) # 10MB blocks
csv_stream = csv.open_csv('ultra_large.csv', read_options=read_options)

# Process batch by batch
for batch in csv_stream:
 # Zero-copy: DuckDB queries Arrow RecordBatch directly
 result = conn.execute("""
 SELECT count(*) as cnt, sum(amount) as total
 FROM batch
 WHERE status = 'completed'
 """).fetchone()
 print(f"Batch result: {result}")

import adbc_driver_duckdb.dbapi as duckdb_adbc

# Connect to DuckDB via ADBC
conn = duckdb_adbc.connect()

# Create table
conn.execute("""
 CREATE TABLE orders AS
 SELECT range AS order_id, 
 random() * 1000 AS amount,
 CASE WHEN random() > 0.5 THEN 'completed' ELSE 'pending' END AS status
 FROM range(1000000)
""")

# Query and get Arrow-formatted results
cur = conn.cursor()
cur.execute("""
 SELECT status, count(*) as cnt, sum(amount) as total
 FROM orders
 GROUP BY status
""")

# Zero-copy retrieval of Arrow data
for batch in cur.fetch_record_batches():
 print(batch)

# Python side: DuckDB processing → Arrow format output
import duckdb
import pyarrow as pa
import pyarrow.ipc as ipc

conn = duckdb.connect()

# Data cleaning
conn.execute("""
 CREATE VIEW cleaned_sales AS
 SELECT 
 sale_id,
 customer_id,
 amount,
 sale_date
 FROM read_parquet('raw_sales/*.parquet')
 WHERE amount > 0 AND customer_id IS NOT NULL
""")

# Export to Arrow IPC file (interchange format)
result = conn.execute("SELECT * FROM cleaned_sales")
arrow_table = result.fetch_arrow_table()
with open('exchange_data.arrow', 'wb') as f:
 writer = ipc.new_file(f, arrow_table.schema)
 writer.write_table(arrow_table)
 writer.close()
print(f"Exported {arrow_table.num_rows} rows to Arrow IPC file")

# Java/Rust side reads this Arrow file directly (zero-copy)

import duckdb
import pyarrow as pa
import pyarrow.parquet as pq
from sklearn.ensemble import RandomForestRegressor
import numpy as np

# DuckDB handles 1 billion log rows
conn = duckdb.connect()

# Feature engineering - all in DuckDB SQL
features = conn.execute("""
 SELECT
 customer_id,
 -- Time features
 date_diff('day', last_purchase_date, current_date) AS days_since_last_purchase,
 -- Aggregate features
 COUNT(*) AS total_orders,
 SUM(amount) AS total_spent,
 AVG(amount) AS avg_order_value,
 STDDEV(amount) AS order_amount_volatility,
 -- Categorical encoding
 CASE payment_method
 WHEN 'credit_card' THEN 1
 WHEN 'debit_card' THEN 2
 WHEN 'paypal' THEN 3
 ELSE 0
 END AS payment_method_code,
 -- Target variable
 CASE WHEN churned = true THEN 1 ELSE 0 END AS label
 FROM customer_events
 WHERE event_date >= '2025-01-01'
 GROUP BY customer_id, last_purchase_date, payment_method, churned
""").fetch_arrow_table() # Zero-copy to Arrow

# Arrow → NumPy (zero-copy for numeric types)
X = np.column_stack([
 features.column('days_since_last_purchase').to_numpy(),
 features.column('total_orders').to_numpy(),
 features.column('total_spent').to_numpy(),
 features.column('avg_order_value').to_numpy(),
 features.column('order_amount_volatility').to_numpy(),
 features.column('payment_method_code').to_numpy(),
])
y = features.column('label').to_numpy()

# Train model
model = RandomForestRegressor(n_estimators=100)
model.fit(X, y)

# Process A: Data Producer (DuckDB → Arrow → shared memory)
import duckdb
import pyarrow as pa
import pyarrow.ipc as ipc

conn = duckdb.connect()
result = conn.execute("SELECT * FROM daily_aggregation")
arrow_table = result.fetch_arrow_table()

# Write Arrow table to shared memory file
with open('/dev/shm/data.arrow', 'wb') as f:
 writer = ipc.new_file(f, arrow_table.schema)
 writer.write_table(arrow_table)
 writer.close()

# Process B: Data Consumer (millisecond read)
import pyarrow.ipc as ipc
with open('/dev/shm/data.arrow', 'rb') as f:
 reader = ipc.open_file(f)
 table = reader.read_all()
print(f"Read {table.num_rows} rows with zero copy from shared memory")

# Best practice example
conn.execute("""
 SELECT 
 -- Only needed columns
 customer_id,
 total_amount
 FROM orders
 WHERE date >= '2025-01-01' -- Predicate pushdown
 ORDER BY total_amount DESC
 LIMIT 1000
""")