feat(query-engine): add opt-in rule-based Spark pushdown optimizer, starting with resample

docs/impulse/docs/config/configuration.md

@@ -141,6 +141,7 @@ Two independent filter families:
 | `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. |
 | `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). |
+| `enable_pushdown`       | `bool`         | `false`                  | When `true`, supported expression subtrees (currently `channel.resample(rate)` on a physical channel) are computed natively in Spark before the pandas evaluation, so fewer rows are shipped to it. A channel is only pushed down when every use of it in a solve batch is; results are unchanged. Meant for downsampling: upsampling produces more rows than the raw channel. See [Spark pushdown](../references/query_engine/query_solvers.md#spark-pushdown-opt-in). |
 | `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). |
 | `measurement_db`        | `str`          | `"MeasurementDB"`        | Registered name of the MeasurementDB implementation to use. The default is the built-in reader; a custom implementation can reshape a different source schema at read time. See [Custom measurement DB](../extensibility/custom_measurement_db.md). |
 | `measurement_db_config` | `dict`         | `null`                   | Keyword arguments for that implementation's config class, merged with the `source` tables (e.g. extra source tables, prefilter keys). |

docs/impulse/docs/references/api/impulse_query_engine/analyze/metadata/time_series_expression.md

@@ -353,6 +353,10 @@ def build(cache: SeriesCache)
 
 Build the time series from cache.
 
+A result the cache already holds for this node (see
+:meth:`SeriesCache.materialized`) is returned as-is, without building the
+operands.
+
 **Arguments**:
 
 - `cache` (`SeriesCache`): Cache containing time series data.

docs/impulse/docs/references/api/impulse_query_engine/analyze/query/optimizer/context.md

@@ -0,0 +1,267 @@
+---
+sidebar_label: context
+title: impulse_query_engine.analyze.query.optimizer.context
+---
+
+What a pushdown rule sees of the samples it computes on.
+
+A rule never handles DataFrames.  The pipeline (:mod:`.pipeline`) evaluates the
+rule's hooks with a :class:`SeriesContext`, whose properties are Spark columns
+describing the current sample of one of the rule's series: its times and value,
+its neighbours, the run (e.g. continuous block) it belongs to, and facts about
+the whole series.  Columns returned by a hook become part of the shared pass
+over the channel data.
+
+
+#### ELEMENT\_TYPE
+
+Element type of the arrays returned by :meth:`PushdownRule.emit`.
+
+
+#### BLOCK
+
+Name of the built-in run of continuous samples.
+
+
+#### run\_col
+
+```python
+def run_col(run: str, name: str) -> str
+```
+
+Name of the pipeline column *name* of the run *run*.
+
+
+## Sample
+
+```python
+class Sample()
+```
+
+One sample as columns: ``tstart`` and ``tend`` as double, and its value as double.
+
+
+#### RunBreak
+
+A run definition: whether the sample *cur* starts a new run after *prev*.  It
+
+is called with ``prev`` columns that are null before the series' first
+sample; a null result counts as a break.
+
+
+## Run
+
+```python
+class Run()
+```
+
+Where the current sample sits in its run, a maximal stretch without a break.
+
+**Arguments**:
+
+- `starts` (`Column`): The sample starts its run.
+- `id` (`Column`): Number of the run within the series, from 1.
+- `start` (`Column`): ``tstart`` of the run's first sample.
+- `next_starts` (`Column`): The next sample starts a new run; null for the series' last sample.
+- `is_last` (`Column`): The run is the last of its series.
+
+#### block\_break
+
+```python
+def block_break(prev: Sample, cur: Sample) -> Column
+```
+
+Continuous blocks: a sample continues its block when it starts where the previous ended.
+
+
+## SeriesContext
+
+```python
+class SeriesContext()
+```
+
+Columns a rule's hooks build on, for the current sample of one of its series.
+
+Every series is one channel's samples of one variant, ordered by
+``(tstart, tend, value)``; all variants of a channel have the same samples.
+Rules running in series order (``ordered``) see neighbours, runs and series
+facts; per-sample rules see only the sample.
+
+**Arguments**:
+
+- `rule` (`PushdownRule`): The rule the hooks belong to.
+- `variants` (`Mapping[int, Pushdown]`): The rule's variants in this solve.
+- `key` (`Sequence[str]`): Columns identifying a series.
+- `order` (`Sequence[str]`): The channel-data ``tstart``, ``tend`` and value columns: the series order.
+- `ordered` (`bool`): Whether the pipeline runs the rule in series order.
+- `runs` (`Mapping[str, str]`): The rule's run names to pipeline run names.
+- `kept` (`bool`): Whether the hook runs on the kept samples (``kept_columns``, ``emit``).
+
+#### kept\_stage
+
+```python
+def kept_stage() -> SeriesContext
+```
+
+This context, for the hooks that run on the kept samples.
+
+
+#### active
+
+```python
+def active() -> Column
+```
+
+The sample belongs to one of the rule's variants.
+
+
+#### name
+
+```python
+def name(column: str) -> str
+```
+
+Pipeline name of the rule's column *column*.
+
+
+#### col
+
+```python
+def col(column: str) -> Column
+```
+
+The rule's column *column*, defined by an earlier hook entry.
+
+
+#### param
+
+```python
+def param(name: str) -> Column
+```
+
+The pushdown parameter *name* of the sample's variant.
+
+
+#### tstart
+
+```python
+def tstart() -> Column
+```
+
+The sample's ``tstart`` as double.
+
+
+#### tend
+
+```python
+def tend() -> Column
+```
+
+The sample's ``tend`` as double.
+
+
+#### value
+
+```python
+def value() -> Column
+```
+
+The sample's value as double.
+
+
+#### sample
+
+```python
+def sample() -> Sample
+```
+
+The current sample.
+
+
+#### element
+
+```python
+def element(tstart: Column | None = None,
+            tend: Column | None = None,
+            value: Column | None = None) -> Column
+```
+
+One emitted row; ``tstart`` / ``tend`` left out keep the sample's own times.
+
+
+#### prev
+
+```python
+def prev() -> Sample
+```
+
+The previous sample of the series; null columns for its first sample.
+
+
+#### next
+
+```python
+def next() -> Sample
+```
+
+The next sample of the series; null columns for its last sample.
+
+
+#### n
+
+```python
+def n() -> Column
+```
+
+Number of samples in the series.
+
+
+#### last
+
+```python
+def last() -> Sample
+```
+
+The series' last sample.
+
+
+#### in\_chunk
+
+```python
+def in_chunk() -> WindowSpec
+```
+
+Window over the sample's chunk in series order, for local neighbour access.
+
+Chunks are consecutive slices of a channel, the same for all its series;
+use :attr:`prev`, :attr:`next` and runs for anything that must see across
+chunk boundaries.
+
+
+#### block
+
+```python
+def block() -> Run
+```
+
+The continuous block the sample belongs to (gaps end blocks).
+
+
+#### run
+
+```python
+def run(name: str) -> Run
+```
+
+The sample's run of the rule's run definition *name* (see :meth:`PushdownRule.runs`).
+
+
+#### block\_end
+
+```python
+def block_end() -> Column
+```
+
+``nanmax`` of the ``tend`` values of the sample's block (kept samples only).
+
+

docs/impulse/docs/references/api/impulse_query_engine/analyze/query/optimizer/pipeline.md

@@ -0,0 +1,75 @@
+---
+sidebar_label: pipeline
+title: impulse_query_engine.analyze.query.optimizer.pipeline
+---
+
+Spark execution of a pushdown plan: one shared pass for all rules.
+
+
+#### CHUNK\_ROWS
+
+Average number of rows per chunk of a channel, counting a sample once per variant.
+
+
+#### apply\_pushdowns
+
+```python
+def apply_pushdowns(rows: DataFrame,
+                    variants: DataFrame,
+                    plan: PushdownPlan,
+                    config,
+                    include_raw: bool = True) -> DataFrame
+```
+
+The rows a solve ships: raw rows (when *include_raw*) and all derived rows.
+
+**Arguments**:
+
+- `rows` (`pyspark.sql.DataFrame`): Channel data ``(container_id, channel_id, tstart, tend, value)``, one row
+per sample.
+- `variants` (`pyspark.sql.DataFrame`): ``(container_id, channel_id, VARIANTS_COL)``: the variant ids each matched
+channel ships, one row per channel (see :meth:`PushdownPlan.channel_variants`).
+Channels without a row are dropped.
+- `plan` (`PushdownPlan`): The solve's pushdown plan.
+- `config` (`SolverConfig`): Provides the internal column names.
+- `include_raw` (`bool`): Whether to keep the raw rows (variant 0).
+
+**Returns**:
+
+`pyspark.sql.DataFrame`: Rows with the columns and types of *rows* plus :data:`VARIANT_COL`.
+
+## SeriesPipeline
+
+```python
+class SeriesPipeline()
+```
+
+The ordered pass shared by all rules that need their series in order.
+
+Chunking and everything that crosses chunk edges runs once per channel; its
+samples fan out into one series per variant only for the rules' hooks.  The
+shuffle therefore carries each sample once, while a task after it processes
+about k rows per shuffled sample of a channel with k variants (chunks are k
+times smaller to keep the rows per chunk at :data:`CHUNK_ROWS`).
+
+Rule runs are computed for every channel of the pass; a rule only reads its
+own runs, on its own series.
+
+**Arguments**:
+
+- `rules` (`Sequence of (PushdownRule, Mapping[int, Pushdown])`): Ordered rules with their variants in this solve.
+- `config` (`SolverConfig`): Provides the internal column names.
+
+#### run
+
+```python
+def run(rows: DataFrame, variants: DataFrame) -> DataFrame
+```
+
+Replace the samples of the rules' variants with their derived rows.
+
+*rows* holds the channel data, one row per sample; *variants* the variant
+ids of each channel (see :func:`apply_pushdowns`).  Only channels shipping a
+variant of the rules are processed.
+
+