Metadata-Version: 2.1
Name: turboframe
Version: 0.2.1
Summary: Fast parallel DataFrame library built on PyArrow for Delta Lake and Fabric
Author: Kushal
License: MIT
Keywords: dataframe,parallel,arrow,delta,fabric
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pyarrow >=12.0.0
Requires-Dist: numpy >=1.24.0
Provides-Extra: all
Requires-Dist: deltalake >=0.15.0 ; extra == 'all'
Requires-Dist: polars >=0.20.0 ; extra == 'all'
Requires-Dist: pandas >=2.0.0 ; extra == 'all'
Requires-Dist: openpyxl >=3.1.0 ; extra == 'all'
Provides-Extra: delta
Requires-Dist: deltalake >=0.15.0 ; extra == 'delta'
Provides-Extra: excel
Requires-Dist: pandas >=2.0.0 ; extra == 'excel'
Requires-Dist: openpyxl >=3.1.0 ; extra == 'excel'
Provides-Extra: polars
Requires-Dist: polars >=0.20.0 ; extra == 'polars'
Provides-Extra: sql
Requires-Dist: pandas >=2.0.0 ; extra == 'sql'

# TurboFrame

A fast, parallel DataFrame library built on **PyArrow**. Designed for **Delta Lake** reads and **parallel GroupBy** operations on **Microsoft Fabric** notebooks — but works everywhere Python runs.

## Why TurboFrame?

| Feature | TurboFrame | Pandas | PySpark |
|---|---|---|---|
| Parallel GroupBy | Yes (multi-core hash partition) | No (single thread) | Yes (cluster) |
| Delta Lake read | Zero-copy via Rust | Slow (via pandas) | Fast (JVM) |
| Predicate pushdown | Yes (skips row groups) | No | Yes |
| CSV read | C++ native (PyArrow) | Python-based | JVM overhead |
| Memory model | Columnar (Arrow) | Row-ish | Distributed |
| Startup time | Instant | Instant | 5-15 seconds |
| Best for | 1GB - 50GB on single node | < 1GB | > 50GB distributed |

---

## Installation

```bash
# Core (parquet, csv, json support included)
pip install turboframe

# With Delta Lake support (recommended for Fabric)
pip install turboframe[delta]

# With Excel support
pip install turboframe[excel]

# With SQL support
pip install turboframe[sql]

# Everything
pip install turboframe[all]
```

---

## Quick Start

```python
from turboframe import TurboFrame

# Create from dict
tf = TurboFrame({"region": ["East", "West", "East"], "sales": [100, 200, 150]})

# Filter + GroupBy + Sort
result = (
    tf.filter("sales > 100")
      .groupby("region")
      .agg({"sales": "sum"})
      .sort("sales_sum", ascending=False)
)

result.show()
# Output:
# region  sales_sum
#   West        200
#   East        150
```

---

## Reading Data

TurboFrame reads from **any** data source. Every reader prints timing info automatically.

### Delta Table (Fastest for Fabric)

```python
# Basic read
tf = TurboFrame.read_delta("/lakehouse/default/Tables/sales")
# [TurboFrame] Read 5,000,000 rows x 12 cols in 1.23s from delta

# Column pruning — only loads 3 columns instead of 50 (huge speedup)
tf = TurboFrame.read_delta(
    "/lakehouse/default/Tables/sales",
    columns=["region", "amount", "status"]
)

# Predicate pushdown — skips Parquet row groups that don't match
# The data never even gets loaded into memory
tf = TurboFrame.read_delta(
    "/lakehouse/default/Tables/sales",
    columns=["region", "amount", "status"],
    filters=[("amount", ">", 1000), ("status", "=", "active")]
)

# Filter with IN operator
tf = TurboFrame.read_delta(
    "/Tables/sales",
    filters=[("region", "in", ["East", "West", "North"])]
)

# OR conditions (list of lists)
tf = TurboFrame.read_delta(
    "/Tables/sales",
    filters=[
        [("region", "=", "East"), ("amount", ">", 500)],    # East AND amount>500
        [("region", "=", "West"), ("amount", ">", 1000)],   # OR West AND amount>1000
    ]
)

# Row limit — prevent out-of-memory on huge tables
tf = TurboFrame.read_delta(
    "/Tables/sales",
    columns=["region", "amount"],
    filters=[("amount", ">", 1000)],
    row_limit=500_000
)

# Time travel — read a specific Delta version
tf = TurboFrame.read_delta("/Tables/sales", version=5)

# Cloud paths
tf = TurboFrame.read_delta("abfss://container@account.dfs.core.windows.net/Tables/sales")
tf = TurboFrame.read_delta("s3://bucket/tables/sales")
```

