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