Add calculated (derived) channels to the query engine and reporting layer

docs/impulse/docs/data_model/gold_layer_event_normalized.md

@@ -165,17 +165,40 @@ channel_mapping_resolution_dimension {
     timestamp _created_at
 }
 
+calculated_channel_dimension {
+    long channel_id
+    int report_id
+    string channel_type
+    string channel_description
+    string channel_expression
+    map identity
+    long definition_hash
+    map attributes
+    timestamp _created_at
+}
+
+calculated_channel_fact {
+    int container_id FK
+    long channel_id FK
+    long tstart
+    long tend
+    double value
+    timestamp _created_at
+}
+
 histogram_fact }o--|| event_dimension: event_id
 histogram2d_fact }o--|| event_dimension: event_id
 stats_aggregator_fact }o--|| event_instance_fact: event_instance_id
 
 histogram_dimension ||--o{ histogram_fact : by_visual_id
 histogram2d_dimension ||--o{ histogram2d_fact : by_visual_id
 stats_aggregator_dimension ||--o{ stats_aggregator_fact: by_visual_id
+calculated_channel_dimension ||--o{ calculated_channel_fact : by_channel_id
 
 measurement_dimension ||--o{ histogram_fact : container_id
 measurement_dimension ||--o{ histogram2d_fact : container_id
 measurement_dimension ||--o{ stats_aggregator_fact : container_id
+measurement_dimension ||--o{ calculated_channel_fact : container_id
 
 measurement_dimension ||--o{ channel_mapping_resolution_dimension : container_id
 
@@ -198,6 +221,7 @@ guaranteed.
 | `{prefix}_histogram2d_fact`      | `container_id`, `visual_id`, `event_id`, `x_bin_id`, `y_bin_id`                        | 2D histogram bin values per container.                       |
 | `{prefix}_stats_aggregator_fact` | `container_id`, `visual_id`, `event_instance_id`, `channel_name`, `aggregation_label`  | Statistics values per signal, event instance, and container. |
 | `{prefix}_event_instance_fact`   | `container_id`, `event_id`, `event_instance_id`                                        | Materialized event occurrences with start/end timestamps.    |
+| `{prefix}_calculated_channel_fact` | `container_id`, `channel_id`, `tstart`                                               | Materialized derived signal — one row per sample interval, in the silver `channels` shape (`tstart`, `tend`, `value`). The channel's identity lives on `calculated_channel_dimension`, joined via `channel_id`. |
 
 ---
 
@@ -211,6 +235,7 @@ guaranteed.
 | `{prefix}_event_dimension`            | `event_id`, `report_id`  | Event definitions (name, expression, required channels).   |
 | `{prefix}_measurement_dimension`      | `container_id`           | Container metadata. Always carries `container_id`, `config_hash`, `_created_at`; additional columns are populated from [`config.measurement_dimensions`](../config/configuration.md#measurement_dimensions-optional). |
 | `{prefix}_channel_mapping_resolution_dimension` | `container_id`, `channel_id`, `channel_alias` | Resolves each channel alias to its physical channel per container (physical join keys, alias `priority`). Written only when the report uses aliased selectors. The `source_unit` / `target_unit` columns are present only when a [`config.unit_conversion_table`](../config/configuration.md) is configured. |
+| `{prefix}_calculated_channel_dimension` | `channel_id`, `report_id` | Calculated-channel definitions (name, TSAL expression, `identity`, `definition_hash`). Written only when the report defines calculated channels. See [Channels](../references/report/channel.md). |
 
 The `channel_mapping_resolution_dimension` table lets BI consumers join a fact
 back to the physical channel that an alias resolved to: join on

docs/impulse/docs/data_model/index.md

@@ -61,6 +61,7 @@ The Gold layer uses a **star schema** with fact and dimension tables. All table
 | `histogram_fact`       | One row per bin per container            | 1D histogram bin values, duration-weighted.               |
 | `histogram2d_fact`     | One row per (x, y) bin per container     | 2D histogram bin values, duration-weighted.               |
 | `stats_aggregator_fact` | One row per signal per event instance    | Descriptive statistics (min, max, mean, median).          |
+| `calculated_channel_fact` | One row per sample interval per container | Materialized derived signal (a *channel*, not a summary), in the silver `channels` shape. |
 
 ### Dimension tables
 
@@ -71,14 +72,16 @@ The Gold layer uses a **star schema** with fact and dimension tables. All table
 | `histogram_dimension`    | Histogram metadata (bins, signal info, units).                     |
 | `histogram2d_dimension`  | 2D histogram metadata (axes, bins, signal info, units).            |
 | `stats_aggregator_dimension` | Statistics metadata (channel names, aggregation labels).       |
+| `calculated_channel_dimension` | Calculated-channel definitions (name, expression, identity). |
 
 ### Join pattern
 
-Fact and dimension tables are connected through three key columns:
+Fact and dimension tables are connected through these key columns:
 
 - **`container_id`** -- links all fact tables to `measurement_dimension`
 - **`event_id`** -- links `event_instance_fact`, `histogram_fact`, and `histogram2d_fact` to `event_dimension`
 - **`visual_id`** -- links each aggregation fact table to its corresponding dimension table
+- **`channel_id`** -- links `calculated_channel_fact` to `calculated_channel_dimension`
 
 `stats_aggregator_fact` additionally joins to `event_instance_fact` via `event_instance_id`, enabling per-interval breakdowns.
 
@@ -93,3 +96,4 @@ Fact and dimension tables are connected through three key columns:
 | **Channel**     | A time-series signal within a container (e.g. "Engine RPM"). Identified by `(container_id, channel_id)`. | `channels`, `channel_metrics`, `channel_tags` |
 | **Event**       | A time window of interest, defined by a condition or spanning the full recording. | `event_dimension`, `event_instance_fact`          |
 | **Aggregation** | A computation over channel data within event windows (histogram, 2D histogram, or statistics).      | `*_fact`, `*_dimension`                          |
+| **Calculated channel** | A derived time-series signal computed from existing channels and materialized at the same per-sample grain. | `calculated_channel_dimension`, `calculated_channel_fact` |

docs/impulse/docs/references/api/impulse_query_engine/analyze/query/channels/calculated_channel.md

@@ -0,0 +1,83 @@
+---
+sidebar_label: calculated_channel
+title: impulse_query_engine.analyze.query.channels.calculated_channel
+---
+
+CalculatedChannel: a labeled derived time-series channel.
+
+
+## CalculatedChannel
+
+```python
+class CalculatedChannel(TimeSeriesExpression)
+```
+
+A derived channel: a wrapped ``TimeSeriesExpression`` plus output identity.
+
+Wraps an expression built from the operator DSL (e.g. ``rpm * 3.6``) that must
+evaluate to a :class:`SampleSeries`.  ``build()`` returns that series' raw
+``(tstarts, tends, values)`` arrays, which
+:meth:`QueryBuilder.solve_calculated_channels` explodes into narrow
+silver-shaped rows plus an ``identity`` ``MapType`` column.
+
+**Arguments**:
+
+- `expr` (`TimeSeriesExpression`): The wrapped expression; must ``build()`` to a ``SampleSeries``.
+- `identity` (`dict of str`): Non-empty identity dict (arbitrary keys, e.g.
+``{"channel_name": "Eng_RPM", "data_key": "TM"}``), emitted per row and
+used to seed the deterministic :attr:`channel_id`.
+
+#### canonical\_identity
+
+```python
+def canonical_identity() -> str
+```
+
+Return a stable string encoding of the identity, used for the id hash.
+
+Keys are sorted so the encoding (and the derived ``channel_id``) is
+independent of kwarg order.
+
+
+#### channel\_id
+
+```python
+def channel_id() -> int
+```
+
+Deterministic output ``channel_id`` derived from the identity.
+
+A CRC32 of :meth:`canonical_identity` masked to a positive int32, so the
+value is stable across runs/processes and fits both ``IntegerType`` and
+``LongType`` source ``channel_id`` columns.  Determinism makes writes
+idempotent and joins predictable; sharing this one derivation across the
+query-engine and reporting layers keeps their ids in lockstep.
+
+
+#### build
+
+```python
+def build(cache: SeriesCache)
+```
+
+Evaluate the wrapped expression and return its raw ``(tstarts, tends, values)``.
+
+Unlike a typical ``TimeSeriesExpression`` (whose ``build`` yields a
+``SampleSeries``), this returns the three parallel ``float64`` arrays
+underlying that series, since the calculated-channels solve path consumes
+the raw samples directly.
+
+
+#### dtype
+
+```python
+def dtype() -> T.DataType
+```
+
+Spark type of ``build()``'s output: the raw ``(tstarts, tends, values)`` arrays.
+
+A struct of three arrays mirroring the tuple ``build()`` returns, with
+element types matching the narrow calculated-channels output
+(``tstart``/``tend`` are ``long``, ``value`` is ``double``).
+
+

docs/impulse/docs/references/api/impulse_query_engine/analyze/query/query_builder.md

@@ -157,6 +157,44 @@ When None, all containers matching query filters are processed (full mode).
 
 `pyspark.sql.DataFrame`: DataFrame containing query results.
 
+#### solve\_calculated\_channels
+
+```python
+def solve_calculated_channels(
+        spark,
+        solver: QuerySolver = BlobSolver(),
+        pre_filtered_containers_df: DataFrame = None) -> DataFrame
+```
+
+Compute calculated channels and return a narrow silver-shaped DataFrame.
+
+Every selection must be a :class:`CalculatedChannel`.  This runs the same
+metadata filter pipeline as :meth:`solve` (resolving the input channels
+each calculated channel depends on), then evaluates each calculated
+channel per container and emits rows in the silver ``channel_data`` shape
+— ``container_id, channel_id, tstart, tend, value`` — plus a single
+``identity`` ``MapType(string, string)`` column holding each channel's
+identity dict.
+
+**Arguments**:
+
+- `spark` (`SparkSession`): Spark session used for query execution.
+- `solver` (`QuerySolver`): Query solver to use.  Must implement ``solve_calculated_channels``
+(``DefaultSolver`` does); the default ``BlobSolver`` does not.
+- `pre_filtered_containers_df` (`DataFrame`): Pre-filtered container metrics for incremental processing.  When
+provided, only these containers are processed; when None, all
+containers matching the query filters are processed.
+
+**Raises**:
+
+- `ValueError`: If any selection is not a ``CalculatedChannel``, or if a wrapped
+expression does not evaluate to a ``SampleSeries``.
+
+**Returns**:
+
+`pyspark.sql.DataFrame`: Narrow DataFrame ``[container_id, channel_id, tstart, tend, value,
+identity]``.
+
 #### toPandas
 
 ```python

docs/impulse/docs/references/api/impulse_query_engine/analyze/query/solvers/default_solver.md

@@ -46,7 +46,8 @@ processing.  Requires an ``is_plausible`` column in the
 silver layer.
 - `raw_encoder` (`RawEncoder`): Which encoder converts RAW point data into intervals for solving.
 ``RawEncoder.RLE`` (default) run-length encodes equal-valued runs;
-``RawEncoder.INTERVAL`` only derives ``tend``.  Only consulted when ``is_raw_data`` is ``True``.
+``RawEncoder.INTERVAL`` only derives ``tend`` and drops exact
+duplicates.  Only consulted when ``is_raw_data`` is ``True``.
 
 #### filter\_container\_tags
 
@@ -288,3 +289,29 @@ the source to the target unit on the fly.
 
 `pyspark.sql.DataFrame`: DataFrame containing results for each container.
 
+#### solve\_calculated\_channels
+
+```python
+def solve_calculated_channels(query, channels_df, selections) -> DataFrame
+```
+
+Solve calculated channels by grouping channels and exploding each result.
+
+Structurally parallels :meth:`solve` — sharing the
+:meth:`_prepare_channels_join` prelude and :meth:`_apply_grouped_map`
+tail — but the grouped-map UDF emits narrow silver-shaped rows (many per
+container) instead of one wide row.  Output columns are
+``[container_id, channel_id, tstart, tend, value, identity]`` where
+``identity`` is a ``MapType(string, string)`` holding the channel's
+identity dict.
+
+**Arguments**:
+
+- `query` (`QueryBuilder`): Query object containing database and filter information.
+- `channels_df` (`pyspark.sql.DataFrame`): Channel-match DataFrame from the filter pipeline.
+- `selections` (`list`): List of ``CalculatedChannel`` selections to evaluate.
+
+**Returns**:
+
+`pyspark.sql.DataFrame`: Narrow DataFrame of calculated-channel samples.
+

docs/impulse/docs/references/api/impulse_query_engine/analyze/query/solvers/query_solver.md

@@ -202,3 +202,27 @@ Stage 6: Solve query.
 
 `pyspark.sql.DataFrame`: DataFrame containing results for each container.
 
+#### solve\_calculated\_channels
+
+```python
+def solve_calculated_channels(query, channels_df, selections) -> DataFrame
+```
+
+Solve calculated channels into a narrow, silver-shaped DataFrame.
+
+Optional stage, parallel to :meth:`solve` but emitting many rows per
+container (the exploded ``SampleSeries`` of each ``CalculatedChannel``)
+instead of one wide row.  Solvers that support calculated channels
+(e.g. ``DefaultSolver``) override this; the base implementation raises.
+
+**Arguments**:
+
+- `query` (`QueryBuilder`): Query object containing database and filter information.
+- `channels_df` (`pyspark.sql.DataFrame`): Channel-match DataFrame from the filter pipeline.
+- `selections` (`list`): List of ``CalculatedChannel`` selections to evaluate.
+
+**Returns**:
+
+`pyspark.sql.DataFrame`: Narrow ``[container_id, channel_id, tstart, tend, value,
+identity]`` DataFrame (``identity`` is a ``MapType(string, string)``).
+

docs/impulse/docs/references/api/impulse_query_engine/analyze/query/solvers/solver_config.md

@@ -37,7 +37,8 @@ RLE
     ``value`` within a container/channel into a single interval (see
     :class:`~impulse_query_engine.analyze.query.solvers.utils.rle_encoder.RleEncoder`).
 INTERVAL
-    Derive ``tend`` from the following sample's timestamp, *without* merging equal-valued runs (see
+    Derive ``tend`` from the following sample's timestamp and drop exact
+    duplicate points, *without* merging equal-valued runs (see
     :class:`~impulse_query_engine.analyze.query.solvers.utils.interval_encoder.IntervalEncoder`).
 
 
@@ -385,6 +386,24 @@ def group_id_col() -> str
 Internal column name for the unit group id on the unit_conversion table.
 
 
+#### timestamp\_col
+
+```python
+def timestamp_col() -> str
+```
+
+Internal column name for the timestamp on the channels table for raw encoded channel data.
+
+
+#### is\_plausible\_col
+
+```python
+def is_plausible_col() -> str
+```
+
+Internal column name for the plausibility flag on the channels table for raw encoded channel data.
+
+
 #### effective\_alias\_join\_keys
 
 ```python