feat(server): let a host enable the xorq stats column-chunk split over HTTP (rows-first s6)

buckaroo/customizations/pd_stats_v2.py

@@ -101,6 +101,25 @@ def _type(is_bool: bool, is_numeric: bool, is_float: bool,
     return "obj"
 
 
+def with_type(flags: dict) -> dict:
+    """``flags``, the ``is_*`` keys a typing stat returns, plus the ``_type``
+    derived from them through the ``_type`` stat."""
+    return {**flags, '_type': _type(**{key.name: flags[key.name] for key in _type._stat_func.requires})}
+
+
+def schema_stats(ser: pd.Series) -> dict:
+    """The stats a column's dtype alone determines: ``dtype``, the ``is_*``
+    flags and ``_type``. ``typing_stats`` runs on a zero-row slice, so no value
+    is read, and ``memory_usage`` is left out because it measures the data.
+
+    An ``object`` column gives the typing nothing to go on, and pandas calls an
+    empty one a string, so every object column types as ``string`` here. Full
+    stats look at the values and may say ``obj``."""
+    flags = typing_stats(ser.iloc[:0])
+    del flags['memory_usage']
+    return with_type(flags)
+
+
 # ============================================================
 # Base Summary Stats (replaces DefaultSummaryStats ColAnalysis)
 # ============================================================

buckaroo/customizations/pl_stats_v2.py

@@ -21,7 +21,7 @@
 from buckaroo.pluggable_analysis_framework.column_filters import is_numeric_not_bool
 
 # Reused unchanged from pd_stats_v2 — operate on stat dict, not raw series
-from buckaroo.customizations.pd_stats_v2 import (_type, computed_default_summary_stats, histogram, cleaning_gen_ops, BaseSummaryResult, NumericStatsResult, HistogramSeriesResult)
+from buckaroo.customizations.pd_stats_v2 import (_type, computed_default_summary_stats, histogram, cleaning_gen_ops, with_type, BaseSummaryResult, NumericStatsResult, HistogramSeriesResult)
 
 
 # ============================================================
@@ -44,16 +44,27 @@ def pl_orig_col_name(ser: RawSeries) -> Any:
      'is_time': bool, 'is_decimal': bool, 'is_binary': bool, 'memory_usage': int})
 
 
-@stat()
-def pl_typing_stats(ser: RawSeries) -> PlTypingResult:
-    """Compute dtype and type flags for a polars column."""
-    dt = ser.dtype
+def pl_dtype_typing(dt: pl.DataType) -> dict:
+    """dtype string and type flags for a polars dtype. Needs a schema only,
+    which is what ``pl_schema_stats`` has to go on."""
     return {'dtype': str(dt), 'is_numeric': dt.is_numeric() and dt.base_type() is not pl.Decimal,
         'is_integer': dt.is_integer(), 'is_float': dt.is_float(), 'is_bool': dt == pl.Boolean,
         'is_datetime': dt.is_temporal() and dt not in (pl.Duration, pl.Time), 'is_timedelta': dt == pl.Duration,
         'is_string': dt in (pl.Utf8, pl.String), 'is_categorical': dt == pl.Categorical or isinstance(dt, pl.Enum),
         'is_period': False, 'is_interval': False, 'is_time': dt == pl.Time, 'is_decimal': dt.base_type() is pl.Decimal,
-        'is_binary': dt == pl.Binary, 'memory_usage': ser.estimated_size()}
+        'is_binary': dt == pl.Binary}
+
+
+@stat()
+def pl_typing_stats(ser: RawSeries) -> PlTypingResult:
+    """Compute dtype and type flags for a polars column."""
+    return {**pl_dtype_typing(ser.dtype), 'memory_usage': ser.estimated_size()}
+
+
+def pl_schema_stats(ser: pl.Series) -> dict:
+    """The stats a polars column's dtype alone determines: ``dtype``, the
+    ``is_*`` flags and ``_type``. Reads ``ser.dtype`` and no value."""
+    return with_type(pl_dtype_typing(ser.dtype))
 
 
 # ============================================================

buckaroo/customizations/xorq_stats_v2.py

