Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -238,7 +238,10 @@ Add an event to the report.

**Raises**:

- `ValueError`: If the event is a ContainerEvent and a ContainerEvent already exists in the report.
- `ValueError`: If the event is a ContainerEvent and a ContainerEvent already exists in the report,
or if an event with the same name, ignoring case, has already been added (event
names identify events in the fact and dimension tables and name the event's
column in the solved data, which Spark resolves case-insensitively).

**Returns**:

Expand Down
5 changes: 3 additions & 2 deletions docs/impulse/docs/references/report/event.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@ Events define time windows within measurement data that scope downstream aggrega
time-series recordings into meaningful intervals -- for example, "engine RPM between 2000 and 5000".

Every event used by an aggregation **must** be registered with the report via `add_event()` before `determine_report()`
is called.
is called. Event names must be unique within a report (ignoring case), and an aggregation must be scoped to the
registered event object itself, not to another object with the same name.

```python
my_report.add_event(my_event)
Expand Down Expand Up @@ -296,7 +297,7 @@ Stores event definitions (one row per event per report).

| Column | Type | Description |
|---------------------|---------------------|-----------------------------------------------------------------------------|
| `event_id` | `int` | Unique event identifier (CRC32 hash of name + expression). |
| `event_id` | `int` | Unique event identifier (CRC32 hash of the event name). |
| `report_id` | `int` | Report identifier. |
| `event_type` | `str` | `"BASIC_EVENT"`, `"CONTAINER_EVENT"`, `"SEQUENCE_OF_EVENTS"`, `"POINTS_IN_TIME_EVENT"`, or `"TIME_WINDOW_EVENT"`. |
| `event_name` | `str` | Event name. |
Expand Down
5 changes: 3 additions & 2 deletions docs/impulse/docs/references/report/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ Either `config` or `config_path` must be provided.
| `get_solver()` | Returns the active `QuerySolver`. | -- |
| `get_sink_config()` | Returns the active `SinkConfig` (e.g. `UnitySinkConfig` with `catalog_name`, `schema_name`, `table_prefix`). | -- |
| `add_page(page)` | Adds a `Page` to the report. | `page`: `Page` instance. |
| `add_event(event)` | Registers an `Event` with the report. All events used by aggregations must be registered. | `event`: `Event` instance. |
| `add_event(event)` | Registers an `Event` with the report. All events used by aggregations must be registered. Event names must be unique (ignoring case). | `event`: `Event` instance. |
| `add_calculated_channel(channel)` | Registers a [`CalculatedChannel`](./channel.md) with the report. | `channel`: `CalculatedChannel` instance. |
| `get_calculated_channels()` | Returns the list of registered calculated channels. | -- |
| `determine_report(is_incremental)` | Computes all events, aggregations, calculated channels, and container dimensions. Results are stored on the report object. | `is_incremental`: `bool` or `None`. Mode hint; overridden by `config.incremental` when present. See [Incremental processing](#incremental-processing). |
Expand All @@ -62,7 +62,8 @@ my_report.persist_results()
```

`determine_report()` validates that every event referenced by an aggregation has been registered with `add_event()`
before computation begins.
before computation begins. The aggregation must reference the registered event object itself; a re-created event with
the same name is rejected.

### Sinkless mode

Expand Down
4 changes: 3 additions & 1 deletion skills/impulse-events/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,9 @@ An event defines time windows within a recording that scope downstream aggregati
expression contains a UDF that declares `container_tags` / `container_metrics`, those values are
injected as usual — the requirement propagates through the event.

**Every event used by an aggregation must be registered with the report before computing:**
**Every event used by an aggregation must be registered with the report before computing.** Event
names must be unique within a report (ignoring case), and the aggregation must be scoped to that
registered object (not a re-created event with the same name):

