Feature/poi draft

[adefabian]

Add POI (point-of-interest) support to TSAL

Summary

Adds a first-class POI concept to Impulse: an externally-produced occurrence log
(e.g. "AEB fired here", one row per occurrence) becomes directly usable in TSAL
expressions, ad-hoc queries, and full reports.

The design lands POI on the core-model types the engine already has, so it
adds no new event class, no new aggregation class, and no gold-schema change. A POI
selection always evaluates to a PointsInTime; a signal's value at an occurrence comes
from sampling the measured channel (channel.where(poi)), not from a POI column.

The feature is inert until source.poi_table is configured — a query or report with no
POI selector behaves exactly as before.

---

Example poi rows

A representative slice (the real table has ~30 columns; only the ones that matter for TSAL
are shown). container_id is the engine join key — the same key used by every other silver
table (if the producer keys rows under a different name, it is mapped to container_id via
column_name_mapping). timestamp is the datetime occurrence time; poi_type is the
kind filter; network / frame / duration are row-filterable spine columns.

container_id	timestamp	poi_type	network	frame	duration	event_type
1001	2024-04-05 14:25:50.426+00	aeb	INFO	943	0.0	computed
1001	2024-04-05 14:27:25.425+00	aeb	INFO	943	0.0	computed
1002	2024-04-05 14:25:49.299+00	aeb	CHASSIS	722	0.900	computed
1002	2024-04-05 14:27:24.389+00	aeb	CHASSIS	694	0.760	computed
1001	2024-04-05 14:27:25.425+00	aeb	CHASSIS	722	0.0	computed
1001	2024-04-05 14:26:10.000+00	ldw	INFO	943	0.0	computed

Reading this table through TSAL:

• q.poi(poi_type="aeb") → every aeb row's timestamp as a PointsInTime; the ldw
  row is excluded.
• Rows 2 and 5 share the same instant (14:27:25.425) on container 1001,
  differing only in network / frame. Dedup collapses them to one instant;
  PoiConfig.dedup_order_by=["network","frame"] deterministically keeps the CHASSIS/722
  row.
• q.poi(poi_type="aeb").having(q.poi_metric("duration") > 0.5) → only the two container
  1002 rows (durations 0.900 and 0.760) survive; the 0.0-duration rows drop.
• To get vehicle speed at each AEB instant, sample the channel — not a POI column:
  q.channel(channel_name="Vehicle Speed Sensor").where(q.poi(poi_type="aeb")).

---

What a user writes

# 1. select occurrences as instants (PointsInTime)
aeb = q.poi(poi_type="aeb")

# 2. row-filter occurrences with a dedicated, typed predicate (no cast needed)
long_aeb = q.poi(poi_type="aeb").having(q.poi_metric("duration") > 5)

# 3. a signal's value AT each occurrence = sample the measured channel
speed_at_aeb = q.channel(channel_name="Vehicle Speed Sensor").where(q.poi(poi_type="aeb"))

# 4. use POI unchanged in the reporting layer
report.add_event(PointsInTimeEvent(name="aeb_activation", expr=aeb))
page.add_aggregation(PointValueAggregator(
    name="speed_at_aeb",
    input_expressions=[q.channel(channel_name="Vehicle Speed Sensor")],
    channel_names=["Vehicle Speed Sensor"],
    event=PointsInTimeEvent(name="aeb_activation", expr=aeb),
))

---

Updated data model

Silver layer — new optional poi table

POI joins the existing silver tables as a new, optional source. Nothing else in the
silver model changes.

Table	Required?	Grain	Purpose
container_metrics	✅	one row / recording	per-recording metadata + time bounds
channel_metrics	✅	one row / channel	per-channel stats; carries channel_name (wide model)
channels	✅	RLE intervals	the time-series data [tstart, tend) → value
container_tags	optional	EAV	key/value metadata per recording
channel_tags	optional	EAV	key/value metadata per channel
poi	optional (new)	one row / occurrence	points of interest, e.g. AEB activations

POI is a pure occurrence log. It carries an occurrence spine (identity, timestamp,
kind, and row-filterable columns like duration). It is not modeled as carrying
snapshot signal values: those are redundant with channels, so a value at an occurrence
is obtained by sampling the measured channel. This keeps a single source of truth and
means the POI table can grow columns without changing how values are read.