@@ -138,6 +138,16 @@ def _type(is_bool: bool, is_integer: bool, is_float: bool, is_datetime: bool, is
     return "obj"
 
 
+def schema_stats(dtype: str) -> dict:
+    """The stats a column's dtype alone determines: ``dtype``, the ``is_*``
+    flags and ``_type``. This is what the xorq dataflow's schema tier
+    publishes, derived through the same stats the pipeline runs."""
+    flags = typing_stats(dtype)
+    return {"dtype": dtype, **flags, "_type": _type(
+        flags["is_bool"], flags["is_integer"], flags["is_float"], flags["is_datetime"],
+        flags["is_string"])}
+
+
 # ============================================================
 # Batched aggregates — one ibis.Expr each, folded into the batch query
 # ============================================================

buckaroo/pluggable_analysis_framework/df_stats_v2.py

@@ -20,15 +20,17 @@
 
 from .col_analysis import AObjs, ColAnalysis
 from .stat_pipeline import StatPipeline, errors_to_errdict
+from .stat_units import StatState, UnitStats
 from .utils import FAST_SUMMARY_WHEN_GREATER
 from .safe_summary_df import output_full_reproduce
 
 
-class DfStatsV2:
+class DfStatsV2(UnitStats):
     """Tie a StatPipeline to a DataFrame, exposing ``.sdf`` and ``.errs``.
 
     Used by DataFlow, autocleaning, and other consumers as the pandas
-    summary-stats executor.
+    summary-stats executor. ``run=False`` builds it without computing anything,
+    for a caller that drives ``plan`` / ``run`` itself.
     """
 
     ap_class = StatPipeline
@@ -39,19 +41,23 @@ def verify_analysis_objects(cls, col_analysis_objs: AObjs) -> None:
         cls.ap_class(col_analysis_objs)
 
     def __init__(self, df_stats_df: pd.DataFrame, col_analysis_objs: AObjs, operating_df_name: str = None,
-            debug: bool = False, skip_columns=None) -> None:
+            debug: bool = False, skip_columns=None, run: bool = True) -> None:
         self.df = self.get_operating_df(df_stats_df, force_full_eval=False)
         self.col_order = self.df.columns
         self.ap = self.ap_class(col_analysis_objs)
         self.operating_df_name = operating_df_name
         self.debug = debug
+        self.state = StatState(self.df, frozenset(skip_columns or ()))
 
-        self.sdf, errors = self.ap.process_df(self.df, self.debug, skip_columns=skip_columns)
-        self.errs = errors_to_errdict(errors)
+        self.sdf = {}
+        self.errs = {}
         self.stat_errors = []
+        if run:
+            self.sdf, errors = self.ap.process_df(self.df, self.debug, skip_columns=skip_columns)
+            self.errs = errors_to_errdict(errors)
 
-        if self.errs:
-            output_full_reproduce(self.errs, self.sdf, operating_df_name)
+            if self.errs:
+                output_full_reproduce(self.errs, self.sdf, operating_df_name)
 
     def get_operating_df(self, df: pd.DataFrame, force_full_eval: bool) -> pd.DataFrame:
         """Downsample large DataFrames for performance."""
@@ -80,8 +86,9 @@ def add_analysis(self, a_obj: Type[ColAnalysis]) -> None:
             self.ap.print_errors(errors + self.stat_errors)
 
 
-class PlDfStatsV2:
-    """Polars summary-stats executor. Uses StatPipeline with @stat polars functions."""
+class PlDfStatsV2(UnitStats):
+    """Polars summary-stats executor. Uses StatPipeline with @stat polars functions.
+    ``run=False`` builds it without computing anything, as for ``DfStatsV2``."""
 
     @classmethod
     def verify_analysis_objects(cls, objs):
@@ -93,14 +100,18 @@ def get_operating_df(self, df):
             return df.sample(n=min(50_000, rows), seed=42)
         return df
 
-    def __init__(self, df, col_analysis_objs, operating_df_name=None, debug=False, skip_columns=None):
+    def __init__(self, df, col_analysis_objs, operating_df_name=None, debug=False, skip_columns=None, run=True):
         self.df = self.get_operating_df(df)
         self.ap = StatPipeline(col_analysis_objs, unit_test=False)