```python
report.add_event(my_event)
Expand Down
4 changes: 3 additions & 1 deletion skills/impulse-reporting/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,9 @@ report.persist_results()

`determine_report()` validates that **every event referenced by an aggregation has been registered
with `add_event()`** before computing — otherwise it raises. Register events even when an aggregation
is the only thing that uses them.
is the only thing that uses them. The aggregation must reference the registered object itself (a
re-created event with the same name raises too), and `add_event()` rejects duplicate event names,
ignoring case.

After `determine_report()`, computed results are also available on the report object without a write —
`report.aggregation_dfs[...]` (used by `impulse-ml`) — which is what makes sinkless mode useful.
Expand Down
63 changes: 45 additions & 18 deletions src/impulse_reporting/core/report.py
Original file line number Diff line number Diff line change
Expand Up @@ -405,7 +405,10 @@ def add_event(self, event: Event):
Raises
------
ValueError
If the event is a ContainerEvent and a ContainerEvent already exists in the report.
If the event is a ContainerEvent and a ContainerEvent already exists in the report,
or if an event with the same name, ignoring case, has already been added (event
names identify events in the fact and dimension tables and name the event's
column in the solved data, which Spark resolves case-insensitively).
"""
if isinstance(event, ContainerEvent) and any(
isinstance(e, ContainerEvent) for e in self.events
Expand All @@ -414,6 +417,16 @@ def add_event(self, event: Event):
"Only one ContainerEvent is allowed per report. "
"A ContainerEvent has already been added to this report."
)
# Names are compared like Spark's case-insensitive column resolution.
existing = next(
(e for e in self.events if e.get_name().lower() == event.get_name().lower()), None
)
if existing is not None:
raise ValueError(
f"Cannot add event '{event.get_name()}': an event named "
f"'{existing.get_name()}' has already been added to this report. "
"Event names must be unique within a report, ignoring case."
)
self.events.append(event)
event.set_report_id(self.report_id)
if isinstance(event, TimeWindowEvent):
Expand Down Expand Up @@ -535,34 +548,48 @@ def _group_aggregations_by_type(self):

def _validate_aggregation_events(self) -> None:
"""
Validate that all events used in aggregations are added to the report.
Validate that every aggregation's event is an event object added to the report.

The event must be the registered object itself, not another object with the same
name: the event fact is derived from the registered event, while the solve scopes the
aggregation to its own event, so a differing copy would produce mismatched rows.

Raises
------
ValueError
If an aggregation uses an event that was not added to the report via add_event().
If an aggregation uses an event that was not added to the report via add_event(),
or a different object with the name (ignoring case) of a registered event.
"""
registered_events = set(self.events)
registered_event_names = {event.get_name() for event in self.events}
# Keyed like add_event's uniqueness check, so there is one event per key.
registered_events = {event.get_name().lower(): event for event in self.events}

missing_events = []
unregistered_events = []

for page in self.pages:
for aggregation in page.aggregations:
event = aggregation.get_event()
if event is not None and event not in registered_events:
event_name = event.get_name()
if event_name not in registered_event_names:
missing_events.append(
f"Aggregation '{aggregation.get_name()}' uses event "
f"'{event_name}' which was not added to the report."
)

if missing_events:
if event is None:
continue
registered = registered_events.get(event.get_name().lower())
if registered is event:
continue
if registered is not None:
unregistered_events.append(
f"Aggregation '{aggregation.get_name()}' uses an event named "
f"'{event.get_name()}' that is a different object from the registered "
f"event '{registered.get_name()}'. Pass the registered event (e.g. "
f"report.get_events_dict()[{registered.get_name()!r}])."
)
else:
unregistered_events.append(
f"Aggregation '{aggregation.get_name()}' uses event "
f"'{event.get_name()}' which was not added to the report."
)

if unregistered_events:
error_message = (
"The following events are used in aggregations but were not added "
"to the report via add_event():\n"
+ "\n".join(f" - {msg}" for msg in missing_events)
"The following aggregations reference events that are not registered with "
"this report:\n" + "\n".join(f" - {msg}" for msg in unregistered_events)
)
raise ValueError(error_message)

Expand Down
Loading
Loading