From 5f8ba9699ee6e157b53771f03d79aa1b3654d66c Mon Sep 17 00:00:00 2001 From: "fabian.ade" Date: Mon, 15 Jun 2026 14:40:10 +0200 Subject: [PATCH 1/2] draft solution for fork pr handling --- .github/workflows/acceptance.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/acceptance.yml b/.github/workflows/acceptance.yml index 96819e2..218c38b 100644 --- a/.github/workflows/acceptance.yml +++ b/.github/workflows/acceptance.yml @@ -37,6 +37,10 @@ jobs: labels: linux-ubuntu-latest permissions: id-token: write + # Fork PRs get no OIDC token / secrets from GitHub, so JFrog auth (and therefore + # dependency installation) cannot run. Skip CI for them; fork PRs are to be tested + # by the reviewer(s) / maintainer(s) before merging. + if: ${{ github.event_name != 'pull_request' || !github.event.pull_request.head.repo.fork }} steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 @@ -63,6 +67,8 @@ jobs: needs: [ not-a-fork, lint ] permissions: id-token: write + # See the note on `lint`: fork PRs cannot authenticate to JFrog, so skip CI for them. + if: ${{ github.event_name != 'pull_request' || !github.event.pull_request.head.repo.fork }} strategy: fail-fast: false matrix: From 77c0594b47fb84c37df1c267ea985c5560734e2f Mon Sep 17 00:00:00 2001 From: "fabian.ade" Date: Tue, 4 Aug 2026 13:24:31 +0200 Subject: [PATCH 2/2] First draft impl --- .../analyze/metadata/poi_expression.py | 189 +++++++++ .../analyze/metadata/poi_selector.py | 217 +++++++++++ .../metadata/time_series_expression.py | 64 +++ .../analyze/query/aggregations/histogram.py | 6 + .../analyze/query/aggregations/histogram2d.py | 10 + .../aggregations/point_value_aggregator.py | 14 + .../query/aggregations/stats_aggregator.py | 8 + .../events/sequence_of_events_expression.py | 6 + .../analyze/query/query_builder.py | 83 +++- .../analyze/query/solvers/blob_solver.py | 5 +- .../analyze/query/solvers/default_solver.py | 256 +++++++++++- .../analyze/query/solvers/in_memory_solver.py | 2 +- .../analyze/query/solvers/query_solver.py | 37 +- .../analyze/query/solvers/series_cache.py | 30 ++ .../analyze/query/solvers/solver_config.py | 55 +++ src/impulse_query_engine/measurement_db.py | 10 + src/impulse_query_engine/schema.py | 23 ++ src/impulse_reporting/config/config_parser.py | 1 + .../query/solvers/default_solver_poi_test.py | 365 ++++++++++++++++++ .../model/expressions/poi_selector_test.py | 194 ++++++++++ 20 files changed, 1567 insertions(+), 8 deletions(-) create mode 100644 src/impulse_query_engine/analyze/metadata/poi_expression.py create mode 100644 src/impulse_query_engine/analyze/metadata/poi_selector.py create mode 100644 tests/impulse_query_engine/unit/analyze/query/solvers/default_solver_poi_test.py create mode 100644 tests/impulse_query_engine/unit/model/expressions/poi_selector_test.py diff --git a/src/impulse_query_engine/analyze/metadata/poi_expression.py b/src/impulse_query_engine/analyze/metadata/poi_expression.py new file mode 100644 index 0000000..01d44a2 --- /dev/null +++ b/src/impulse_query_engine/analyze/metadata/poi_expression.py @@ -0,0 +1,189 @@ +"""A dedicated predicate DSL for POI (point-of-interest) row filtering. + +This replaces the earlier reuse of :class:`TagExpression` for POI. A POI table is a +**wide, natively-typed** table (``duration double``, ``occurrences int``, ``poi_type +varchar`` …), so — unlike the EAV tag tables whose ``value`` column is always a string — +POI predicates need **no caller-supplied cast**: the column type comes from the table. + +The model mirrors :class:`MetricExpression` (the accessor for wide ``container_metrics`` +columns), which is the right analogue for a wide typed table, rather than +:class:`TagExpression` (EAV). ``QueryBuilder.poi_metric("duration") > 5`` is to the POI +table what ``QueryBuilder.metric("duration_ms") > 5`` is to ``container_metrics``. + +Two classes: + +- :class:`PoiMetricSelector` — a reference to a POI column, produced by ``q.poi_metric``. + Comparison operators on it build a :class:`PoiPredicate`. +- :class:`PoiPredicate` — the small predicate AST. It satisfies the two contracts the + solver needs: :meth:`get_selector_expr` (compiles to a Spark ``Column`` so + ``DefaultSolver.filter_poi`` can push it down) and a stable :meth:`__str__` (so a + ``PoiSelector`` carrying it hashes to a stable ``selector_id`` / definition hash). +""" + +from __future__ import annotations + +import operator +from typing import Any + +import pyspark.sql.functions as F +from pyspark.sql.column import Column + +# Operator → SQL-ish symbol, for a stable, readable ``__str__`` that feeds the definition +# hash. Kept explicit (not ``operator.__name__``) so the string form never drifts. +_OP_SYMBOL = { + operator.eq: "==", + operator.ne: "!=", + operator.gt: ">", + operator.ge: ">=", + operator.lt: "<", + operator.le: "<=", + operator.and_: "&", + operator.or_: "|", +} + + +class PoiPredicate: + """A comparison/boolean predicate over POI columns. + + Built by comparing a :class:`PoiMetricSelector` (``q.poi_metric("duration") > 5``) and + combined with ``&`` / ``|``. Consumed by ``DefaultSolver.filter_poi`` via + :meth:`get_selector_expr`. + """ + + def __init__(self, op, left, right): + self.op = op + self.left = left + self.right = right + + def get_selector_expr(self) -> Column: + """Compile this predicate to a Spark ``Column`` for pushdown in ``filter_poi``. + + A :class:`PoiMetricSelector` operand becomes a (optionally cast) ``F.col``; a + :class:`PoiPredicate` operand recurses; anything else is a literal. + """ + return self.op(self._compile(self.left), self._compile(self.right)) + + @staticmethod + def _compile(operand): + if isinstance(operand, PoiMetricSelector): + return operand.column_expr() + if isinstance(operand, PoiPredicate): + return operand.get_selector_expr() + return operand + + def required_columns(self) -> set[str]: + """Return the set of POI column names this predicate references.""" + cols: set[str] = set() + for operand in (self.left, self.right): + if isinstance(operand, PoiMetricSelector): + cols.add(operand.name) + elif isinstance(operand, PoiPredicate): + cols |= operand.required_columns() + return cols + + def __and__(self, other: "PoiPredicate") -> "PoiPredicate": + return PoiPredicate(operator.and_, self, other) + + def __or__(self, other: "PoiPredicate") -> "PoiPredicate": + return PoiPredicate(operator.or_, self, other) + + def __str__(self) -> str: + return f"({self._str(self.left)} {_OP_SYMBOL.get(self.op, '?')} {self._str(self.right)})" + + @staticmethod + def _str(operand) -> str: + return str(operand) if isinstance(operand, (PoiMetricSelector, PoiPredicate)) else repr(operand) + + def as_dict(self) -> dict[str, Any]: + return { + "type": "PoiPredicate", + "op": _OP_SYMBOL.get(self.op, "?"), + "left": self.left.as_dict() if hasattr(self.left, "as_dict") else self.left, + "right": self.right.as_dict() if hasattr(self.right, "as_dict") else self.right, + } + + +class PoiMetricSelector: + """A reference to a single POI column, produced by ``QueryBuilder.poi_metric(name)``. + + Comparison operators build a :class:`PoiPredicate`. An optional ``cast_type`` is + accepted for parity with ``q.metric``/``q.tag`` and for the rare case of a POI column + whose physical type is not what the comparison needs, but it is **not required**: POI + columns are natively typed, so ``q.poi_metric("duration") > 5`` compares numerically + with no cast (unlike EAV tags, where the string ``value`` column forced a cast). + + Parameters + ---------- + name : str + The POI column name (after ``PoiConfig.column_name_mapping``). + cast_type : str or None, optional + Optional Spark type to cast the column to before comparison. Defaults to ``None`` + (use the column's native type). + """ + + def __init__(self, name: str, cast_type: str | None = None): + self.name = name + self.cast_type = cast_type + + def column_expr(self) -> Column: + col = F.col(self.name) + if self.cast_type is not None: + col = col.cast(self.cast_type) + return col + + def __eq__(self, other) -> PoiPredicate: + return PoiPredicate(operator.eq, self, other) + + def __ne__(self, other) -> PoiPredicate: + return PoiPredicate(operator.ne, self, other) + + def __gt__(self, other) -> PoiPredicate: + return PoiPredicate(operator.gt, self, other) + + def __ge__(self, other) -> PoiPredicate: + return PoiPredicate(operator.ge, self, other) + + def __lt__(self, other) -> PoiPredicate: + return PoiPredicate(operator.lt, self, other) + + def __le__(self, other) -> PoiPredicate: + return PoiPredicate(operator.le, self, other) + + __hash__ = None # predicates are not hashable; PoiMetricSelector isn't either + + def __str__(self) -> str: + cast = f", cast={self.cast_type}" if self.cast_type else "" + return f"PoiMetricSelector<{self.name}{cast}>" + + def as_dict(self) -> dict[str, Any]: + return {"type": "PoiMetricSelector", "name": self.name, "cast_type": self.cast_type} + + +def poi_kind_predicate(**kind_filters) -> PoiPredicate: + """Build the ANDed equality predicate behind ``QueryBuilder.poi(**kind_filters)``. + + ``poi(poi_type="aeb", event_type="computed")`` becomes + ``(poi_type == "aeb") & (event_type == "computed")`` as :class:`PoiPredicate` nodes. + + Parameters + ---------- + **kind_filters : dict + Column-name → value equality pairs identifying the POI rows. + + Returns + ------- + PoiPredicate + The combined predicate. + + Raises + ------ + ValueError + If no kind filter is given (a bare ``poi()`` would match the whole table). + """ + pred: PoiPredicate | None = None + for key, value in kind_filters.items(): + eq = PoiMetricSelector(key) == str(value) + pred = eq if pred is None else (pred & eq) + if pred is None: + raise ValueError("poi(...) needs at least one kind filter, e.g. poi(poi_type='aeb')") + return pred diff --git a/src/impulse_query_engine/analyze/metadata/poi_selector.py b/src/impulse_query_engine/analyze/metadata/poi_selector.py new file mode 100644 index 0000000..17cbf43 --- /dev/null +++ b/src/impulse_query_engine/analyze/metadata/poi_selector.py @@ -0,0 +1,217 @@ +"""The POI (point of interest) TSAL expression leaf. + +A ``PoiSelector`` is a **sibling** of :class:`TimeSeriesSelector` — it subclasses +:class:`TimeSeriesExpression` so it composes with every operator (``+ - * / & |``) and +every core-model method reachable through ``TimeSeriesExpression.__getattr__``, but it is +deliberately **not** a ``TimeSeriesSelector``. See ``POI_PROPOSAL_REVIEW.md`` §2 for why: +``TimeSeriesSelector`` carries the ``RequiresDeserialization`` marker, and the +``toPandas()`` deserialization gate (``query_builder.py``) fires on that marker's +``isinstance`` — routing POI results (``array``) through +``SampleSeries.deserialize`` → ``lz4f.decompress`` and crashing. Not inheriting avoids +the whole problem. + +**POI is a pure occurrence log (Option D).** A ``PoiSelector`` always evaluates to a +:class:`PointsInTime` — the instants at which the matching occurrences happened. It does +**not** carry the POI table's own snapshot columns (``vehicle_wheel_speed`` etc.) as +values: those are redundant with the measured ``channels``, so a value *at* an occurrence +is obtained by sampling the real channel — ``q.channel("Vehicle Speed Sensor").where(poi)`` +— which is typed by the channel and needs no POI-attribute machinery. + +Row filtering uses a **dedicated POI predicate** (:mod:`poi_expression`), not the EAV +``TagExpression``: + +- ``q.poi(poi_type="aeb")`` — kind filter (equality kwargs). +- ``q.poi(poi_type="aeb").having(q.poi_metric("duration") > 5)`` — an extra row predicate. + +``build(cache)`` returns an empty :class:`PointsInTime` when the cache has no POI data — +which is exactly what makes ``build(EmptyTimeSeriesCache())`` work as a data-free type +probe, so ``require_evaluation_type`` can validate a POI expression at construction time +with no Spark. +""" + +from __future__ import annotations + +import zlib +from typing import TYPE_CHECKING, Any + +import numpy as np + +from impulse_query_engine.analyze.metadata.poi_expression import ( + PoiPredicate, + poi_kind_predicate, +) +from impulse_query_engine.analyze.metadata.time_series_expression import ( + TimeSeriesExpression, +) +from impulse_query_engine.model.series.points_in_time import PointsInTime + +if TYPE_CHECKING: + from impulse_query_engine.analyze.query.solvers.series_cache import SeriesCache + + +class PoiSelector(TimeSeriesExpression): + """A TSAL leaf selecting points of interest from the configured POI table. + + Always evaluates to :class:`PointsInTime`. + + Parameters + ---------- + kind_predicate : PoiPredicate + The ANDed equality predicate identifying which POI rows this selector matches + (e.g. ``poi_type == "aeb"``). Built by :func:`poi_expression.poi_kind_predicate` + from the ``**kind_filters`` passed to ``QueryBuilder.poi``. + having : list of PoiPredicate or None, optional + Extra row predicates (e.g. ``q.poi_metric("duration") > 5``), ANDed with the kind + predicate and pushed down Spark-side in ``DefaultSolver.filter_poi``. Normally set + through :meth:`having` rather than the constructor. + """ + + def __init__( + self, + kind_predicate: PoiPredicate, + *, + having: list[PoiPredicate] | None = None, + ): + self._kind_predicate = kind_predicate + self._having = list(having) if having else [] + super().__init__(is_single_signal=True) + + # ------------------------------------------------------------------ + # Fluent row filtering + # ------------------------------------------------------------------ + + def having(self, *predicates: PoiPredicate) -> "PoiSelector": + """Return a new ``PoiSelector`` with extra row predicate(s) ANDed in. + + Named ``having`` rather than ``where`` deliberately: on a + :class:`TimeSeriesExpression`, ``where`` already means "sample this series at those + points" (``channel.where(poi)``), so reusing it as a row filter would give one word + two meanings. ``having`` reads as "restrict the source's rows" and returns a new + immutable selector, so it chains: ``q.poi(poi_type="aeb").having(a).having(b)``. + + Parameters + ---------- + *predicates : PoiPredicate + Row predicates over POI columns, e.g. ``q.poi_metric("duration") > 5``. + + Returns + ------- + PoiSelector + A new selector; the original is unchanged. + """ + return PoiSelector( + self._kind_predicate, having=[*self._having, *predicates] + ) + + # ------------------------------------------------------------------ + # Identity / definition hashing + # ------------------------------------------------------------------ + + @property + def _predicate(self) -> PoiPredicate: + """The full row predicate: kind equality ANDed with any ``having`` predicates.""" + pred = self._kind_predicate + for extra in self._having: + pred = pred & extra + return pred + + @property + def selector_id(self) -> int: + """crc32 of the predicate — matches ``TimeSeriesSelector.selector_id`` derivation. + + Kept consistent with the channel selectors so the same + ``QuerySolver._build_selector_id_expr`` ``WHEN…THEN`` machinery can tag POI rows + with a ``selector_id`` in Spark. + """ + return zlib.crc32(str(self._predicate).encode()) + + def __str__(self) -> str: + """String form — feeds the definition hash. + + Includes the full row predicate (kind + ``having``) so two differently-filtered + POIs of the same kind are treated as **distinct definitions** by the incremental + definition-hash comparator. + """ + return f"PoiSelector<{self._predicate}>" + + # ------------------------------------------------------------------ + # The type-correct core + # ------------------------------------------------------------------ + + def build(self, cache: SeriesCache) -> PointsInTime: + """Resolve this POI selector against *cache* into a :class:`PointsInTime`. + + Returns an empty ``PointsInTime`` when the cache has no POI rows — this is what + makes ``build(EmptyTimeSeriesCache())`` a valid, data-free type probe. + ``EmptyTimeSeriesCache`` and the definition-time probe therefore always land here + with an empty frame; the solver's ``TimeSeriesCache`` overrides ``resolve_poi`` to + return this container's matching, deduped POI instants. + + Parameters + ---------- + cache : SeriesCache + Cache exposing ``resolve_poi(selection)`` (a concrete default on + :class:`SeriesCache` returning an empty frame). + + Returns + ------- + PointsInTime + The occurrence instants (sorted, unique). + """ + rows = cache.resolve_poi(self) + if rows is None or len(rows) == 0: + return PointsInTime.empty() + ts = np.asarray(rows["ts"], dtype=np.float64) + # np.unique returns sorted + unique, satisfying PointsInTime's assume_unique contract. + return PointsInTime(np.unique(ts)) + + def dtype(self): + """Spark wire type — owned by the core-model type, never hardcoded here.""" + return PointsInTime.empty().dtype() + + # ------------------------------------------------------------------ + # TimeSeriesExpression contract (the abstract methods) + # ------------------------------------------------------------------ + + def get_required_tag_exprs(self) -> set: + # POI predicates are not TagExpressions; nothing consumes tag *expressions* for a + # POI leaf (get_selectors() == [] keeps it out of the channel-tag pivot), so the + # set is empty. required_columns() below exposes the referenced POI columns. + return set() + + def required_tags(self) -> set[str]: + # The POI columns this selector references. Only ever unioned by aggregation + # wrappers; never reaches the channel-tag pivot (get_selectors() == []). + return self._predicate.required_columns() + + def get_selector_expr(self): + # The Spark predicate. Applied in DefaultSolver.filter_poi, keyed by selector_id; + # the channel pipeline never sees this because get_selectors() returns []. + return self._predicate.get_selector_expr() + + def get_selectors(self) -> list: + # Empty by design: keeps POI out of filter_channel_tags / filter_channel_metrics. + return [] + + def get_poi_selectors(self) -> list["PoiSelector"]: + # The parallel collection path, mirrored by TimeSeriesExpression.get_poi_selectors. + return [self] + + # ------------------------------------------------------------------ + # Serialization + # ------------------------------------------------------------------ + + def as_dict(self) -> dict[str, Any]: + obj = TimeSeriesExpression.as_dict(self) + obj["type"] = "PoiSelector" + obj["kind_predicate"] = self._kind_predicate.as_dict() + obj["having"] = [p.as_dict() for p in self._having] + return obj + + +def poi_expr(**kind_filters) -> PoiPredicate: + """Alias for :func:`poi_expression.poi_kind_predicate`. + + Kept as the builder behind ``QueryBuilder.poi``. + """ + return poi_kind_predicate(**kind_filters) diff --git a/src/impulse_query_engine/analyze/metadata/time_series_expression.py b/src/impulse_query_engine/analyze/metadata/time_series_expression.py index 0977dc3..a7e5cab 100644 --- a/src/impulse_query_engine/analyze/metadata/time_series_expression.py +++ b/src/impulse_query_engine/analyze/metadata/time_series_expression.py @@ -244,6 +244,54 @@ def get_selectors(self) -> list["TimeSeriesSelector"]: """ pass + def get_poi_selectors(self) -> list[Any]: + """Return all POI leaves reachable from this expression. + + Concrete default returning ``[]`` so every existing expression node is + POI-free unless it overrides this (``PoiSelector`` returns ``[self]``; + ``TimeSeriesOp`` and ``TimeSeriesAliasSelector`` walk their children). This is + the parallel of :meth:`get_selectors` for the POI pipeline: POI leaves return + ``[]`` from ``get_selectors`` so they stay out of the channel-match stages, and + are collected here instead. + + Returns + ------- + list + POI selector leaves (``PoiSelector`` instances); possibly with duplicates, + deduplicated by :meth:`collect_poi_selectors`. + """ + return [] + + @staticmethod + def collect_poi_selectors(expressions: Iterable[Any]) -> list[Any]: + """Collect deduplicated POI leaves from a list of expressions. + + Mirrors :meth:`collect_selectors`: walks each ``TimeSeriesExpression``'s + :meth:`get_poi_selectors`, skips non-expressions, and deduplicates by + ``selector_id`` preserving discovery order. + + Parameters + ---------- + expressions : Iterable[Any] + Items to walk; non-``TimeSeriesExpression`` entries are skipped. + + Returns + ------- + list + Deduplicated POI selectors in discovery order. + """ + selectors: list[Any] = [] + seen_ids: set = set() + for expression in expressions: + if not isinstance(expression, TimeSeriesExpression): + continue + for selector in expression.get_poi_selectors(): + if selector.selector_id in seen_ids: + continue + seen_ids.add(selector.selector_id) + selectors.append(selector) + return selectors + @staticmethod def collect_selectors( expressions: Iterable[Any], @@ -882,6 +930,12 @@ def get_selectors(self) -> list["TimeSeriesSelector"]: result.extend(alias.get_selectors()) return result + def get_poi_selectors(self) -> list[Any]: + result: list[Any] = [] + for alias in self._aliases: + result.extend(alias.get_poi_selectors()) + return result + def __str__(self): """ String representation. @@ -992,6 +1046,16 @@ def get_selectors(self) -> list["TimeSeriesSelector"]: result.extend(kwarg.get_selectors()) return result + def get_poi_selectors(self) -> list[Any]: + result: list[Any] = [] + for arg in self.args: + if isinstance(arg, TimeSeriesExpression): + result.extend(arg.get_poi_selectors()) + for kwarg in self.kwargs.values(): + if isinstance(kwarg, TimeSeriesExpression): + result.extend(kwarg.get_poi_selectors()) + return result + def build(self, cache: SeriesCache): """ Build the time series from cache. diff --git a/src/impulse_query_engine/analyze/query/aggregations/histogram.py b/src/impulse_query_engine/analyze/query/aggregations/histogram.py index d5eef9b..0e7130f 100644 --- a/src/impulse_query_engine/analyze/query/aggregations/histogram.py +++ b/src/impulse_query_engine/analyze/query/aggregations/histogram.py @@ -144,6 +144,9 @@ def get_required_tag_exprs(self) -> set[TagExpression]: def get_selectors(self) -> list[TimeSeriesSelector]: return self.selection.get_selectors() + def get_poi_selectors(self) -> list: + return self.selection.get_poi_selectors() + class HistogramCustomWeights(Aggregation): def __init__( @@ -287,3 +290,6 @@ def get_required_tag_exprs(self) -> set[TagExpression]: def get_selectors(self) -> list[TimeSeriesSelector]: return self.selection.get_selectors() + self.weights.get_selectors() + + def get_poi_selectors(self) -> list: + return self.selection.get_poi_selectors() + self.weights.get_poi_selectors() diff --git a/src/impulse_query_engine/analyze/query/aggregations/histogram2d.py b/src/impulse_query_engine/analyze/query/aggregations/histogram2d.py index 44bedb4..68f7968 100644 --- a/src/impulse_query_engine/analyze/query/aggregations/histogram2d.py +++ b/src/impulse_query_engine/analyze/query/aggregations/histogram2d.py @@ -133,6 +133,9 @@ def get_required_tag_exprs(self) -> set[TagExpression]: def get_selectors(self) -> list[TimeSeriesSelector]: return self.x_selection.get_selectors() + self.y_selection.get_selectors() + def get_poi_selectors(self) -> list: + return self.x_selection.get_poi_selectors() + self.y_selection.get_poi_selectors() + class Histogram2DCustomWeights(Aggregation): """Class representing a 2D histogram aggregation in a report with custom weights.""" @@ -310,3 +313,10 @@ def get_selectors(self) -> list[TimeSeriesSelector]: + self.y_selection.get_selectors() + self.weights_expr.get_selectors() ) + + def get_poi_selectors(self) -> list: + return ( + self.x_selection.get_poi_selectors() + + self.y_selection.get_poi_selectors() + + self.weights_expr.get_poi_selectors() + ) diff --git a/src/impulse_query_engine/analyze/query/aggregations/point_value_aggregator.py b/src/impulse_query_engine/analyze/query/aggregations/point_value_aggregator.py index 726bf75..5da980f 100644 --- a/src/impulse_query_engine/analyze/query/aggregations/point_value_aggregator.py +++ b/src/impulse_query_engine/analyze/query/aggregations/point_value_aggregator.py @@ -195,3 +195,17 @@ def get_selectors(self) -> list[TimeSeriesSelector]: if self.event_expression is not None: result.extend(self.event_expression.get_selectors()) return result + + def get_poi_selectors(self) -> list: + """Return all POI leaves reachable from the input and event expressions. + + This is what lets a POI scope a ``PointValueAggregator``: the event + expression is a ``PoiSelector`` (a ``PointsInTime``), and the report's + POI collection walks into it here so ``filter_poi`` runs for it. + """ + result: list = [] + for expr in self.input_expressions: + result.extend(expr.get_poi_selectors()) + if self.event_expression is not None: + result.extend(self.event_expression.get_poi_selectors()) + return result diff --git a/src/impulse_query_engine/analyze/query/aggregations/stats_aggregator.py b/src/impulse_query_engine/analyze/query/aggregations/stats_aggregator.py index 18c7f5a..4b736b9 100644 --- a/src/impulse_query_engine/analyze/query/aggregations/stats_aggregator.py +++ b/src/impulse_query_engine/analyze/query/aggregations/stats_aggregator.py @@ -590,6 +590,14 @@ def get_selectors(self) -> list[TimeSeriesSelector]: result.extend(self.event_expression.get_selectors()) return result + def get_poi_selectors(self) -> list: + result: list = [] + for expr in self.input_expressions: + result.extend(expr.get_poi_selectors()) + if self.event_expression is not None: + result.extend(self.event_expression.get_poi_selectors()) + return result + def weighted_median(self, durations, values): """Calculate duration-weighted median for RLE compressed data.""" # Extract the slice diff --git a/src/impulse_query_engine/analyze/query/events/sequence_of_events_expression.py b/src/impulse_query_engine/analyze/query/events/sequence_of_events_expression.py index 36dd498..b33d7ae 100644 --- a/src/impulse_query_engine/analyze/query/events/sequence_of_events_expression.py +++ b/src/impulse_query_engine/analyze/query/events/sequence_of_events_expression.py @@ -104,6 +104,12 @@ def get_selectors(self) -> list[TimeSeriesSelector]: result.extend(expr.get_selectors()) return result + def get_poi_selectors(self) -> list: + result: list = [] + for expr in self.expressions: + result.extend(expr.get_poi_selectors()) + return result + def get_selector_expr(self): """ Return the combined selector expression (OR of all children). diff --git a/src/impulse_query_engine/analyze/query/query_builder.py b/src/impulse_query_engine/analyze/query/query_builder.py index 26ac47e..8f0a2c9 100644 --- a/src/impulse_query_engine/analyze/query/query_builder.py +++ b/src/impulse_query_engine/analyze/query/query_builder.py @@ -5,6 +5,8 @@ from pyspark.sql import DataFrame from impulse_query_engine.analyze.metadata.metric_expression import MetricSelector +from impulse_query_engine.analyze.metadata.poi_expression import PoiMetricSelector +from impulse_query_engine.analyze.metadata.poi_selector import PoiSelector, poi_expr from impulse_query_engine.analyze.metadata.tag_expression import TagSelector from impulse_query_engine.analyze.metadata.time_series_expression import ( RequiresDeserialization, @@ -38,6 +40,10 @@ def __init__(self, db: "impulse_query_engine.analyze.MeasurementDB"): self.selections = [] self.result_objects = [] self.result_dtypes = [] + # Per-solve POI frame. Set inside _run_filter_pipeline and cleared at its top so a + # frame scoped to one container set can never leak into the next solve() — the + # correctness rule from POI_PROPOSAL_REVIEW.md §3.2. Never a long-lived cache. + self._poi_df = None def where(self, *args): """ @@ -161,6 +167,60 @@ def channel_with_alias(self, **kwargs) -> TimeSeriesSelector: expr = expr & (TagSelector(k) == str(arg)) return TimeSeriesSelector(expr, uses_alias=True) + def poi(self, **kind_filters) -> PoiSelector: + """ + Create a POI (point-of-interest) selector. + + A POI is a point in time from the configured ``poi_table`` (one row per + occurrence, e.g. "AEB fired here"). It always evaluates to :class:`PointsInTime` — + the instants of the matching occurrences. + + Row filtering: + + - equality kwargs pick the occurrences: ``q.poi(poi_type="aeb")``; + - extra predicates go through :meth:`PoiSelector.having` with :meth:`poi_metric`: + ``q.poi(poi_type="aeb").having(q.poi_metric("duration") > 5)``. + + To read a signal's value *at* each occurrence, sample the measured channel rather + than a POI column: ``q.channel("Vehicle Speed Sensor").where(q.poi(poi_type="aeb"))``. + + Parameters + ---------- + **kind_filters : dict + Column → value equality pairs identifying the rows, e.g. ``poi_type="aeb"``. + + Returns + ------- + PoiSelector + The POI expression leaf. + """ + return PoiSelector(poi_expr(**kind_filters)) + + def poi_metric(self, name: str, cast_type: str | None = None) -> PoiMetricSelector: + """ + Create a POI column selector for row-filtering, à la :meth:`metric`. + + ``q.poi_metric("duration") > 5`` builds a predicate over the wide, typed POI + table — the analogue of ``q.metric("duration_ms") > 5`` over ``container_metrics``. + Because POI columns are natively typed, **no cast is needed** for numeric + comparisons (unlike ``q.tag``, whose EAV string ``value`` forces one). Pass to + :meth:`PoiSelector.having`. + + Parameters + ---------- + name : str + The POI column name (after ``PoiConfig.column_name_mapping``). + cast_type : str or None, optional + Optional Spark cast, for the rare column whose physical type is not what the + comparison needs. Defaults to ``None`` (native type). + + Returns + ------- + PoiMetricSelector + A column reference; compare it to build a predicate. + """ + return PoiMetricSelector(name, cast_type=cast_type) + def select(self, *args) -> Self: """ Set the selection expressions for the query. @@ -238,7 +298,13 @@ def solve( channel_metrics_df = self._run_filter_pipeline(spark, solver, pre_filtered_containers_df) - return solver.solve(self, channel_metrics_df, self.selections, self.result_dtypes) + return solver.solve( + self, + channel_metrics_df, + self.selections, + self.result_dtypes, + poi_df=self._poi_df, + ) def _run_filter_pipeline(self, spark, solver, pre_filtered_containers_df) -> DataFrame: """Run the shared metadata filter pipeline and return the channel-match frame. @@ -250,6 +316,12 @@ def _run_filter_pipeline(self, spark, solver, pre_filtered_containers_df) -> Dat ``solver`` call. Returns the ``(container_id, channel_id, selector_ids …)`` DataFrame identifying the channels selected by the current selections. """ + # Clear any POI frame from a previous solve BEFORE running the pipeline: it is + # scoped to that solve's container set, and this QueryBuilder is reused across the + # two differently-scoped solves inside one determine_report() (see + # POI_PROPOSAL_REVIEW.md §3.2). Never carry it over. + self._poi_df = None + # extract selectors upfront direct_selectors = TimeSeriesExpression.collect_selectors( self.selections, uses_alias=False @@ -257,6 +329,7 @@ def _run_filter_pipeline(self, spark, solver, pre_filtered_containers_df) -> Dat aliased_selectors = TimeSeriesExpression.collect_selectors( self.selections, uses_alias=True ) + poi_selectors = TimeSeriesExpression.collect_poi_selectors(self.selections) # create Query tags_df = solver.filter_container_tags(spark, self) @@ -268,6 +341,14 @@ def _run_filter_pipeline(self, spark, solver, pre_filtered_containers_df) -> Dat spark, self.db, channel_tags_df, direct_selectors ) + # POI stage: resolve POI rows for the *container* frame (metrics_df) — already + # container-filtered and incremental-scoped. Keyed off the stage-2 frame rather + # than the channel-match frame so a POI-only query (no q.channel(...)) still has a + # non-empty container set (POI_PROPOSAL_REVIEW.md §3.1). No-op / None on solvers + # that don't support POI, so POI-free queries are unaffected. + if len(poi_selectors) > 0: + self._poi_df = solver.filter_poi(spark, self.db, metrics_df, poi_selectors) + if len(aliased_selectors) > 0: # Aliased resolution must run against the full tag-filtered container # set (metrics_df). diff --git a/src/impulse_query_engine/analyze/query/solvers/blob_solver.py b/src/impulse_query_engine/analyze/query/solvers/blob_solver.py index 6e5bced..4d8ffd1 100644 --- a/src/impulse_query_engine/analyze/query/solvers/blob_solver.py +++ b/src/impulse_query_engine/analyze/query/solvers/blob_solver.py @@ -128,10 +128,13 @@ def _solve_container(row, selections=None, base_uri=None): result[s._alias] = s.build(cache) return result - def solve(self, query, channels_df, selections, dtypes=None): + def solve(self, query, channels_df, selections, dtypes=None, poi_df=None): """ Solve the query by grouping channels and applying selections. + ``poi_df`` is accepted for interface compatibility with + :meth:`QuerySolver.solve` and ignored — ``BlobSolver`` does not support POI. + Parameters ---------- query : QueryBuilder diff --git a/src/impulse_query_engine/analyze/query/solvers/default_solver.py b/src/impulse_query_engine/analyze/query/solvers/default_solver.py index 5e2e5d7..269c8e3 100644 --- a/src/impulse_query_engine/analyze/query/solvers/default_solver.py +++ b/src/impulse_query_engine/analyze/query/solvers/default_solver.py @@ -24,7 +24,13 @@ class TimeSeriesCache(SeriesCache): - def __init__(self, pdf, col_map: dict[str, str]): + def __init__( + self, + pdf, + col_map: dict[str, str], + poi_pdf=None, + poi_col_map: dict[str, str] | None = None, + ): """ Initialize the TimeSeriesCache. @@ -39,12 +45,23 @@ def __init__(self, pdf, col_map: dict[str, str]): the caller's frame is reordered by ``(cid, ch, ts)`` and its index reset. The sort makes each channel's rows contiguous, which the constructor exploits to build a ``(cid, ch)`` → - ``(start, stop)`` range index for :meth:`load_blob`. + ``(start, stop)`` range index for :meth:`load_blob`. May be + **empty** (a POI-only container in the cogroup path); the range + index is then empty and only :meth:`resolve_poi` returns data. col_map : dict[str, str] Mapping with keys ``"cid"``, ``"ch"``, ``"ts"``, ``"te"``, ``"val"``, ``"conv"`` to the actual column names in *pdf*. The ``"conv"`` column is optional in *pdf*. + poi_pdf : pd.DataFrame, optional + This container's resolved POI rows from ``filter_poi`` — already + time-base-normalized (integer-µs ``ts``), deduped, and tagged with + a ``selector_id``. ``None`` when the query has no POI selectors. + poi_col_map : dict[str, str], optional + Mapping with keys ``"cid"``, ``"ts"``, ``"sel"`` to the POI-frame + column names. Required when *poi_pdf* is given. """ + self._poi_pdf = poi_pdf + self._poi_col_map = poi_col_map or {} self._cid_col = col_map["cid"] self._ch_col = col_map["ch"] self._ts_col = col_map["ts"] @@ -98,6 +115,35 @@ def resolve(self, selection): idx = selection._expr.build_pandas(self.mdf) return self.mdf[idx] + def resolve_poi(self, selection): + """Return this container's POI rows matching *selection*, renamed to ``ts``. + + The Spark-side ``filter_poi`` already evaluated every POI predicate and tagged + each surviving row with the matching ``selector_id``, so here we only select the + rows carrying this selector's id — the UDF never sees POI rows it cannot use. The + resolved-instant column is exposed as ``ts`` (what :meth:`PoiSelector.build` + reads). + + Parameters + ---------- + selection : PoiSelector + The asking POI leaf; ``selection.selector_id`` selects its rows. + + Returns + ------- + pandas.DataFrame + Rows with a ``ts`` column (integer µs). Empty when this container has no + matching POI rows. + """ + if self._poi_pdf is None or len(self._poi_pdf) == 0: + return pd.DataFrame(columns=["ts"]) + sel_col = self._poi_col_map["sel"] + ts_col = self._poi_col_map["ts"] + rows = self._poi_pdf[self._poi_pdf[sel_col] == selection.selector_id] + if ts_col != "ts": + rows = rows.rename(columns={ts_col: "ts"}) + return rows + def load_blob(self, mid, cid, uses_alias: bool = False): """ Load a time series blob from the DataFrame. @@ -898,6 +944,125 @@ def _compute_conversion_factors(self, spark, query, channels_df: DataFrame) -> D return channels_df + # ------------------------------------------------------------------ + # POI stage + # ------------------------------------------------------------------ + + def filter_poi(self, spark, db, container_df, poi_selectors) -> DataFrame | None: + """Resolve POI rows for the selected containers into a narrow, solve-ready frame. + + Modeled on :meth:`filter_aliased_channel_metrics`: apply + ``column_name_mapping`` → ``project_id`` → per-table ``filters``, restrict to the + candidate containers, resolve the ``ts_column`` datetime to integer microseconds, + and dedup to one row per ``(selector, container, instant)`` with a deterministic + total order. + + The output is narrow (``container_id``, ``ts``, ``selector_id``) and each row is + tagged with the ``selector_id`` of the POI leaf whose predicate it satisfied, so + the per-container cache can hand each selector exactly its rows without + re-evaluating predicates. + + Parameters + ---------- + spark : SparkSession + Spark session used for query execution. + db : MeasurementDB + Measurement database; reads ``db.poi(spark)``. + container_df : pyspark.sql.DataFrame + Stage-2 container frame — already container-filtered and incremental-scoped. + poi_selectors : list[PoiSelector] + POI leaves collected from the selections. + + Returns + ------- + pyspark.sql.DataFrame or None + ``None`` when there are no POI selectors. Otherwise + ``(container_id, ts, selector_id)``. + """ + if not poi_selectors: + return None + + poi_cfg = self.config.poi + container_id_col = self.config.container_id_col + ts_col = self.config.poi_ts_col + sel_col = self.config.poi_selector_id_col + + poi = db.poi(spark) + poi = self._apply_column_mapping(poi, poi_cfg.column_name_mapping) + + if self.config.project_id is not None and self.config.project_id_col in poi.columns: + poi = poi.where(F.col(self.config.project_id_col) == self.config.project_id) + for col_name, value in poi_cfg.filters.items(): + poi = poi.where(F.col(col_name) == value) + + # Keep only rows matching some selector; tag each with its selector_id. Rows + # matching several selectors are exploded so each selector sees its own copy. + poi = poi.where(self._build_expr(poi_selectors)) + sel_when = F.array( + *[ + F.when(s.get_selector_expr(), F.lit(s.selector_id)) + for s in poi_selectors + ] + ) + poi = poi.withColumn( + sel_col, + F.explode(F.array_compact(sel_when)), + ) + + # Restrict to candidate containers with a left_semi join: filters without + # widening, and no POI row can be duplicated by a repeated container key. This is + # what replaces the old inner-join container binding (POI_PROPOSAL_REVIEW.md §2). + poi = poi.join( + F.broadcast(container_df.select(container_id_col).distinct()), + on=[container_id_col], + how="left_semi", + ) + + poi = self._resolve_poi_time_base(poi) + poi = self._dedup_poi(poi, poi_selectors) + + # POI is a pure occurrence log (Option D): ship only keys, the resolved instant, + # and the selector id. Values at an occurrence come from sampling the measured + # channel (channel.where(poi)), not from POI columns, so no attribute columns. + return poi.select(container_id_col, ts_col, sel_col) + + def _resolve_poi_time_base(self, poi) -> DataFrame: + """Add the internal integer-microsecond ``ts`` column from ``PoiConfig.ts_column``. + + ``ts_column`` must be a datetime / Spark ``timestamp`` column (enforced by + documentation, see the silver-layer schema docs). It is read directly as an + absolute instant via ``unix_micros`` — no unit or per-container-origin assumptions. + A non-timestamp column would silently resolve to nonsensical instants, which is why + the datetime requirement is stated explicitly. + """ + ts_col = self.config.poi_ts_col + return poi.withColumn( + ts_col, F.unix_micros(F.col(self.config.poi.ts_column).cast("timestamp")) + ) + + def _dedup_poi(self, poi, poi_selectors) -> DataFrame: + """Collapse to one row per ``(selector_id, container, instant)``, deterministically. + + Two POI rows can share an instant; the gold event id is + ``crc32(cid::name::start::end)`` with ``start == end`` for a point, so duplicates + collide in the MERGE. The dedup order is ``PoiConfig.dedup_order_by`` (any columns + present on the frame) then a stable fallback, giving a *total* order so the + surviving row is deterministic across shuffles. + """ + container_id_col = self.config.container_id_col + ts_col = self.config.poi_ts_col + sel_col = self.config.poi_selector_id_col + + order_cols = [ + F.col(c).asc_nulls_last() + for c in self.config.poi.dedup_order_by + if c in poi.columns + ] + order_cols.append(F.col(ts_col).asc_nulls_last()) + dedup_window = Window.partitionBy(sel_col, container_id_col, ts_col).orderBy(*order_cols) + poi = poi.withColumn("_poi_rank", F.row_number().over(dedup_window)) + return poi.where(F.col("_poi_rank") == 1).drop("_poi_rank") + # ------------------------------------------------------------------ # Solve # ------------------------------------------------------------------ @@ -932,7 +1097,41 @@ def _solve_udf(pdf, selections: Iterable, col_map: dict[str, str]) -> pd.DataFra result[s._alias] = [res] return pd.DataFrame(result) - def solve(self, query, channels_df, selections, dtypes) -> DataFrame: + @staticmethod + def _solve_udf_with_poi( + chan_pdf, + poi_pdf, + selections: Iterable, + col_map: dict[str, str], + poi_col_map: dict[str, str], + ) -> pd.DataFrame: + """Cogroup UDF: solve one container from its channel frame *and* its POI frame. + + Cogroup is full-outer, so exactly one of the two frames may be empty: + + - channels-only container → ``poi_pdf`` empty (channels still solve), + - POI-only container → ``chan_pdf`` empty (POI still solves). + + The container id is read from whichever frame is non-empty, so a POI-only + container does not ``IndexError`` on ``chan_pdf[cid].iloc[0]``. + """ + cid_col = col_map["cid"] + src = chan_pdf if len(chan_pdf) > 0 else poi_pdf + src_cid_col = cid_col if len(chan_pdf) > 0 else poi_col_map["cid"] + result = {cid_col: [src[src_cid_col].iloc[0]]} + cache = TimeSeriesCache( + chan_pdf, col_map=col_map, poi_pdf=poi_pdf, poi_col_map=poi_col_map + ) + for s in selections: + res = s.build(cache) + if hasattr(res, "serialize") and callable(res.serialize): + res = res.serialize() + elif hasattr(res, "get_data") and callable(res.get_data): + res = res.get_data() + result[s._alias] = [res] + return pd.DataFrame(result) + + def solve(self, query, channels_df, selections, dtypes, poi_df=None) -> DataFrame: """ Solve the query by grouping channels and applying selections. @@ -943,6 +1142,10 @@ def solve(self, query, channels_df, selections, dtypes) -> DataFrame: the grouped-map UDF so that time-series values are converted from the source to the target unit on the fly. + When *poi_df* is present the solve forks to a ``cogroup`` of the channel data and + the POI frame by ``container_id`` — full-outer, so channel-only and POI-only + containers both still emit a row. A plain ``GROUPED_MAP`` is used otherwise. + Parameters ---------- query : QueryBuilder @@ -953,6 +1156,8 @@ def solve(self, query, channels_df, selections, dtypes) -> DataFrame: List of selection expressions to apply. dtypes : list List of data types for each selection. + poi_df : pyspark.sql.DataFrame, optional + Resolved POI frame from :meth:`filter_poi`, or ``None``. Returns ------- @@ -961,8 +1166,13 @@ def solve(self, query, channels_df, selections, dtypes) -> DataFrame: """ col_map = self.config.col_map q, joined_df, container_count = self._prepare_channels_join(query, channels_df) - schema = self._build_solve_output_schema(q, selections, dtypes) + + if poi_df is not None: + return self._apply_cogrouped_map( + query, joined_df, poi_df, selections, schema, col_map + ) + solve_udf = F.pandas_udf( partial(DefaultSolver._solve_udf, selections=selections, col_map=col_map), schema, @@ -970,6 +1180,44 @@ def solve(self, query, channels_df, selections, dtypes) -> DataFrame: ) return self._apply_grouped_map(joined_df, container_count, schema, solve_udf) + def _apply_cogrouped_map( + self, query, joined_df, poi_df, selections, schema, col_map + ) -> DataFrame: + """Cogroup channel data with the POI frame by container and run the POI UDF. + + The container set is the **union** of channel-bearing and POI-bearing containers + — derived from the cogroup itself rather than the channel-match frame — so a + POI-only query (no ``q.channel(...)``) still produces rows. This is the fix for + the silent-empty POI-only query in ``POI_PROPOSAL_REVIEW.md`` §3.1: a plain + ``GROUPED_MAP`` short-circuits when the channel-match frame is empty, but a + cogroup whose right side carries the POI containers does not. + """ + container_id_col = self.config.container_id_col + poi_col_map = self.config.poi_col_map + + # container_count is the union of both sides; if it is zero there is genuinely + # nothing to solve (no channels AND no POI), so an empty frame is correct. + container_count = ( + joined_df.select(container_id_col) + .union(poi_df.select(container_id_col)) + .distinct() + .count() + ) + if container_count == 0: + return self.spark.createDataFrame([], schema=schema) + + left = joined_df.repartition(container_count, container_id_col).groupBy(container_id_col) + right = poi_df.repartition(container_count, container_id_col).groupBy(container_id_col) + return left.cogroup(right).applyInPandas( + partial( + DefaultSolver._solve_udf_with_poi, + selections=selections, + col_map=col_map, + poi_col_map=poi_col_map, + ), + schema=schema, + ) + def _prepare_channels_join(self, query, channels_df) -> tuple[DataFrame, DataFrame, int]: """Shared prelude for :meth:`solve` and :meth:`solve_calculated_channels`. diff --git a/src/impulse_query_engine/analyze/query/solvers/in_memory_solver.py b/src/impulse_query_engine/analyze/query/solvers/in_memory_solver.py index e85d914..1452f0d 100644 --- a/src/impulse_query_engine/analyze/query/solvers/in_memory_solver.py +++ b/src/impulse_query_engine/analyze/query/solvers/in_memory_solver.py @@ -23,5 +23,5 @@ def filter_channel_tags(self, spark, db, container_df, selectors) -> DataFrame: def filter_channel_metrics(self, spark, db, channel_df, selectors) -> DataFrame: raise NotImplementedError - def solve(self, query, channels_df, selections, dtypes=None): + def solve(self, query, channels_df, selections, dtypes=None, poi_df=None): raise NotImplementedError diff --git a/src/impulse_query_engine/analyze/query/solvers/query_solver.py b/src/impulse_query_engine/analyze/query/solvers/query_solver.py index bfa3504..ad4ae10 100644 --- a/src/impulse_query_engine/analyze/query/solvers/query_solver.py +++ b/src/impulse_query_engine/analyze/query/solvers/query_solver.py @@ -392,8 +392,38 @@ def filter_candidates(self, query, channel_df) -> DataFrame: """ pass + def filter_poi(self, spark, db: MeasurementDB, container_df, poi_selectors): + """ + Optional POI stage: resolve POI rows for the selected containers. + + Non-abstract and returns ``None`` by default — mirroring the no-op + :meth:`filter_candidates` — so solvers that do not support POI (``BlobSolver``, + ``InMemorySolver``) need no changes and a POI-free query is unaffected. Solvers + that support POI (``DefaultSolver``) override this to produce a narrow, per- + container, deduped, time-base-resolved POI frame tagged with ``selector_id``. + + Parameters + ---------- + spark : SparkSession + Spark session used for query execution. + db : MeasurementDB + Measurement database for table access (reads ``db.poi(spark)``). + container_df : pyspark.sql.DataFrame + The stage-2 container frame — already container-filtered and + incremental-scoped. POI is restricted to these containers with a + ``left_semi`` join so no POI row is dropped or duplicated. + poi_selectors : list + The ``PoiSelector`` leaves collected from the selections. + + Returns + ------- + pyspark.sql.DataFrame or None + ``None`` when there are no POI selectors or the solver does not support POI. + """ + return None + @abc.abstractmethod - def solve(self, query, channels_df, selections, dtypes): + def solve(self, query, channels_df, selections, dtypes, poi_df=None): """ Stage 6: Solve query. @@ -407,6 +437,11 @@ def solve(self, query, channels_df, selections, dtypes): List of selection expressions to apply. dtypes : list List of data types for each selection. + poi_df : pyspark.sql.DataFrame, optional + The resolved POI frame from :meth:`filter_poi`, or ``None`` when the query + has no POI selectors. When present, the solver co-groups it with the channel + data by ``container_id`` (full-outer) so POI-only and channel-only containers + both still produce results. Returns ------- diff --git a/src/impulse_query_engine/analyze/query/solvers/series_cache.py b/src/impulse_query_engine/analyze/query/solvers/series_cache.py index 081d53b..f43dd23 100644 --- a/src/impulse_query_engine/analyze/query/solvers/series_cache.py +++ b/src/impulse_query_engine/analyze/query/solvers/series_cache.py @@ -23,6 +23,36 @@ def resolve(self, selection) -> pd.DataFrame: """ pass + # todo discuss why is this necessary + def resolve_poi(self, selection) -> pd.DataFrame: + """Return the POI rows matching *selection* for the current container. + + Deliberately **concrete, not** ``@abstractmethod`` (the PR #30 precedent): the + default returns an empty frame, so every existing cache — including any written + outside this repo — keeps working unchanged, and + :class:`~impulse_query_engine.analyze.query.solvers.empty_cache.EmptyTimeSeriesCache` + inherits it, which is what makes ``build(EmptyTimeSeriesCache())`` a valid, + data-free type probe for a :class:`PoiSelector`. + + The solver's per-container cache (``TimeSeriesCache`` in ``default_solver.py``) + overrides this to return the rows that survived the Spark-side ``filter_poi`` + stage, matched by ``selection.selector_id``. + + Parameters + ---------- + selection : Any + The asking POI selector (a ``PoiSelector``). Carries ``selector_id`` and, + when set, ``_attribute``. + + Returns + ------- + pandas.DataFrame + Empty by default. An overriding cache returns a frame with at least a + ``ts`` column (integer microseconds) and, when the selector names an + ``attribute``, that attribute column. + """ + return pd.DataFrame(columns=["ts"]) + @abstractmethod def load_blob(self, mid, cid, uses_alias: bool = False) -> SampleSeries: """ diff --git a/src/impulse_query_engine/analyze/query/solvers/solver_config.py b/src/impulse_query_engine/analyze/query/solvers/solver_config.py index 0ca7113..42862ab 100644 --- a/src/impulse_query_engine/analyze/query/solvers/solver_config.py +++ b/src/impulse_query_engine/analyze/query/solvers/solver_config.py @@ -85,6 +85,33 @@ class JoinKey(BaseModel): metrics_col: str +class PoiConfig(TableConfig): + """``TableConfig`` for the optional POI table. + + Follows :class:`ChannelMappingConfig` in extending :class:`TableConfig` with a few + typed fields. ``column_name_mapping`` handles container binding without a join: a + producer that keys POI rows on ``recording_session_id`` maps it to ``container_id``. + + Attributes + ---------- + ts_column : str + Physical column (after ``column_name_mapping``) holding the POI occurrence time. + **Must be a datetime / Spark ``timestamp`` column** — the solver reads it directly + as an absolute instant (via ``unix_micros``) with no unit or origin assumptions. + This is enforced by convention/documentation, not by the engine: a non-timestamp + column would resolve to nonsensical instants. See the POI section of the silver + layer schema docs. + dedup_order_by : list[str] + Deterministic tiebreak columns for the "one row per (kind, instant)" dedup. Two + POI rows can share an instant and collide in the gold MERGE (event ids are + ``crc32(cid::name::start::end)`` with ``start == end`` for a point), so the dedup + needs a *total* order. + """ + + ts_column: str = "timestamp" + dedup_order_by: list[str] = [] + + class ChannelMappingConfig(TableConfig): """``TableConfig`` plus an optional alias-resolution join-key spec. @@ -133,6 +160,10 @@ class SolverConfig(BaseModel): Column mappings and filters for the channel data table. unit_conversion : TableConfig Column mappings and filters for the unit conversion table. + poi : PoiConfig + Column mappings, filters, time base and dedup order for the optional POI + (point-of-interest) table. Inert unless a ``poi_table`` is configured on the + database. """ project_id: str | None = None @@ -144,6 +175,7 @@ class SolverConfig(BaseModel): channel_mapping: ChannelMappingConfig = ChannelMappingConfig() channels: TableConfig = TableConfig() unit_conversion: TableConfig = TableConfig() + poi: PoiConfig = PoiConfig() # ------------------------------------------------------------------ # Class methods @@ -369,6 +401,20 @@ def effective_alias_join_keys(self) -> list[tuple[str, str]]: ] return [(jk.mapping_col, jk.metrics_col) for jk in self.channel_mapping.join_keys] + @property + def poi_ts_col(self) -> str: + """Internal column name for the resolved POI instant (integer microseconds). + + This is the *normalized* column ``filter_poi`` produces from the physical + ``PoiConfig.ts_column`` after time-base resolution — not the physical column. + """ + return "ts" + + @property + def poi_selector_id_col(self) -> str: + """Internal column name tagging each POI row with the selector that matched it.""" + return "selector_id" + @property def col_map(self) -> dict[str, str]: """Short-key → internal-column-name mapping for UDFs and caches.""" @@ -380,3 +426,12 @@ def col_map(self) -> dict[str, str]: "val": self.value_col, "conv": self.conversion_factor_col, } + + @property + def poi_col_map(self) -> dict[str, str]: + """Short-key → internal-column-name mapping for the POI cache/UDF path.""" + return { + "cid": self.container_id_col, + "ts": self.poi_ts_col, + "sel": self.poi_selector_id_col, + } diff --git a/src/impulse_query_engine/measurement_db.py b/src/impulse_query_engine/measurement_db.py index c0ba27c..0e1cf03 100644 --- a/src/impulse_query_engine/measurement_db.py +++ b/src/impulse_query_engine/measurement_db.py @@ -16,6 +16,7 @@ def __init__( channels_uri=None, channel_mapping_table=None, unit_conversion_table=None, + poi_table=None, table_locations: str = "external_locations", ): self.container_tags_table = container_tags_table @@ -25,6 +26,7 @@ def __init__( self.channels_uri = channels_uri self.channel_mapping_table = channel_mapping_table self.unit_conversion_table = unit_conversion_table + self.poi_table = poi_table self.table_locations = table_locations self.debug_tables = None @@ -34,6 +36,7 @@ def for_unity_catalog( core_schema_name: str = "core", channel_mapping_table: str | None = None, unit_conversion_table: str | None = None, + poi_table: str | None = None, ): return MeasurementDBConfig( container_tags_table=f"{catalog_name}.{core_schema_name}.container_tags", @@ -43,6 +46,7 @@ def for_unity_catalog( channels_uri=f"{catalog_name}.{core_schema_name}.channels", channel_mapping_table=channel_mapping_table, unit_conversion_table=unit_conversion_table, + poi_table=poi_table, table_locations="unity_catalog", ) @@ -64,6 +68,7 @@ def for_debug(debug_tables): unit_conversion_table=( "unit_conversion" if "unit_conversion" in debug_tables else None ), + poi_table="poi" if "poi" in debug_tables else None, table_locations="debug", ) cfg.debug_tables = debug_tables @@ -113,6 +118,11 @@ def unit_conversion(self, spark) -> DataFrame: raise ValueError("unit_conversion_table is not configured") return self._read_table(spark, self.config.unit_conversion_table) + def poi(self, spark) -> DataFrame: + if self.config.poi_table is None: + raise ValueError("poi_table is not configured") + return self._read_table(spark, self.config.poi_table) + def channel_uri(self): return self.config.channels_uri diff --git a/src/impulse_query_engine/schema.py b/src/impulse_query_engine/schema.py index 8323182..1bcb7a1 100644 --- a/src/impulse_query_engine/schema.py +++ b/src/impulse_query_engine/schema.py @@ -70,3 +70,26 @@ T.StructField("value", T.DoubleType()), ] ) + +# Optional POI (point-of-interest) table. One row per occurrence (e.g. "AEB fired here"). +# A POI is a point in time — no duration, no sample rate, no value that persists between +# entries. Under Option D, POI is a pure occurrence log: it always evaluates to a +# PointsInTime, and a signal's value *at* an occurrence comes from sampling the measured +# channel (``q.channel(...).where(q.poi(...))``), not from a POI column. Non-spine columns +# (``poi_type``, ``duration``, ``event_type``, …) are row-filterable via +# ``q.poi(...).having(q.poi_metric("duration") > 5)``. This schema mirrors the external +# ``tech_rds_dev.poi.poi`` table and is used for fixtures. +POI_SCHEMA = T.StructType( + [ + T.StructField("container_id", T.LongType(), nullable=False), + # Occurrence time. This is PoiConfig.ts_column and MUST be a datetime / timestamp + # column: the solver reads it directly as an absolute instant (unix_micros), with + # no unit or origin assumptions. + T.StructField("timestamp", T.TimestampType()), + # Kind discriminator: q.poi(poi_type="aeb") filters on this. + T.StructField("poi_type", T.StringType()), + # Row-filterable spine columns (q.poi_metric(...) in a .having(...) predicate). + T.StructField("duration", T.DoubleType()), + T.StructField("event_type", T.StringType()), + ] +) diff --git a/src/impulse_reporting/config/config_parser.py b/src/impulse_reporting/config/config_parser.py index 6bff951..aa5979c 100644 --- a/src/impulse_reporting/config/config_parser.py +++ b/src/impulse_reporting/config/config_parser.py @@ -149,6 +149,7 @@ class Source(BaseModel): channels_uri: Annotated[str, AfterValidator(is_valid_table_name)] channel_mapping_table: Annotated[str, AfterValidator(is_valid_table_name)] | None = None unit_conversion_table: Annotated[str, AfterValidator(is_valid_table_name)] | None = None + poi_table: Annotated[str, AfterValidator(is_valid_table_name)] | None = None class UnitySink(BaseModel): diff --git a/tests/impulse_query_engine/unit/analyze/query/solvers/default_solver_poi_test.py b/tests/impulse_query_engine/unit/analyze/query/solvers/default_solver_poi_test.py new file mode 100644 index 0000000..0ab1190 --- /dev/null +++ b/tests/impulse_query_engine/unit/analyze/query/solvers/default_solver_poi_test.py @@ -0,0 +1,365 @@ +# pylint: disable=missing-function-docstring, redefined-outer-name +"""End-to-end POI tests for DefaultSolver against the real 30-column POI table shape. + +These run real Spark (the session-scoped ``spark`` fixture) over an in-memory POI table +modeled on ``tech_rds_dev.poi.poi`` — ``recording_session_id`` as the natural key (bound +to ``container_id`` via ``column_name_mapping``), ``poi_type`` as the kind discriminator, +and ``timestamp`` (a datetime column) as the occurrence time. + +Covered: +- ``filter_poi``: column-mapped container binding, predicate tagging by ``selector_id``, + ``left_semi`` container restriction, datetime ts_column → epoch-µs resolution, dedup. +- ``solve`` cogroup fork: channel+POI, channel-only, and POI-only containers all emit. +- The POI-only *query* (no ``q.channel(...)``) returns rows — the ``container_count`` + union fix. +- The ``attribute`` path → ``PointsInTimeSeries`` and its channel comparison. +""" + +import datetime + +import numpy as np +import pyspark.sql.types as T +import pytest +from pyspark.sql import SparkSession + +from impulse_query_engine.analyze.metadata.time_series_expression import ( + TimeSeriesExpression, +) +from impulse_query_engine.analyze.query.solvers.default_solver import DefaultSolver +from impulse_query_engine.analyze.query.solvers.solver_config import ( + PoiConfig, + SolverConfig, +) +from impulse_query_engine.measurement_db import MeasurementDB, MeasurementDBConfig +from tests.conftest import mock_workspace_client, spark # noqa: F401 + +_US_PER_SEC = 1_000_000 + +# The full external POI schema (tech_rds_dev.poi.poi): recording_session_id is the natural +# key, there is no container_id column. Only a representative subset of the 30 columns is +# populated with values in the fixtures; the rest are present so the schema is faithful. +POI_TABLE_SCHEMA = T.StructType( + [ + T.StructField("provider", T.StringType()), + T.StructField("runid", T.StringType()), + T.StructField("tcid", T.StringType()), + T.StructField("dt", T.StringType()), + T.StructField("recording_session_id", T.StringType()), + T.StructField("time", T.DoubleType()), + T.StructField("timestamp", T.TimestampType()), + T.StructField("poi_type", T.StringType()), + T.StructField("value", T.StringType()), + T.StructField("network", T.StringType()), + T.StructField("ecu", T.StringType()), + T.StructField("frame", T.StringType()), + T.StructField("processed", T.BooleanType()), + T.StructField("occurrences", T.IntegerType()), + T.StructField("duration", T.DoubleType()), + T.StructField("created_at", T.TimestampType()), + T.StructField("longitude", T.DoubleType()), + T.StructField("latitude", T.DoubleType()), + T.StructField("dtc_state", T.ShortType()), + T.StructField("life_situation_dj", T.DoubleType()), + T.StructField("odometer_dj", T.DoubleType()), + T.StructField("low_beam_state", T.DoubleType()), + T.StructField("high_beam_state", T.DoubleType()), + T.StructField("odometer", T.DoubleType()), + T.StructField("vehicle_wheel_speed", T.DoubleType()), + T.StructField("window_wiper_status", T.DoubleType()), + T.StructField("front_fog_light_status", T.DoubleType()), + T.StructField("aeb_state", T.DoubleType()), + T.StructField("event_type", T.StringType()), + T.StructField("poi_id", T.StringType()), + ] +) + +# Container metrics: two containers keyed by the same container_id the POI rows bind to +# (via column_name_mapping). start_dt is unused by POI now but kept as realistic metadata. +CONTAINER_METRICS_SCHEMA = T.StructType( + [ + T.StructField("container_id", T.StringType(), nullable=False), + T.StructField("start_dt", T.TimestampType()), + ] +) + +CHANNELS_SCHEMA = T.StructType( + [ + T.StructField("container_id", T.StringType(), nullable=False), + T.StructField("channel_id", T.IntegerType(), nullable=False), + T.StructField("tstart", T.LongType(), nullable=False), + T.StructField("tend", T.LongType(), nullable=False), + T.StructField("value", T.DoubleType()), + ] +) + +CHANNEL_METRICS_SCHEMA = T.StructType( + [ + T.StructField("container_id", T.StringType(), nullable=False), + T.StructField("channel_id", T.IntegerType(), nullable=False), + T.StructField("channel_name", T.StringType()), + ] +) + +# Two recording sessions used as container keys. +SID_A = "38004ebff4cfdefa3f458eb4ef25f62c8ebb936c259a4f204cef36073d8ad703" +SID_B = "d4f5ee6b81159f688ce096d4bcd0fd720db93e8117ae1b9d3a9180109c72b0d0" + + +_ANCHOR = datetime.datetime(2024, 4, 5, 14, 0, 0, tzinfo=datetime.timezone.utc) + + +def _ts(seconds: float) -> datetime.datetime: + """A UTC datetime `seconds` after a fixed epoch anchor (for absolute time base).""" + return _ANCHOR + datetime.timedelta(seconds=seconds) + + +def _abs_us(seconds: float) -> int: + """The absolute epoch-microsecond value the solver resolves for ``_ts(seconds)``. + + The absolute time base uses ``unix_micros``, so an instant is epoch µs — not + ``seconds * 1e6``. Assertions compare against this. + """ + return int(round(_ts(seconds).timestamp() * _US_PER_SEC)) + + +def _poi_row(sid, poi_type, ts_seconds, *, network="INFO", frame="943", wheel_speed=0.0, + duration=0.0, rel_time=0.0): + """One faithful 30-column POI row; only the fields the tests read are meaningful.""" + return ( + "hmt", "03867", "9559", "2024-04-05", sid, + rel_time, _ts(ts_seconds), poi_type, None, network, None, frame, + False, 1, duration, _ts(0), 0.911, 48.6, 1, 5771344.0, 0.0, None, None, + 7800.0, wheel_speed, None, None, 3.0, "computed", None, + ) + + +@pytest.fixture +def poi_db(spark: SparkSession, mock_workspace_client) -> MeasurementDB: + """A MeasurementDB with a POI table, container_metrics, channels, channel_metrics. + + POI rows are keyed by ``recording_session_id`` (mapped to ``container_id``). Container + A has two AEB occurrences (one duplicated at the same instant to exercise dedup) plus + a channel; container B has one AEB occurrence and NO channel (POI-only container). + """ + poi_rows = [ + # container A: two distinct AEB instants, plus a duplicate of the first instant + # differing only in network/frame — must dedup to one row per (kind, instant). + # The 200s occurrence has duration=5.0 so a having(duration > 4) filter keeps it. + _poi_row(SID_A, "aeb", 100.0, network="INFO", frame="943", duration=0.0), + _poi_row(SID_A, "aeb", 100.0, network="CHASSIS", frame="722", duration=0.0), + _poi_row(SID_A, "aeb", 200.0, network="INFO", frame="943", duration=5.0), + # a different poi_type in A — must not match poi_type="aeb" + _poi_row(SID_A, "ldw", 150.0), + # container B: one AEB, no channel data at all (POI-only container in the cogroup) + _poi_row(SID_B, "aeb", 300.0, network="CHASSIS", frame="694", duration=0.0), + ] + poi = spark.createDataFrame(poi_rows, schema=POI_TABLE_SCHEMA) + + container_metrics = spark.createDataFrame( + [(SID_A, _ts(0)), (SID_B, _ts(0))], schema=CONTAINER_METRICS_SCHEMA + ) + # Only container A has a channel; B is POI-only. Intervals are epoch-aligned to the + # POI instants so channel.where(poi) can sample at them: speed is 20 up to 150s then + # 80, so at the 100s AEB speed=20 and at the 200s AEB speed=80. + channels = spark.createDataFrame( + [ + (SID_A, 10, _abs_us(0), _abs_us(150), 20.0), + (SID_A, 10, _abs_us(150), _abs_us(400), 80.0), + ], + schema=CHANNELS_SCHEMA, + ) + channel_metrics = spark.createDataFrame( + [(SID_A, 10, "Vehicle Speed Sensor")], schema=CHANNEL_METRICS_SCHEMA + ) + + tables = { + "container_metrics": container_metrics, + "channels": channels, + "channel_metrics": channel_metrics, + "poi": poi, + } + cfg = MeasurementDBConfig.for_debug(tables) + return MeasurementDB(cfg, ws=mock_workspace_client) + + +def _poi_solver(spark) -> DefaultSolver: + """A DefaultSolver whose POI config binds recording_session_id → container_id. + + ``ts_column`` is the datetime ``timestamp`` column, resolved to integer µs. + """ + return DefaultSolver( + spark, + config=SolverConfig( + poi=PoiConfig( + column_name_mapping={"recording_session_id": "container_id"}, + ts_column="timestamp", + dedup_order_by=["network", "frame"], + ) + ), + ) + + +# --------------------------------------------------------------------------- +# filter_poi +# --------------------------------------------------------------------------- + + +class TestFilterPoi: + def test_binds_container_and_tags_selector_and_dedups( + self, spark: SparkSession, poi_db: MeasurementDB + ): + solver = _poi_solver(spark) + query = poi_db.query + aeb = query.poi(poi_type="aeb") + query.select(aeb) + + tags_df = solver.filter_container_tags(spark, query) + container_df = solver.filter_container_metrics(spark, query, tags_df) + poi_selectors = TimeSeriesExpression.collect_poi_selectors(query.selections) + result = solver.filter_poi(spark, poi_db, container_df, poi_selectors) + + rows = result.collect() + # container A: instants 100 and 200 (the duplicate at 100 deduped away); + # container B: instant 300. The ldw row is filtered out. => 3 rows. + assert len(rows) == 3 + assert {"container_id", "ts", "selector_id"}.issubset(set(result.columns)) + # every row carries the asking selector's id + assert {r.selector_id for r in rows} == {aeb.selector_id} + # instants resolved to integer microseconds (absolute) + by_container = {} + for r in rows: + by_container.setdefault(r.container_id, []).append(r.ts) + assert sorted(by_container[SID_A]) == [_abs_us(100), _abs_us(200)] + assert sorted(by_container[SID_B]) == [_abs_us(300)] + + def test_ts_column_datetime_resolves_to_epoch_micros( + self, spark: SparkSession, poi_db: MeasurementDB + ): + # The datetime ts_column is read directly via unix_micros: instants equal the + # epoch-µs of the timestamp column, with no unit/origin math. + solver = _poi_solver(spark) + query = poi_db.query + query.select(query.poi(poi_type="aeb")) + tags_df = solver.filter_container_tags(spark, query) + container_df = solver.filter_container_metrics(spark, query, tags_df) + poi_selectors = TimeSeriesExpression.collect_poi_selectors(query.selections) + rows = solver.filter_poi(spark, poi_db, container_df, poi_selectors).collect() + assert {r.ts for r in rows} == {_abs_us(100), _abs_us(200), _abs_us(300)} + + def test_container_filter_restricts_poi_rows( + self, spark: SparkSession, poi_db: MeasurementDB + ): + # Restrict the container frame to A only; B's POI row must not leak through. + solver = _poi_solver(spark) + query = poi_db.query + query.select(query.poi(poi_type="aeb")) + container_df = solver.filter_container_metrics( + spark, query, solver.filter_container_tags(spark, query) + ) + container_a = container_df.where(container_df.container_id == SID_A) + poi_selectors = TimeSeriesExpression.collect_poi_selectors(query.selections) + rows = solver.filter_poi(spark, poi_db, container_a, poi_selectors).collect() + assert {r.container_id for r in rows} == {SID_A} + + +# --------------------------------------------------------------------------- +# solve — the cogroup fork +# --------------------------------------------------------------------------- + + +class TestSolvePoi: + def _solve(self, spark, db, *selections): + solver = _poi_solver(spark) + query = db.query + query.select(*selections) + result = query.solve(spark, solver=solver) + return {r.container_id: r for r in result.collect()} + + def test_poi_only_query_returns_rows_for_all_containers( + self, spark: SparkSession, poi_db: MeasurementDB + ): + """A POI-only query (no q.channel) must return rows — the container_count fix.""" + query = poi_db.query + aeb = query.poi(poi_type="aeb").alias("aeb") + rows = self._solve(spark, poi_db, aeb) + # Both containers appear, including B which has NO channel data at all. + assert set(rows.keys()) == {SID_A, SID_B} + # A's points: 100s and 200s as epoch µs; B's: 300s as epoch µs. + assert sorted(rows[SID_A]["aeb"]) == [_abs_us(100), _abs_us(200)] + assert sorted(rows[SID_B]["aeb"]) == [_abs_us(300)] + + def test_channel_and_poi_together(self, spark: SparkSession, poi_db: MeasurementDB): + """Channel-only container B still solves its channel expr; A gets both.""" + query = poi_db.query + aeb = query.poi(poi_type="aeb").alias("aeb") + ch = query.channel(channel_name="Vehicle Speed Sensor").mean().alias("speed_mean") + rows = self._solve(spark, poi_db, aeb, ch) + # A has both channel and POI. + assert sorted(rows[SID_A]["aeb"]) == [_abs_us(100), _abs_us(200)] + assert rows[SID_A]["speed_mean"] is not None + # B is POI-only: it still appears with its POI, and a null channel mean. + assert sorted(rows[SID_B]["aeb"]) == [_abs_us(300)] + + def test_having_row_filter_on_poi_metric( + self, spark: SparkSession, poi_db: MeasurementDB + ): + """having(q.poi_metric(...) > x) filters occurrences Spark-side before solving.""" + query = poi_db.query + # Only A's 200s AEB has duration=5.0 > 4; the 100s occurrences (duration 0) drop. + long_aeb = ( + query.poi(poi_type="aeb") + .having(query.poi_metric("duration") > 4.0) + .alias("long_aeb") + ) + rows = self._solve(spark, poi_db, long_aeb) + assert sorted(rows[SID_A]["long_aeb"]) == [_abs_us(200)] + # B's AEB has duration 0, so B contributes no instants (but may still appear empty). + assert rows.get(SID_B) is None or list(rows[SID_B]["long_aeb"]) == [] + + def test_value_at_occurrence_via_channel_where( + self, spark: SparkSession, poi_db: MeasurementDB + ): + """Option D: a signal's value AT each occurrence = channel.where(poi), not a POI column.""" + query = poi_db.query + # Vehicle speed sampled at each AEB instant in A: 100s→20 (speed<150s), 200s→80. + speed_at_aeb = ( + query.channel(channel_name="Vehicle Speed Sensor") + .where(query.poi(poi_type="aeb")) + .alias("speed_at_aeb") + ) + rows = self._solve(spark, poi_db, speed_at_aeb) + # PointsInTimeSeries serialized as [ts, value] pairs. + pairs = {int(ts): val for ts, val in rows[SID_A]["speed_at_aeb"]} + assert pairs[_abs_us(100)] == pytest.approx(20.0) + assert pairs[_abs_us(200)] == pytest.approx(80.0) + + def test_dedup_collapses_same_instant_to_one_point( + self, spark: SparkSession, poi_db: MeasurementDB + ): + """Two AEB rows at the same instant (differing network/frame) → one instant.""" + query = poi_db.query + aeb = query.poi(poi_type="aeb").alias("aeb") + rows = self._solve(spark, poi_db, aeb) + # A's 100s instant had two colliding rows; dedup leaves exactly one → 2 instants. + assert sorted(rows[SID_A]["aeb"]) == [_abs_us(100), _abs_us(200)] + assert len(rows[SID_A]["aeb"]) == 2 + + +# --------------------------------------------------------------------------- +# backward compatibility: a POI table present but unused +# --------------------------------------------------------------------------- + + +def test_poi_table_present_but_unused_is_inert( + spark: SparkSession, poi_db: MeasurementDB +): + """A query with no POI selector must behave exactly as if no POI table existed.""" + solver = _poi_solver(spark) + query = poi_db.query + ch = query.channel(channel_name="Vehicle Speed Sensor").mean().alias("m") + query.select(ch) + result = query.solve(spark, solver=solver) + # No cogroup path taken (poi_df is None); only container A has this channel. + rows = {r.container_id: r.m for r in result.collect()} + assert SID_A in rows + assert rows[SID_A] is not None diff --git a/tests/impulse_query_engine/unit/model/expressions/poi_selector_test.py b/tests/impulse_query_engine/unit/model/expressions/poi_selector_test.py new file mode 100644 index 0000000..05dd198 --- /dev/null +++ b/tests/impulse_query_engine/unit/model/expressions/poi_selector_test.py @@ -0,0 +1,194 @@ +"""PoiSelector unit tests — the data-free type probe and the TSAL contract. + +These run with no Spark and no data: everything hinges on ``build(EmptyTimeSeriesCache())`` +returning an empty :class:`PointsInTime`, which is what lets events validate a POI +expression at construction time. + +Under Option D, POI is a pure occurrence log — a ``PoiSelector`` always evaluates to +``PointsInTime`` (no attribute / ``PointsInTimeSeries`` path). Row filtering uses the +dedicated POI predicate DSL (``q.poi_metric(...)`` in a ``.having(...)`` predicate), not +``TagExpression``. +""" + +# pylint: disable=missing-function-docstring, redefined-outer-name + +import numpy as np +import pandas as pd +import pytest + +from impulse_query_engine.analyze.metadata.poi_expression import ( + PoiMetricSelector, + PoiPredicate, + poi_kind_predicate, +) +from impulse_query_engine.analyze.metadata.poi_selector import PoiSelector, poi_expr +from impulse_query_engine.analyze.metadata.time_series_expression import TimeSeriesExpression +from impulse_query_engine.analyze.query.solvers.empty_cache import EmptyTimeSeriesCache +from impulse_query_engine.analyze.query.solvers.series_cache import SeriesCache +from impulse_query_engine.model.series.points_in_time import PointsInTime + + +class _FakePoiCache(SeriesCache): + """A minimal cache that serves a fixed POI frame, ignoring the selector.""" + + def __init__(self, rows: pd.DataFrame): + self._rows = rows + + def resolve(self, selection): + return [] + + def resolve_poi(self, selection): + return self._rows + + def load_blob(self, mid, cid, uses_alias: bool = False): + return None + + +# --- the type probe: build on an empty cache yields an empty PointsInTime -------------- + + +def test_build_on_empty_cache_is_points_in_time(): + poi = PoiSelector(poi_expr(poi_type="aeb")) + result = poi.build(EmptyTimeSeriesCache()) + assert isinstance(result, PointsInTime) + assert len(result) == 0 + + +def test_evaluation_type_is_always_points_in_time(): + assert PoiSelector(poi_expr(poi_type="aeb")).evaluation_type() is PointsInTime + # ...even with a having predicate attached + filtered = PoiSelector(poi_expr(poi_type="aeb")).having(PoiMetricSelector("duration") > 5) + assert filtered.evaluation_type() is PointsInTime + + +# --- require_evaluation_type gating (the PointsInTimeEvent / consumer path) ------------- + + +def test_require_evaluation_type_accepts_points(): + PoiSelector(poi_expr(poi_type="aeb")).require_evaluation_type( + PointsInTime, owner="PointsInTimeEvent" + ) # must not raise + + +# --- the sibling contract: stays out of the channel pipeline --------------------------- + + +def test_get_selectors_empty_keeps_poi_out_of_channel_pipeline(): + poi = PoiSelector(poi_expr(poi_type="aeb")) + assert poi.get_selectors() == [] + assert TimeSeriesExpression.collect_selectors([poi]) == [] + + +def test_get_poi_selectors_collects_self(): + poi = PoiSelector(poi_expr(poi_type="aeb")) + assert poi.get_poi_selectors() == [poi] + assert TimeSeriesExpression.collect_poi_selectors([poi]) == [poi] + + +def test_poi_selectors_collected_through_operators(): + a = PoiSelector(poi_expr(poi_type="aeb")) + b = PoiSelector(poi_expr(poi_type="ldw")) + composed = a & b + assert len(TimeSeriesExpression.collect_poi_selectors([composed])) == 2 + + +def test_collect_poi_selectors_dedups_by_id(): + a = PoiSelector(poi_expr(poi_type="aeb")) + a2 = PoiSelector(poi_expr(poi_type="aeb")) # same predicate → same selector_id + assert len(TimeSeriesExpression.collect_poi_selectors([a, a2])) == 1 + + +# --- having(): fluent, immutable, definition-hash-distinct ----------------------------- + + +def test_having_returns_new_selector_and_is_immutable(): + base = PoiSelector(poi_expr(poi_type="aeb")) + filtered = base.having(PoiMetricSelector("duration") > 5) + assert filtered is not base + # base is unchanged + assert base.selector_id == PoiSelector(poi_expr(poi_type="aeb")).selector_id + + +def test_having_changes_definition_hash(): + base = PoiSelector(poi_expr(poi_type="aeb")) + filtered = base.having(PoiMetricSelector("duration") > 5) + assert filtered.selector_id != base.selector_id + + +def test_having_chains(): + base = PoiSelector(poi_expr(poi_type="aeb")) + two = base.having(PoiMetricSelector("duration") > 5).having(PoiMetricSelector("occurrences") == 1) + # both predicates recorded, distinct from single-having + one = base.having(PoiMetricSelector("duration") > 5) + assert two.selector_id != one.selector_id + assert "occurrences" in two.required_tags() + assert "duration" in two.required_tags() + + +def test_required_tags_exposes_referenced_poi_columns(): + poi = PoiSelector(poi_expr(poi_type="aeb")).having(PoiMetricSelector("duration") > 5) + assert poi.required_tags() == {"poi_type", "duration"} + + +# --- identity / definition hashing ----------------------------------------------------- + + +def test_selector_id_stable_for_same_predicate(): + a = PoiSelector(poi_expr(poi_type="aeb")) + b = PoiSelector(poi_expr(poi_type="aeb")) + assert a.selector_id == b.selector_id + assert isinstance(a.selector_id, int) + + +def test_str_includes_predicate(): + poi = PoiSelector(poi_expr(poi_type="aeb")).having(PoiMetricSelector("duration") > 5) + s = str(poi) + assert "poi_type" in s and "duration" in s + + +def test_dtype_is_points_in_time(): + assert PoiSelector(poi_expr(poi_type="aeb")).dtype() == PointsInTime.empty().dtype() + + +# --- build against real POI rows ------------------------------------------------------- + + +def test_build_dedups_and_sorts_instants(): + rows = pd.DataFrame({"ts": [30, 10, 10, 20]}) + result = PoiSelector(poi_expr(poi_type="aeb")).build(_FakePoiCache(rows)) + assert isinstance(result, PointsInTime) + np.testing.assert_array_equal(result.tstarts, np.array([10.0, 20.0, 30.0])) + + +def test_build_empty_frame_is_empty_points(): + result = PoiSelector(poi_expr(poi_type="aeb")).build(_FakePoiCache(pd.DataFrame({"ts": []}))) + assert isinstance(result, PointsInTime) + assert len(result) == 0 + + +# --- the predicate DSL (poi_expression) ------------------------------------------------ + + +def test_poi_kind_predicate_requires_a_filter(): + with pytest.raises(ValueError, match="at least one kind filter"): + poi_kind_predicate() + + +def test_poi_kind_predicate_ands_equalities_and_lists_columns(): + pred = poi_kind_predicate(poi_type="aeb", event_type="computed") + assert isinstance(pred, PoiPredicate) + assert pred.required_columns() == {"poi_type", "event_type"} + + +def test_poi_metric_predicate_no_cast_needed(): + # a numeric comparison builds a PoiPredicate referencing the column; no cast_type given + pred = PoiMetricSelector("duration") > 5 + assert isinstance(pred, PoiPredicate) + assert pred.required_columns() == {"duration"} + + +def test_predicate_str_is_stable(): + p1 = str(PoiMetricSelector("duration") > 5) + p2 = str(PoiMetricSelector("duration") > 5) + assert p1 == p2 + assert "duration" in p1 and ">" in p1