-        self.sdf, errors = self.ap.process_df(self.df, debug, skip_columns=skip_columns)
-        self.errs = errors_to_errdict(errors)
+        self.state = StatState(self.df, frozenset(skip_columns or ()))
+        self.sdf = {}
+        self.errs = {}
         self.stat_errors = []
-        if self.errs:
-            output_full_reproduce(self.errs, self.sdf, operating_df_name)
+        if run:
+            self.sdf, errors = self.ap.process_df(self.df, debug, skip_columns=skip_columns)
+            self.errs = errors_to_errdict(errors)
+            if self.errs:
+                output_full_reproduce(self.errs, self.sdf, operating_df_name)
 
     def add_analysis(self, a_obj):
         """Add a new analysis class interactively."""

buckaroo/pluggable_analysis_framework/stat_pipeline.py

@@ -9,16 +9,18 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
 
 import pandas as pd
 
+from buckaroo.dataflow.df_types import DataFrameLike
 from buckaroo.df_util import old_col_new_col
 
 from . import perf_log
 from .col_analysis import ColAnalysis, ErrDict, SDType
 from .stat_func import (StatFunc, RawSeries, SampledSeries, RawDataFrame, XorqExpr, XorqExecute, RAW_MARKER_TYPES, MISSING, collect_stat_funcs)
 from .stat_result import Ok, Err, UpstreamError, StatError, StatResult, resolve_accumulator
+from .stat_units import Fragment, StatAccumulator, StatState, StatUnit, UnitPipeline, columns_in_scope, prioritized
 from .typed_dag import build_typed_dag, build_column_dag, DAGConfigError
 from .utils import PERVERSE_DF
 
@@ -172,7 +174,7 @@ def _execute_stat_func(sf: StatFunc, accumulator: Dict[str, StatResult], column_
                     inputs=kwargs.copy())
 
 