Two things are configured, not inferred:

• Occurrence timestamp (PoiConfig.ts_column, default timestamp) — must be a
  datetime / Spark timestamp column. The solver reads it directly as an absolute
  instant via unix_micros, with no unit or per-container-origin assumptions. This is
  enforced by documentation, not the engine: pointing it at an epoch-integer or a
  relative-seconds column would silently resolve to nonsensical instants (the ms-vs-µs,
  "8-years-off" hazard). Producers that store the time otherwise cast it to a timestamp
  in the view/table they expose as poi.
• Dedup order (PoiConfig.dedup_order_by) — two POI rows can share an instant; a
  deterministic total order picks the surviving row so the gold event_instance_id
  (crc32(cid::name::start::end), start == end for a point) stays unique.

Container binding is a rename, not a join. POI rows are keyed by the producer's own key
(e.g. recording_session_id); PoiConfig.column_name_mapping maps it to container_id.
POI is then restricted to the query's containers with a left_semi join (filters without
widening or duplicating rows).

Gold layer — unchanged

Nothing new lands in gold. POI reaches the star schema through the existing tables:

Table	POI rows
event_instance_fact	one per occurrence (start_ts == end_ts) via PointsInTimeEvent
event_dimension	one per event; POI metadata in the existing attributes map
stats_aggregator_fact	one per (channel, occurrence) via PointValueAggregator

Querying needs nothing POI-specific: join event_instance_fact to event_dimension on
event_id, filter event_type = 'POINTS_IN_TIME_EVENT'.

Config

• Source.poi_table — a flat catalog.schema.table string (nested models would break
  the MeasurementDBConfig(**dict(config.source)) splat, so report.py needs no change).
• SolverConfig.poi: PoiConfig — column mapping, filters, datetime ts_column, dedup order.
• MeasurementDBConfig.poi_table + MeasurementDB.poi(spark) accessor; wired into
  for_unity_catalog and for_debug.

---

Capabilities of the POI filter

q.poi(**kwargs) builds a PoiSelector (always → PointsInTime). Row filtering uses a
dedicated, typed POI predicate DSL (q.poi_metric), not the EAV TagExpression.

Capability	Syntax	Notes
Kind filter (equality)	q.poi(poi_type="aeb")	kwargs are ANDed
Multi-column equality	q.poi(poi_type="aeb", network="INFO")	poi_type=='aeb' AND network=='INFO'
Row predicate (>, <, ==, …)	q.poi(poi_type="aeb").having(q.poi_metric("duration") > 5)	no cast needed — POI columns are natively typed
Chained predicates (AND)	.having(a).having(b)	immutable; returns a new selector each time
OR within a filter	.having((q.poi_metric("network")=="INFO") | (q.poi_metric("network")=="CHASSIS"))	| / & on predicates
Set algebra on instants	q.poi(poi_type="aeb") & q.poi(poi_type="ldw")	PointsInTime operators
Temporal shaping	q.poi(poi_type="aeb").expand(w)	→ Intervals for BasicEvent / histogram scope
Value at each occurrence	q.channel(name).where(q.poi(...))	→ PointsInTimeSeries, typed by the channel
Container scoping (orthogonal)	q.where(q.tag("brand")=="Seat")	POI left_semi-restricted to survivors

Why a dedicated predicate (not TagExpression)

The POI table is wide and natively typed, so — unlike the EAV tag tables whose value
column is always a string — POI needs no caller-supplied cast. q.poi_metric("duration") > 5
is the POI analogue of q.metric("duration_ms") > 5 (wide typed column), and it removes the
q.tag misnomer and the string-cast footgun for POI. This is consistent with the rest of
Impulse: TagExpression, MetricExpression, and the new PoiPredicate all support &/|;
kwargs on q.channel / q.poi both fold to AND.

Row filtering vs. sampling — having, not where

Row filtering uses .having(...), deliberately not .where(...): on any
TimeSeriesExpression, where already means "sample this series at these points"
(channel.where(poi)). having reads as "restrict the source's rows" and returns a new
immutable selector.

The solve path

• POI leaves collect via a parallel get_poi_selectors() (POI returns get_selectors() == []
  so it never enters the channel-match pipeline).
