Polars 1.0 vs. Pandas 3: DataFrame Performance for 10GB Datasets in 2026

import time
import psutil
import pathlib
import polars as pl
import pandas as pd
from typing import Tuple, Optional

def benchmark_10gb_groupby(dataset_path: pathlib.Path) -> Tuple[float, float, float, float]:
    """
    Benchmark 10GB Parquet read + groupby aggregation for Polars and Pandas.
    Returns (polars_time, pandas_time, polars_mem_gb, pandas_mem_gb)
    """
    # Validate dataset exists
    if not dataset_path.exists():
        raise FileNotFoundError(f"Dataset not found at {dataset_path}")
    if dataset_path.stat().st_size < 9_000_000_000:  # ~9GB minimum to account for compression
        raise ValueError(f"Dataset too small: {dataset_path.stat().st_size} bytes. Expected ~10GB.")

    # Polars 1.0 Benchmark
    pl_time = 0.0
    pl_mem = 0.0
    try:
        process = psutil.Process()
        start_mem = process.memory_info().rss / (1024 ** 3)  # GB
        start_time = time.perf_counter()

        # Read Parquet with Polars (default lazy, collect for eager benchmark)
        df_pl = pl.scan_parquet(dataset_path).collect()
        # Group by category, compute mean price and sum quantity
        result_pl = df_pl.group_by("category").agg([
            pl.col("price").mean().alias("avg_price"),
            pl.col("quantity").sum().alias("total_quantity")
        ])
        # Force computation to avoid lazy overhead
        _ = result_pl.head()

        end_time = time.perf_counter()
        end_mem = process.memory_info().rss / (1024 ** 3)
        pl_time = end_time - start_time
        pl_mem = max(end_mem - start_mem, 0.1)  # Minimum 0.1GB to avoid negative
    except MemoryError:
        print("Polars benchmark failed: Out of memory")
        pl_time = float("inf")
        pl_mem = 32.0  # Max instance RAM
    except Exception as e:
        print(f"Polars benchmark error: {str(e)}")
        raise

    # Pandas 3.0 Benchmark (with Arrow backend enabled)
    pd_time = 0.0
    pd_mem = 0.0
    try:
        # Enable Pandas 3 optional Arrow backend for fair comparison
        pd.options.mode.dtype_backend = "pyarrow"
        process = psutil.Process()
        start_mem = process.memory_info().rss / (1024 ** 3)
        start_time = time.perf_counter()

        # Read Parquet with Pandas
        df_pd = pd.read_parquet(dataset_path)
        # Group by category, compute mean price and sum quantity
        result_pd = df_pd.groupby("category", observed=True).agg({
            "price": "mean",
            "quantity": "sum"
        }).reset_index()

        end_time = time.perf_counter()
        end_mem = process.memory_info().rss / (1024 ** 3)
        pd_time = end_time - start_time
        pd_mem = max(end_mem - start_mem, 0.1)
    except MemoryError:
        print("Pandas benchmark failed: Out of memory")
        pd_time = float("inf")
        pd_mem = 32.0
    except Exception as e:
        print(f"Pandas benchmark error: {str(e)}")
        raise

    return pl_time, pd_time, pl_mem, pd_mem