-class StatPipeline:
+class StatPipeline(UnitPipeline):
     """Top-level orchestrator for the pluggable analysis framework.
 
     Accepts a mix of:
@@ -276,9 +278,59 @@ def process_column(self, column_name: str, column_dtype, raw_series=None, sample
 
         return resolve_accumulator(accumulator, column_name, col_key_to_func)
 
+    @staticmethod
+    def _skipped(state: StatState, pairs: Sequence[Tuple[Any, str]]) -> set:
+        """The columns of ``pairs`` named in ``state.skip_columns``, by original
+        or rewritten name."""
+        return {orig for orig, rewritten in pairs
+            if orig in state.skip_columns or rewritten in state.skip_columns}
+
+    def plan(self, state: StatState) -> List[StatUnit]:
+        """One unit per column that is not skipped, in column order (or the
+        order ``state.priority`` asks for). A column's stats read only its own
+        column, so no unit waits for another. Nothing is computed. These
+        pipelines have only the full tier."""
+        if state.tier != "full":
+            raise ValueError(f"the pandas and polars pipelines have only the full tier, not {state.tier!r}")
+        if len(state.data) == 0:
+            return []
+        pairs = columns_in_scope(state)
+        skip = self._skipped(state, pairs)
+        active = prioritized(state, [pair for pair in pairs if pair[0] not in skip])
+        return [StatUnit(id=f"column:{orig}", columns=(orig,), phase="column") for orig, _rewritten in active]
+
+    def new_accumulator(self, state: StatState) -> StatAccumulator:
+        """The accumulator for a run of ``state``. A skipped column is entered
+        with its two names, as ``process_df`` always did, and a frame with no
+        rows has no entries at all."""
+        if self.record_timings:
+            self.timings = []
+        if len(state.data) == 0:
+            return StatAccumulator(state)
+        pairs = columns_in_scope(state)
+        acc = StatAccumulator(state, columns=[orig for orig, _rewritten in pairs], rewritten=dict(pairs))
+        for orig in self._skipped(state, pairs):
+            acc.results[orig] = {'orig_col_name': orig, 'rewritten_col_name': acc.rewritten[orig]}
+        return acc
+
+    def run(self, unit: StatUnit, acc: StatAccumulator) -> Fragment:
+        """Run one column's stats through the stat DAG and return them as
+        ``{orig_col: stats}``. The errors go on ``acc.errors``."""
+        df = acc.state.data
+        fragment: Fragment = {}
+        for orig_col_name in unit.columns:
+            rewritten_col_name = acc.rewritten[orig_col_name]
+            ser = df[orig_col_name]
+            col_result, col_errors = self.process_column(column_name=rewritten_col_name, column_dtype=ser.dtype,
+                raw_series=ser, sampled_series=ser, raw_dataframe=df,
+                initial_stats={'orig_col_name': orig_col_name, 'rewritten_col_name': rewritten_col_name})
+            fragment[orig_col_name] = col_result
+            acc.record({orig_col_name: col_result}, col_errors)
+        return fragment
+
     def process_df(self, df: pd.DataFrame, debug: bool = False,
                    skip_columns=None) -> Tuple[SDType, List[StatError]]:
-        """Process all columns of a DataFrame.
+        """Process all columns of a DataFrame: the planned units, run in order.
 
         ``skip_columns`` names columns whose summary stats are supplied
         externally (e.g. via ``init_sd`` — reused from a source dataframe in a
@@ -292,35 +344,15 @@ def process_df(self, df: pd.DataFrame, debug: bool = False,
         if len(df) == 0:
             return {}, []
 
-        if self.record_timings:
-            self.timings = []
-
-        skip = set(skip_columns or ())
-        summary: SDType = {}
-        all_errors: List[StatError] = []
-
-        for orig_col_name, rewritten_col_name in old_col_new_col(df):
-            if orig_col_name in skip or rewritten_col_name in skip:
-                # Provided externally — keep the column, skip computation.
-                summary[rewritten_col_name] = {
-                    'orig_col_name': orig_col_name, 'rewritten_col_name': rewritten_col_name}
-                continue
-            ser = df[orig_col_name]
-            col_dtype = ser.dtype
-
-            col_result, col_errors = self.process_column(column_name=rewritten_col_name, column_dtype=col_dtype,
-                raw_series=ser, sampled_series=ser, raw_dataframe=df,
-                initial_stats={'orig_col_name': orig_col_name, 'rewritten_col_name': rewritten_col_name})
-
-            summary[rewritten_col_name] = col_result
-            all_errors.extend(col_errors)
+        acc = self.run_all(StatState(df, frozenset(skip_columns or ())))
+        summary: SDType = {acc.rewritten[orig]: stats for orig, stats in acc.sd().items()}
 
         if self.record_timings and perf_log.enabled() and not self._suppress_perf_summary:
             rec = perf_log.PerfRecorder(label=f"stats rows={len(df)} cols={len(summary)}")
             rec.extend_timings("pandas/polars", self.timings)
             rec.summary()
 
-        return summary, all_errors
+        return summary, acc.errors
 
     def unit_test(self) -> Tuple[bool, List[StatError]]:
         """Test the pipeline against PERVERSE_DF."""
@@ -432,3 +464,31 @@ def errors_to_errdict(errors: List[StatError]) -> ErrDict:
         err_key = (se.column, se.stat_func.name if se.stat_func else "unknown")
         errs[err_key] = (se.error, None)
     return errs
+
+
+def schema_sd(df: DataFrameLike, column_typing: Callable[[Any], Dict[str, Any]],
+              skip_columns=None) -> SDType:
+    """The summary dict for ``df`` from its schema alone, with no stat run.
+
+    Shaped like ``StatPipeline.process_df``'s output: one entry per column,
+    keyed by the rewritten name, holding ``orig_col_name``,
+    ``rewritten_col_name``, ``length`` and whatever ``column_typing`` returns
+    for the column. ``column_typing`` is handed the column's series and may read
+    its dtype only; it is what keeps this from touching a value. As in
+    ``process_df``, a frame with no rows gives an empty dict and a column in
+    ``skip_columns`` gets only its two names, so its typing comes from
+    ``init_sd``.
+    """
+    length = len(df)
+    if length == 0:
+        return {}
+    skip = set(skip_columns or ())
+    summary: SDType = {}
+    for orig_col_name, rewritten_col_name in old_col_new_col(df):
+        col_stats: Dict[str, Any] = {
+            'orig_col_name': orig_col_name, 'rewritten_col_name': rewritten_col_name}
+        if orig_col_name not in skip and rewritten_col_name not in skip:
+            col_stats.update(column_typing(df[orig_col_name]))
+            col_stats['length'] = length
+        summary[rewritten_col_name] = col_stats
+    return summary