• DefaultSolver.filter_poi pushes the predicate down in Spark, tags each surviving row with
  its selector_id, resolves the datetime ts_column to epoch µs, and dedups.
• solve co-groups channel data with the POI frame per container (full-outer), so
  channel-only, POI-only, and mixed containers all still produce rows — including a
  POI-only query (no q.channel(...)), which the naive path would silently return empty.

---

Files changed

New (production)

File	LOC	Contents
analyze/metadata/poi_expression.py	~190	PoiMetricSelector, PoiPredicate, poi_kind_predicate — the dedicated POI predicate DSL
analyze/metadata/poi_selector.py	~217	PoiSelector (→ PointsInTime), having(), the type probe

Modified (production, additive — no signature breaks)

File	Δ	What
analyze/query/solvers/default_solver.py	+281	filter_poi (+ time-base/dedup helpers), TimeSeriesCache.resolve_poi, cogroup solve fork (_solve_udf_with_poi, _apply_cogrouped_map)
analyze/query/solvers/solver_config.py	+55	PoiConfig (datetime ts_column, dedup_order_by), poi_ts_col/poi_selector_id_col/poi_col_map
analyze/query/query_builder.py	+83	poi() + poi_metric() accessors, POI stage in _run_filter_pipeline, poi_df passed to solve
analyze/metadata/time_series_expression.py	+64	concrete get_poi_selectors() + collect_poi_selectors(), walked by TimeSeriesOp / alias
analyze/query/solvers/series_cache.py	+30	concrete (non-abstract) resolve_poi() default so all caches + the type probe keep working
analyze/query/solvers/query_solver.py	+37	non-abstract filter_poi() no-op; poi_df kwarg on solve
aggregations/{histogram,histogram2d,stats_aggregator,point_value_aggregator}.py, events/sequence_of_events_expression.py	+44	get_poi_selectors() walkers so POI reaches the solver through an aggregation/event
measurement_db.py	+10	poi_table field, poi() accessor, factory wiring
schema.py	+24	POI_SCHEMA for fixtures
analyze/query/solvers/{blob,in_memory}_solver.py	+7	poi_df kwarg for interface compatibility
config/config_parser.py	+1	Source.poi_table

Not touched: the core model (points_in_time.py, intervals.py, sample_series.py),
all of impulse_reporting/events/ and aggregations/ class definitions, every gold schema,
and report.py (the flat poi_table is picked up by the existing config splat).

Tests (all green)

File	Cases	Covers
unit/model/expressions/poi_selector_test.py	20	type probe, having immutability/chaining, predicate DSL, dedup, definition hashing (no Spark)
unit/analyze/query/solvers/default_solver_poi_test.py	9	filter_poi (container bind, tagging, left_semi, datetime ts_column → epoch µs, dedup), cogroup fork, POI-only query, having row filter, channel.where(poi) values, inert-when-unused
integration/poi_report_test.py	1	full Report: PointsInTimeEvent → event_instance_fact (points), POI-scoped PointValueAggregator → facts

Docs / demos

• demos/poi_and_filters_walkthrough.ipynb — self-contained (local Spark + Delta), inline-mocks
  every source incl. the 30-column POI table, walks every filter and TSAL object, builds an
  example report; executes top to bottom.
• POI_PROPOSAL_REVIEW.md — review of the original design proposal + rationale for the
  decisions taken here.

---

Follow-ups (intentionally out of scope)

• POI-with-duration → Intervals. Deferred; would add an explicit .as_intervals()
  structural switch and a PoiConfig overlap policy. No core-model or gold change needed
  when added.
• One column-predicate DSL. There are now three (tag, metric, poi_metric); unifying
  them into a shared ColumnExpression is a separate cleanup, deliberately not coupled to
  this PR.
• q.poi(predicate) positional for pure top-level OR (parity with q.where(predicate));
  today a pure OR is built via .having((a) | (b)) or by constructing PoiSelector(pred).

This pull request and its description were written by Isaac.

Test Plan

• Unit tests added/updated
• Manual testing completed
• Documentation updated (if applicable)

Checklist

• Code follows project style guidelines
• Self-review completed
• No new linter warnings introduced