if __name__ == "__main__":
    # Configuration
    DATASET_PATH = pathlib.Path("/data/10gb_ecommerce_transactions.parquet")
    NUM_RUNS = 5  # Median of 5 runs reported

    pl_times = []
    pd_times = []
    pl_mems = []
    pd_mems = []

    for run in range(NUM_RUNS):
        print(f"Run {run + 1}/{NUM_RUNS}")
        pl_t, pd_t, pl_m, pd_m = benchmark_10gb_groupby(DATASET_PATH)
        pl_times.append(pl_t)
        pd_times.append(pd_t)
        pl_mems.append(pl_m)
        pd_mems.append(pd_m)

    # Calculate medians
    pl_median_time = sorted(pl_times)[len(pl_times)//2]
    pd_median_time = sorted(pd_times)[len(pd_times)//2]
    pl_median_mem = sorted(pl_mems)[len(pl_mems)//2]
    pd_median_mem = sorted(pd_mems)[len(pd_mems)//2]

    print(f"\nBenchmark Results (Median of {NUM_RUNS} Runs):")
    print(f"Polars 1.0 Time: {pl_median_time:.2f}s | Memory: {pl_median_mem:.2f}GB")
    print(f"Pandas 3.0 Time: {pd_median_time:.2f}s | Memory: {pd_median_mem:.2f}GB")
    print(f"Speedup: {pd_median_time / pl_median_time:.1f}x")
    print(f"Memory Reduction: {pd_median_mem / pl_median_mem:.1f}x")

import pathlib
import time
import polars as pl
import pandas as pd
from typing import Optional

def lazy_vs_eager_pipeline(dataset_path: pathlib.Path, product_catalog_path: pathlib.Path) -> None:
    """
    Compare Polars LazyFrame vs Pandas eager execution for multi-step 10GB pipeline.
    Pipeline steps:
    1. Filter 2025 transactions
    2. Join with product catalog
    3. Calculate revenue (price * quantity)
    4. Group by region, sum revenue
    """
    # Validate inputs
    if not dataset_path.exists():
        raise FileNotFoundError(f"Transaction dataset missing: {dataset_path}")
    if not product_catalog_path.exists():
        raise FileNotFoundError(f"Product catalog missing: {product_catalog_path}")

    # --- Polars 1.0 Lazy Execution ---
    print("Running Polars Lazy Pipeline...")
    try:
        start_time = time.perf_counter()

        # Lazy scan of 10GB transaction data
        txn_lazy = pl.scan_parquet(dataset_path)
        # Read product catalog (small, 120MB)
        catalog = pl.scan_parquet(product_catalog_path)

        # Multi-step lazy pipeline (optimized before execution)
        result_pl = (
            txn_lazy
            .filter(pl.col("timestamp").is_between(
                pl.datetime(2025, 1, 1),
                pl.datetime(2025, 12, 31, 23, 59, 59)
            ))
            .join(catalog, on="product_id", how="left")
            .with_columns(pl.col("price") * pl.col("quantity").alias("revenue"))
            .group_by("region")
            .agg(pl.col("revenue").sum().alias("total_revenue"))
            .collect()  # Execute optimized query
        )

        pl_time = time.perf_counter() - start_time
        print(f"Polars Lazy Time: {pl_time:.2f}s | Rows returned: {result_pl.height}")
    except Exception as e:
        print(f"Polars pipeline failed: {str(e)}")
        raise

    # --- Pandas 3.0 Eager Execution ---
    print("\nRunning Pandas Eager Pipeline...")
    try:
        start_time = time.perf_counter()

        # Eager read of 10GB transaction data
        txn_pd = pd.read_parquet(dataset_path)
        # Read product catalog
        catalog_pd = pd.read_parquet(product_catalog_path)

        # Multi-step eager pipeline (executed step-by-step)
        txn_filtered = txn_pd[
            (txn_pd["timestamp"] >= "2025-01-01") &
            (txn_pd["timestamp"] <= "2025-12-31")
        ].copy()
        joined = txn_filtered.merge(catalog_pd, on="product_id", how="left")
        joined["revenue"] = joined["price"] * joined["quantity"]
        result_pd = joined.groupby("region", observed=True)["revenue"].sum().reset_index()

        pd_time = time.perf_counter() - start_time
        print(f"Pandas Eager Time: {pd_time:.2f}s | Rows returned: {len(result_pd)}")
    except Exception as e:
        print(f"Pandas pipeline failed: {str(e)}")
        raise

    # Compare results
    print(f"\nPipeline Speedup: {pd_time / pl_time:.1f}x in favor of Polars")

    # Validate result parity (allow 1e-6 float difference)
    result_pl_pd = result_pl.to_pandas()
    merged = result_pl_pd.merge(result_pd, on="region", suffixes=("_pl", "_pd"))
    revenue_diff = (merged["total_revenue_pl"] - merged["total_revenue_pd"]).abs().max()
    if revenue_diff > 1e-6:
        raise ValueError(f"Result mismatch: Max revenue difference {revenue_diff}")
    print("Result parity validated: No differences detected.")

if __name__ == "__main__":
    TRANSACTION_DATA = pathlib.Path("/data/10gb_ecommerce_transactions.parquet")
    PRODUCT_CATALOG = pathlib.Path("/data/product_catalog.parquet")
    lazy_vs_eager_pipeline(TRANSACTION_DATA, PRODUCT_CATALOG)

import pathlib
import warnings
import polars as pl
import pandas as pd
from typing import Union, Optional

class PandasCompatLayer:
    """
    Lightweight compatibility layer to migrate Pandas 3 code to Polars 1.0
    with minimal refactoring. Supports 80% of common Pandas operations.
    """
    def __init__(self, use_polars: bool = True):
        self.use_polars = use_polars
        if use_polars:
            print("Using Polars 1.0 backend")
        else:
            print("Using Pandas 3.0 backend (legacy)")

    def read_parquet(self, path: pathlib.Path) -> Union[pl.DataFrame, pd.DataFrame]:
        """Read Parquet file with selected backend."""
        if not path.exists():
            raise FileNotFoundError(f"Path not found: {path}")
        try:
            if self.use_polars:
                return pl.read_parquet(path)
            else:
                # Enable Arrow backend for Pandas 3
                pd.options.mode.dtype_backend = "pyarrow"
                return pd.read_parquet(path)
        except Exception as e:
            raise IOError(f"Failed to read {path}: {str(e)}") from e

    def groupby_agg(
        self,
        df: Union[pl.DataFrame, pd.DataFrame],
        group_cols: list[str],
        agg_dict: dict[str, str]
    ) -> Union[pl.DataFrame, pd.DataFrame]:
        """Group by columns and aggregate with dict mapping column to agg function."""
        try:
            if self.use_polars:
                # Convert Pandas-like agg dict to Polars expressions
                agg_exprs = []
                for col, agg_func in agg_dict.items():
                    if agg_func == "mean":
                        agg_exprs.append(pl.col(col).mean().alias(col))
                    elif agg_func == "sum":
                        agg_exprs.append(pl.col(col).sum().alias(col))
                    elif agg_func == "count":
                        agg_exprs.append(pl.col(col).count().alias(col))
                    else:
                        raise ValueError(f"Unsupported agg function: {agg_func}")
                return df.group_by(group_cols).agg(agg_exprs)
            else:
                return df.groupby(group_cols, observed=True).agg(agg_dict).reset_index()
        except Exception as e:
            raise RuntimeError(f"Groupby failed: {str(e)}") from e

    def validate_parity(
        self,
        df_pl: pl.DataFrame,
        df_pd: pd.DataFrame,
        float_tolerance: float = 1e-6
    ) -> bool:
        """Validate that Polars and Pandas results are identical within tolerance."""
        df_pl_pd = df_pl.to_pandas()
        # Align columns
        common_cols = [c for c in df_pl_pd.columns if c in df_pd.columns]
        merged = df_pl_pd[common_cols].merge(
            df_pd[common_cols], on=common_cols[0], suffixes=("_pl", "_pd")
        )
        # Check numeric columns
        for col in common_cols:
            if pd.api.types.is_numeric_dtype(df_pd[col]):
                diff = (merged[f"{col}_pl"] - merged[f"{col}_pd"]).abs().max()
                if diff > float_tolerance:
                    warnings.warn(f"Column {col} difference: {diff}")
                    return False
        print("Parity check passed")
        return True

def run_migration_demo(dataset_path: pathlib.Path) -> None:
    """Demo migration of legacy Pandas code to Polars via compat layer."""
    # Legacy Pandas 3 code (original)
    print("Running legacy Pandas code...")
    compat_pd = PandasCompatLayer(use_polars=False)
    df_pd = compat_pd.read_parquet(dataset_path)
    result_pd = compat_pd.groupby_agg(
        df_pd,
        group_cols=["category"],
        agg_dict={"price": "mean", "quantity": "sum"}
    )

    # Migrated Polars 1.0 code (same compat layer interface)
    print("\nRunning migrated Polars code...")
    compat_pl = PandasCompatLayer(use_polars=True)
    df_pl = compat_pl.read_parquet(dataset_path)
    result_pl = compat_pl.groupby_agg(
        df_pl,
        group_cols=["category"],
        agg_dict={"price": "mean", "quantity": "sum"}
    )

    # Validate parity
    print("\nValidating result parity...")
    compat_pl.validate_parity(result_pl, result_pd)

if __name__ == "__main__":
    DATASET = pathlib.Path("/data/10gb_ecommerce_transactions.parquet")
    run_migration_demo(DATASET)

# Good: Define full lazy pipeline before collecting
lazy_result = (
    pl.scan_parquet("10gb_transactions.parquet")
    .filter(pl.col("timestamp").dt.year() == 2025)
    .join(pl.scan_parquet("products.parquet"), on="product_id")
    .with_columns(pl.col("price") * pl.col("quantity").alias("revenue"))
    .group_by("region")
    .agg(pl.col("revenue").sum())
    .collect()  # Only collect after all steps are defined

# Bad: Collect after every step
df = pl.read_parquet("10gb_transactions.parquet")  # Eager read, no optimization
df = df.filter(pl.col("timestamp").dt.year() == 2025)
df = df.join(pl.read_parquet("products.parquet"), on="product_id")
# ... etc, loses all lazy optimization benefits

# Enable Arrow backend for all new Pandas 3 DataFrames
pd.options.mode.dtype_backend = "pyarrow"

# Read Parquet with Arrow dtypes
df_pd = pd.read_parquet("10gb_transactions.parquet")
print(df_pd.dtypes)  # All ArrowDtype: int64[pyarrow], string[pyarrow], etc.

# Zero-copy conversion to Polars
df_pl = pl.from_pandas(df_pd)
print(df_pl.dtypes)  # Matching Arrow dtypes, no data copy

import psutil
import time

def profile_memory(func):
    """Decorator to profile peak memory usage of a function."""
    def wrapper(*args, **kwargs):
        process = psutil.Process()
        start_mem = process.memory_info().rss / (1024 ** 3)
        start_time = time.perf_counter()
        result = func(*args, **kwargs)
        end_time = time.perf_counter()
        end_mem = process.memory_info().rss / (1024 ** 3)
        peak_mem = end_mem - start_mem
        print(f"Runtime: {end_time - start_time:.2f}s | Peak Memory: {peak_mem:.2f}GB")
        return result
    return wrapper

@profile_memory
def run_pipeline():
    # Your 10GB pipeline here
    pass