diff --git a/demos/FIXING_DUCKDB_TABLE_ERROR.md b/demos/FIXING_DUCKDB_TABLE_ERROR.md
new file mode 100644
index 0000000..3992e0a
--- /dev/null
+++ b/demos/FIXING_DUCKDB_TABLE_ERROR.md
@@ -0,0 +1,181 @@
+# Fixing DuckDB "Table Does Not Exist" Error in Nested Joins
+
+## Problem
+
+When using `DataFrameHandler` with Ibis/DuckDB backend for joins, you encountered:
+```
+Catalog Error: Table with name lf_litdmfcxzm does not exist!
+```
+
+## Root Causes
+
+### 1. Multiple Handler Creation (Primary Issue)
+**Don't create new `DataFrameHandler` instances unnecessarily!** Each handler creation:
+- Triggers expensive introspection
+- Creates new temporary table references
+- Can orphan table references across operations
+
+### 2. Multiple Sessions (Secondary Issue)
+**Each new DuckDB session creates its own isolated catalog**. When you:
+
+1. Create DataFrame A in session 1 → generates temp table `lf_abc123`
+2. Create DataFrame B in session 2 → generates temp table `lf_def456`
+3. Join A and B → creates reference to both tables
+4. Create new `DataFrameHandler` on joined result → **new session 3**
+5. Try to access data → ❌ **tables from sessions 1 & 2 don't exist in session 3!**
+
+The problem occurs because:
+- `pyarrow_to_leanframe()` created a **new** DuckDB connection each time
+- Each DataFrame had references to tables in **different** DuckDB instances
+- Joined results referenced tables that were orphaned across sessions
+
+## Solutions
+
+### Solution 1: Reuse Handlers (MOST IMPORTANT!)
+
+**Create handler ONCE, reuse everywhere:**
+
+```python
+# ❌ BAD: Creating new handlers repeatedly
+def join_dataframes(df1, df2, join_path):
+ handler1 = DataFrameHandler(df1) # Creates temp tables
+ # ... extract fields ...
+ handler2 = DataFrameHandler(df1) # ❌ WASTEFUL! Re-creates everything
+ # ... more operations ...
+ handler3 = DataFrameHandler(joined) # ❌ Can fail with table refs
+
+# ✅ GOOD: Create once, reuse
+customer_handler = DataFrameHandler(customers_df) # Create ONCE (schema introspection)
+customer_handler.show_structure() # Inspect cached metadata (fast!)
+print(handler.extracted_fields) # Access cached schema info
+
+extracted_df = customer_handler.extract_nested_fields() # Compute when needed (functional!)
+join_on_nested(customer_handler, orders_df, ...) # Pass handler (metadata), not data
+```
+
+**Key principles**:
+- Handler caches **IMMUTABLE schema metadata** (fast, thread-safe)
+- Handler does **NOT cache data** (functional operations, no stale state)
+- Create handler once per DataFrame, reuse for multiple operations
+
+### Solution 2: Shared Session Pattern
+
+**Use a single shared session for all operations in your workflow:**
+
+```python
+import ibis
+import leanframe
+
+# Global shared session
+_shared_session = None
+
+def get_shared_session():
+ """Get or create a shared leanframe session for all operations."""
+ global _shared_session
+ if _shared_session is None:
+ backend = ibis.duckdb.connect()
+ _shared_session = leanframe.Session(backend=backend)
+ return _shared_session
+
+# Usage
+session = get_shared_session()
+df1 = session.DataFrame(pandas_df1) # Uses same backend
+df2 = session.DataFrame(pandas_df2) # Uses same backend
+joined = join_operation(df1, df2) # Tables exist in same catalog!
+```
+
+## Key Benefits
+
+✅ **Single catalog**: All temporary tables exist in one DuckDB instance
+✅ **Consistent references**: Joins maintain valid table references
+✅ **No orphaned tables**: All DataFrames share the same backend
+✅ **Proper lifecycle**: Tables persist for the session duration
+
+## Alternative: Materialize Before Creating New Handlers
+
+If you can't use a shared session, materialize the data:
+
+```python
+# Instead of creating new handler on joined result:
+joined_handler = DataFrameHandler(joined_df) # ❌ May fail
+
+# Materialize first:
+joined_pandas = joined_df.to_pandas() # ✅ Materializes data
+# Then work with pandas data directly
+```
+
+## Implementation in demo_nested_joins.py
+
+**Before (Broken):**
+```python
+# ❌ Created new handlers repeatedly
+def prepare_for_join(df, join_path):
+ handler = DataFrameHandler(df) # New handler every call!
+ # ...
+
+def join_operation():
+ joined_df = join_dataframes(...)
+ handler = DataFrameHandler(joined_df) # Another new handler!
+```
+
+**After (Fixed):**
+```python
+# ✅ Create handlers once, reuse them
+customer_handler = DataFrameHandler(customers_df) # Once!
+customer_handler.show_structure()
+
+# Pass handler to functions, don't recreate
+joined_df = NestedJoinHelper.join_on_nested(
+ left_handler=customer_handler, # Reuse existing handler!
+ right_df=orders_df,
+ left_join_path="profile.contact.email",
+ ...
+)
+
+# ✅ For sessions: Use shared session
+session = get_shared_session() # Reuse same session
+customers_df = session.DataFrame(customer_pandas)
+orders_df = session.DataFrame(order_pandas) # Same catalog
+```
+
+## When This Matters
+
+This issue affects:
+- ✅ **Joins across DataFrames** from different creation points
+- ✅ **Creating handlers on joined results**
+- ✅ **Any multi-step DataFrame operations** in DuckDB
+- ❌ Not an issue with BigQuery backend (server-side tables)
+
+## Best Practices
+
+### Universal (DuckDB & BigQuery):
+1. **✅ Create handlers ONCE per DataFrame** - Don't recreate unnecessarily
+2. **✅ Pass handlers as parameters** - Reuse existing instances in functions
+3. **✅ Avoid creating handlers on intermediate results** - Use the original handler's extracted_df instead
+
+### DuckDB Specific:
+4. **✅ Use shared session** - One session for all related DataFrames
+5. **✅ Materialize if needed** - Use `.to_pandas()` before creating new handlers on complex results
+
+### BigQuery Production:
+- Handler reuse still matters for **performance** (avoid re-introspection)
+- Session isolation less critical (server-side tables)
+- But shared session is still **good practice** for consistency
+
+## Performance Impact
+
+**Creating unnecessary handlers costs:**
+- 🐌 Re-introspection of schema (can be slow on large nested structures)
+- 🐌 DuckDB temp table management overhead
+- 💾 Memory for duplicate schema metadata
+
+**Handler reuse gives:**
+- 🚀 Instant access to cached schema metadata
+- 🚀 No redundant schema analysis
+- 🚀 Clean table reference management
+- ✨ Better code clarity
+
+**Note on data operations:**
+- `extract_nested_fields()` is a **functional operation** - always computes fresh results
+- No data caching = no stale state, safe for parallel operations
+- Only schema metadata is cached (immutable, thread-safe)
\ No newline at end of file
diff --git a/demos/NESTED_JOINS_STRATEGY.md b/demos/NESTED_JOINS_STRATEGY.md
new file mode 100644
index 0000000..28c60b3
--- /dev/null
+++ b/demos/NESTED_JOINS_STRATEGY.md
@@ -0,0 +1,233 @@
+## Joining on Nested Columns: Strategy & Implementation Guide
+
+### Problem: How to join DataFrames on nested columns at arbitrary depth?
+
+You have nested structures in BigQuery/Ibis and want to join on deeply nested fields like `customer.profile.contact.email` with regular columns like `order.customer_email`.
+
+---
+
+## 🏗️ NEW ARCHITECTURE: NestedHandler + DataFrameHandler
+
+We've refactored to a clean two-class architecture following Single Responsibility Principle:
+
+### DataFrameHandler (Individual Wrapper)
+- Wraps **ONE** leanframe DataFrame
+- Introspects and caches schema metadata
+- Provides extraction methods for nested fields
+- Functional operations (no data caching)
+
+### NestedHandler (Orchestrator)
+- Manages **MULTIPLE** DataFrameHandler instances
+- Coordinates cross-DataFrame operations (joins, etc.)
+- Fluent API: `add()` → `join_on_nested()`
+- Clean separation of concerns
+
+### Quick Example:
+```python
+from leanframe.core.nested_handler import NestedHandler
+
+# Create orchestrator
+handler = NestedHandler()
+
+# Add DataFrames (creates DataFrameHandler automatically)
+handler.add("customers", customers_df)
+handler.add("orders", orders_df)
+
+# Perform join using names
+joined_df = handler.join_on_nested(
+ left="customers",
+ right="orders",
+ left_path="profile.contact.email",
+ right_column="customer_email",
+ how="inner"
+)
+
+# Result can be added back for chaining
+handler.add("customer_orders", joined_df)
+```
+
+---
+
+## ✅ RECOMMENDED APPROACH: Selective Flattening
+
+### Core Strategy:
+1. **Keep data nested by default** (efficient storage/transfer)
+2. **Extract only join columns** (minimal flattening)
+3. **Use NestedHandler for orchestration** (manages multiple DataFrames)
+4. **DataFrameHandler for auto-discovery** (handles arbitrary depth)
+5. **Join on flattened keys** (BigQuery-compatible)
+6. **Preserve nested structure in results** (best of both worlds)
+
+### Why This Works Best:
+- ✅ **BigQuery Native**: Works with BigQuery's SQL engine
+- ✅ **Efficient**: Only extracts what's needed for joins
+- ✅ **Flexible**: Handles any nesting depth dynamically
+- ✅ **Scalable**: Avoids massive column explosion
+- ✅ **Ibis Compatible**: Uses standard Ibis join operations
+
+---
+
+## 🔧 Implementation Pattern
+
+### 1. Dynamic Structure Discovery
+```python
+from leanframe.core.frame import DataFrameHandler
+
+# Create handler once - introspects schema, caches metadata
+handler = DataFrameHandler(nested_df)
+
+# Fast metadata access (cached, no computation)
+print(handler.extracted_fields)
+# {'profile.contact.email': 'profile_contact_email',
+# 'profile.contact.region': 'profile_contact_region', ...}
+
+handler.show_structure() # Fast - displays cached metadata
+
+# Key Design Principle:
+# - Handler caches IMMUTABLE schema metadata (thread-safe, efficient)
+# - Handler does NOT cache data (functional operations, no stale state)
+# - Create once per DataFrame, reuse for multiple operations
+```
+
+### 2. Selective Extraction for Joins
+```python
+class NestedJoinHelper:
+ @staticmethod
+ def prepare_for_join(handler: DataFrameHandler, join_path: str, prefix: str = ""):
+ """Extract only the nested field needed for joining.
+
+ Args:
+ handler: DataFrameHandler instance (reuse to avoid re-introspection!)
+ join_path: Nested path like "profile.contact.email"
+ prefix: Optional prefix for extracted column name
+
+ Design Note:
+ - Handler caches schema metadata (fast lookup in extracted_fields)
+ - extract_nested_fields() is FUNCTIONAL (no data caching)
+ - Reuse handler across multiple operations for efficiency
+ """
+ if join_path in handler.extracted_fields:
+ extracted_name = handler.extracted_fields[join_path] # Fast metadata lookup
+ final_name = f"{prefix}_{extracted_name}" if prefix else extracted_name
+
+ # Functional operation - computes fresh result each time
+ extracted_df = handler.extract_nested_fields()
+ return extracted_df, final_name
+ else:
+ raise ValueError(f"Join path '{join_path}' not found")
+```
+
+### 3. Ibis Join Operation
+```python
+def join_on_nested(left_handler, right_df, left_nested_path, right_column, how="inner"):
+ """Join on nested column by extracting it first.
+
+ Args:
+ left_handler: DataFrameHandler instance (pass existing, don't create new!)
+ right_df: DataFrame to join with
+ left_nested_path: Nested path to extract and join on
+ right_column: Flat column name in right_df
+ how: Join type ("inner", "left", "outer", etc.)
+
+ Design Pattern:
+ - Pass handler as parameter (reuse existing instance)
+ - Extract fields functionally (no cached state)
+ - Efficient: Schema already analyzed in handler
+ """
+ # Prepare left side - uses cached metadata, computes fresh data
+ left_prepared, left_join_col = NestedJoinHelper.prepare_for_join(
+ left_handler, left_nested_path, "left"
+ )
+
+ # Join using Ibis
+ joined_table = left_prepared._data.join(
+ right_df._data,
+ predicates=[(left_join_col, right_column)],
+ how=how
+ )
+
+ return DataFrame(joined_table)
+```
+
+### 4. Complete Example
+```python
+# Create handler once - caches schema metadata
+customer_handler = DataFrameHandler(customers_df)
+customer_handler.show_structure() # Fast - uses cached metadata
+
+# Join using handler - functional data extraction
+joined_df = NestedJoinHelper.join_on_nested(
+ left_handler=customer_handler, # Pass handler (not DataFrame!)
+ right_df=orders_df, # Has customer_email
+ left_nested_path="profile.contact.email", # Uses cached metadata lookup
+ right_join_column="customer_email",
+ how="inner"
+)
+
+# Reuse same handler for multiple operations - no re-introspection needed!
+# Result preserves nested structures + adds flat join columns
+```
+
+---
+
+## 📊 Strategy Comparison
+
+| Approach | Pros | Cons | BigQuery Fit |
+|----------|------|------|--------------|
+| **🚀 Selective Flattening** | ✅ Efficient
✅ Flexible
✅ Preserves structure | ⚠️ Requires extraction logic | ⭐⭐⭐⭐⭐ |
+| **🐌 Full Flattening** | ✅ Simple joins | ❌ Column explosion
❌ Hard to re-nest | ⭐⭐⭐ |
+| **🔧 Native Struct Joins** | ✅ No flattening | ❌ Limited BigQuery support
❌ Complex syntax | ⭐⭐ |
+
+---
+
+## 🎯 Key Benefits of Our Approach
+
+### For Your Use Case:
+1. **DataFrameHandler discovers structure automatically** - works with any nesting
+2. **Selective extraction** - only flatten join keys, keep rest nested
+3. **BigQuery native** - uses standard SQL joins under the hood
+4. **Preserves performance** - minimal data movement and processing
+5. **Handles arbitrary depth** - works with `level1.level2.level3.field` joins
+
+### Real-World Example:
+```python
+# Customer data: profile.contact.email (nested 2 levels deep)
+# Order data: customer_email (flat)
+# Result: Efficient join preserving customer's nested profile structure
+```
+
+---
+
+## 📝 Implementation Notes
+
+### When to Use This Pattern:
+- ✅ Joining on nested fields of any depth
+- ✅ BigQuery/Ibis backend
+- ✅ Want to preserve nested structures
+- ✅ Need flexible, dynamic solutions
+
+### Alternative Simple Cases:
+- **Single-level nesting**: Can use direct Ibis field access: `table["struct_col"]["field"]`
+- **Known structure**: Can pre-extract specific fields manually
+- **Full flattening OK**: Use `handler.extract_nested_fields()` for all operations
+
+### Performance Considerations:
+- **Schema introspection**: One-time cost when creating handler (cached as metadata)
+- **Data extraction**: Functional operation - computes fresh each time (no stale state)
+- **Handler reuse**: Avoid re-introspection by passing handlers as parameters
+- **Storage efficiency**: Nested data stays compact
+- **Query efficiency**: BigQuery optimizes struct field access
+- **Network efficiency**: Less data transfer vs full flattening
+- **Thread safety**: Metadata caching is safe; no data caching = no race conditions
+
+---
+
+## 🔍 Demo Results
+
+The demo shows successful:
+- ✅ **Auto-discovery** of nested structures (`profile.contact.email`, `order_details.shipping.method`)
+- ✅ **Dynamic extraction** of join keys without manual field specification
+- ✅ **Join execution** using standard Ibis operations
+- ✅ **Structure preservation** of non-join nested columns
+
+**Recommendation**: Use this selective flattening pattern with DataFrameHandler for robust, efficient nested joins in BigQuery environments.
\ No newline at end of file
diff --git a/demos/README.md b/demos/README.md
new file mode 100644
index 0000000..8f27300
--- /dev/null
+++ b/demos/README.md
@@ -0,0 +1,139 @@
+# LeanFrame Demos
+
+This directory contains demonstration scripts for LeanFrame's capabilities.
+
+## 📚 Main Demos (Start Here!)
+
+### 🎯 **demo_nested_handler_backend.py** - COMPREHENSIVE DEMO ⭐
+
+**The definitive guide to NestedHandler with backend integration.**
+
+This demo shows TWO approaches for different user needs:
+
+**APPROACH 1: Convenience `join()` Method (Regular Users)**
+- ✅ Simple, intuitive API
+- ✅ Automatic nested field extraction
+- ✅ Multi-table, multi-column joins
+- ✅ All join types supported (inner, left, right, outer, cross)
+- ✅ No need to understand Ibis internals
+
+**APPROACH 2: `prepare()` + Direct Ibis (Power Users)**
+- ⚡ Maximum flexibility
+- ⚡ Full control over SQL operations
+- ⚡ WHERE, HAVING, window functions
+- ⚡ Complex multi-table queries
+- ⚡ Everything BigQuery/SQL supports
+
+**Also covers:**
+- Backend integration (DuckDB)
+- Table push/pull operations
+- Backend reference management
+- DataFrame lifecycle (in-memory ↔ backend)
+- Join provenance tracking
+
+**Run it:**
+```bash
+python demos/demo_nested_handler_backend.py
+```
+
+---
+
+## 📖 Additional Examples
+
+### demo_flexible_joins.py
+**Status:** Consolidated into `demo_nested_handler_backend.py`
+
+Contains additional advanced examples:
+- Three-table joins
+- Window functions (ROW_NUMBER, RANK, etc.)
+- Complex GROUP BY queries
+- Various join types
+
+**Note:** For most use cases, see `demo_nested_handler_backend.py` instead.
+
+### demo_dynamic_nested_handler.py
+Shows dynamic creation and manipulation of nested DataFrames.
+
+---
+
+## 🚀 Quick Start Guide
+
+### For Regular Users (Simple Joins)
+```python
+from leanframe.core.nested_handler import NestedHandler
+
+# Create handler
+handler = NestedHandler()
+handler.add("customers", customers_df)
+handler.add("orders", orders_df)
+
+# Simple join - handles nested fields automatically
+# Use natural dot notation for nested paths (converted internally to underscores)
+result = handler.join(
+ tables={"c": "customers", "o": "orders"},
+ on=[("c", "profile.contact.email", "o", "shipping.recipient.email")],
+ how="inner"
+)
+```
+
+### For Power Users (Complex SQL)
+```python
+# Extract nested fields first
+customers_flat = handler.prepare("customers")
+orders_flat = handler.prepare("orders")
+
+# Use direct Ibis for full SQL power
+joined = customers_flat._data.join(
+ orders_flat._data,
+ predicates=[customers_flat._data.email == orders_flat._data.customer_email],
+ how="inner"
+)
+
+# Add WHERE clause
+filtered = joined.filter(joined.amount > 100) # type: ignore
+
+# Add GROUP BY
+summary = filtered.group_by("customer_id").aggregate(
+ total=filtered.amount.sum(), # type: ignore
+ count=filtered.order_id.count() # type: ignore
+)
+
+from leanframe.core.frame import DataFrame
+result = DataFrame(summary)
+```
+
+---
+
+## 🧪 Testing
+
+All features demonstrated here are covered by the test suite:
+- `tests/unit/nested_data/test_prepare_method.py` - Tests for `prepare()`
+- `tests/unit/nested_data/test_join_convenience.py` - Tests for `join()`
+- Other nested data tests for core functionality
+
+---
+
+## 📝 Documentation
+
+For more details on the architecture and design decisions, see:
+- `NESTED_JOINS_STRATEGY.md` - Join implementation strategy
+- API documentation (coming soon)
+
+---
+
+## ❓ Questions?
+
+**Which demo should I run?**
+→ Start with `demo_nested_handler_backend.py` - it's comprehensive!
+
+**I just want simple joins, what do I use?**
+→ Use the `join()` convenience method (APPROACH 1)
+
+**I need complex SQL queries, what do I use?**
+→ Use `prepare()` + direct Ibis (APPROACH 2)
+
+**Can I mix both approaches?**
+→ Yes! Use `join()` for simple cases, `prepare()` for complex ones.
+
+**Where are the tests?**
+→ See `tests/unit/nested_data/` for comprehensive test coverage.
diff --git a/demos/demo_flexible_joins.py b/demos/demo_flexible_joins.py
new file mode 100644
index 0000000..e472d8d
--- /dev/null
+++ b/demos/demo_flexible_joins.py
@@ -0,0 +1,303 @@
+#!/usr/bin/env python3
+"""
+Demo: Flexible SQL-like Joins with NestedHandler
+
+⚠️ NOTE: This demo has been CONSOLIDATED into demo_nested_handler_backend.py
+ See that file for the LATEST and most comprehensive examples of:
+ - APPROACH 1: Convenience join() method (regular users)
+ - APPROACH 2: prepare() + Ibis (power users with full SQL)
+
+This file is kept for reference showing additional advanced examples:
+- Three-table joins
+- Window functions
+- Complex GROUP BY queries
+- Different join types
+
+For the main demo, see: demos/demo_nested_handler_backend.py
+
+IMPORTANT: Column naming convention
+- User-facing: Use dot notation for nested paths (e.g., "profile.contact.email")
+- Internal: After prepare(), columns use underscores (e.g., "profile_contact_email")
+- In this demo: We show the AFTER prepare() state, so you see underscore names
+
+---
+
+Original Description:
+This demonstrates the NEW architecture where:
+1. NestedHandler prepares DataFrames (extracts nested fields)
+2. Ibis handles all SQL operations (joins, WHERE, HAVING, windows, etc.)
+3. No artificial limitations - full SQL flexibility!
+
+Key Benefits:
+- Join ANY number of tables
+- Use ANY join type (inner, left, right, outer, cross, semi, anti)
+- Add WHERE clauses, HAVING, window functions
+- Multi-column joins
+- Complex predicates
+- Everything SQL/BigQuery supports!
+"""
+
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import ibis
+import pandas as pd
+import leanframe
+from leanframe.core.frame import DataFrame
+from leanframe.core.nested_handler import NestedHandler
+from demos.utils.create_nested_data import (
+ create_customers_for_join,
+ create_orders_for_join
+)
+
+
+def main():
+ print("=" * 70)
+ print("🚀 Flexible SQL-like Joins with NestedHandler")
+ print("=" * 70)
+
+ # Setup
+ backend = ibis.duckdb.connect()
+ session = leanframe.Session(backend=backend)
+ nested = NestedHandler()
+
+ # ===================================================================
+ # STEP 1: Prepare Data
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 1: Prepare Test Data")
+ print("=" * 70)
+
+ # Create nested DataFrames
+ customers_df = create_customers_for_join()
+ orders_df = create_orders_for_join()
+
+ # Add third table - products
+ products_pd = pd.DataFrame({
+ 'product_id': [1, 2, 3],
+ 'name': ['Widget', 'Gadget', 'Doohickey'],
+ 'category': ['Electronics', 'Electronics', 'Hardware'],
+ 'price': [29.99, 149.99, 9.99]
+ })
+ products_df = session.DataFrame(products_pd)
+
+ # Add to NestedHandler
+ nested.add("customers", session.DataFrame(customers_df.to_pandas()))
+ nested.add("orders", session.DataFrame(orders_df.to_pandas()))
+ nested.add("products", products_df)
+
+ print("\n✅ Added 3 tables:")
+ print(" - customers: nested profile.contact.email")
+ print(" - orders: nested shipping.recipient.email")
+ print(" - products: flat structure")
+
+ # ===================================================================
+ # STEP 2: Simple Two-Table Join
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 2: Simple Two-Table Join (NEW Approach)")
+ print("=" * 70)
+
+ print("\n📋 OLD way (limited, deprecated):")
+ print(" result = nested.join_on_both_nested(...)")
+ print(" ❌ Only 2 tables")
+ print(" ❌ Limited join types")
+ print(" ❌ No WHERE, HAVING, etc.")
+
+ print("\n📋 NEW way (full SQL power):")
+ print(" 1. Prepare: Extract nested fields")
+ print(" 2. Join: Use Ibis directly")
+ print(" 3. Query: Add WHERE, GROUP BY, etc.")
+
+ # Step 1: Prepare - extract nested fields
+ print("\n1️⃣ Prepare DataFrames (extract nested fields):")
+ customers_flat = nested.prepare("customers")
+ orders_flat = nested.prepare("orders")
+
+ print(f" Customers: {len(customers_flat.columns)} columns")
+ print(f" Orders: {len(orders_flat.columns)} columns")
+
+ # Step 2: Join using Ibis
+ print("\n2️⃣ Join using Ibis (full SQL flexibility):")
+ # Note: After prepare(), nested fields are flattened with underscores
+ # e.g., "profile.contact.email" → "profile_contact_email" (column name)
+ # For simple joins, you could use: nested.join(..., on=[("c", "profile.contact.email", ...)])
+ joined = customers_flat._data.join(
+ orders_flat._data,
+ predicates=[
+ customers_flat._data.profile_contact_email ==
+ orders_flat._data.shipping_recipient_email
+ ],
+ how="inner"
+ )
+
+ result = DataFrame(joined)
+ print(f" ✅ Joined: {len(result.columns)} columns, {joined.count().execute()} rows")
+
+ # Step 3: Add WHERE clause
+ print("\n3️⃣ Add WHERE clause (filter results):")
+ filtered = joined.filter(joined.amount > 100) # type: ignore
+ result_filtered = DataFrame(filtered) # noqa
+ print(f" ✅ Filtered (amount > 100): {filtered.count().execute()} rows")
+
+ # ===================================================================
+ # STEP 3: Three-Table Join
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 3: Three-Table Join (IMPOSSIBLE with old methods!)")
+ print("=" * 70)
+
+ print("\n🎯 Goal: customers ⋈ orders ⋈ products")
+ print(" Old methods: ❌ Can't do this!")
+ print(" New approach: ✅ Easy!")
+
+ # Modify orders to have product_id
+ orders_with_products_pd = orders_flat.to_pandas()
+ orders_with_products_pd['product_id'] = [1, 2, 1, 3, 2] # Match order IDs
+ orders_with_products = session.DataFrame(orders_with_products_pd)
+
+ print("\n1️⃣ First join: customers ⋈ orders")
+ step1 = customers_flat._data.join(
+ orders_with_products._data,
+ predicates=[
+ customers_flat._data.profile_contact_email ==
+ orders_with_products._data.shipping_recipient_email
+ ],
+ how="inner"
+ )
+
+ print("\n2️⃣ Second join: (customers ⋈ orders) ⋈ products")
+ step2 = step1.join(
+ products_df._data,
+ predicates=[step1.product_id == products_df._data.product_id],
+ how="inner"
+ )
+
+ three_table_result = DataFrame(step2)
+ print(f" ✅ Three-table join: {len(three_table_result.columns)} columns")
+ print(f" ✅ Result: {step2.count().execute()} rows")
+
+ # ===================================================================
+ # STEP 4: Complex Query with GROUP BY
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 4: Complex Query (GROUP BY + Aggregation)")
+ print("=" * 70)
+
+ print("\n🎯 Calculate total sales per customer")
+
+ # Group and aggregate
+ grouped = step2.group_by("customer_id").aggregate(
+ total_amount=step2.amount.sum(), # type: ignore
+ order_count=step2.order_id.count(), # type: ignore
+ avg_price=step2.price.mean() # type: ignore
+ )
+
+ grouped_result = DataFrame(grouped)
+ print(f" ✅ Grouped by customer: {grouped.count().execute()} customers")
+ print("\n Sample results:")
+ sample = grouped_result.to_pandas().head(3)
+ for _, row in sample.iterrows():
+ print(f" Customer {row['customer_id']}: "
+ f"{row['order_count']} orders, "
+ f"${row['total_amount']:.2f} total, "
+ f"${row['avg_price']:.2f} avg")
+
+ # ===================================================================
+ # STEP 5: Different Join Types
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 5: Different Join Types (ALL supported!)")
+ print("=" * 70)
+
+ print("\n✅ Supported join types:")
+ join_types = ["inner", "left", "right", "outer", "cross", "semi", "anti"]
+ for jt in join_types:
+ print(f" • {jt}")
+
+ print("\n📝 Example: LEFT JOIN (keep all customers)")
+ left_join = customers_flat._data.join(
+ orders_flat._data,
+ predicates=[
+ customers_flat._data.profile_contact_email ==
+ orders_flat._data.shipping_recipient_email
+ ],
+ how="left"
+ )
+ left_result = DataFrame(left_join)
+ print(f" ✅ LEFT JOIN: {left_result._data.count().execute()} rows (includes customers without orders)")
+
+ # ===================================================================
+ # STEP 6: Window Functions
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 6: Window Functions (BigQuery/SQL feature)")
+ print("=" * 70)
+
+ print("\n🎯 Add row numbers within each customer's orders")
+
+ # Use Ibis window functions
+ window = ibis.window(group_by=step2.customer_id, order_by=step2.amount.desc())
+ with_rank = step2.select(
+ step2.customer_id,
+ step2.order_id,
+ step2.amount,
+ order_rank=ibis.row_number().over(window)
+ )
+
+ rank_result = DataFrame(with_rank)
+ print(f" ✅ Added ranking: {rank_result._data.count().execute()} rows")
+ print("\n Sample with rankings:")
+ sample = rank_result.to_pandas().head(5)
+ for _, row in sample.iterrows():
+ print(f" Customer {row['customer_id']}, "
+ f"Order {row['order_id']}: "
+ f"${row['amount']:.2f} (rank #{row['order_rank']})")
+
+ # ===================================================================
+ # Summary
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("✅ Summary: Why This Architecture is Better")
+ print("=" * 70)
+
+ print("\n🎯 Key Advantages:")
+ print(" 1. ✅ Join ANY number of tables (not just 2)")
+ print(" 2. ✅ ALL join types: inner, left, right, outer, cross, semi, anti")
+ print(" 3. ✅ Multi-column joins")
+ print(" 4. ✅ Complex predicates (AND, OR, etc.)")
+ print(" 5. ✅ WHERE clauses")
+ print(" 6. ✅ GROUP BY + aggregations (SUM, COUNT, AVG, etc.)")
+ print(" 7. ✅ HAVING clauses")
+ print(" 8. ✅ Window functions (ROW_NUMBER, RANK, LAG, LEAD, etc.)")
+ print(" 9. ✅ UNION, INTERSECT, EXCEPT")
+ print(" 10. ✅ Everything BigQuery/SQL supports!")
+
+ print("\n📐 Design Pattern:")
+ print(" ┌─────────────────────────────────────────────┐")
+ print(" │ NestedHandler: Prepare nested data │")
+ print(" │ (Extract nested fields → flat columns) │")
+ print(" └─────────────┬───────────────────────────────┘")
+ print(" │")
+ print(" ▼")
+ print(" ┌─────────────────────────────────────────────┐")
+ print(" │ Ibis/Backend: Handle ALL SQL operations │")
+ print(" │ (Joins, WHERE, GROUP BY, windows, etc.) │")
+ print(" └─────────────────────────────────────────────┘")
+
+ print("\n🔧 Simple Workflow:")
+ print(" 1. handler.prepare('table_name') # Extract nested → flat")
+ print(" 2. Use Ibis operations directly # Full SQL power")
+ print(" 3. Result is ready for BigQuery # Or any backend")
+
+ print("\n💡 Future: handler.join() convenience method")
+ print(" Will wrap this pattern for common cases")
+ print(" But you ALWAYS have direct Ibis access for complex queries!")
+
+ backend.disconnect()
+ print("\n🔌 Disconnected from database")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/demos/demo_nested_handler_backend.py b/demos/demo_nested_handler_backend.py
new file mode 100644
index 0000000..7c96f5f
--- /dev/null
+++ b/demos/demo_nested_handler_backend.py
@@ -0,0 +1,365 @@
+#!/usr/bin/env python3
+"""
+Comprehensive Demo: NestedHandler with Backend Integration
+
+This demo shows TWO approaches for working with nested data:
+
+APPROACH 1: Convenience join() method (Regular Users)
+- Simple, intuitive API for common join cases
+- Automatic nested field extraction
+- Multi-table joins with clean syntax
+
+APPROACH 2: prepare() + Direct Ibis (Power Users)
+- Maximum flexibility for complex queries
+- WHERE, HAVING, window functions, CTEs
+- Full SQL power when you need it
+
+Also demonstrates:
+- Backend integration (load/save from DuckDB)
+- Backend reference management (table qualifiers)
+- Lineage tracking
+- In-memory → backend → in-memory lifecycle
+
+Key Architectural Principles:
+- One NestedHandler = One Backend Connection
+- DataFrameHandler owns its backend reference (table_qualifier property)
+- NestedHandler prepares data, Ibis executes SQL
+"""
+
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import ibis
+import leanframe
+from leanframe.core.nested_handler import NestedHandler
+from demos.utils.create_nested_data import (
+ create_customers_for_join,
+ create_orders_for_join
+)
+
+
+def main():
+ print("=" * 70)
+ print("🚀 Comprehensive NestedHandler Demo with Backend Integration")
+ print("=" * 70)
+
+ # ===================================================================
+ # STEP 1-2: Data Preparation - Create DataFrames and Push to DB
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 1-2: Data Preparation")
+ print("=" * 70)
+
+ print("\n📊 Creating nested DataFrames...")
+ customers_df = create_customers_for_join()
+ orders_df = create_orders_for_join()
+
+ print("✅ Created two DataFrames:")
+ print(" - Customers: email at profile.contact.email (nested 2 levels)")
+ print(" - Orders: email at shipping.recipient.email (nested 2 levels, different path!)")
+
+ # Connect to local DuckDB
+ print("\n🔌 Connecting to local DuckDB...")
+ backend = ibis.duckdb.connect()
+ session = leanframe.Session(backend=backend)
+
+ # Push to database
+ print("\n💾 Pushing tables to local database...")
+
+ # Convert to pandas temporarily for database insertion
+ customers_pandas = customers_df.to_pandas()
+ orders_pandas = orders_df.to_pandas()
+
+ # Register tables in DuckDB
+ backend.create_table("customers", customers_pandas, overwrite=True)
+ backend.create_table("orders", orders_pandas, overwrite=True)
+
+ print("✅ Tables created in database:")
+ print(" - customers (4 records)")
+ print(" - orders (5 records)")
+
+ # ===================================================================
+ # STEP 3: Create NestedHandler
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 3: Create NestedHandler Orchestrator")
+ print("=" * 70)
+
+ nested = NestedHandler()
+ print("✅ Created NestedHandler instance")
+ print(f" Current state: {nested}")
+
+ # ===================================================================
+ # STEP 4-5: Load from DB and Add to Handler
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 4-5: Load Tables from Database → DataFrameHandlers → NestedHandler")
+ print("=" * 70)
+
+ print("\n📥 Loading 'customers' table from database...")
+ customers_table = backend.table("customers")
+ customers_lf = session.DataFrame(customers_table.to_pandas())
+
+ # Add to NestedHandler with table qualifier for lineage
+ nested.add(
+ name="customers",
+ df=customers_lf,
+ table_qualifier="local_duckdb.main.customers" # Full qualifier
+ )
+
+ print("\n📥 Loading 'orders' table from database...")
+ orders_table = backend.table("orders")
+ orders_lf = session.DataFrame(orders_table.to_pandas())
+
+ nested.add(
+ name="orders",
+ df=orders_lf,
+ table_qualifier="local_duckdb.main.orders"
+ )
+
+ print(f"\n✅ NestedHandler now manages: {nested}")
+
+ # ===================================================================
+ # STEP 5A: APPROACH 1 - Convenience join() Method (Regular Users)
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 5A: APPROACH 1 - Convenience join() Method")
+ print("=" * 70)
+
+ print("\n🎯 For Regular Users: Simple, intuitive API")
+ print(" ✅ Automatic nested field extraction")
+ print(" ✅ Clean multi-table join syntax")
+ print(" ✅ No need to know about prepare() or Ibis")
+
+ print("\n🔗 Joining customers and orders on email...")
+ print(" - Customers: profile.contact.email (nested)")
+ print(" - Orders: shipping.recipient.email (nested)")
+
+ # Use convenience join() method - handles nested extraction automatically!
+ # Note: You can use dot notation naturally - it gets converted to underscores internally
+ joined_df = nested.join(
+ tables={"c": "customers", "o": "orders"},
+ on=[("c", "profile.contact.email", "o", "shipping.recipient.email")],
+ how="inner"
+ )
+
+ print("\n✅ Join complete!")
+ print(f" Result DataFrame has {len(joined_df.columns)} columns")
+
+ # Add result to handler for tracking
+ nested.add("customer_orders_simple", joined_df,
+ table_qualifier="joined(local_duckdb.main.customers⋈local_duckdb.main.orders)")
+ print(" Added to handler: 'customer_orders_simple'")
+ print(f" NestedHandler now has: {len(nested)} DataFrames")
+
+ # ===================================================================
+ # STEP 5B: APPROACH 2 - prepare() + Ibis (Power Users)
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 5B: APPROACH 2 - prepare() + Direct Ibis")
+ print("=" * 70)
+
+ print("\n⚡ For Power Users: Maximum flexibility")
+ print(" ✅ Full control over SQL operations")
+ print(" ✅ WHERE, HAVING, window functions")
+ print(" ✅ Complex multi-table queries")
+
+ print("\n🔧 Step 1: Prepare DataFrames (extract nested fields)")
+ customers_flat = nested.prepare("customers", verbose=False)
+ orders_flat = nested.prepare("orders", verbose=False)
+ print(f" Customers prepared: {len(customers_flat.columns)} columns")
+ print(f" Orders prepared: {len(orders_flat.columns)} columns")
+
+ print("\n🔧 Step 2: Use direct Ibis operations")
+ # Now we have full Ibis power!
+ c_ibis = customers_flat._data
+ o_ibis = orders_flat._data
+
+ # Join on email
+ joined_ibis = c_ibis.join(
+ o_ibis,
+ predicates=[c_ibis.profile_contact_email == o_ibis.shipping_recipient_email],
+ how="inner"
+ )
+
+ # Can add WHERE clause (filter high-value orders)
+ print(" Adding WHERE clause: amount > 100")
+ # Note: Ibis operations - type checker may show warnings but code works
+ filtered_ibis = joined_ibis.filter(joined_ibis.amount > 100) # type: ignore
+
+ # Can add GROUP BY and aggregations
+ print(" Adding GROUP BY: customer_id")
+ grouped_ibis = filtered_ibis.group_by("customer_id").aggregate(
+ total_amount=filtered_ibis.amount.sum(), # type: ignore
+ order_count=filtered_ibis.order_id.count() # type: ignore
+ )
+
+ from leanframe.core.frame import DataFrame as LeanDF
+ power_user_result = LeanDF(grouped_ibis)
+
+ print("\n✅ Power user query complete!")
+ print(f" Result has {len(power_user_result.columns)} columns")
+ print(" With WHERE, GROUP BY, and aggregations")
+
+ # Add to handler
+ nested.add("customer_summary", power_user_result,
+ table_qualifier="aggregated(customer_orders)")
+ print(" Added to handler: 'customer_summary'")
+ print(f" NestedHandler now has: {len(nested)} DataFrames")
+
+ # ===================================================================
+ # STEP 6: Store Results in Database
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 6: Store Results Back to Database")
+ print("=" * 70)
+
+ # Store the simple join result
+ print("\n💾 Writing convenience join result to database...")
+ simple_result_handler = nested.get("customer_orders_simple")
+ simple_pandas = simple_result_handler.original_df.to_pandas()
+ backend.create_table("customer_orders_simple", simple_pandas, overwrite=True)
+ print("✅ Created table: customer_orders_simple")
+ print(f" Rows: {len(simple_pandas)}")
+
+ # Store the power user result
+ print("\n� Writing power user summary to database...")
+ summary_pandas = power_user_result.to_pandas()
+ backend.create_table("customer_summary", summary_pandas, overwrite=True)
+ print("✅ Created table: customer_summary")
+ print(f" Rows: {len(summary_pandas)}")
+
+ # Show comparison
+ print("\n📊 Results comparison:")
+ print(f" Simple join: {len(simple_pandas)} rows, {len(simple_pandas.columns)} columns")
+ print(f" Power user: {len(summary_pandas)} rows, {len(summary_pandas.columns)} columns (aggregated)")
+
+ # Show sample data from simple join
+ print("\n📋 Sample from simple join (first 2 records):")
+ sample = simple_pandas.head(2)
+ for i, (idx, row) in enumerate(sample.iterrows(), 1):
+ print(f"\n Record {i}:")
+ print(f" Customer ID: {row.get('customer_id', 'N/A')}")
+ print(f" Customer Name: {row.get('profile_name', 'N/A')}")
+ print(f" Email: {row.get('profile_contact_email', 'N/A')}")
+ print(f" Order ID: {row.get('order_id', 'N/A')}")
+ print(f" Order Amount: ${row.get('amount', 0):.2f}")
+
+ # ===================================================================
+ # STEP 7: Backend Reference Management - NEW Architecture!
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 7: Backend Reference Management (Property-Based)")
+ print("=" * 70)
+
+ print("\n📚 NEW Storage Architecture:")
+ print(" Each DataFrameHandler now OWNS its backend reference!")
+ print()
+ print(" Old way (NestedHandler managed references):")
+ print(" ❌ Tight coupling")
+ print(" ❌ DataFrame can't update its own backend status")
+ print()
+ print(" NEW way (DataFrameHandler manages its own reference):")
+ print(" ✅ Each handler has table_qualifier property")
+ print(" ✅ Can be None (in-memory) or backend identifier")
+ print(" ✅ Independent backend status updates")
+ print(" ✅ Clean separation of concerns")
+
+ print("\n🔍 Current backend status:")
+ nested.show_backend_status()
+
+ print("\n📋 Detailed handler information:")
+ for df_name in nested.list_dataframes():
+ handler = nested.get(df_name)
+ backend_info = handler.get_backend_info()
+ print(f"\n {df_name}:")
+ print(f" {handler}")
+ print(f" Has backend table: {handler.has_backend_table()}")
+ if handler.has_backend_table():
+ print(f" Qualifier: {handler.table_qualifier}")
+ print(f" Parsed: {backend_info}")
+
+ # ===================================================================
+ # STEP 8: Demonstrate Backend Reference Updates
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("STEP 8: Backend Reference Updates (NEW Feature)")
+ print("=" * 70)
+
+ print("\n💡 Demonstrating independent backend reference updates...")
+
+ # Create an in-memory DataFrame
+ print("\n1️⃣ Create in-memory DataFrame (no backend):")
+ import pandas as pd
+ temp_df = session.DataFrame(pd.DataFrame({
+ 'id': [1, 2, 3],
+ 'value': [10, 20, 30]
+ }))
+ nested.add("temp_data", temp_df) # No table_qualifier
+
+ temp_handler = nested.get("temp_data")
+ print(f" Status: {temp_handler.has_backend_table()} (no backend table)")
+
+ # Simulate saving to backend
+ print("\n2️⃣ Save to backend database:")
+ temp_pandas = temp_df.to_pandas()
+ backend.create_table("temp_data", temp_pandas, overwrite=True)
+
+ # Update the backend reference
+ temp_handler.set_table_qualifier("local_duckdb.main.temp_data")
+ print(f" Status: {temp_handler.has_backend_table()} (now has backend table!)")
+
+ # Show updated status
+ print("\n3️⃣ Show all backend status after update:")
+ nested.show_backend_status()
+
+ # Simulate dropping from backend
+ print("\n4️⃣ Drop from backend (DataFrame still in memory):")
+ backend.drop_table("temp_data")
+ temp_handler.set_table_qualifier(None) # Clear backend reference
+ print(f" Status: {temp_handler.has_backend_table()} (back to in-memory)")
+
+ print("\n5️⃣ Final backend status:")
+ nested.show_backend_status()
+
+ print("\n✨ Benefits of NEW Architecture:")
+ print(" ✅ Independent backend reference management")
+ print(" ✅ DataFrames can update their own status")
+ print(" ✅ Clean separation: NestedHandler orchestrates, handlers manage state")
+ print(" ✅ Support for in-memory → backend → in-memory lifecycle")
+ print(" ✅ Join provenance tracking (lineage from source tables)")
+ print(" ✅ No tight coupling between orchestrator and backend state")
+
+ # ===================================================================
+ # Summary
+ # ===================================================================
+ print("\n" + "=" * 70)
+ print("✅ Demo Complete!")
+ print("=" * 70)
+
+ print("\n📊 Summary:")
+ print(" - Created 2 nested DataFrames with email in different nested paths")
+ print(" - Pushed to local DuckDB: customers, orders")
+ print(" - Loaded from DB into NestedHandler (with table qualifiers)")
+ print(" - Joined on nested email: profile.contact.email ⋈ shipping.recipient.email")
+ print(" - Result auto-added to handler: 'customer_orders'")
+ print(" - Saved joined result back to database")
+ print(" - Demonstrated backend reference updates (save/drop lifecycle)")
+
+ print("\n🎯 Key Architectural Insights:")
+ print(" 1. DataFrameHandler owns its backend reference (table_qualifier property)")
+ print(" 2. NestedHandler orchestrates operations, handlers manage state")
+ print(" 3. Backend references can be None (in-memory) or qualified names")
+ print(" 4. Handlers can independently update backend status via set_table_qualifier()")
+ print(" 5. Clean lifecycle: in-memory → save (set qualifier) → drop (clear qualifier)")
+ print(" 6. Join results track lineage: joined(table1⋈table2)")
+ print(" 7. Auto-add results enable fluent chaining of operations")
+
+ # Cleanup
+ backend.disconnect()
+ print("\n🔌 Disconnected from database")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/demos/utils/create_nested_data.py b/demos/utils/create_nested_data.py
index 675a175..b79a87a 100644
--- a/demos/utils/create_nested_data.py
+++ b/demos/utils/create_nested_data.py
@@ -236,6 +236,106 @@ def create_deeply_nested_dataframe() -> DataFrame:
return pyarrow_to_leanframe(pa_table)
+def create_customers_for_join() -> DataFrame:
+ """
+ Create customers DataFrame with nested email for join testing.
+
+ Structure: customer_id, profile.contact.email (nested 2 levels)
+ """
+ data = {
+ "customer_id": [1, 2, 3, 4],
+ "profile": [
+ {
+ "name": "Alice Johnson",
+ "contact": {"email": "alice@example.com", "phone": "555-1001"}
+ },
+ {
+ "name": "Bob Smith",
+ "contact": {"email": "bob@example.com", "phone": "555-1002"}
+ },
+ {
+ "name": "Charlie Brown",
+ "contact": {"email": "charlie@example.com", "phone": "555-1003"}
+ },
+ {
+ "name": "Diana Prince",
+ "contact": {"email": "diana@example.com", "phone": "555-1004"}
+ }
+ ]
+ }
+
+ contact_schema = pa.struct([
+ pa.field("email", pa.string()),
+ pa.field("phone", pa.string())
+ ])
+
+ profile_schema = pa.struct([
+ pa.field("name", pa.string()),
+ pa.field("contact", contact_schema)
+ ])
+
+ schema = pa.schema([
+ pa.field("customer_id", pa.int64()),
+ pa.field("profile", profile_schema)
+ ])
+
+ pa_table = pa.Table.from_pydict(data, schema=schema)
+ return pyarrow_to_leanframe(pa_table)
+
+
+def create_orders_for_join() -> DataFrame:
+ """
+ Create orders DataFrame with differently nested email for join testing.
+
+ Structure: order_id, shipping.recipient.email (nested 2 levels, different path!)
+ """
+ data = {
+ "order_id": [101, 102, 103, 104, 105],
+ "shipping": [
+ {
+ "recipient": {"email": "alice@example.com", "name": "Alice J."},
+ "address": "123 Main St"
+ },
+ {
+ "recipient": {"email": "bob@example.com", "name": "Bob S."},
+ "address": "456 Oak Ave"
+ },
+ {
+ "recipient": {"email": "alice@example.com", "name": "Alice J."},
+ "address": "123 Main St"
+ },
+ {
+ "recipient": {"email": "charlie@example.com", "name": "Charlie B."},
+ "address": "789 Pine Rd"
+ },
+ {
+ "recipient": {"email": "bob@example.com", "name": "Bob S."},
+ "address": "456 Oak Ave"
+ }
+ ],
+ "amount": [299.99, 150.00, 75.50, 420.00, 89.99]
+ }
+
+ recipient_schema = pa.struct([
+ pa.field("email", pa.string()),
+ pa.field("name", pa.string())
+ ])
+
+ shipping_schema = pa.struct([
+ pa.field("recipient", recipient_schema),
+ pa.field("address", pa.string())
+ ])
+
+ schema = pa.schema([
+ pa.field("order_id", pa.int64()),
+ pa.field("shipping", shipping_schema),
+ pa.field("amount", pa.float64())
+ ])
+
+ pa_table = pa.Table.from_pydict(data, schema=schema)
+ return pyarrow_to_leanframe(pa_table)
+
+
# Convenience functions for backward compatibility
if __name__ == "__main__":
print("=== Centralized Nested Data Creation Demo ===\n")
diff --git a/docs/architecture/NESTED_HANDLER_VISION.md b/docs/architecture/NESTED_HANDLER_VISION.md
new file mode 100644
index 0000000..51915a3
--- /dev/null
+++ b/docs/architecture/NESTED_HANDLER_VISION.md
@@ -0,0 +1,226 @@
+# NestedHandler Architecture - Design Vision
+
+## Critical Design Principle
+
+**NestedHandler prepares data. Ibis/Backend executes SQL.**
+
+### Why This Matters
+
+Your feedback was spot-on:
+1. ❌ Multiple `join_on_*` methods are unnecessary complexity
+2. ❌ Two-table limitation is artificial
+3. ❌ Need support for ALL SQL features (WHERE, HAVING, windows, etc.)
+4. ✅ Should delegate to Ibis (which already supports everything!)
+
+## New Architecture
+
+```
+┌─────────────────────────────────────────────────────┐
+│ NestedHandler: Data Preparation Layer │
+│ │
+│ Responsibilities: │
+│ • Analyze nested structure │
+│ • Extract nested fields → flat columns │
+│ • Manage multiple DataFrames │
+│ • Track backend references │
+└──────────────────┬──────────────────────────────────┘
+ │
+ │ prepare() returns flat DataFrame
+ │
+ ▼
+┌─────────────────────────────────────────────────────┐
+│ Ibis/Backend: SQL Execution Layer │
+│ │
+│ Full SQL Power: │
+│ • Joins: ANY number of tables, ALL types │
+│ • WHERE, HAVING clauses │
+│ • GROUP BY + aggregations │
+│ • Window functions │
+│ • UNION, INTERSECT, EXCEPT │
+│ • CTEs (Common Table Expressions) │
+│ • Everything BigQuery/SQL supports! │
+└─────────────────────────────────────────────────────┘
+```
+
+## Key Methods
+
+### `prepare(name, fields=None, verbose=False)`
+Extract nested fields from a DataFrame, returning a flattened version.
+
+```python
+# Extract all nested fields
+customers_flat = handler.prepare("customers")
+
+# Extract specific fields only
+customers_flat = handler.prepare("customers", fields=["profile.email", "profile.name"])
+```
+
+### Direct Ibis Operations
+After `prepare()`, use full Ibis/SQL power:
+
+```python
+# Simple join
+result = table1._data.join(table2._data, predicates=[...], how="inner")
+
+# Complex multi-table join
+step1 = customers._data.join(orders._data, ...)
+step2 = step1.join(products._data, ...)
+step3 = step2.join(regions._data, ...)
+
+# Add WHERE
+filtered = step3.filter(step3.amount > 100)
+
+# Add GROUP BY
+grouped = filtered.group_by("region").aggregate(
+ total=filtered.amount.sum(),
+ count=filtered.order_id.count()
+)
+
+# Add HAVING
+having_filtered = grouped.filter(grouped.total > 1000)
+
+# Window functions
+with_rank = grouped.select(
+ grouped.region,
+ grouped.total,
+ rank=ibis.row_number().over(window)
+)
+```
+
+## Usage Pattern
+
+### 1. Add DataFrames
+```python
+handler = NestedHandler()
+handler.add("customers", customers_df, table_qualifier="db.sales.customers")
+handler.add("orders", orders_df, table_qualifier="db.sales.orders")
+handler.add("products", products_df)
+```
+
+### 2. Prepare (Extract Nested Fields)
+```python
+# NestedHandler handles the complex part: nested field extraction
+customers_flat = handler.prepare("customers") # Extracts profile.email → profile_email
+orders_flat = handler.prepare("orders") # Extracts shipping.email → shipping_email
+products_flat = handler.prepare("products") # Already flat, returns as-is
+```
+
+### 3. Use Ibis for SQL Operations
+```python
+# Ibis handles the SQL part: all operations BigQuery supports
+joined = customers_flat._data.join(
+ orders_flat._data,
+ predicates=[customers_flat._data.profile_email == orders_flat._data.shipping_email],
+ how="inner"
+).join(
+ products_flat._data,
+ predicates=[orders_flat._data.product_id == products_flat._data.product_id],
+ how="left"
+).filter(
+ lambda t: t.amount > 100
+).group_by("customer_id").aggregate(
+ total=lambda t: t.amount.sum(),
+ count=lambda t: t.order_id.count()
+)
+
+result = DataFrame(joined)
+```
+
+## Why This is Better
+
+### ✅ Separation of Concerns
+- **NestedHandler**: Knows about nested structures (domain-specific)
+- **Ibis**: Knows about SQL operations (already mature, tested)
+
+### ✅ No Artificial Limitations
+- Join ANY number of tables
+- Use ANY join type Ibis supports
+- Add ANY SQL clause (WHERE, HAVING, etc.)
+- Use window functions, CTEs, everything!
+
+### ✅ Future-Proof
+- When BigQuery adds new features, they work immediately (via Ibis)
+- No need to wrap every SQL operation
+- NestedHandler focuses on its unique value: nested data
+
+### ✅ Composable
+```python
+# Each step is a standard DataFrame
+customers_flat = handler.prepare("customers")
+orders_flat = handler.prepare("orders")
+
+# Combine however you want
+result1 = customers_flat._data.join(orders_flat._data, ...)
+result2 = customers_flat._data.filter(...)
+result3 = result1.union(result2)
+```
+
+## Future: Convenience Method
+
+Eventually we might add:
+```python
+# Convenience wrapper for common cases
+result = handler.join(
+ tables={"c": "customers", "o": "orders", "p": "products"},
+ on=[("c", "email", "o", "customer_email"),
+ ("o", "product_id", "p", "product_id")],
+ how="inner"
+)
+```
+
+**BUT** this is just sugar over `prepare()` + Ibis operations. Power users can always go direct!
+
+## Migration Path
+
+### Old Code (Deprecated)
+```python
+result = handler.join_on_both_nested(
+ left="customers",
+ right="orders",
+ left_path="profile.email",
+ right_path="shipping.email",
+ how="inner"
+)
+```
+
+### New Code (Flexible)
+```python
+# Prepare
+customers = handler.prepare("customers")
+orders = handler.prepare("orders")
+
+# Join with full SQL power
+result_ibis = customers._data.join(
+ orders._data,
+ predicates=[customers._data.profile_email == orders._data.shipping_email],
+ how="inner"
+)
+
+# Add more operations as needed
+filtered = result_ibis.filter(result_ibis.amount > 100)
+grouped = filtered.group_by("customer_id").aggregate(...)
+
+result = DataFrame(filtered) # Or grouped, or any Ibis expression
+```
+
+## Implementation Status
+
+- ✅ `prepare()` method implemented
+- ✅ Backend reference management (property-based)
+- ✅ Demo showing flexible joins
+- ✅ Documentation
+- ⏳ Convenience `join()` method (future)
+- ❌ Old `join_on_*` methods deprecated (raise DeprecationWarning)
+
+## Examples
+
+See:
+- `demos/demo_flexible_joins.py` - Comprehensive examples
+- `docs/architecture/backend_reference_management.md` - Architecture details
+
+## Summary
+
+**The right tool for the right job:**
+- NestedHandler: Expert at nested data preparation
+- Ibis: Expert at SQL operations
+- Together: Full BigQuery/pandas emulation with no limitations!
diff --git a/docs/architecture/backend_reference_management.md b/docs/architecture/backend_reference_management.md
new file mode 100644
index 0000000..3bfba2a
--- /dev/null
+++ b/docs/architecture/backend_reference_management.md
@@ -0,0 +1,273 @@
+# Backend Reference Management Architecture
+
+## Overview
+
+This document explains how leanframe's `DataFrameHandler` and `NestedHandler` manage backend table references, enabling DataFrames to track their storage location independently.
+
+## Architecture Evolution
+
+### Previous Design (Centralized)
+- **Problem**: NestedHandler managed table qualifiers separately in `_table_qualifiers` dict
+- **Issue**: Tight coupling between orchestrator and backend state
+- **Limitation**: DataFrame couldn't update its own backend reference
+
+### Current Design (Property-Based)
+- **Solution**: DataFrameHandler owns its backend reference as a property
+- **Benefit**: Clean separation of concerns
+- **Feature**: Independent backend status updates
+
+## Core Components
+
+### DataFrameHandler
+
+Each `DataFrameHandler` instance manages:
+- Its DataFrame and nested structure analysis
+- Its backend table reference (optional `table_qualifier`)
+- Methods to query and update backend status
+
+```python
+handler = DataFrameHandler(df, table_qualifier="mydb.sales.customers")
+
+# Query backend status
+handler.table_qualifier # "mydb.sales.customers"
+handler.has_backend_table() # True
+handler.get_backend_info() # Parsed qualifier details
+
+# Update backend reference
+handler.set_table_qualifier("new.location.customers") # Update
+handler.set_table_qualifier(None) # Clear (in-memory)
+```
+
+### NestedHandler
+
+The orchestrator manages multiple DataFrameHandlers:
+- Stores handlers by user-friendly names
+- Coordinates operations (joins, etc.)
+- Accesses backend references via handler properties
+
+```python
+nested = NestedHandler()
+
+# Add with backend reference
+nested.add("customers", df, table_qualifier="mydb.sales.customers")
+
+# Access handler and its backend info
+handler = nested.get("customers")
+print(handler.table_qualifier) # "mydb.sales.customers"
+
+# Show all backend statuses
+nested.show_backend_status()
+```
+
+## Key Properties & Methods
+
+### DataFrameHandler
+
+#### `table_qualifier` Property
+```python
+@property
+def table_qualifier(self) -> str | None:
+ """Get backend table qualifier or None for in-memory DataFrame."""
+```
+
+- **Returns**: Backend identifier (e.g., `"project.dataset.table"`) or `None`
+- **Example**: `"local_duckdb.main.customers"`
+
+#### `set_table_qualifier()` Method
+```python
+def set_table_qualifier(self, qualifier: str | None):
+ """Update backend reference. Call when DataFrame saved/dropped."""
+```
+
+- **Usage**: Update when backend state changes
+- **Events**: Save to backend, load from backend, drop table
+- **Prints**: Status messages showing reference changes
+
+#### `has_backend_table()` Method
+```python
+def has_backend_table(self) -> bool:
+ """Check if DataFrame has backend table reference."""
+```
+
+- **Returns**: `True` if backend table exists, `False` for in-memory
+- **Use Case**: Conditional logic based on backend status
+
+#### `get_backend_info()` Method
+```python
+def get_backend_info(self) -> dict[str, str | None]:
+ """Get parsed backend information."""
+```
+
+- **Returns**: Dict with `qualifier`, `project`, `dataset`, `table`, `type`
+- **Parsing**: Attempts to parse standard `project.dataset.table` format
+- **Example**:
+ ```python
+ {
+ 'qualifier': 'myproject.sales.customers',
+ 'project': 'myproject',
+ 'dataset': 'sales',
+ 'table': 'customers',
+ 'type': 'backend'
+ }
+ ```
+
+### NestedHandler
+
+#### `show_backend_status()` Method
+```python
+def show_backend_status(self):
+ """Display backend status for all DataFrames."""
+```
+
+- **Output**: Pretty-printed list showing each DataFrame's backend status
+- **Symbols**: ✅ for backend tables, ⚪ for in-memory DataFrames
+
+## Lifecycle Examples
+
+### 1. In-Memory → Backend → In-Memory
+
+```python
+# Start with in-memory DataFrame
+nested = NestedHandler()
+nested.add("temp", df) # No table_qualifier
+
+handler = nested.get("temp")
+print(handler.has_backend_table()) # False
+
+# Save to backend
+backend.create_table("temp", df.to_pandas())
+handler.set_table_qualifier("mydb.main.temp")
+print(handler.has_backend_table()) # True
+
+# Drop from backend (DataFrame still in memory)
+backend.drop_table("temp")
+handler.set_table_qualifier(None)
+print(handler.has_backend_table()) # False
+```
+
+### 2. Load from Backend
+
+```python
+# Load from database with qualifier
+df = session.read_sql_table("customers")
+nested.add("customers", df, table_qualifier="mydb.sales.customers")
+
+handler = nested.get("customers")
+print(handler.table_qualifier) # "mydb.sales.customers"
+```
+
+### 3. Join Result Lineage
+
+```python
+# Join tracks source tables in qualifier
+nested.add("customers", customers_df, table_qualifier="db.sales.customers")
+nested.add("orders", orders_df, table_qualifier="db.sales.orders")
+
+result, name = nested.join_on_nested(
+ left="customers",
+ right="orders",
+ left_path="email",
+ right_column="customer_email"
+)
+
+handler = nested.get("customers_orders_joined")
+print(handler.table_qualifier)
+# Output: "joined(db.sales.customers⋈db.sales.orders)"
+```
+
+## Design Benefits
+
+### ✅ Clean Separation of Concerns
+- **DataFrameHandler**: Manages individual DataFrame state (including backend reference)
+- **NestedHandler**: Orchestrates operations across DataFrames
+- **No Duplication**: Single source of truth for backend reference
+
+### ✅ Independent Updates
+- DataFrame can update its own backend status
+- No need to notify or update orchestrator
+- Natural flow: save → update reference, drop → clear reference
+
+### ✅ Flexible Storage
+- `None` for in-memory DataFrames
+- Qualified names for backend tables: `"project.dataset.table"`
+- Custom formats for lineage: `"joined(table1⋈table2)"`
+
+### ✅ Lineage Tracking
+- Join results track source tables
+- Easy to see data provenance
+- Supports complex workflows with multiple joins
+
+### ✅ Backend Agnostic
+- Works with any backend (DuckDB, BigQuery, etc.)
+- Qualifier format adapts to backend conventions
+- Parser supports multiple formats (3-part, 2-part, 1-part)
+
+## Best Practices
+
+### 1. Set Qualifier When Loading from Backend
+```python
+df = load_from_database("customers")
+nested.add("customers", df, table_qualifier="mydb.sales.customers")
+```
+
+### 2. Update After Saving to Backend
+```python
+# Create in-memory
+nested.add("result", result_df) # No qualifier
+
+# Save and update
+backend.create_table("result", result_df.to_pandas())
+nested.get("result").set_table_qualifier("mydb.temp.result")
+```
+
+### 3. Clear After Dropping from Backend
+```python
+backend.drop_table("temp")
+nested.get("temp").set_table_qualifier(None)
+# DataFrame still available in memory!
+```
+
+### 4. Check Before Backend Operations
+```python
+handler = nested.get("data")
+if handler.has_backend_table():
+ # Can reload from backend
+ backend.table(handler.table_qualifier)
+else:
+ # In-memory only
+ print("No backend table available")
+```
+
+## Future Enhancements
+
+### Automatic Synchronization
+- Notification system for backend events (save/drop)
+- Auto-update references when backend changes
+- Callback registration for lifecycle events
+
+### Smart Reload
+- Reload DataFrame from backend using qualifier
+- Refresh stale data
+- Validate backend table still exists
+
+### Advanced Lineage
+- Track full operation history
+- Serialize/deserialize lineage graphs
+- Visualize data provenance
+
+### Backend Validation
+- Verify qualifier matches backend schema
+- Auto-detect qualifier format
+- Validate table exists before setting reference
+
+## Summary
+
+The property-based backend reference management architecture provides:
+
+1. **Ownership**: DataFrameHandler owns its backend reference
+2. **Independence**: Can update without coordinating with NestedHandler
+3. **Flexibility**: Supports in-memory, backend, and hybrid workflows
+4. **Lineage**: Tracks data provenance through qualifiers
+5. **Simplicity**: Clean API with intuitive methods
+
+This design enables complex workflows while maintaining clean separation of concerns and allowing DataFrames to independently manage their backend state.
diff --git a/leanframe/__init__.py b/leanframe/__init__.py
index 257b0da..ec66097 100644
--- a/leanframe/__init__.py
+++ b/leanframe/__init__.py
@@ -15,11 +15,13 @@
"""LeanFrame provides a DataFrame API for BigQuery."""
from leanframe.core.session import Session
-from leanframe.core.frame import DynamicNestedHandler
+from leanframe.core.frame import DataFrameHandler
+from leanframe.core.nested_handler import NestedHandler
from leanframe.version import __version__
__all__ = [
"__version__",
"Session",
- "DynamicNestedHandler",
+ "DataFrameHandler",
+ "NestedHandler",
]
diff --git a/leanframe/core/frame.py b/leanframe/core/frame.py
index 10caf76..eabf83f 100644
--- a/leanframe/core/frame.py
+++ b/leanframe/core/frame.py
@@ -19,7 +19,6 @@
import ibis
import ibis.expr.types as ibis_types
import pandas as pd
-from typing import Any, Iterator
from leanframe.core.dtypes import convert_ibis_to_pandas
@@ -100,31 +99,79 @@ def to_pandas(self) -> pd.DataFrame:
"""
# TODO: replace prints by logging, ask TIM about logger usage
-class DynamicNestedHandler:
+class DataFrameHandler:
"""
- Completely dynamic nested data handler that can introspect any DataFrame
- and automatically extract nested fields of arbitrary depth.
-
+ Wrapper for a single leanframe DataFrame that handles nested column introspection.
+
+ This class wraps ONE DataFrame and provides metadata about its nested structure
+ along with functional operations for extracting nested fields.
+
+ Design Philosophy - Stateless Data, Cached Metadata:
+ - Wraps a SINGLE leanframe DataFrame (not multiple DataFrames)
+ - Caches IMMUTABLE schema metadata (nested field mappings, column types)
+ - Does NOT cache data or extraction results
+ - All data operations return NEW DataFrame objects (functional style)
+ - Thread-safe for concurrent operations
+
Features:
- - Automatic nested structure detection
- - Multi-level nesting support
- - Dynamic field extraction
+ - Automatic nested structure detection via schema introspection
+ - Multi-level nesting support (configurable depth)
+ - Dynamic field extraction (computed on-demand)
- Preserves non-nested columns
- - Dictionary-like record access
- Efficient columnar operations
+
+ Usage Pattern:
+ # Create handler for a single DataFrame (introspects schema once)
+ handler = DataFrameHandler(customers_df)
+
+ # Access cached schema metadata (fast, immutable)
+ print(handler.extracted_fields) # {'person.name': 'person_name', ...}
+ handler.show_structure()
+
+ # Compute data operations (functional, returns new objects)
+ extracted_df = handler.extract_nested_fields() # No caching!
+ another_df = handler.extract_nested_fields() # Computes again
+
+ # For multi-DataFrame operations, use NestedHandler orchestrator
+ # (See NestedHandler class for joins and other cross-DataFrame operations)
"""
- def __init__(self, lf_df: DataFrame, max_depth: int = 10):
+ def __init__(
+ self,
+ lf_df: DataFrame,
+ max_depth: int = 10,
+ table_qualifier: str | None = None
+ ):
+ """
+ Initialize handler by introspecting DataFrame schema.
+
+ Args:
+ lf_df: leanframe DataFrame to analyze
+ max_depth: Maximum nesting depth to analyze (prevents infinite recursion)
+ table_qualifier: Optional backend table identifier (e.g., "project.dataset.table")
+ Can be None for in-memory DataFrames. Can be updated later
+ via set_table_qualifier() method.
+
+ Note:
+ Constructor only analyzes SCHEMA (metadata), does not extract data.
+ This makes handler creation fast and allows metadata reuse.
+ """
+ # Store reference to original DataFrame (not a copy!)
self.original_df = lf_df
+
+ # Configuration
self.max_depth = max_depth
- self.nested_fields: dict[str, dict] = {} # Maps original_path -> extracted_column_name
- self.struct_columns: set[str] = set() # Track which columns are structs
- self.extracted_df = lf_df # Initialize with original, will be replaced
- self._arrow_cache = None
-
- # Perform dynamic introspection
+
+ # Backend reference - can be None for in-memory DataFrames
+ # This is mutable and can be updated when DataFrame is saved to backend
+ self._table_qualifier: str | None = table_qualifier
+
+ # Cached schema metadata (immutable after introspection)
+ self.nested_fields: dict[str, dict] = {} # Maps path -> field metadata
+ self.struct_columns: set[str] = set() # Tracks struct columns
+
+ # Perform schema introspection (builds metadata cache)
self._introspect_structure()
- self._extract_all_nested_fields()
def _introspect_structure(self):
"""Dynamically introspect the DataFrame to find all nested structures."""
@@ -204,8 +251,58 @@ def _analyze_struct_column_native(
else:
print(f"{indent} ⚠️ Struct type has no accessible fields attribute")
- def _extract_all_nested_fields(self):
- """Extract all discovered nested fields into a flat DataFrame."""
+ def _extract_nested_fields_silent(self) -> DataFrame:
+ """
+ Internal method: Extract nested fields without printing.
+ Used by backward compatibility methods.
+ """
+ ibis_table = self.original_df._data
+ select_exprs = []
+
+ # Add all non-struct columns first
+ for col_name in ibis_table.columns:
+ if col_name not in self.struct_columns:
+ select_exprs.append(ibis_table[col_name])
+
+ # Add all extracted nested fields
+ for field_path, field_info in self.nested_fields.items():
+ try:
+ field_expr = field_info["expression"]
+ extracted_name = field_info["extracted_name"]
+ extraction_expr = field_expr.name(extracted_name)
+ select_exprs.append(extraction_expr)
+ except Exception:
+ pass # Skip fields that can't be extracted
+
+ # Create and return the new DataFrame
+ if select_exprs:
+ extracted_table = ibis_table.select(*select_exprs)
+ return DataFrame(extracted_table)
+ else:
+ return self.original_df
+
+ def extract_nested_fields(self, verbose: bool = True) -> DataFrame:
+ """
+ Extract all discovered nested fields into a flat DataFrame.
+
+ IMPORTANT: This is a FUNCTIONAL operation that returns a NEW DataFrame.
+ Results are NOT cached - each call computes fresh extraction.
+ This ensures thread-safety and prevents stale data issues.
+
+ Args:
+ verbose: If True, prints extraction progress. Set False for silent operation.
+
+ Returns:
+ NEW DataFrame with flattened structure (does not modify original)
+
+ Example:
+ handler = DynamicNestedHandler(nested_df)
+ flat1 = handler.extract_nested_fields() # Computes extraction
+ flat2 = handler.extract_nested_fields() # Computes again (no cache!)
+ """
+ if not verbose:
+ return self._extract_nested_fields_silent()
+
print("\n🚀 Extracting all nested fields...")
ibis_table = self.original_df._data
@@ -237,92 +334,150 @@ def _extract_all_nested_fields(self):
print(f"\n📊 Summary: {extracted_count} nested fields extracted")
- # Create the new DataFrame
+ # Create and return the new DataFrame (no state storage!)
if select_exprs:
extracted_table = ibis_table.select(*select_exprs)
- self.extracted_df = DataFrame(extracted_table)
+ result = DataFrame(extracted_table)
else:
- self.extracted_df = self.original_df
-
- print(f" Final DataFrame columns: {len(self.extracted_df.columns)} total")
-
- def _get_arrow_table(self):
- """Get cached PyArrow table for efficient access."""
- if self._arrow_cache is None:
- reader = self.extracted_df._data.to_pyarrow()
- self._arrow_cache = reader
- return self._arrow_cache
+ result = self.original_df
- # Public API - same as before but now completely dynamic
+ print(f" Final DataFrame columns: {len(result.columns)} total")
+ return result
- def get_column(self, column_name: str) -> list:
- """Get entire column - works for both original and extracted columns."""
- table = self._get_arrow_table()
- if column_name not in table.column_names:
- available = ", ".join(table.column_names)
- raise KeyError(f"Column '{column_name}' not found. Available: {available}")
- return table[column_name].to_pylist()
-
- @property
- def columns(self) -> list[str]:
- """All available column names (original + extracted)."""
- return self._get_arrow_table().column_names
+ # Public API - Functional design without state
@property
def original_columns(self) -> list[str]:
- """Original DataFrame column names."""
+ """
+ Get original DataFrame column names.
+
+ Returns:
+ List of column names from the original DataFrame
+ """
return self.original_df.columns.tolist()
@property
def extracted_fields(self) -> dict[str, str]:
- """Mapping of original nested paths to extracted column names."""
+ """
+ Get mapping of nested paths to extracted column names.
+
+ This returns CACHED SCHEMA METADATA, not data.
+ Example: {'person.name': 'person_name', 'person.age': 'person_age'}
+
+ Returns:
+ Dictionary mapping nested paths to flattened column names
+ """
return {
path: info["extracted_name"] for path, info in self.nested_fields.items()
}
+
+ def get_extracted_column_name(self, nested_path: str) -> str | None:
+ """
+ Look up the extracted column name for a nested path.
+
+ Args:
+ nested_path: Nested path like 'person.address.city'
+
+ Returns:
+ Extracted column name like 'person_address_city', or None if not found
+ """
+ return self.extracted_fields.get(nested_path)
- def get_record(self, index: int) -> dict[str, Any]:
- """Get single record as dictionary."""
- table = self._get_arrow_table()
- if index >= len(table):
- raise IndexError(f"Index {index} out of range (0-{len(table) - 1})")
-
- row = table.slice(index, 1)
- record = {}
- for i, col_name in enumerate(row.column_names):
- column = row.column(i)
- record[col_name] = column[0].as_py() if len(column) > 0 else None
-
- return record
-
- def get_all_records(self, limit: int | None = None) -> list[dict[str, Any]]:
- """Get all records as list of dictionaries."""
- table = self._get_arrow_table()
- if limit and limit < len(table):
- table = table.slice(0, limit)
- return table.to_pylist()
-
- def filter_by(self, column_name: str, value: Any) -> "DynamicNestedHandler":
- """Filter records using pure ibis operations."""
- if column_name not in self.columns:
- raise KeyError(f"Column '{column_name}' not found")
-
- ibis_table = self.extracted_df._data
- filtered_table = ibis_table.filter(ibis_table[column_name] == value)
- filtered_df = DataFrame(filtered_table)
-
- # Create new handler with filtered data - skip introspection since structure is same
- new_handler = DynamicNestedHandler.__new__(DynamicNestedHandler)
- new_handler.original_df = filtered_df
- new_handler.extracted_df = filtered_df
- new_handler.nested_fields = self.nested_fields
- new_handler.struct_columns = self.struct_columns
- new_handler._arrow_cache = None
- new_handler.max_depth = self.max_depth
-
- return new_handler
+ # Backward compatibility methods - compute on-demand (NO caching!)
+ @property
+ def columns(self) -> list[str]:
+ """
+ Get column names from extracted DataFrame (computed on-demand).
+
+ WARNING: This performs extraction on EVERY call - use sparingly.
+ For efficiency, call extract_nested_fields() once and reuse the result.
+ """
+ extracted = self._extract_nested_fields_silent()
+ return extracted.columns.tolist()
+
+ def get_column(self, column_name: str) -> list:
+ """
+ Get entire column data (computed on-demand, not cached).
+
+ WARNING: This extracts and materializes data on EVERY call.
+ For efficiency, call extract_nested_fields() once and work with that.
+ """
+ extracted = self._extract_nested_fields_silent()
+ pandas_df = extracted.to_pandas()
+ if column_name not in pandas_df.columns:
+ available = ", ".join(pandas_df.columns)
+ raise KeyError(f"Column '{column_name}' not found. Available: {available}")
+ return pandas_df[column_name].tolist()
+
+ def get_record(self, index: int) -> dict:
+ """
+ Get single record as dictionary (computed on-demand).
+
+ WARNING: Performs extraction and materialization on EVERY call.
+ Not efficient for iterating over records - use extract_nested_fields() instead.
+ """
+ extracted = self._extract_nested_fields_silent()
+ pandas_df = extracted.to_pandas()
+ if index >= len(pandas_df):
+ raise IndexError(f"Index {index} out of range (0-{len(pandas_df) - 1})")
+ return pandas_df.iloc[index].to_dict()
+
+ def __len__(self) -> int:
+ """
+ Get number of records from original DataFrame.
+
+ Note: Uses original DataFrame to avoid extraction overhead.
+ """
+ # Use original DataFrame to avoid extraction overhead
+ pandas_df = self.original_df.to_pandas()
+ return len(pandas_df)
+
+ def __getitem__(self, index: int) -> dict:
+ """Get record by index."""
+ return self.get_record(index)
+
+ def __iter__(self):
+ """Iterate over records."""
+ for i in range(len(self)):
+ yield self.get_record(i)
+
+ def __contains__(self, key: str) -> bool:
+ """Check if column exists in extracted fields."""
+ return key in self.columns
+
+ def keys(self):
+ """Get all column names."""
+ return self.columns
+
+ def items(self):
+ """Iterate over (column_name, column_data) pairs."""
+ for key in self.keys():
+ yield key, self.get_column(key)
+
+ def values(self):
+ """Iterate over column data."""
+ for key in self.keys():
+ yield self.get_column(key)
+
+ def get(self, key: str, default=None):
+ """Dictionary-like get with default value."""
+ try:
+ return self.get_column(key)
+ except KeyError:
+ return default
def show_structure(self):
- """Display the complete DataFrame structure analysis."""
+ """
+ Display the complete DataFrame structure analysis.
+
+ Shows CACHED SCHEMA METADATA:
+ - Original columns and their types
+ - Discovered nested fields and their paths
+ - Expected flattened column structure
+ - Record count from original DataFrame
+
+ This is a read-only view of cached metadata, not data extraction.
+ """
print("\n📋 DYNAMIC STRUCTURE ANALYSIS")
print("=" * 50)
@@ -337,48 +492,134 @@ def show_structure(self):
for original_path, extracted_name in self.extracted_fields.items():
print(f" 🔗 {original_path} → {extracted_name}")
- print(f"\nFinal flattened columns: {len(self.columns)}")
- for col in self.columns:
+ # Show what the flattened structure would look like
+ expected_columns = [col for col in self.original_columns if col not in self.struct_columns]
+ expected_columns.extend(self.extracted_fields.values())
+ print(f"\nFlattened columns ({len(expected_columns)} total):")
+ for col in expected_columns:
print(f" 📊 {col}")
- table = self._get_arrow_table()
- print(f"\nRecords: {len(table)}")
-
- def __len__(self) -> int:
- return len(self._get_arrow_table())
-
- def __getitem__(self, index: int) -> dict[str, Any]:
- return self.get_record(index)
+ # Show record count from original
+ pandas_preview = self.original_df.to_pandas()
+ print(f"\nRecords: {len(pandas_preview)}")
- def __iter__(self) -> Iterator[dict[str, Any]]:
- for i in range(len(self)):
- yield self.get_record(i)
+ # Backend reference management
+
+ @property
+ def table_qualifier(self) -> str | None:
+ """
+ Get the backend table qualifier for this DataFrame.
+
+ Returns:
+ Backend table identifier (e.g., "project.dataset.table") or None
+ if this is an in-memory DataFrame not yet persisted to backend.
+
+ Example:
+ handler = DataFrameHandler(df)
+ print(handler.table_qualifier) # None (in-memory)
+
+ # After saving to backend
+ handler.set_table_qualifier("mydb.sales.customers")
+ print(handler.table_qualifier) # "mydb.sales.customers"
+ """
+ return self._table_qualifier
+
+ def set_table_qualifier(self, qualifier: str | None):
+ """
+ Update the backend table qualifier for this DataFrame.
+
+ This should be called when:
+ - DataFrame is saved to a backend table
+ - DataFrame is loaded from a backend table
+ - Backend table is dropped (set to None)
+ - DataFrame is renamed in the backend
+
+ Args:
+ qualifier: New backend table identifier or None to clear
+
+ Example:
+ handler = DataFrameHandler(in_memory_df)
+
+ # Save to backend
+ backend.create_table("customers", df.to_pandas())
+ handler.set_table_qualifier("mydb.main.customers")
+
+ # Later, if table is dropped
+ backend.drop_table("customers")
+ handler.set_table_qualifier(None) # DataFrame still in memory
+ """
+ old_qualifier = self._table_qualifier
+ self._table_qualifier = qualifier
+
+ if old_qualifier != qualifier:
+ if qualifier is None:
+ print(f"🔗 Cleared backend reference (was: {old_qualifier})")
+ elif old_qualifier is None:
+ print(f"🔗 Set backend reference: {qualifier}")
+ else:
+ print(f"🔗 Updated backend reference: {old_qualifier} → {qualifier}")
+
+ def has_backend_table(self) -> bool:
+ """
+ Check if this DataFrame has a backend table reference.
+
+ Returns:
+ True if DataFrame has a backend table, False if in-memory only
+ """
+ return self._table_qualifier is not None
+
+ def get_backend_info(self) -> dict[str, str | None]:
+ """
+ Get information about the backend storage.
+
+ Returns:
+ Dictionary with backend information:
+ - 'qualifier': Full table qualifier or None
+ - 'project': Project name (if qualifier follows project.dataset.table pattern)
+ - 'dataset': Dataset name (if applicable)
+ - 'table': Table name (if applicable)
+ - 'type': 'backend' or 'memory'
+
+ Example:
+ info = handler.get_backend_info()
+ # {'qualifier': 'myproject.sales.customers',
+ # 'project': 'myproject',
+ # 'dataset': 'sales',
+ # 'table': 'customers',
+ # 'type': 'backend'}
+ """
+ if self._table_qualifier is None:
+ return {
+ 'qualifier': None,
+ 'project': None,
+ 'dataset': None,
+ 'table': None,
+ 'type': 'memory'
+ }
+
+ # Try to parse standard format: project.dataset.table
+ parts = self._table_qualifier.split('.')
+ info: dict[str, str | None] = {
+ 'qualifier': self._table_qualifier,
+ 'project': None,
+ 'dataset': None,
+ 'table': None,
+ 'type': 'backend'
+ }
+
+ if len(parts) == 3:
+ info['project'] = parts[0]
+ info['dataset'] = parts[1]
+ info['table'] = parts[2]
+ elif len(parts) == 2:
+ info['dataset'] = parts[0]
+ info['table'] = parts[1]
+ elif len(parts) == 1:
+ info['table'] = parts[0]
+
+ return info
def __repr__(self) -> str:
- return f"DynamicNestedHandler({len(self)} records, {len(self.original_columns)} → {len(self.columns)} columns)"
-
- # Dictionary-like interface for column access
- def get(self, key: str, default=None):
- """Dictionary-like get with default value"""
- try:
- return self.get_column(key)
- except KeyError:
- return default
-
- def keys(self):
- """Get all column names"""
- return self.columns
-
- def items(self):
- """Iterate over column names and their data"""
- for key in self.keys():
- yield key, self.get_column(key)
-
- def values(self):
- """Get all column data"""
- for key in self.keys():
- yield self.get_column(key)
-
- def __contains__(self, key: str) -> bool:
- """Check if column exists"""
- return key in self.columns
+ num_extracted = len(self.nested_fields)
+ backend_info = " [in-memory]" if not self.has_backend_table() else f" [{self._table_qualifier}]"
+ return f"DataFrameHandler({len(self.original_columns)} cols → {num_extracted} nested fields{backend_info})"
diff --git a/leanframe/core/nested_handler.py b/leanframe/core/nested_handler.py
new file mode 100644
index 0000000..0709236
--- /dev/null
+++ b/leanframe/core/nested_handler.py
@@ -0,0 +1,531 @@
+# Copyright 2025 Google LLC, LeanFrame Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+NestedHandler - Orchestrator for multi-DataFrame operations with nested columns
+
+This module provides the NestedHandler class which manages multiple DataFrames
+and coordinates operations between them, particularly joins on nested columns.
+"""
+
+from __future__ import annotations
+
+from leanframe.core.frame import DataFrame, DataFrameHandler
+
+
+class NestedHandler:
+ """
+ Orchestrator for managing multiple DataFrames with nested columns.
+
+ This class:
+ - Manages multiple DataFrameHandler instances (one per DataFrame)
+ - Provides operations across DataFrames (joins, etc.)
+ - Maintains relationships between DataFrames
+ - Tracks DataFrames by name for easy reference
+
+ Design Philosophy:
+ - Each DataFrame gets its own DataFrameHandler (single responsibility)
+ - NestedHandler coordinates operations between handlers
+ - Handlers are created lazily when DataFrames are added
+ - Results can be added back to the context for chaining operations
+
+ Usage Pattern:
+ # Create orchestrator
+ handler = NestedHandler()
+
+ # Add DataFrames (creates DataFrameHandler internally)
+ handler.add("customers", customers_df)
+ handler.add("orders", orders_df)
+
+ # Perform operations using named references
+ joined_df = handler.join_on_nested(
+ left="customers",
+ right="orders",
+ left_path="profile.contact.email",
+ right_column="customer_email",
+ how="inner"
+ )
+
+ # Add result back for further operations
+ handler.add("customer_orders", joined_df)
+
+ Example Workflow:
+ handler = NestedHandler()
+ handler.add("customers", customers_df)
+ handler.add("orders", orders_df)
+ handler.add("regions", regions_df)
+
+ # Chain operations
+ handler.join_on_nested("customers", "regions", ...)
+ handler.join_on_nested("customer_regions", "orders", ...)
+ """
+
+ def __init__(self):
+ """Initialize empty NestedHandler."""
+ # Maps name -> DataFrameHandler
+ self._handlers: dict[str, DataFrameHandler] = {}
+
+ # Optional: Track join relationships for lineage/debugging
+ self._relationships: list[dict] = []
+
+ def add(
+ self,
+ name: str,
+ df: DataFrame,
+ max_depth: int = 10,
+ table_qualifier: str | None = None
+ ) -> DataFrameHandler:
+ """
+ Add a leanframe DataFrame to the handler context.
+
+ This creates a DataFrameHandler for the DataFrame and stores it
+ under the given name for later reference in operations.
+
+ Args:
+ name: Unique identifier for this DataFrame
+ df: leanframe DataFrame to add
+ max_depth: Maximum nesting depth to analyze (passed to DataFrameHandler)
+ table_qualifier: Optional backend table identifier (e.g., "project.dataset.table")
+ The handler will track this as its backend reference.
+
+ Returns:
+ The created DataFrameHandler (for direct access if needed)
+
+ Raises:
+ ValueError: If name already exists
+
+ Example:
+ handler = NestedHandler()
+
+ # Without table qualifier (in-memory DataFrame)
+ customer_handler = handler.add("customers", customers_df)
+ print(customer_handler.table_qualifier) # None
+
+ # With table qualifier for backend reference
+ handler.add("orders", orders_df, table_qualifier="mydb.sales.orders")
+
+ # Can access handler directly or via name
+ print(customer_handler.nested_fields)
+ print(handler.get("customers").nested_fields) # Same thing
+
+ # Check backend status
+ orders_handler = handler.get("orders")
+ print(orders_handler.table_qualifier) # "mydb.sales.orders"
+ print(orders_handler.has_backend_table()) # True
+ """
+ if name in self._handlers:
+ raise ValueError(
+ f"DataFrame '{name}' already exists. "
+ f"Use remove('{name}') first or choose a different name."
+ )
+
+ print(f"\n📦 Adding DataFrame '{name}' to NestedHandler...")
+ if table_qualifier:
+ print(f" Backend table: {table_qualifier}")
+
+ # Create handler with table qualifier - handler owns this reference now
+ df_handler = DataFrameHandler(df, max_depth=max_depth, table_qualifier=table_qualifier)
+ self._handlers[name] = df_handler
+
+ print(f"✅ Added '{name}' with {len(df_handler.original_columns)} columns "
+ f"({len(df_handler.nested_fields)} nested fields discovered)")
+
+ return df_handler
+
+ def get(self, name: str) -> DataFrameHandler:
+ """
+ Get a DataFrameHandler by name.
+
+ Args:
+ name: Name of the DataFrame
+
+ Returns:
+ DataFrameHandler for the named DataFrame
+
+ Raises:
+ KeyError: If name not found
+ """
+ if name not in self._handlers:
+ available = list(self._handlers.keys())
+ raise KeyError(
+ f"DataFrame '{name}' not found. "
+ f"Available: {available}"
+ )
+ return self._handlers[name]
+
+ def remove(self, name: str):
+ """
+ Remove a DataFrame from the handler.
+
+ Args:
+ name: Name of the DataFrame to remove
+
+ Raises:
+ KeyError: If name not found
+ """
+ if name not in self._handlers:
+ raise KeyError(f"DataFrame '{name}' not found")
+ del self._handlers[name]
+ print(f"🗑️ Removed DataFrame '{name}' from NestedHandler")
+
+ def list_dataframes(self) -> list[str]:
+ """
+ List all DataFrame names in the handler.
+
+ Returns:
+ List of DataFrame names
+ """
+ return list(self._handlers.keys())
+
+ def show_backend_status(self):
+ """
+ Show backend table status for all DataFrames.
+
+ Displays which DataFrames have backend tables and which are in-memory only.
+
+ Example output:
+ 📊 Backend Status for 4 DataFrames:
+ ✅ customers → myproject.sales.customers
+ ✅ orders → myproject.sales.orders
+ ⚪ processed → in-memory (no backend table)
+ ✅ joined_result → joined(customers⋈orders)
+ """
+ print(f"\n📊 Backend Status for {len(self._handlers)} DataFrames:")
+ for name, handler in self._handlers.items():
+ if handler.has_backend_table():
+ print(f"✅ {name} → {handler.table_qualifier}")
+ else:
+ print(f"⚪ {name} → in-memory (no backend table)")
+
+ def show_structure(self, name: str | None = None):
+ """
+ Show structure of one or all DataFrames.
+
+ Args:
+ name: Specific DataFrame name, or None to show all
+ """
+ if name is not None:
+ print(f"\n📊 Structure of '{name}':")
+ self.get(name).show_structure()
+ else:
+ print(f"\n📊 NestedHandler contains {len(self._handlers)} DataFrames:")
+ for df_name in self._handlers.keys():
+ print(f"\n--- {df_name} ---")
+ self._handlers[df_name].show_structure()
+
+ # Data preparation - extract nested fields for operations
+
+ def prepare(
+ self,
+ name: str,
+ fields: list[str] | None = None,
+ verbose: bool = False
+ ) -> DataFrame:
+ """
+ Prepare a DataFrame by extracting specified nested fields.
+
+ This is a preprocessing step before operations like joins. The handler
+ analyzes nested structure and extracts fields, returning a flattened
+ DataFrame ready for SQL-like operations.
+
+ Args:
+ name: Name of the DataFrame to prepare
+ fields: List of nested paths to extract (e.g., ['profile.email', 'address.city'])
+ If None, extracts all discovered nested fields
+ verbose: Whether to print extraction details
+
+ Returns:
+ DataFrame with nested fields extracted as flat columns
+
+ Example:
+ # Prepare customers with specific fields
+ customers_flat = handler.prepare(
+ "customers",
+ fields=['profile.contact.email', 'profile.name']
+ )
+
+ # Prepare with all nested fields
+ orders_flat = handler.prepare("orders")
+
+ # Now use prepared DataFrames with standard operations
+ result = customers_flat.join(orders_flat, ...)
+
+ Note:
+ This does NOT modify the original DataFrame in the handler.
+ It returns a new DataFrame with extracted fields.
+ """
+ df_handler = self.get(name)
+
+ if fields is None:
+ # Extract all nested fields
+ return df_handler.extract_nested_fields(verbose=verbose)
+ else:
+ # Validate all requested fields exist
+ for path in fields:
+ if path not in df_handler.extracted_fields:
+ raise ValueError(
+ f"Path '{path}' not found in nested structure. "
+ f"Available: {list(df_handler.extracted_fields.keys())}"
+ )
+
+ # Extract all nested fields first
+ extracted = df_handler.extract_nested_fields(verbose=verbose)
+
+ # Build list of columns to keep:
+ # 1. All original non-nested columns (keep regular columns)
+ # 2. Only the requested nested fields (by their extracted names)
+ needed_cols = []
+
+ # Check which original columns are top-level nested structs
+ # nested_fields has paths like "profile.contact.email", not "profile"
+ # So we need to check if a column is the root of any nested path
+ nested_root_columns = set()
+ for nested_path in df_handler.nested_fields.keys():
+ # Get the top-level column name (before first dot)
+ root = nested_path.split('.')[0]
+ nested_root_columns.add(root)
+
+ # Add non-nested original columns (exclude nested struct columns)
+ for col in df_handler.original_columns:
+ if col not in nested_root_columns:
+ needed_cols.append(col)
+
+ # Add requested nested fields (by their extracted names)
+ for path in fields:
+ extracted_name = df_handler.extracted_fields[path]
+ needed_cols.append(extracted_name)
+
+ # Select only the needed columns
+ return DataFrame(extracted._data.select(*needed_cols))
+
+ def join(
+ self,
+ tables: dict[str, str | list[str] | None],
+ on: list[tuple[str, str, str, str]] | None = None,
+ predicates: list | None = None,
+ how: str = "inner"
+ ) -> DataFrame:
+ """
+ Convenience method for SQL-like joins on multiple tables.
+
+ This is syntactic sugar over prepare() + Ibis operations. It:
+ 1. Prepares DataFrames (extracts nested fields if specified)
+ 2. Builds join predicates
+ 3. Delegates to Ibis for execution
+ 4. Returns result DataFrame
+
+ For complex queries (WHERE, HAVING, windows), use prepare() + direct Ibis!
+
+ Args:
+ tables: Dict mapping alias -> DataFrame name or list of fields to extract
+ - str value: Use DataFrame by name, extract all nested fields
+ - list value: Extract only specified nested fields
+
+ Examples:
+ - {"c": "customers"} # All fields from customers
+ - {"c": "customers", "o": "orders"} # All fields from both
+ - {"c": ["profile.email", "name"]} # Only specific fields
+
+ on: List of join conditions as (table1_alias, col1, table2_alias, col2) tuples
+ Each tuple specifies: (left_table, left_column, right_table, right_column)
+
+ Examples:
+ - [("c", "customer_id", "o", "customer_id")] # Single condition
+ - [("c", "email", "o", "customer_email"), # Multi-column join
+ ("c", "region", "o", "region")]
+
+ predicates: Alternative to 'on': provide raw Ibis predicates for complex conditions
+ If provided, 'on' is ignored
+
+ how: Join type - 'inner', 'left', 'right', 'outer', 'cross', 'semi', 'anti'
+ All Ibis join types supported (default: 'inner')
+
+ Returns:
+ Joined DataFrame (leanframe.DataFrame wrapping Ibis table)
+
+ Raises:
+ ValueError: If tables dict is empty, aliases invalid, or join conditions malformed
+ KeyError: If referenced DataFrames don't exist in handler
+
+ Examples:
+ # Simple two-table join
+ result = handler.join(
+ tables={"c": "customers", "o": "orders"},
+ on=[("c", "customer_id", "o", "customer_id")],
+ how="inner"
+ )
+
+ # Join with selective nested field extraction
+ # Note: Use dot notation naturally - gets converted to underscores internally
+ result = handler.join(
+ tables={
+ "c": ["profile.contact.email", "profile.name"], # Only these fields
+ "o": "orders" # All fields
+ },
+ on=[("c", "profile.contact.email", "o", "customer.email")], # Dot notation OK!
+ how="left"
+ )
+
+ # Three-table join with multiple conditions
+ result = handler.join(
+ tables={"c": "customers", "o": "orders", "p": "products"},
+ on=[
+ ("c", "customer_id", "o", "customer_id"),
+ ("o", "product_id", "p", "product_id")
+ ],
+ how="inner"
+ )
+
+ # Cross join (no conditions needed)
+ result = handler.join(
+ tables={"c": "customers", "p": "products"},
+ how="cross"
+ )
+
+ # Continue with Ibis operations on result
+ filtered = result._data.filter(result._data.amount > 100)
+ grouped = filtered.group_by("region").aggregate(total=filtered.amount.sum())
+ final = DataFrame(grouped)
+
+ Note:
+ For complex queries with WHERE, HAVING, window functions, etc.,
+ use prepare() directly and compose Ibis operations:
+
+ customers = handler.prepare("customers")
+ orders = handler.prepare("orders")
+
+ # Full Ibis power available
+ result = customers._data.join(orders._data, ...).filter(...).group_by(...)
+ """
+ if not tables:
+ raise ValueError("Must provide at least one table in 'tables' dict")
+
+ if not on and not predicates and how != "cross":
+ raise ValueError(
+ "Must provide either 'on' conditions or 'predicates', "
+ "or use how='cross' for cross join"
+ )
+
+ print(f"\n🔗 Joining {len(tables)} table(s) using '{how}' join...")
+
+ # Step 1: Prepare all tables (extract nested fields as needed)
+ prepared_tables: dict[str, DataFrame] = {}
+
+ for alias, spec in tables.items():
+ if isinstance(spec, str):
+ # It's a DataFrame name - prepare with all fields
+ print(f" 📋 Preparing '{alias}' from '{spec}' (all fields)")
+ prepared_tables[alias] = self.prepare(spec, verbose=False)
+ elif isinstance(spec, list):
+ # It's a list of field paths - selective extraction
+ # Need to extract the DataFrame name from the alias
+ # For now, assume the first table entry is the reference
+ # This is a limitation - might need to adjust API
+ raise NotImplementedError(
+ "Selective field extraction syntax not yet implemented.\n"
+ "Use prepare() with fields parameter, then pass DataFrame name:\n"
+ " handler.add('c_prepared', handler.prepare('customers', fields=[...]))\n"
+ " result = handler.join(tables={'c': 'c_prepared', ...}, ...)"
+ )
+ else:
+ raise ValueError(
+ f"Invalid table spec for alias '{alias}': {type(spec)}. "
+ f"Expected str (DataFrame name) or list (field paths)"
+ )
+
+ # Step 2: Build the join chain
+ # Start with the first table
+ aliases = list(prepared_tables.keys())
+ if len(aliases) == 0:
+ raise ValueError("No tables to join")
+
+ # Get first table
+ result_table = prepared_tables[aliases[0]]._data
+ print(f" ✅ Starting with '{aliases[0]}': {len(result_table.columns)} columns")
+
+ # Join with remaining tables sequentially
+ for i in range(1, len(aliases)):
+ right_alias = aliases[i]
+ right_table = prepared_tables[right_alias]._data
+
+ print(f" 🔗 Joining with '{right_alias}': {len(right_table.columns)} columns")
+
+ # Build predicates for this join
+ if predicates is not None:
+ # Use raw Ibis predicates (advanced usage)
+ join_predicates = predicates
+ elif on:
+ # Build predicates from 'on' conditions
+ # Filter conditions that involve the current right table
+ join_predicates = []
+ for condition in on:
+ if len(condition) != 4:
+ raise ValueError(
+ f"Invalid join condition: {condition}. "
+ f"Expected (table1_alias, col1, table2_alias, col2)"
+ )
+
+ left_alias, left_col, right_alias_cond, right_col = condition
+
+ # Check if this condition involves the current right table
+ if right_alias_cond == right_alias:
+ # Convert dot notation to underscore (user convenience)
+ # User can write "profile.contact.email" and we convert to "profile_contact_email"
+ left_col_normalized = left_col.replace(".", "_")
+ right_col_normalized = right_col.replace(".", "_")
+
+ # Build Ibis predicate
+ # Need to access the correct table - tricky with chained joins
+ # For now, use column name directly
+ join_predicates.append(
+ (result_table[left_col_normalized], right_table[right_col_normalized])
+ )
+ else:
+ join_predicates = []
+
+ # Perform the join
+ if join_predicates:
+ result_table = result_table.join(
+ right_table,
+ predicates=join_predicates,
+ how=how # type: ignore
+ )
+ else:
+ # Cross join (no predicates)
+ result_table = result_table.join(
+ right_table,
+ how=how # type: ignore
+ )
+
+ result_df = DataFrame(result_table)
+ print(f" ✅ Join complete: {len(result_df.columns)} total columns")
+
+ return result_df
+
+ # Legacy join methods - DEPRECATED
+ # Use prepare() + direct Ibis operations instead for full SQL flexibility
+
+
+ def __repr__(self) -> str:
+ count = len(self._handlers)
+ names = ", ".join(self._handlers.keys()) if count > 0 else "empty"
+ return f"NestedHandler({count} DataFrames: {names})"
+
+ def __len__(self) -> int:
+ """Return number of managed DataFrames."""
+ return len(self._handlers)
+
+ def __contains__(self, name: str) -> bool:
+ """Check if a DataFrame name exists in the handler."""
+ return name in self._handlers
diff --git a/tests/unit/nested_data/test_dynamic_access.py b/tests/unit/nested_data/test_dynamic_access.py
index 01f3139..9bb83d0 100644
--- a/tests/unit/nested_data/test_dynamic_access.py
+++ b/tests/unit/nested_data/test_dynamic_access.py
@@ -1,18 +1,18 @@
#!/usr/bin/env python3
-"""Test the dictionary-like access interface of the DynamicNestedHandler"""
+"""Test the dictionary-like access interface of the DataFrameHandler"""
from demos.utils.create_nested_data import (
create_simple_nested_dataframe,
create_extended_nested_dataframe,
)
-from leanframe.core.frame import DynamicNestedHandler
+from leanframe.core.frame import DataFrameHandler
def test_dictionary_access():
- """Test dictionary-like access interface of DynamicNestedHandler."""
+ """Test dictionary-like access interface of DataFrameHandler."""
# Create a simple nested DataFrame
df = create_simple_nested_dataframe()
- handler = DynamicNestedHandler(df)
+ handler = DataFrameHandler(df)
# Test keys are available
keys = list(handler.keys())
@@ -52,7 +52,7 @@ def test_dictionary_access():
def test_extended_structure_access():
"""Test dictionary access with extended nested structure."""
extended_df = create_extended_nested_dataframe()
- extended_handler = DynamicNestedHandler(extended_df)
+ extended_handler = DataFrameHandler(extended_df)
# Test extended keys include address fields
keys = list(extended_handler.keys())
diff --git a/tests/unit/nested_data/test_dynamic_nested_handler.py b/tests/unit/nested_data/test_dynamic_nested_handler.py
index c60110a..6cca862 100644
--- a/tests/unit/nested_data/test_dynamic_nested_handler.py
+++ b/tests/unit/nested_data/test_dynamic_nested_handler.py
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
-Tests for DynamicNestedHandler comprehensive functionality.
+Tests for DataFrameHandler comprehensive functionality.
This tests a truly dynamic handler that can work with ANY nested DataFrame structure.
Key features:
@@ -12,13 +12,13 @@
"""
from demos.utils.create_nested_data import create_simple_nested_dataframe, create_extended_nested_dataframe, create_deeply_nested_dataframe
-from leanframe.core.frame import DynamicNestedHandler
+from leanframe.core.frame import DataFrameHandler
def test_basic_usage():
- """Test basic DynamicNestedHandler usage and structure inspection."""
+ """Test basic DataFrameHandler usage and structure inspection."""
df = create_simple_nested_dataframe(5)
- handler = DynamicNestedHandler(df)
+ handler = DataFrameHandler(df)
# Test basic structure
assert len(handler.original_columns) == 3 # id, person, contact
@@ -49,9 +49,9 @@ def test_basic_usage():
def test_data_access():
- """Test data access patterns of DynamicNestedHandler."""
+ """Test data access patterns of DataFrameHandler."""
df = create_simple_nested_dataframe(3)
- handler = DynamicNestedHandler(df)
+ handler = DataFrameHandler(df)
# Test column-wise access
names = handler.get_column("person_name")
@@ -77,25 +77,27 @@ def test_data_access():
def test_filtering():
- """Test filtering functionality of DynamicNestedHandler."""
+ """Test filtering functionality using extracted DataFrame (stateless approach)."""
df = create_simple_nested_dataframe(5)
- handler = DynamicNestedHandler(df)
-
- # Test filtering by age - returns a new handler
- filtered_handler = handler.filter_by("person_age", 30)
- assert isinstance(filtered_handler, DynamicNestedHandler)
-
+ handler = DataFrameHandler(df)
+
+ # Extract the flattened DataFrame
+ extracted_df = handler.extract_nested_fields(verbose=False)
+
+ # Filter using pandas (simpler for testing)
+ filtered_pandas = extracted_df.to_pandas()
+ filtered_by_age = filtered_pandas[filtered_pandas["person_age"] == 30]
+
# Test the filtered results
- if len(filtered_handler) > 0:
- for record in filtered_handler:
- assert record["person_age"] == 30
+ if len(filtered_by_age) > 0:
+ assert all(filtered_by_age["person_age"] == 30)
def test_different_structures():
- """Test DynamicNestedHandler with different nested structures."""
+ """Test DataFrameHandler with different nested structures."""
# Extended structure with address
extended_df = create_extended_nested_dataframe(2)
- extended_handler = DynamicNestedHandler(extended_df)
+ extended_handler = DataFrameHandler(extended_df)
expected_extended_columns = [
"id",
@@ -119,9 +121,9 @@ def test_different_structures():
def test_deep_nesting():
- """Test DynamicNestedHandler with deeply nested structures."""
+ """Test DataFrameHandler with deeply nested structures."""
deep_df = create_deeply_nested_dataframe()
- deep_handler = DynamicNestedHandler(deep_df)
+ deep_handler = DataFrameHandler(deep_df)
# Verify deep structure handling
assert len(deep_handler.columns) >= 5 # Should have multiple levels extracted
@@ -142,9 +144,9 @@ def test_deep_nesting():
def test_handler_capabilities():
- """Test general capabilities and edge cases of DynamicNestedHandler."""
+ """Test general capabilities and edge cases of DataFrameHandler."""
df = create_simple_nested_dataframe(2)
- handler = DynamicNestedHandler(df)
+ handler = DataFrameHandler(df)
# Test length
assert len(handler) == 2
diff --git a/tests/unit/nested_data/test_join_convenience.py b/tests/unit/nested_data/test_join_convenience.py
new file mode 100644
index 0000000..13d6287
--- /dev/null
+++ b/tests/unit/nested_data/test_join_convenience.py
@@ -0,0 +1,280 @@
+"""Test the convenience join() method in NestedHandler."""
+
+import pytest
+import ibis
+import pandas as pd
+
+from leanframe.core.frame import DataFrame
+from leanframe.core.nested_handler import NestedHandler
+
+
+@pytest.fixture
+def duckdb_backend():
+ """Create a DuckDB backend for testing."""
+ return ibis.duckdb.connect()
+
+
+@pytest.fixture
+def sample_data(duckdb_backend):
+ """Create sample test data with nested and flat tables.
+
+ IMPORTANT: All tables use the SAME backend connection to ensure
+ they can be joined within a single NestedHandler.
+ """
+ # Customers with nested profile
+ customers_data = pd.DataFrame({
+ 'customer_id': [1, 2, 3],
+ 'profile': [
+ {'name': 'Alice', 'contact': {'email': 'alice@example.com', 'phone': '555-0001'}},
+ {'name': 'Bob', 'contact': {'email': 'bob@example.com', 'phone': '555-0002'}},
+ {'name': 'Charlie', 'contact': {'email': 'charlie@example.com', 'phone': '555-0003'}},
+ ]
+ })
+
+ # Orders (flat)
+ orders_data = pd.DataFrame({
+ 'order_id': [101, 102, 103, 104],
+ 'customer_id': [1, 2, 1, 3],
+ 'amount': [100.0, 200.0, 150.0, 75.0]
+ })
+
+ # Products (flat)
+ products_data = pd.DataFrame({
+ 'product_id': [1, 2, 3],
+ 'name': ['Widget', 'Gadget', 'Doohickey'],
+ 'price': [29.99, 149.99, 9.99]
+ })
+
+ # Create all tables on the SAME backend
+ customers_table = duckdb_backend.create_table('test_customers', customers_data, temp=True)
+ orders_table = duckdb_backend.create_table('test_orders', orders_data, temp=True)
+ products_table = duckdb_backend.create_table('test_products', products_data, temp=True)
+
+ return {
+ 'backend': duckdb_backend, # Include backend for tests that need it
+ 'customers': DataFrame(customers_table),
+ 'orders': DataFrame(orders_table),
+ 'products': DataFrame(products_table)
+ }
+
+
+def test_join_two_tables_simple(sample_data):
+ """Test simple two-table inner join."""
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+ handler.add("orders", sample_data['orders'])
+
+ # Join on customer_id
+ result = handler.join(
+ tables={"c": "customers", "o": "orders"},
+ on=[("c", "customer_id", "o", "customer_id")],
+ how="inner"
+ )
+
+ # Verify result
+ result_pd = result.to_pandas()
+ assert len(result_pd) == 4 # 4 orders
+ assert 'customer_id' in result_pd.columns
+ assert 'order_id' in result_pd.columns
+ assert 'amount' in result_pd.columns
+ assert 'profile_name' in result_pd.columns # Nested field extracted
+ assert 'profile_contact_email' in result_pd.columns
+
+
+def test_join_handles_nested_extraction(sample_data):
+ """Test that join automatically extracts nested fields."""
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+ handler.add("orders", sample_data['orders'])
+
+ result = handler.join(
+ tables={"c": "customers", "o": "orders"},
+ on=[("c", "customer_id", "o", "customer_id")],
+ how="left"
+ )
+
+ columns = result.columns
+
+ # Check nested fields were extracted
+ assert 'profile_name' in columns
+ assert 'profile_contact_email' in columns
+ assert 'profile_contact_phone' in columns
+
+ # Check original nested column is gone
+ assert 'profile' not in columns
+
+
+def test_join_three_tables(sample_data):
+ """Test three-table join."""
+ # Add product_id to orders for this test
+ orders_pd = sample_data['orders'].to_pandas()
+ orders_pd['product_id'] = [1, 2, 1, 3]
+
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+
+ # CRITICAL: Use the SAME backend from sample_data fixture
+ # All DataFrames in one NestedHandler must share the same backend!
+ backend = sample_data['backend']
+ orders_table = backend.create_table('orders_with_products', orders_pd, temp=True)
+ handler.add("orders", DataFrame(orders_table))
+ handler.add("products", sample_data['products'])
+
+ # Three-table join
+ result = handler.join(
+ tables={"c": "customers", "o": "orders", "p": "products"},
+ on=[
+ ("c", "customer_id", "o", "customer_id"),
+ ("o", "product_id", "p", "product_id")
+ ],
+ how="inner"
+ )
+
+ # Verify all tables are joined
+ result_pd = result.to_pandas()
+ assert 'customer_id' in result_pd.columns
+ assert 'order_id' in result_pd.columns
+ assert 'product_id' in result_pd.columns
+ assert 'profile_name' in result_pd.columns
+ assert 'name' in result_pd.columns # Product name
+ assert 'price' in result_pd.columns
+
+
+def test_join_different_types(sample_data):
+ """Test different join types."""
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+ handler.add("orders", sample_data['orders'])
+
+ # Inner join
+ inner = handler.join(
+ tables={"c": "customers", "o": "orders"},
+ on=[("c", "customer_id", "o", "customer_id")],
+ how="inner"
+ )
+ inner_pd = inner.to_pandas()
+ assert len(inner_pd) == 4 # Only customers with orders
+
+ # Left join - all customers
+ left = handler.join(
+ tables={"c": "customers", "o": "orders"},
+ on=[("c", "customer_id", "o", "customer_id")],
+ how="left"
+ )
+ left_pd = left.to_pandas()
+ # Should include all customers (3) × their orders
+ # Customer 1 has 2 orders, customer 2 has 1, customer 3 has 1
+ assert len(left_pd) == 4
+
+
+def test_join_empty_tables_dict_raises_error(sample_data):
+ """Test that empty tables dict raises error."""
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+
+ with pytest.raises(ValueError, match="Must provide at least one table"):
+ handler.join(tables={}, on=[], how="inner")
+
+
+def test_join_missing_conditions_raises_error(sample_data):
+ """Test that missing join conditions raises error for non-cross joins."""
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+ handler.add("orders", sample_data['orders'])
+
+ with pytest.raises(ValueError, match="Must provide either"):
+ handler.join(
+ tables={"c": "customers", "o": "orders"},
+ how="inner"
+ )
+
+
+def test_join_nonexistent_dataframe_raises_error(sample_data):
+ """Test that referencing nonexistent DataFrame raises error."""
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+
+ with pytest.raises(KeyError, match="not found"):
+ handler.join(
+ tables={"c": "customers", "o": "nonexistent"},
+ on=[("c", "customer_id", "o", "customer_id")],
+ how="inner"
+ )
+
+
+def test_join_result_can_be_further_processed(sample_data):
+ """Test that join result can be used for further Ibis operations."""
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+ handler.add("orders", sample_data['orders'])
+
+ # Join
+ result = handler.join(
+ tables={"c": "customers", "o": "orders"},
+ on=[("c", "customer_id", "o", "customer_id")],
+ how="inner"
+ )
+
+ # Continue with Ibis operations
+ filtered = result._data.filter(result._data.amount > 100)
+ filtered_df = DataFrame(filtered)
+
+ filtered_pd = filtered_df.to_pandas()
+ assert len(filtered_pd) == 2 # Only 2 orders > 100
+ assert all(filtered_pd['amount'] > 100)
+
+
+def test_join_multi_column_conditions(sample_data):
+ """Test join with multiple column conditions."""
+ # CRITICAL: Use the SAME backend from sample_data fixture
+ backend = sample_data['backend']
+
+ table1_data = pd.DataFrame({
+ 'id': [1, 2, 3],
+ 'region': ['US', 'EU', 'US'],
+ 'value': [10, 20, 30]
+ })
+
+ table2_data = pd.DataFrame({
+ 'id': [1, 2, 1],
+ 'region': ['US', 'EU', 'EU'],
+ 'amount': [100, 200, 150]
+ })
+
+ table1 = backend.create_table('t1', table1_data, temp=True)
+ table2 = backend.create_table('t2', table2_data, temp=True)
+
+ handler = NestedHandler()
+ handler.add("t1", DataFrame(table1))
+ handler.add("t2", DataFrame(table2))
+
+ # Multi-column join
+ result = handler.join(
+ tables={"a": "t1", "b": "t2"},
+ on=[
+ ("a", "id", "b", "id"),
+ ("a", "region", "b", "region")
+ ],
+ how="inner"
+ )
+
+ result_pd = result.to_pandas()
+ # Should only match rows where BOTH id AND region match
+ assert len(result_pd) == 2 # (1, US) and (2, EU)
+
+
+def test_join_cross_join(sample_data):
+ """Test cross join (Cartesian product)."""
+ handler = NestedHandler()
+ handler.add("customers", sample_data['customers'])
+ handler.add("products", sample_data['products'])
+
+ # Cross join - no conditions needed
+ result = handler.join(
+ tables={"c": "customers", "p": "products"},
+ how="cross"
+ )
+
+ result_pd = result.to_pandas()
+ # 3 customers × 3 products = 9 rows
+ assert len(result_pd) == 9
diff --git a/tests/unit/nested_data/test_nested_handler.py b/tests/unit/nested_data/test_nested_handler.py
index 3ff985d..e809012 100644
--- a/tests/unit/nested_data/test_nested_handler.py
+++ b/tests/unit/nested_data/test_nested_handler.py
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
-"""Test the DynamicNestedHandler from its proper location in leanframe.core"""
+"""Test the DataFrameHandler from its proper location in leanframe.core"""
-from leanframe.core.frame import DynamicNestedHandler
+from leanframe.core.frame import DataFrameHandler
from demos.utils.create_nested_data import (
create_simple_nested_dataframe,
create_extended_nested_dataframe,
@@ -12,7 +12,7 @@
def test_simple_nested_structure():
"""Test dynamic handler with simple nested structure (person + contact)."""
simple_df = create_simple_nested_dataframe(3)
- handler = DynamicNestedHandler(simple_df)
+ handler = DataFrameHandler(simple_df)
# Test basic properties
assert len(handler) == 3
@@ -31,7 +31,7 @@ def test_simple_nested_structure():
def test_extended_nested_structure():
"""Test dynamic handler with extended nested structure (person + contact + address)."""
extended_df = create_extended_nested_dataframe(3)
- handler = DynamicNestedHandler(extended_df)
+ handler = DataFrameHandler(extended_df)
# Test extended properties
assert len(handler) == 3
@@ -52,7 +52,7 @@ def test_extended_nested_structure():
def test_deeply_nested_structure():
"""Test dynamic handler with deeply nested structure."""
deep_df = create_deeply_nested_dataframe()
- handler = DynamicNestedHandler(deep_df)
+ handler = DataFrameHandler(deep_df)
# Test deep nesting handling (create_deeply_nested_dataframe creates 2 records by default)
assert len(handler) == 2
@@ -90,7 +90,7 @@ def test_handler_adaptability():
]
for i, df in enumerate(structures):
- handler = DynamicNestedHandler(df)
+ handler = DataFrameHandler(df)
# Each should work regardless of structure
assert len(handler) >= 2
diff --git a/tests/unit/nested_data/test_prepare_method.py b/tests/unit/nested_data/test_prepare_method.py
new file mode 100644
index 0000000..4b641a5
--- /dev/null
+++ b/tests/unit/nested_data/test_prepare_method.py
@@ -0,0 +1,175 @@
+"""Test the new prepare() method in NestedHandler."""
+
+import pytest
+import ibis
+import pandas as pd
+
+from leanframe.core.frame import DataFrame
+from leanframe.core.nested_handler import NestedHandler
+
+
+@pytest.fixture
+def duckdb_backend():
+ """Create a DuckDB backend for testing."""
+ return ibis.duckdb.connect()
+
+
+@pytest.fixture
+def nested_customers_df(duckdb_backend):
+ """Create a DataFrame with nested customer data."""
+ data = pd.DataFrame({
+ 'customer_id': [1, 2, 3],
+ 'profile': [
+ {'name': 'Alice', 'contact': {'email': 'alice@example.com', 'phone': '555-0001'}},
+ {'name': 'Bob', 'contact': {'email': 'bob@example.com', 'phone': '555-0002'}},
+ {'name': 'Charlie', 'contact': {'email': 'charlie@example.com', 'phone': '555-0003'}},
+ ]
+ })
+
+ table = duckdb_backend.create_table('test_customers', data, temp=True)
+ return DataFrame(table)
+
+
+def test_prepare_extracts_all_nested_fields(nested_customers_df):
+ """Test that prepare() extracts all nested fields by default."""
+ handler = NestedHandler()
+ handler.add("customers", nested_customers_df)
+
+ # Prepare should extract all nested fields
+ prepared = handler.prepare("customers")
+
+ # Check that nested fields were extracted
+ columns = prepared.columns
+ assert 'customer_id' in columns # Original flat column
+ assert 'profile_name' in columns # Extracted from profile.name
+ assert 'profile_contact_email' in columns # Extracted from profile.contact.email
+ assert 'profile_contact_phone' in columns # Extracted from profile.contact.phone
+
+ # Original nested column should not be in prepared DataFrame
+ assert 'profile' not in columns
+
+
+def test_prepare_with_specific_fields(nested_customers_df):
+ """Test that prepare() can extract specific fields only."""
+ handler = NestedHandler()
+ handler.add("customers", nested_customers_df)
+
+ # Prepare with specific fields
+ prepared = handler.prepare(
+ "customers",
+ fields=['profile.contact.email', 'profile.name']
+ )
+
+ columns = prepared.columns
+
+ # Should have requested fields
+ assert 'profile_contact_email' in columns
+ assert 'profile_name' in columns
+
+ # Should have original flat columns
+ assert 'customer_id' in columns
+
+ # Should NOT have unrequested nested fields
+ assert 'profile_contact_phone' not in columns
+
+
+def test_prepare_nonexistent_field_raises_error(nested_customers_df):
+ """Test that prepare() raises error for nonexistent fields."""
+ handler = NestedHandler()
+ handler.add("customers", nested_customers_df)
+
+ with pytest.raises(ValueError, match="not found in nested structure"):
+ handler.prepare("customers", fields=['profile.nonexistent.field'])
+
+
+def test_prepare_nonexistent_dataframe_raises_error():
+ """Test that prepare() raises error for nonexistent DataFrame."""
+ handler = NestedHandler()
+
+ with pytest.raises(KeyError, match="not found"):
+ handler.prepare("nonexistent")
+
+
+def test_prepare_does_not_modify_original(nested_customers_df):
+ """Test that prepare() doesn't modify the original DataFrame in handler."""
+ handler = NestedHandler()
+ handler.add("customers", nested_customers_df)
+
+ # Get original handler
+ original_handler = handler.get("customers")
+ original_columns = original_handler.original_columns
+
+ # Prepare
+ prepared = handler.prepare("customers")
+
+ # Original should be unchanged
+ assert original_handler.original_columns == original_columns
+ assert 'profile' in original_handler.original_columns
+
+ # Prepared should be different
+ assert 'profile' not in prepared.columns
+
+
+def test_prepare_enables_direct_ibis_joins(nested_customers_df, duckdb_backend):
+ """Test that prepared DataFrames can be joined using direct Ibis operations."""
+ # Create orders DataFrame
+ orders_data = pd.DataFrame({
+ 'order_id': [101, 102, 103],
+ 'customer_email': ['alice@example.com', 'bob@example.com', 'alice@example.com'],
+ 'amount': [100.0, 200.0, 150.0]
+ })
+ orders_table = duckdb_backend.create_table('test_orders', orders_data, temp=True)
+ orders_df = DataFrame(orders_table)
+
+ # Add both to handler
+ handler = NestedHandler()
+ handler.add("customers", nested_customers_df)
+ handler.add("orders", orders_df)
+
+ # Prepare both
+ customers_flat = handler.prepare("customers")
+ orders_flat = handler.prepare("orders")
+
+ # Join using direct Ibis operations
+ joined = customers_flat._data.join(
+ orders_flat._data,
+ predicates=[
+ customers_flat._data.profile_contact_email == orders_flat._data.customer_email
+ ],
+ how="inner"
+ )
+
+ result = DataFrame(joined)
+
+ # Verify join worked
+ result_pd = result.to_pandas()
+ assert len(result_pd) == 3 # 3 orders
+ assert 'customer_id' in result_pd.columns
+ assert 'order_id' in result_pd.columns
+ assert 'amount' in result_pd.columns
+ assert 'profile_contact_email' in result_pd.columns
+
+
+def test_prepare_with_flat_dataframe(duckdb_backend):
+ """Test that prepare() works with already-flat DataFrames."""
+ # Create flat DataFrame (no nested columns)
+ data = pd.DataFrame({
+ 'id': [1, 2, 3],
+ 'name': ['Alice', 'Bob', 'Charlie'],
+ 'age': [25, 30, 35]
+ })
+
+ table = duckdb_backend.create_table('test_flat', data, temp=True)
+ flat_df = DataFrame(table)
+
+ handler = NestedHandler()
+ handler.add("flat", flat_df)
+
+ # Prepare should work even without nested fields
+ prepared = handler.prepare("flat")
+
+ # Should have all original columns
+ columns = prepared.columns
+ assert 'id' in columns
+ assert 'name' in columns
+ assert 'age' in columns