diff --git a/docs/impulse/docs/references/api/impulse_reporting/core/report.md b/docs/impulse/docs/references/api/impulse_reporting/core/report.md index 450b2970..3d4a8fb2 100644 --- a/docs/impulse/docs/references/api/impulse_reporting/core/report.md +++ b/docs/impulse/docs/references/api/impulse_reporting/core/report.md @@ -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**: diff --git a/docs/impulse/docs/references/report/event.md b/docs/impulse/docs/references/report/event.md index fa4671be..0f34764a 100644 --- a/docs/impulse/docs/references/report/event.md +++ b/docs/impulse/docs/references/report/event.md @@ -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) @@ -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. | diff --git a/docs/impulse/docs/references/report/index.md b/docs/impulse/docs/references/report/index.md index a5087fa0..9fc0e3ed 100644 --- a/docs/impulse/docs/references/report/index.md +++ b/docs/impulse/docs/references/report/index.md @@ -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). | @@ -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 diff --git a/skills/impulse-events/SKILL.md b/skills/impulse-events/SKILL.md index 542caddf..3736a094 100644 --- a/skills/impulse-events/SKILL.md +++ b/skills/impulse-events/SKILL.md @@ -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) diff --git a/skills/impulse-reporting/SKILL.md b/skills/impulse-reporting/SKILL.md index 65014f20..fe3030c3 100644 --- a/skills/impulse-reporting/SKILL.md +++ b/skills/impulse-reporting/SKILL.md @@ -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. diff --git a/src/impulse_reporting/core/report.py b/src/impulse_reporting/core/report.py index 82955e8a..1a6b6f9f 100644 --- a/src/impulse_reporting/core/report.py +++ b/src/impulse_reporting/core/report.py @@ -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 @@ -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): @@ -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) diff --git a/tests/impulse_reporting/unit/core/report_test.py b/tests/impulse_reporting/unit/core/report_test.py index 7d067b3a..a3e9fb5b 100644 --- a/tests/impulse_reporting/unit/core/report_test.py +++ b/tests/impulse_reporting/unit/core/report_test.py @@ -13,7 +13,9 @@ from impulse_reporting.core.page import Page from impulse_reporting.core.report import Report from impulse_reporting.events.basic_event import BasicEvent +from impulse_reporting.events.container_event import ContainerEvent from impulse_reporting.events.event_types import EventType +from impulse_reporting.events.time_window_event import TimeWindowEvent from impulse_reporting.persist.report_storage import SinkConfig DUMMY_CONFIG = { @@ -351,6 +353,120 @@ def test_create_solver_delta_with_solver_config(): assert solver.config.value_col == "value" +class TestAddEvent: + """Tests for add_event method.""" + + def test_duplicate_event_name_raises_error(self): + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + report.add_event(BasicEvent(name="rpm_event", expr=ts_expr > 0)) + + with pytest.raises(ValueError, match="'rpm_event' has already been added"): + report.add_event(BasicEvent(name="rpm_event", expr=ts_expr > 100)) + + assert len(report.get_events()) == 1 + + def test_duplicate_event_name_of_other_type_raises_error(self): + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + report.add_event(BasicEvent(name="window", expr=ts_expr > 0)) + + with pytest.raises(ValueError, match="'window' has already been added"): + report.add_event(TimeWindowEvent(name="window", window_length=60_000)) + + def test_same_event_added_twice_raises_error(self): + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + event = BasicEvent(name="rpm_event", expr=ts_expr > 0) + report.add_event(event) + + with pytest.raises(ValueError, match="'rpm_event' has already been added"): + report.add_event(event) + + def test_duplicate_event_name_ignoring_case_raises_error(self): + # Spark resolves the solved event columns case-insensitively. + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + report.add_event(BasicEvent(name="RPM_high", expr=ts_expr > 0)) + + with pytest.raises( + ValueError, match="Cannot add event 'rpm_high': an event named 'RPM_high'" + ): + report.add_event(BasicEvent(name="rpm_high", expr=ts_expr > 0)) + + def test_names_distinct_in_spark_are_allowed(self): + # Spark resolves these as two distinct columns, so both names are accepted. + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + report.add_event(BasicEvent(name="Drehzahl_groß", expr=ts_expr > 0)) + report.add_event(BasicEvent(name="Drehzahl_gross", expr=ts_expr > 0)) + + assert len(report.get_events()) == 2 + + def test_second_container_event_keeps_container_event_error(self): + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + report.add_event(ContainerEvent(name="measurement")) + + with pytest.raises(ValueError, match="Only one ContainerEvent is allowed per report"): + report.add_event(ContainerEvent(name="measurement")) + + def test_distinct_event_names_are_all_registered(self): + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + events = [ + BasicEvent(name="rpm_event", expr=ts_expr > 0), + ContainerEvent(name="measurement"), + TimeWindowEvent(name="window", window_length=60_000), + ] + for event in events: + report.add_event(event) + + assert report.get_events() == events + assert report.get_events_dict() == {event.get_name(): event for event in events} + + class TestValidateAggregationEvents: """Tests for _validate_aggregation_events method.""" @@ -382,7 +498,7 @@ def test_statistics_event_not_registered_raises_error(self): assert "test_event" in str(exc_info.value) assert "test_stats" in str(exc_info.value) - assert "not added to the report" in str(exc_info.value) + assert "uses event 'test_event' which was not added to the report" in str(exc_info.value) def test_histogram_event_not_registered_raises_error(self): report: Report = Report( @@ -568,6 +684,136 @@ def test_same_event_used_in_multiple_aggregations(self): report._validate_aggregation_events() + def test_same_name_different_definition_event_raises_error(self): + # The event fact would tile 60 s windows while the solve tiles 10 s windows. + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + report.add_event(TimeWindowEvent(name="w", window_length=60_000)) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + stats = StatsAggregator( + name="window_stats", + input_expressions=[ts_expr], + channel_names=["test_signal"], + statistics=["mean"], + event=TimeWindowEvent(name="w", window_length=10_000), + ) + + page = Page(page_number=1) + page.add_aggregation(stats) + report.add_page(page) + + with pytest.raises(ValueError) as exc_info: + report._validate_aggregation_events() + + error_msg = str(exc_info.value) + assert "window_stats" in error_msg + assert "'w' that is a different object" in error_msg + + def test_same_name_identical_copy_event_raises_error(self): + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + report.add_event(BasicEvent(name="rpm_event", expr=ts_expr > 0)) + + hist = HistogramDuration( + name="test_histogram", + base_expr=ts_expr, + bins=[0.0, 100.0], + event=BasicEvent(name="rpm_event", expr=ts_expr > 0), + ) + + page = Page(page_number=1) + page.add_aggregation(hist) + report.add_page(page) + + with pytest.raises(ValueError, match="'rpm_event' that is a different object"): + report._validate_aggregation_events() + + def test_name_differing_only_by_case_counts_as_copy(self): + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + report.add_event(BasicEvent(name="RPM_high", expr=ts_expr > 0)) + + stats = StatsAggregator( + name="rpm_stats", + input_expressions=[ts_expr], + channel_names=["test_signal"], + statistics=["mean"], + event=BasicEvent(name="rpm_high", expr=ts_expr > 0), + ) + + page = Page(page_number=1) + page.add_aggregation(stats) + report.add_page(page) + + with pytest.raises(ValueError) as exc_info: + report._validate_aggregation_events() + + error_msg = str(exc_info.value) + assert ( + "'rpm_high' that is a different object from the registered event 'RPM_high'" + in error_msg + ) + assert "report.get_events_dict()['RPM_high']" in error_msg + + def test_unregistered_and_copied_events_are_listed_together(self): + report: Report = Report( + name="my_report", + spark=None, + workspace_client=create_autospec(WorkspaceClient), + config=DUMMY_CONFIG, + ) + + ts_expr = TimeSeriesSelector(TagSelector("name") == "test_signal") + report.add_event(BasicEvent(name="registered_event", expr=ts_expr > 0)) + + copied = StatsAggregator( + name="copied_stats", + input_expressions=[ts_expr], + channel_names=["test_signal"], + statistics=["min"], + event=BasicEvent(name="registered_event", expr=ts_expr > 0), + ) + unregistered = StatsAggregator( + name="unregistered_stats", + input_expressions=[ts_expr], + channel_names=["test_signal"], + statistics=["max"], + event=BasicEvent(name="other_event", expr=ts_expr > 0), + ) + + page = Page(page_number=1) + page.add_aggregation(copied) + page.add_aggregation(unregistered) + report.add_page(page) + + with pytest.raises(ValueError) as exc_info: + report._validate_aggregation_events() + + error_msg = str(exc_info.value) + assert "'copied_stats' uses an event named 'registered_event'" in error_msg + assert ( + "'unregistered_stats' uses event 'other_event' which was not added to the report" + in error_msg + ) + class TestValidateFullRecalculationScope: """Tests for _validate_full_recalculation_scope method."""