#### Filter Operators for `read_delta`

| Operator | Example |
|---|---|
| `=` or `==` | `("region", "=", "East")` |
| `!=` | `("status", "!=", "cancelled")` |
| `>` | `("amount", ">", 1000)` |
| `>=` | `("date", ">=", "2024-01-01")` |
| `<` | `("quantity", "<", 10)` |
| `<=` | `("score", "<=", 100)` |
| `in` | `("region", "in", ["East", "West"])` |
| `not in` | `("status", "not in", ["cancelled", "draft"])` |

### Huge Tables (Batch Processing)

For tables too large to fit in memory, process in batches:

```python
# Process a 100M row table in 200K row chunks — never OOM
results = TurboFrame.read_delta_batched(
    "/Tables/huge_table",
    batch_fn=lambda batch: batch.groupby("region").agg({"amount": "sum"}),
    columns=["region", "amount"],
    filters=[("status", "=", "active")],
    batch_size=200_000
)
# [TurboFrame] Processed 100,000,000 rows in 500 batches in 45.67s

# Merge partial results
final = results.groupby("region").agg({"amount_sum": "sum"})
final.show()
```

### CSV / TSV

```python
tf = TurboFrame.read_csv("data/sales.csv")
tf = TurboFrame.read_csv("data/export.tsv", delimiter="\t")
tf = TurboFrame.read_csv("data/big.csv", columns=["id", "amount"])  # column pruning
```

### Parquet

```python
tf = TurboFrame.read_parquet("data/sales.parquet")
tf = TurboFrame.read_parquet("data/partitioned/", columns=["region", "amount"])
```

### JSON

```python
# Newline-delimited JSON (JSONL) — C++ fast reader
tf = TurboFrame.read_json("data/events.jsonl")

# Standard JSON array: [{"a": 1}, {"a": 2}]
tf = TurboFrame.read_json_array("data/records.json")
```

### Excel

```python
tf = TurboFrame.read_excel("report.xlsx")
tf = TurboFrame.read_excel("report.xlsx", sheet_name="Q4 Data")
tf = TurboFrame.read_excel("report.xlsx", columns=["region", "revenue"])
```

### SQL Database

```python
# Via pyodbc / SQLAlchemy (uses pandas internally)
import pyodbc
conn = pyodbc.connect("DRIVER={ODBC Driver 18 for SQL Server};SERVER=...")
tf = TurboFrame.read_sql("SELECT * FROM sales WHERE year = 2024", conn)

# Via ADBC — fastest (reads directly into Arrow, no pandas)
import adbc_driver_postgresql.dbapi
conn = adbc_driver_postgresql.dbapi.connect("postgresql://user:pass@host/db")
tf = TurboFrame.read_sql_arrow("SELECT * FROM sales", conn)
```

### From DataFrames

```python
# Pandas
import pandas as pd
tf = TurboFrame(pd.DataFrame({"a": [1, 2, 3]}))
tf = TurboFrame.from_pandas(my_pandas_df)

# Spark (on Fabric notebooks)
spark_df = spark.sql("SELECT * FROM lakehouse.sales")
tf = TurboFrame(spark_df)
tf = TurboFrame.from_spark(spark_df)

# Polars (zero-copy)
import polars as pl
tf = TurboFrame(pl.read_csv("data.csv"))
tf = TurboFrame.from_polars(my_polars_df)

# Dict
tf = TurboFrame({"region": ["E", "W"], "sales": [100, 200]})
tf = TurboFrame.from_dict({"region": ["E", "W"], "sales": [100, 200]})

# List of dicts
tf = TurboFrame([{"name": "Alice", "score": 95}, {"name": "Bob", "score": 87}])
tf = TurboFrame.from_records([{"name": "Alice", "score": 95}])
```

