docs(adr): ADR-002 rows-first delivery with server-pushed summary stats

buckaroo/buckaroo_widget.py

@@ -31,7 +31,7 @@
 from .serialization_utils import EMPTY_DF_WHOLE, check_and_fix_df, pd_to_obj, to_parquet, send_infinite_resp
 from .dataflow.dataflow import CustomizableDataflow
 from .dataflow.dataflow_extras import (Sampling, exception_protect)
-from .dataflow.styling_core import (ComponentConfig, InitSD, OverrideColumnConfig, PinnedRowConfig, StylingAnalysis, merge_column_config, EMPTY_DFVIEWER_CONFIG)
+from .dataflow.styling_core import (ComponentConfig, InitSD, OverrideColumnConfig, PinnedRowConfig, StylingAnalysis, EMPTY_DFVIEWER_CONFIG)
 from .dataflow.autocleaning import PandasAutocleaning
 from pathlib import Path
 
@@ -359,39 +359,10 @@ def _handle_widget_change(self, change):
             _bk_flash("_handle_widget_change EXIT — processed_df is None")
             return
 
-        # df_data_dict is still hardcoded for now
-        # eventually processed_df will be able to add or alter values of df_data_dict
-        # correlation would be added, filtered would probably be altered
-
-        # to expedite processing maybe future provided dfs from
-        # postprcoessing could default to empty until that is
-        # selected, optionally
-
-        #note this needs to be empty so that we can do the infinite stuff
-        self.df_data_dict = {'main': [],
-            'all_stats': self._sd_to_jsondf(merged_sd),
-            'empty': []}
-
-        temp_display_args = {}
-        for display_name, A_Klass in self.dataflow.df_display_klasses.items():
-            df_viewer_config = A_Klass.get_dfviewer_config(merged_sd, processed_df)
-            base_column_config = df_viewer_config['column_config']
-            df_viewer_config['column_config'] =  merge_column_config(
-                base_column_config, self.dataflow.processed_df, self.dataflow.column_config_overrides)
-            disp_arg = {'data_key': A_Klass.data_key,
-                        #'df_viewer_config': json.loads(json.dumps(df_viewer_config)),
-                        'df_viewer_config': df_viewer_config,
-                        'summary_stats_key': A_Klass.summary_stats_key}
-            temp_display_args[display_name] = disp_arg
-
-        if self.dataflow.pinned_rows is not None:
-            temp_display_args['main']['df_viewer_config']['pinned_rows'] = self.dataflow.pinned_rows
-        if self.dataflow.extra_grid_config:
-            temp_display_args['main']['df_viewer_config']['extra_grid_config'] = self.dataflow.extra_grid_config
-        if self.dataflow.component_config:
-            temp_display_args['main']['df_viewer_config']['component_config'] = self.dataflow.component_config
-
-        self.df_display_args = temp_display_args
+        # The dataflow is built with skip_main_serial, so 'main' stays empty for
+        # the infinite rows.
+        self.df_data_dict = self.dataflow._build_df_data_dict(processed_df, merged_sd)
+        self.df_display_args = self.dataflow._build_df_display_args(processed_df, merged_sd)
         _bk_flash("_handle_widget_change EXIT (df_display_args → JS)")
 
 

buckaroo/customizations/xorq_stats_v2.py

@@ -139,6 +139,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/stat_pipeline.py

@@ -23,6 +23,16 @@
 from .utils import PERVERSE_DF
 
 
+def with_stat(inputs: list, stat_func_or_class) -> list:
+    """``inputs`` with ``stat_func_or_class`` added: a class of the same name is
+    replaced, anything else is appended. The rule both pipelines' ``add_stat``
+    and ``CustomizableDataflow.add_analysis`` follow."""
+    if isinstance(stat_func_or_class, type):
+        inputs = [inp for inp in inputs
+            if not (isinstance(inp, type) and inp.__name__ == stat_func_or_class.__name__)]
+    return [*inputs, stat_func_or_class]
+
+
 def _normalize_inputs(inputs: list) -> List[StatFunc]:
     """Convert a mixed list of StatFunc, @stat functions, and stat-group classes to StatFuncs.
 
@@ -349,15 +359,7 @@ def add_stat(self, stat_func_or_class) -> Tuple[bool, List[StatError]]:
 
         Validates the DAG and runs unit test against PERVERSE_DF.
         """
-        new_inputs = list(self._original_inputs)
-
-        # Remove existing with same name if re-adding
-        if isinstance(stat_func_or_class, type):
-            new_inputs = [
-                inp for inp in new_inputs
-                if not (isinstance(inp, type) and inp.__name__ == stat_func_or_class.__name__)
-            ]
-        new_inputs.append(stat_func_or_class)
+        new_inputs = with_stat(self._original_inputs, stat_func_or_class)
 
         try:
             new_funcs = _normalize_inputs(new_inputs)

buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py

@@ -38,7 +38,7 @@
 from .stat_cache import (CachedScope, CachedStatError, StatCache, length_stat_id, make_scope_id,
     stat_hashes)
 from .stat_func import XorqColumn, XorqExpr, XorqExecute, RAW_MARKER_TYPES, StatFunc
-from .stat_pipeline import _execute_stat_func, _normalize_inputs, errors_to_errdict
+from .stat_pipeline import _execute_stat_func, _normalize_inputs, errors_to_errdict, with_stat
 from .stat_result import Err, Ok, StatError, StatResult, resolve_accumulator
 from .typed_dag import build_column_dag, build_typed_dag
 from .utils import PERVERSE_DF
@@ -667,18 +667,7 @@ def add_stat(self, stat_func_or_class) -> Tuple[bool, List[StatError]]:
         against). Validates the DAG; returns ``(True, [])`` on success
         or ``(False, [config_error])`` if the DAG can't be built.
         """
-        new_inputs = list(self._original_inputs)
-
-        if isinstance(stat_func_or_class, type):
-            new_inputs = [
-                inp
-                for inp in new_inputs
-                if not (
-                    isinstance(inp, type)
-                    and inp.__name__ == stat_func_or_class.__name__
-                )
-            ]
-        new_inputs.append(stat_func_or_class)
+        new_inputs = with_stat(self._original_inputs, stat_func_or_class)
 
         try:
             new_funcs = _normalize_inputs(new_inputs)