Rename query_engine.batch_size to max_channels_per_batch

docs/impulse/docs/config/configuration.md

@@ -139,7 +139,7 @@ Two independent filter families:
 | `data_type`             | `str`          | `"RLE"`                  | `"RLE"` (intervals `[tstart, tend)`) or `"RAW"` (raw timestamps; converted to intervals at query time by the configured `raw_encoder`). |
 | `raw_encoder`           | `str`          | `null` (resolves to `"RLE"` when `data_type = "RAW"`) | How RAW point data is converted to intervals. `"RLE"` (the default) run-length encodes on the fly to reduce memory consumption: consecutive samples with the same value collapse into a single `[tstart, tend)` interval per run. `"INTERVAL"` keeps every original sample, only deriving `tend` from the next sample's timestamp and dropping exact duplicate points — use it when downstream analysis needs all original timestamps. Only takes effect with `data_type = "RAW"`; ignored for `"RLE"` input. See [How Impulse interprets intervals](../data_model/silver_layer_schema.md#raw-format) for validity semantics and what counts as a duplicate point. |
 | `drop_implausible_data` | `bool`         | `false`                  | When `true`, drops `channels` rows where `is_plausible = false`. Requires `data_type = "RAW"`; combining with `"RLE"` raises a validation error. |
-| `batch_size`            | `int`          | `500`                    | Maximum number of selectors solved per batch. Applies to events, aggregations, and calculated channels.                     |
+| `max_channels_per_batch` | `int`         | `500`                    | Caps how many distinct channels are read in one solve batch, to bound per-solve memory. Counts channel selections (each distinct `channel(...)` / `poi_channel(...)` in your expressions). Applies to events, aggregations, and calculated channels; for calculated channels it counts the *input* channels they read, not the number of calculated channels. The cap is a packing target, not a hard limit: an expression that on its own reads more channels than the cap still runs, in a batch of its own. Must be `>= 1`. The old name `batch_size` is a **deprecated alias** (still accepted, warns, removed in a future release); setting both raises a validation error. |
 | `max_containers_per_batch` | `int`       | `null`                   | Caps how many containers are solved at once, to bound per-solve memory. `null` disables the cap; when set it must be `>= 1`. On an **incremental** run, `Report.run()` commits at most this many upserted containers per iteration and loops until the population is drained (see [incremental](#incremental-optional)). On a **full** run, and when a changed definition recomputes over all containers, the population is split into chunks of about this size and solved chunk by chunk. Chunk sizes are approximate (the cap is a memory heuristic, not an exact bound). |
 | `solver_config`         | `SolverConfig` | `null`                   | Per-table column mappings, per-table equality filters, and project scoping. Set `project_id` to scope reads by project — it is applied to `container_tags` (if configured), `container_metrics`, and `channel_mapping` (if configured), so it works in both narrow EAV and wide-only data models. Omit it when you don't need project scoping. See [Solver column mappings and filters](#solver-column-mappings-and-filters). |
 

docs/impulse/docs/references/api/impulse_reporting/channels/calculated_channel.md

@@ -20,8 +20,10 @@ the narrow result to a gold fact table, and update it incrementally.
 Structurally parallels :class:`BasicEvent` (holds an aliased expression,
 name-derived id, SHA-256 definition hash). Like the other entity types, its
 narrow solve is batched from :class:`Report` (see
-``Report._solve_calculated_channels_batched``): the channels are partitioned by
-``batch_size``, each batch solved via ``QueryBuilder.solve_calculated_channels``
+``Report._solve_calculated_channels_batched``): the channels are packed into
+batches of at most ``max_channels_per_batch`` distinct input channels (a channel
+whose own inputs exceed the cap gets a batch of its own); each batch is solved
+via ``QueryBuilder.solve_calculated_channels``
 and persisted as a temp table, then the batches are unioned into a narrow
 ``solved_df``. :meth:`determine_calculated_channels` shapes that already-solved
 df, mirroring ``determine_aggregations`` / ``determine_events``.

docs/impulse/docs/references/report/channel.md

@@ -71,7 +71,7 @@ speed_kmh = CalculatedChannel(
 ### How it works
 
 1. The `expr` is evaluated per measurement container by the solver, yielding a `SampleSeries` for each container.
-2. The channels' narrow solve is **batched from the `Report`** (the same pattern events and aggregations use): the channels are partitioned by `query_engine.batch_size`, each batch is solved via `QueryBuilder.solve_calculated_channels` and persisted as a temporary table (`__impulse_temp_{run_id}_{batch_idx}`), and the batches are unioned into the final `calculated_channel_fact`. Multi-channel expressions (e.g. `rpm + speed`) are time-base synchronized automatically; emitted intervals are the intersection.
+2. The channels' narrow solve is **batched from the `Report`** (the same pattern events and aggregations use): the channels are packed into batches of at most `query_engine.max_channels_per_batch` distinct input channels (a calculated channel whose own inputs exceed the cap gets a batch of its own), each batch is solved via `QueryBuilder.solve_calculated_channels` and persisted as a temporary table (`__impulse_temp_{run_id}_{batch_idx}`), and the batches are unioned into the final `calculated_channel_fact`. Multi-channel expressions (e.g. `rpm + speed`) are time-base synchronized automatically; emitted intervals are the intersection.
 3. Each series is exploded into one row per sample interval: `container_id, channel_id, tstart, tend, value`. The `identity` is **not** on the fact — it lives on the dimension, joined via `channel_id`.
 4. The `channel_id` is a deterministic hash of the sorted `identity` — the same value appears in both the fact and dimension tables, so they join directly.
 5. The sample rows are written to the `calculated_channel_fact` table; the channel definition (name, description, expression, identity) is written to the `calculated_channel_dimension` table.

skills/impulse-channels/SKILL.md

@@ -94,8 +94,8 @@ across selections.
 ## Batching (in a Report)
 
 Inside a `Report`, calculated channels are solved in **batches**, the same pattern events and
-aggregations use (see `impulse-reporting`). The channels are partitioned by `query_engine.batch_size`
-(max unique input selectors per batch), each batch is solved via `solve_calculated_channels` and
+aggregations use (see `impulse-reporting`). The channels are partitioned by `query_engine.max_channels_per_batch`
+(max distinct input channels per batch), each batch is solved via `solve_calculated_channels` and
 persisted as a temp table (`__impulse_temp_{run_id}_{batch_idx}` in the sink schema, or a Spark temp
 view when sinkless), and the batches are unioned into the final `calculated_channel_fact`. The temp
 tables share the `__impulse_temp_*` prefix, so they are cleaned up like the aggregation/event batches

skills/impulse-config/SKILL.md

@@ -101,7 +101,7 @@ Omit entirely for `DefaultSolver` + `data_type="RLE"`.
 | `data_type`             | `"RLE"`            | `"RLE"` for interval-encoded `[tstart, tend)` samples; `"RAW"` for `(timestamp, value)` samples.  |
 | `raw_encoder`           | `null` (→ `"RLE"`) | How RAW point data becomes intervals. `"RLE"` collapses equal-valued runs (default); `"INTERVAL"` keeps every sample, only deriving `tend` + dropping exact duplicates. Only used when `data_type="RAW"`. |
 | `drop_implausible_data` | `false`            | Drop `channels` rows where `is_plausible = false`. **Requires `data_type="RAW"`** (RLE raises).   |
-| `batch_size`            | `500`              | Max selectors solved per batch. Applies to events, aggregations, and calculated channels.         |
+| `max_channels_per_batch` | `500`             | Max distinct channels (`channel(...)` / `poi_channel(...)` selections) read per solve batch. Applies to events, aggregations, and calculated channels (counts their *input* channels). `batch_size` is a **deprecated alias** (warns; removed in a future release). |
 | `solver_config`         | `null`             | Per-table column mappings, per-table filters, and project scoping (below).                        |
 
 ## solver_config (optional) — adapt to an existing layout

src/impulse_reporting/channels/calculated_channel.py

@@ -47,8 +47,10 @@ class CalculatedChannel:
     Structurally parallels :class:`BasicEvent` (holds an aliased expression,
     name-derived id, SHA-256 definition hash). Like the other entity types, its
     narrow solve is batched from :class:`Report` (see
-    ``Report._solve_calculated_channels_batched``): the channels are partitioned by
-    ``batch_size``, each batch solved via ``QueryBuilder.solve_calculated_channels``
+    ``Report._solve_calculated_channels_batched``): the channels are packed into
+    batches of at most ``max_channels_per_batch`` distinct input channels (a channel
+    whose own inputs exceed the cap gets a batch of its own); each batch is solved
+    via ``QueryBuilder.solve_calculated_channels``
     and persisted as a temp table, then the batches are unioned into a narrow
     ``solved_df``. :meth:`determine_calculated_channels` shapes that already-solved
     df, mirroring ``determine_aggregations`` / ``determine_events``.

src/impulse_reporting/config/config_parser.py

@@ -1,9 +1,11 @@
+import os
 import re
 import warnings
 from datetime import datetime
 from enum import Enum, StrEnum
 from typing import Annotated
 
+import pydantic
 from pydantic import AfterValidator, BaseModel, field_validator, model_validator
 
 from impulse_query_engine.analyze.query.solvers.solver_config import RawEncoder, SolverConfig
@@ -348,6 +350,13 @@ class ContainerFilters(BaseModel):
     metric_filters: list[list[MetricFilter]] = []
 
 
+_BATCH_SIZE_DEPRECATION_MSG = (
+    "query_engine.batch_size is deprecated; use max_channels_per_batch instead. "
+    "batch_size will be removed in a future release."
+)
+_PYDANTIC_DIR = os.path.dirname(pydantic.__file__)
+
+
 class QueryEngine(BaseModel):
     """
     Configuration for the query engine solver.
@@ -399,9 +408,66 @@ class QueryEngine(BaseModel):
     drop_implausible_data: bool = False
     raw_encoder: RawEncoder | None = None
     solver_config: SolverConfig | None = None
-    batch_size: int = 500
+    max_channels_per_batch: int = 500
     max_containers_per_batch: int | None = None
 
+    @model_validator(mode="before")
+    @classmethod
+    def _resolve_deprecated_batch_size(cls, data):
+        """Accept the deprecated ``batch_size`` alias for ``max_channels_per_batch``.
+
+        ``batch_size`` was renamed to ``max_channels_per_batch`` (it caps the number
+        of distinct ``TimeSeriesSelector`` selections per solve batch, not a row count).
+        The old name keeps working for now but warns; setting both is an error.
+        """
+        if not isinstance(data, dict):
+            return data
+        if "batch_size" in data:
+            if "max_channels_per_batch" in data:
+                raise ValueError(
+                    "Set only one of query_engine.max_channels_per_batch or the deprecated "
+                    "alias query_engine.batch_size, not both."
+                )
+            # FutureWarning, not DeprecationWarning: the latter is hidden by default when
+            # raised from library code, so notebook/job users would never see it.
+            warnings.warn(
+                _BATCH_SIZE_DEPRECATION_MSG,
+                FutureWarning,
+                # stacklevel=1 (the warn call itself): inside a pydantic model_validator the
+                # frames above are pydantic internals, so a higher level would mislead.
+                stacklevel=1,
+            )
+            # Copy so the caller's config dict is left untouched (it may be reused).
+            data = dict(data)
+            data["max_channels_per_batch"] = data.pop("batch_size")
+        return data
+
+    @property
+    def batch_size(self) -> int:
+        """Deprecated alias for :attr:`max_channels_per_batch`."""
+        warnings.warn(_BATCH_SIZE_DEPRECATION_MSG, FutureWarning, stacklevel=2)
+        return self.max_channels_per_batch
+
+    @batch_size.setter
+    def batch_size(self, value: int) -> None:
+        # Skip pydantic's __setattr__ frames so the warning points at the caller.
+        warnings.warn(
+            _BATCH_SIZE_DEPRECATION_MSG,
+            FutureWarning,
+            stacklevel=2,
+            skip_file_prefixes=(_PYDANTIC_DIR,),
+        )
+        # Assignment skips field validators (no validate_assignment), so check explicitly.
+        self.max_channels_per_batch = self._validate_max_channels_per_batch(value)
+
+    @field_validator("max_channels_per_batch", mode="after")
+    @classmethod
+    def _validate_max_channels_per_batch(cls, value: int) -> int:
+        """Max distinct ``TimeSeriesSelector`` selections (by ``selector_id``) per solve batch."""
+        if value < 1:
+            raise ValueError("max_channels_per_batch must be >= 1.")
+        return value
+
     @field_validator("max_containers_per_batch", mode="after")
     @classmethod
     def _validate_max_containers_per_batch(cls, value: int | None) -> int | None:

src/impulse_reporting/core/report.py

@@ -910,7 +910,7 @@ def _solve_expressions_batched(
             expressions=expressions,
             query=self.query,
             solver=self.solver,
-            batch_size=self.config.query_engine.batch_size,
+            max_channels_per_batch=self.config.query_engine.max_channels_per_batch,
             has_sink=self._has_sink,
             catalog=getattr(self.config, "unity_sink", None) and self.config.unity_sink.catalog,
             schema=getattr(self.config, "unity_sink", None) and self.config.unity_sink.schema,
@@ -933,7 +933,7 @@ def _solve_calculated_channels_batched(
             qe_channels=qe_channels,
             query=self.query,
             solver=self.solver,
-            batch_size=self.config.query_engine.batch_size,
+            max_channels_per_batch=self.config.query_engine.max_channels_per_batch,
             has_sink=self._has_sink,
             catalog=getattr(self.config, "unity_sink", None) and self.config.unity_sink.catalog,
             schema=getattr(self.config, "unity_sink", None) and self.config.unity_sink.schema,

src/impulse_reporting/core/report_utils.py

@@ -37,19 +37,21 @@
 
 def build_batches(
     expressions: list[TimeSeriesExpression],
-    batch_size: int,
+    max_channels_per_batch: int,
 ) -> list[list[TimeSeriesExpression]]:
     """Selector-aware best-fit-decreasing bin packing.
 
-    Groups expressions that share ``TimeSeriesSelector`` instances to
+    Groups expressions that share selectors (same ``selector_id``) to
     maximise data locality and minimise cross-batch selector duplication.
 
     Parameters
     ----------
     expressions : list[TimeSeriesExpression]
         Expressions to partition into batches.
-    batch_size : int
-        Maximum number of unique ``TimeSeriesSelector`` instances per batch.
+    max_channels_per_batch : int
+        Maximum number of distinct selectors per batch, deduplicated by the
+        content-based ``selector_id`` (selectors built from the same tags count
+        once). A single expression that alone exceeds the cap gets its own batch.
 
     Returns
     -------
@@ -72,17 +74,17 @@ def build_batches(
     all_selector_ids: set[int] = set()
     for s in selector_ids.values():
         all_selector_ids |= s
-    if len(all_selector_ids) <= batch_size:
+    if len(all_selector_ids) <= max_channels_per_batch:
         return [list(expressions)]
 
     # --- Phase 3: Best-Fit Decreasing (BFD) bin-packing.
     #
     # Sort expressions by the number of selectors they use (heaviest first).
     # Then for each expression find the existing batch where it causes the
     # smallest growth of the selector set (= highest overlap) without
-    # exceeding batch_size.  If no batch can accommodate it, open a new one.
+    # exceeding max_channels_per_batch.  If no batch can accommodate it, open a new one.
     #
-    # Why BFD?  Expressions that share the same TimeSeriesSelector objects
+    # Why BFD?  Expressions that share the same selectors (same selector_id)
     # (e.g. same channel/signal) are naturally packed together, maximising
     # data locality during the Spark solve step and minimising the number
     # of redundant selector reads across batches.
@@ -99,7 +101,7 @@ def build_batches(
         # Find the batch with the most selector overlap that still fits
         for bi in range(len(final_batches)):
             combined = batch_selector_ids[bi] | expr_sels
-            if len(combined) <= batch_size:
+            if len(combined) <= max_channels_per_batch:
                 overlap = len(batch_selector_ids[bi] & expr_sels)
                 if overlap > best_overlap:
                     best_overlap = overlap
@@ -566,7 +568,7 @@ def solve_expressions_batched(
     expressions: list[TimeSeriesExpression],
     query: QueryBuilder,
     solver: QuerySolver,
-    batch_size: int,
+    max_channels_per_batch: int,
     *,
     has_sink: bool = False,
     catalog: str = None,
@@ -592,8 +594,9 @@ def solve_expressions_batched(
         Query builder instance.
     solver : QuerySolver
         Query solver instance.
-    batch_size : int
-        Maximum number of unique selectors per batch (passed to ``build_batches``).
+    max_channels_per_batch : int
+        Maximum number of distinct selectors (by ``selector_id``) per batch (passed to
+        ``build_batches``).
     has_sink : bool
         Whether a Unity Catalog sink is configured.
     catalog : str, optional
@@ -629,7 +632,9 @@ def _solve_selector_batches(pre_filter: DataFrame | None) -> DataFrame:
                 catalog,
                 schema,
             )
-            for batch_idx, batch_exprs in enumerate(build_batches(expressions, batch_size))
+            for batch_idx, batch_exprs in enumerate(
+                build_batches(expressions, max_channels_per_batch)
+            )
         ]
         dfs = [spark.table(name) for name in batch_names]
         result = dfs[0]
@@ -665,7 +670,7 @@ def solve_calculated_channels_batched(
     qe_channels: list,
     query: QueryBuilder,
     solver: QuerySolver,
-    batch_size: int,
+    max_channels_per_batch: int,
     *,
     has_sink: bool = False,
     catalog: str = None,
@@ -693,8 +698,9 @@ def solve_calculated_channels_batched(
         Query builder instance.
     solver : QuerySolver
         Query solver implementing ``solve_calculated_channels``.
-    batch_size : int
-        Maximum number of unique selectors per batch (passed to ``build_batches``).
+    max_channels_per_batch : int
+        Maximum number of distinct selectors (by ``selector_id``) per batch (passed to
+        ``build_batches``).
     has_sink : bool
         Whether a Unity Catalog sink is configured.
     catalog : str, optional
@@ -728,7 +734,9 @@ def _solve_selector_batches(pre_filter: DataFrame | None) -> DataFrame:
                 catalog,
                 schema,
             )
-            for batch_idx, batch_channels in enumerate(build_batches(qe_channels, batch_size))
+            for batch_idx, batch_channels in enumerate(
+                build_batches(qe_channels, max_channels_per_batch)
+            )
         ]
         dfs = [spark.table(name) for name in batch_names]
         result = dfs[0]