---

## Operations

### Filter

```python
tf.filter("amount > 1000")
tf.filter("region == 'East'")
tf.filter("status != 'cancelled'")

# Chain multiple filters (AND logic)
result = (
    tf.filter("amount > 1000")
      .filter("region == 'East'")
      .filter("quantity >= 10")
)
```

### Select / Drop / Rename Columns

```python
tf.select(["region", "amount", "date"])
tf.drop("internal_id")
tf.drop(["col1", "col2", "col3"])
tf.rename({"amt": "amount", "qty": "quantity"})
```

### Add / Replace Column

```python
tf.with_column("tax", [10, 20, 30])          # from list
tf.with_column("flag", "active")              # broadcast scalar
tf.with_column("doubled", some_arrow_array)   # from Arrow array
```

### Sort

```python
tf.sort("amount")                                  # ascending
tf.sort("amount", ascending=False)                 # descending
tf.sort(["region", "amount"], ascending=[True, False])  # multi-column
```

### GroupBy (Parallel)

This is the core feature. GroupBy uses **hash partitioning** across all CPU cores:

1. Data is split into N partitions (one per CPU core) by hashing the group key
2. Each core aggregates its partition independently
3. Results are concatenated (no merge needed — hash guarantees no key overlap)

```python
# Single aggregation
result = tf.groupby("region").agg({"sales": "sum"})

# Multiple aggregations
result = tf.groupby("region").agg({
    "sales": "sum",
    "quantity": "mean",
    "order_id": "count"
})

# Multiple group keys
result = tf.groupby(["region", "year"]).agg({
    "sales": "sum",
    "profit": "mean"
})

# Shorthand
tf.groupby("region").count()
tf.groupby("region").sum(["sales", "quantity"])
```

#### Supported Aggregations

| Aggregation | Description |
|---|---|
| `sum` | Sum of values |
| `mean` | Average |
| `min` | Minimum |
| `max` | Maximum |
| `count` | Count non-null values |
| `std` | Standard deviation |
| `var` | Variance |
| `median` | Approximate median |
| `nunique` | Count distinct values |

### Join

```python
sales = TurboFrame.read_delta("/Tables/sales")
regions = TurboFrame.read_delta("/Tables/regions")

sales.join(regions, on="region_id", how="inner")
sales.join(regions, on="region_id", how="left")
sales.join(regions, on="region_id", how="right")
sales.join(regions, on="region_id", how="outer")
```

### Head / Tail / Sample

```python
tf.head(10)
tf.tail(10)
tf.sample(100, seed=42)
```

### Stats & Inspection

```python
tf.shape              # (1000000, 12)
tf.columns            # ["region", "amount", ...]
tf.dtypes             # {"region": string, "amount": float64, ...}
tf.describe()         # {"amount": {"mean": 500, "min": 1, "max": 999, ...}}
tf.unique("region")   # ["East", "West", "North"]
tf.value_counts("region")
tf.count_nulls()      # {"region": 0, "amount": 5}
tf.is_empty()         # False
tf.show(20)           # pretty-print first 20 rows
```

---

## Writing Data

```python
# To Delta table
tf.to_delta("/Tables/output", mode="overwrite")
tf.to_delta("/Tables/output", mode="append")

# To files
tf.to_parquet("output.parquet")
tf.to_csv("output.csv")
tf.to_json("output.json")

# To other frameworks
pdf = tf.to_pandas()
pldf = tf.to_polars()
sdf = tf.to_spark(spark)
arrow = tf.to_arrow()
d = tf.to_dict()
```

---

## Complete Pipeline Examples

### Example 1: Sales Analysis on Fabric

```python
from turboframe import TurboFrame

# Read with column pruning + predicate pushdown
tf = TurboFrame.read_delta(
    "/lakehouse/default/Tables/transactions",
    columns=["region", "category", "amount", "date", "status"],
    filters=[("status", "=", "completed"), ("amount", ">", 0)]
)

# Parallel groupby
summary = (
    tf.groupby(["region", "category"])
      .agg({"amount": "sum", "date": "count"})
      .sort("amount_sum", ascending=False)
)

summary.show(20)

# Write results back to Delta
summary.to_delta("/lakehouse/default/Tables/sales_summary")
```

### Example 2: Spark Pre-filter + TurboFrame Analytics

```python
# Best pattern for huge tables on Fabric:
# Spark filters across the cluster, TurboFrame does fast local analytics

spark_df = spark.sql("""
    SELECT region, category, amount, quantity
    FROM lakehouse.transactions
    WHERE year = 2024 AND status = 'completed'
""")

tf = TurboFrame(spark_df)

result = (
    tf.groupby(["region", "category"])
      .agg({"amount": "sum", "quantity": "mean"})
      .sort("amount_sum", ascending=False)
)

# Back to Spark for writing
spark_result = result.to_spark(spark)
spark_result.write.format("delta").mode("overwrite").saveAsTable("summary")
```

### Example 3: CSV Analysis

```python
tf = TurboFrame.read_csv("exports/sales_2024.csv", columns=["region", "amount", "product"])

top_regions = (
    tf.filter("amount > 500")
      .groupby("region")
      .agg({"amount": "sum", "product": "count"})
      .rename({"amount_sum": "total_sales", "product_count": "num_orders"})
      .sort("total_sales", ascending=False)
)

top_regions.show()
top_regions.to_csv("reports/top_regions.csv")
```

### Example 4: Join + Aggregate

```python
orders = TurboFrame.read_delta("/Tables/orders", columns=["order_id", "customer_id", "amount"])
customers = TurboFrame.read_delta("/Tables/customers", columns=["customer_id", "segment"])

result = (
    orders.join(customers, on="customer_id", how="left")
          .groupby("segment")
          .agg({"amount": "sum", "order_id": "count"})
          .sort("amount_sum", ascending=False)
)

result.show()
```

### Example 5: Process Huge Table in Batches

```python
# 100M row table — never loads more than 200K rows at a time
results = TurboFrame.read_delta_batched(
    "/Tables/all_transactions",
    batch_fn=lambda b: b.groupby("region").agg({"amount": "sum", "qty": "count"}),
    columns=["region", "amount", "qty"],
    filters=[("year", ">=", 2023)],
    batch_size=200_000
)

# Final merge of partial results
final = results.groupby("region").agg({"amount_sum": "sum", "qty_count": "sum"})
final.show()
```

---

## Fabric Notebook Setup

### Option 1: Install from PyPI (Recommended)

```python
# Cell 1
%pip install turboframe[delta]

# Cell 2
from turboframe import TurboFrame
tf = TurboFrame.read_delta("/lakehouse/default/Tables/your_table")
```

### Option 2: Fabric Environment (Persistent — no reinstall each session)

1. Workspace -> **+ New** -> **Environment**
2. **Public Libraries** -> **Add from PyPI** -> `turboframe[delta]`
3. Click **Publish**
4. In Notebook -> **Environment** dropdown -> select your environment

### Option 3: From .whl file

```python
# Upload turboframe-2.1.0-py3-none-any.whl to lakehouse Files/
%pip install /lakehouse/default/Files/turboframe-2.1.0-py3-none-any.whl
```

---

## Performance Tips

1. **Always use `columns=`** when reading — loading 3 columns instead of 50 is 10x faster
2. **Use `filters=` in `read_delta()`** — skips entire Parquet row groups at the storage level
3. **Filter early** — apply `.filter()` before `.groupby()` to reduce data volume
4. **For huge tables** — use `read_delta_batched()` or let Spark pre-filter first
5. **Use `row_limit=`** as a safety net against OOM
6. **Worker count** — auto-detects CPU cores. Override with `max_workers=8` if needed

---

## Version

Current: **2.1.0**

## License

MIT
