From 4e0c05ea8df159a259eabd34cead10287b9b0c80 Mon Sep 17 00:00:00 2001 From: "tom.bonfert" Date: Wed, 14 Oct 2026 12:42:12 +0200 Subject: [PATCH 1/3] feat(reporting): deprecate `determine_report()`/`persist_results()` in favor of `run()` - Mark `Report.determine_report()` and `Report.persist_results()` as deprecated with `FutureWarning`; keep behavior unchanged. - Refactor `Report.run()` to call internal `_determine_report()`/`_persist_results()` so the public two-step API can warn independently. - Update README, demos, docs, and skills to use `run()` (and `run(persist_results=False)` for compute-only/sinkless workflows). - Migrate tests to `run()` or the internal helpers where deprecation warnings are not the subject under test. --- README.md | 3 +- demos/getting_started.ipynb | 3 +- demos/ml_feature_engineering.ipynb | 4 +- demos/reporting_pipeline.ipynb | 6 +- docs/impulse/docs/config/configuration.md | 8 +- docs/impulse/docs/motivation.md | 5 +- .../impulse_reporting/config/config_parser.md | 4 +- .../api/impulse_reporting/core/report.md | 75 ++++---- .../impulse/docs/references/report/channel.md | 2 +- docs/impulse/docs/references/report/event.md | 2 +- docs/impulse/docs/references/report/index.md | 33 ++-- docs/impulse/docs/tutorial/demo.md | 13 +- skills/impulse-config/SKILL.md | 6 +- skills/impulse-ml/SKILL.md | 8 +- skills/impulse-reporting/SKILL.md | 40 ++-- skills/impulse/SKILL.md | 3 +- src/impulse_reporting/config/config_parser.py | 4 +- src/impulse_reporting/core/report.py | 104 ++++++++--- .../batched_report_pipeline_test.py | 24 ++- .../integration/calculated_channel_test.py | 47 ++--- ...annel_mapping_resolution_dimension_test.py | 13 +- .../container_and_sequence_event_test.py | 7 +- .../integration/container_event_test.py | 7 +- .../integration/custom_statistics_test.py | 4 +- .../incremental_container_id_mapping_test.py | 3 +- .../integration/incremental_report_test.py | 55 ++---- .../max_containers_per_batch_test.py | 11 +- .../point_value_aggregator_test.py | 13 +- .../integration/points_in_time_event_test.py | 4 +- .../integration/raw_encoder_test.py | 2 +- .../integration/simple_report_test.py | 23 +-- .../integration/statistics_test.py | 20 +- .../string_container_id_report_test.py | 5 +- .../integration/time_window_event_test.py | 29 ++- .../unit/core/report_incremental_test.py | 171 ++++++++++++------ .../unit/core/report_utils_test.py | 2 +- 36 files changed, 401 insertions(+), 362 deletions(-) diff --git a/README.md b/README.md index 4fcf0921..641886c1 100644 --- a/README.md +++ b/README.md @@ -77,8 +77,7 @@ page.add_aggregation( report.add_page(page) # Compute and persist to the Gold-layer star schema. -report.determine_report() -report.persist_results() +report.run() ``` ## AI Assistant Skills diff --git a/demos/getting_started.ipynb b/demos/getting_started.ipynb index 975645e8..5f336e1b 100644 --- a/demos/getting_started.ipynb +++ b/demos/getting_started.ipynb @@ -91,8 +91,7 @@ ")\n", "report.add_page(page)\n", "\n", - "report.determine_report()\n", - "report.persist_results()\n" + "report.run()\n" ], "id": "3c9aea0cf4a3ecc" }, diff --git a/demos/ml_feature_engineering.ipynb b/demos/ml_feature_engineering.ipynb index cb6e884f..81591c2f 100644 --- a/demos/ml_feature_engineering.ipynb +++ b/demos/ml_feature_engineering.ipynb @@ -529,7 +529,7 @@ "source": [ "# 5. Compute Statistics\n", "\n", - "`determine_report()` runs the statistics computation\n", + "`run(persist_results=False)` runs the statistics computation\n", "in parallel across all containers via Spark Pandas UDFs.\n", "No Gold-layer write is needed here since we retrieve\n", "the results directly from memory." @@ -553,7 +553,7 @@ }, "outputs": [], "source": [ - "report.determine_report()\n", + "report.run(persist_results=False)\n", "print(\"Statistics computed\")" ] }, diff --git a/demos/reporting_pipeline.ipynb b/demos/reporting_pipeline.ipynb index cc93a062..4cfe25de 100644 --- a/demos/reporting_pipeline.ipynb +++ b/demos/reporting_pipeline.ipynb @@ -778,8 +778,7 @@ "source": [ "# 7. Compute & Persist\n", "\n", - "- `determine_report()` — parallel execution\n", - "- `persist_results()` — writes star schema" + "`run()` computes everything in parallel and writes the star schema." ] }, { @@ -800,8 +799,7 @@ }, "outputs": [], "source": [ - "report.determine_report()\n", - "report.persist_results()\n", + "report.run()\n", "print(f\"Report persisted to {pfx}_*\")" ] }, diff --git a/docs/impulse/docs/config/configuration.md b/docs/impulse/docs/config/configuration.md index d8199953..462ecaec 100644 --- a/docs/impulse/docs/config/configuration.md +++ b/docs/impulse/docs/config/configuration.md @@ -80,7 +80,7 @@ Defines where gold-layer tables are written. | `catalog` | `str` | Yes | Target catalog name. | | `schema` | `str` | Yes | Target schema name. | | `table_prefix` | `str` | Yes | Prefix for all generated table names. | -| `cleanup_temp_tables` | `bool` | No | Drop the batch-solving `__impulse_temp_*` tables from this schema after `persist_results()` completes successfully. Defaults to `false`. Overridable per call via `persist_results(cleanup_temp_tables=...)`. | +| `cleanup_temp_tables` | `bool` | No | Drop the batch-solving `__impulse_temp_*` tables from this schema after `run()` persists successfully. Defaults to `false`. Overridable per call via `run(cleanup_temp_tables=...)`. | Output tables are named `{table_prefix}_{entity}` (e.g. `my_report_histogram_fact`). @@ -88,9 +88,9 @@ Output tables are named `{table_prefix}_{entity}` (e.g. ### Sinkless reports `unity_sink` is optional. When omitted, the report runs in **sinkless -mode**: `determine_report()` still computes events, aggregations, and -container dimensions and exposes them on the report object, but -`persist_results()` becomes a no-op. Useful for ad-hoc analysis, +mode**: `run()` still computes events, aggregations, and container +dimensions and exposes them on the report object, but writes nothing. +Useful for ad-hoc analysis, notebooks, and tests where writing to Unity Catalog is not desired. --- diff --git a/docs/impulse/docs/motivation.md b/docs/impulse/docs/motivation.md index 6d86c60d..2daf32d1 100644 --- a/docs/impulse/docs/motivation.md +++ b/docs/impulse/docs/motivation.md @@ -71,11 +71,10 @@ Every measurement data analysis follows a recognizable pattern: Impulse's **Report** orchestrator standardizes this entire workflow. A JSON configuration defines the data source and sink. Events and aggregations are declared as composable objects. -Two method calls execute the full pipeline: +One method call executes the full pipeline: ```python -my_report.determine_report() # compute all events and aggregations -my_report.persist_results() # write results to the Gold layer +my_report.run() # compute all events and aggregations and write them to the Gold layer ``` This reduces a typical analysis from hundreds of lines of ad-hoc Spark code to a concise, declarative report definition. diff --git a/docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md b/docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md index dcaca28e..848069ff 100644 --- a/docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md +++ b/docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md @@ -106,7 +106,7 @@ Configuration for data sink location in Unity Catalog. - `schema` (`str`): Target schema name for output tables. - `table_prefix` (`str`): Prefix to use for generated output table names. - `cleanup_temp_tables` (`bool`): When ``True``, the intermediate ``__impulse_temp_*`` tables written to this -sink during batch solving are dropped after ``persist_results()`` completes +sink during batch solving are dropped after ``Report.run()`` persists successfully. Defaults to ``False`` (temp tables are retained for inspection and only cleared at the start of the next report run). @@ -286,7 +286,7 @@ incremental. In full mode this is a no-op, since everything recomputes over all containers anyway. Names that do not match any registered entity are rejected at the start of -``Report.determine_report`` (fail fast on typos / stale names). +``Report.run`` (fail fast on typos / stale names). **Arguments**: 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 3d4a8fb2..207ca616 100644 --- a/docs/impulse/docs/references/api/impulse_reporting/core/report.md +++ b/docs/impulse/docs/references/api/impulse_reporting/core/report.md @@ -305,23 +305,14 @@ Get the list of calculated channels associated with the report. def persist_results(cleanup_temp_tables: bool | None = None) ``` -Persist report results using appropriate strategy based on definition changes. +Deprecated: use :meth:`run`, which determines and persists in one call. -Uses tracked state from determine_report() to decide persistence strategy: -- Changed definitions: replaceWhere (atomic delete + insert) -- Unchanged definitions: MERGE (upsert) +Persists the results of a prior :meth:`determine_report` call; behavior is +unchanged. Unlike :meth:`run`, it does not drain capped incremental batches. **Arguments**: -- `cleanup_temp_tables` (`bool`): Whether to drop the batch-solving ``__impulse_temp_*`` tables from the -sink schema after persistence completes successfully. -- True/False: use this value, overriding the config flag. -- None (default): fall back to ``config.unity_sink.cleanup_temp_tables`` - (which itself defaults to False). - -**Returns**: - -`None`: +- `cleanup_temp_tables` (`bool | None`): Same as the :meth:`run` argument. #### run @@ -333,21 +324,38 @@ def run(is_incremental: bool = None, Determine and persist a report, draining all container batches. -Wraps :meth:`determine_report` + :meth:`persist_results` (both remain usable -standalone). With ``max_containers_per_batch`` set on an incremental run, ``run()`` -loops: each iteration commits at most that many upserted containers, and the loop -continues until a run observes at most that many remaining (the last batch clears -the rest). After the first iteration all definition hashes are current, so later -iterations recompute only unchanged entities over the next batch of new containers. -Without a cap (or in full mode) it is a single determine+persist pass. +The entry point for executing a report. Computes events, aggregations, calculated +channels, and container dimensions, then writes them to the gold layer. Computed +results stay accessible on the report's attributes (e.g. ``aggregation_dfs``, +``event_dfs``), so ``run(persist_results=False)`` is the compute-only path. Without +a configured sink nothing is written. + +Supports incremental processing with definition-hash-based optimization: +- Changed definitions trigger full reprocessing (all containers) +- Unchanged definitions use incremental processing (only new/updated containers) + +With ``max_containers_per_batch`` set on an incremental run, ``run()`` loops: each +iteration commits at most that many upserted containers, and the loop continues +until a run observes at most that many remaining (the last batch clears the rest). +After the first iteration all definition hashes are current, so later iterations +recompute only unchanged entities over the next batch of new containers. Without a +cap (or in full mode) it is a single determine+persist pass. **Arguments**: -- `is_incremental` (`bool`): Forwarded to :meth:`determine_report` (config still overrides it). +- `is_incremental` (`bool`): Hint for processing mode. Overwritten by config when incremental +config is provided. +- True: Request incremental processing (if gold layer exists) +- False: Force full processing +- None: Use config value (default: full processing) - `persist_results` (`bool`): When True (default), persist after determining and drain the batches. When -False, run :meth:`determine_report` once without persisting (no iteration — -nothing commits, so the batch cannot advance). -- `cleanup_temp_tables` (`bool | None`): Forwarded to :meth:`persist_results`. +False, determine once without persisting (no iteration — nothing commits, so +the batch cannot advance). +- `cleanup_temp_tables` (`bool | None`): Whether to drop the batch-solving ``__impulse_temp_*`` tables from the +sink schema after persistence completes successfully. +- True/False: use this value, overriding the config flag. +- None (default): fall back to ``config.unity_sink.cleanup_temp_tables`` + (which itself defaults to False). #### determine\_report @@ -355,23 +363,12 @@ nothing commits, so the batch cannot advance). def determine_report(is_incremental: bool = None) ``` -Determine and process events, aggregations, and container dimensions for the report. +Deprecated: use :meth:`run`, or ``run(persist_results=False)`` to compute only. -Results are accessible in the report's attributes. - -Supports incremental processing with definition-hash-based optimization: -- Changed definitions trigger full reprocessing (all containers) -- Unchanged definitions use incremental processing (only new/updated containers) +Computes the report without persisting; results are accessible in the report's +attributes and behavior is unchanged. **Arguments**: -- `is_incremental` (`bool`): Hint for processing mode. Overwritten by config when incremental -config is provided. -- True: Request incremental processing (if gold layer exists) -- False: Force full processing -- None: Use config value (default: full processing) - -**Returns**: - -`None`: +- `is_incremental` (`bool`): Same as the :meth:`run` argument. diff --git a/docs/impulse/docs/references/report/channel.md b/docs/impulse/docs/references/report/channel.md index aea3632d..a9c6faa5 100644 --- a/docs/impulse/docs/references/report/channel.md +++ b/docs/impulse/docs/references/report/channel.md @@ -23,7 +23,7 @@ those values are injected as usual — the requirement propagates through the ca ::: Every calculated channel used in a report **must** be registered with the report via -`add_calculated_channel()` before `determine_report()` is called. +`add_calculated_channel()` before `run()` is called. ```python my_report.add_calculated_channel(my_channel) diff --git a/docs/impulse/docs/references/report/event.md b/docs/impulse/docs/references/report/event.md index 0f34764a..99241d68 100644 --- a/docs/impulse/docs/references/report/event.md +++ b/docs/impulse/docs/references/report/event.md @@ -8,7 +8,7 @@ title: Events Events define time windows within measurement data that scope downstream aggregations. An event segments continuous 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()` +Every event used by an aggregation **must** be registered with the report via `add_event()` before `run()` 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. diff --git a/docs/impulse/docs/references/report/index.md b/docs/impulse/docs/references/report/index.md index 9fc0e3ed..47ebe84b 100644 --- a/docs/impulse/docs/references/report/index.md +++ b/docs/impulse/docs/references/report/index.md @@ -43,8 +43,7 @@ Either `config` or `config_path` must be provided. | `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). | -| `persist_results(cleanup_temp_tables)` | Writes all computed results (fact and dimension tables) to the configured Gold layer sink. | `cleanup_temp_tables`: `bool` or `None`. When truthy, drops the batch-solving `__impulse_temp_*` tables after a successful write; `None` (default) falls back to `config.unity_sink.cleanup_temp_tables`. | +| `run(is_incremental, persist_results, cleanup_temp_tables)` | Computes all events, aggregations, calculated channels, and container dimensions and writes them (fact and dimension tables) to the configured Gold layer sink. Results are also stored on the report object. On a capped incremental run it loops until all containers are processed. | `is_incremental`: `bool` or `None`. Mode hint; overridden by `config.incremental` when present. See [Incremental processing](#incremental-processing). `persist_results`: `bool`, default `True`. `False` computes without writing. `cleanup_temp_tables`: `bool` or `None`. When truthy, drops the batch-solving `__impulse_temp_*` tables after a successful write; `None` (default) falls back to `config.unity_sink.cleanup_temp_tables`. | ### Execution workflow @@ -54,22 +53,28 @@ Either `config` or `config_path` must be provided. my_report.add_event(my_event) my_report.add_page(page) -# 3. Compute -my_report.determine_report() - -# 4. Persist -my_report.persist_results() +# 3. Compute and persist +my_report.run() ``` -`determine_report()` validates that every event referenced by an aggregation has been registered with `add_event()` -before computation begins. The aggregation must reference the registered event object itself; a re-created event with +`run()` validates that every event referenced by an aggregation has been registered with `add_event()` +before computation begins. + +:::note Deprecated: `determine_report()` / `persist_results()` +The two-step `determine_report()` + `persist_results()` still works but emits a `FutureWarning` and will be +removed in a future release. Replace the pair with `run()`, and a standalone `determine_report()` with +`run(persist_results=False)`. Unlike `run()`, a single `determine_report()` + `persist_results()` pass does not +drain a capped incremental run. The `persist_results` argument of `run()` is a flag and is unrelated to the +deprecated method of the same name. +::: The aggregation must reference the registered event object itself; a re-created event with the same name is rejected. ### Sinkless mode -If no `unity_sink` is configured, the report runs without persistence. `determine_report()` still computes events, -aggregations, and dimensions and exposes them on the report object, but `persist_results()` becomes a no-op. This is -useful for ad-hoc analysis, notebooks, and tests where writing to Unity Catalog is not desired. +If no `unity_sink` is configured, the report runs without persistence. `run()` still computes events, +aggregations, and dimensions and exposes them on the report object, but writes nothing. To compute without writing +while a sink is configured, call `run(persist_results=False)`. This is useful for ad-hoc analysis, notebooks, and +tests where writing to Unity Catalog is not desired. --- @@ -81,7 +86,7 @@ validated using Pydantic models. See ## Incremental processing -Incremental processing lets `determine_report()` skip containers and +Incremental processing lets `run()` skip containers and definitions it already handled on a prior run. Turn it on via `incremental.enabled` in config (see [Configuration › incremental](../../config/configuration.md#incremental-optional)), @@ -100,7 +105,7 @@ or pass `is_incremental=True` at call time. 1. No gold layer yet? Run full. Nothing to compare against. 2. `config.incremental` set? `config.incremental.enabled` wins. -3. Otherwise the `is_incremental` argument to `determine_report()` wins. +3. Otherwise the `is_incremental` argument to `run()` wins. 4. Neither set? Run full. The first run of a new report is always full. Subsequent runs pick up where the last one left off. diff --git a/docs/impulse/docs/tutorial/demo.md b/docs/impulse/docs/tutorial/demo.md index ee5a9133..391e8d47 100644 --- a/docs/impulse/docs/tutorial/demo.md +++ b/docs/impulse/docs/tutorial/demo.md @@ -280,22 +280,21 @@ my_report.add_calculated_channel(avg_temp_channel) - `identity` is any non-empty dict (keys are arbitrary, e.g. `channel_name` / `data_key`); it seeds the deterministic `channel_id` and is stored on `calculated_channel_dimension` (as a `map`, joined to the fact via `channel_id`) — not repeated on every fact row. -- Registration mirrors events: `add_calculated_channel()` before `determine_report()`. +- Registration mirrors events: `add_calculated_channel()` before `run()`. --- ## Step 5: Compute and persist -Two calls execute the full pipeline: +One call executes the full pipeline: ```python -my_report.determine_report() -my_report.persist_results() +my_report.run() ``` -- `determine_report()` resolves all TSAL expressions, computes event instances, runs all aggregations, and solves - calculated channels across all containers matched by the configuration. -- `persist_results()` writes fact and dimension tables to the Gold layer in Unity Catalog. +`run()` resolves all TSAL expressions, computes event instances, runs all aggregations, and solves calculated +channels across all containers matched by the configuration. It then writes fact and dimension tables to the Gold +layer in Unity Catalog. To compute without writing, call `run(persist_results=False)`. --- diff --git a/skills/impulse-config/SKILL.md b/skills/impulse-config/SKILL.md index 81568e03..9d0b9952 100644 --- a/skills/impulse-config/SKILL.md +++ b/skills/impulse-config/SKILL.md @@ -71,10 +71,10 @@ Where gold tables are written. Output tables are named `{table_prefix}_{entity}` | `catalog` | Yes | Target catalog. | | `schema` | Yes | Target schema. | | `table_prefix` | Yes | Prefix for tables. | -| `cleanup_temp_tables` | No | Drop the batch-solving `__impulse_temp_*` tables from the schema after `persist_results()` succeeds. Defaults to `false`. Overridable per call via `persist_results(cleanup_temp_tables=...)`. | +| `cleanup_temp_tables` | No | Drop the batch-solving `__impulse_temp_*` tables from the schema after `run()` persists successfully. Defaults to `false`. Overridable per call via `run(cleanup_temp_tables=...)`. | -**Sinkless mode:** omit `unity_sink` entirely. `determine_report()` still computes everything and -exposes it on the report object, but `persist_results()` becomes a no-op. Use it for ad-hoc analysis, +**Sinkless mode:** omit `unity_sink` entirely. `run()` still computes everything and exposes it on +the report object, but writes nothing. Use it for ad-hoc analysis, ML feature extraction, notebooks, and tests. (See `impulse-analyze` and `impulse-ml`.) ## container_filters (optional) diff --git a/skills/impulse-ml/SKILL.md b/skills/impulse-ml/SKILL.md index efb30977..5ef0a800 100644 --- a/skills/impulse-ml/SKILL.md +++ b/skills/impulse-ml/SKILL.md @@ -39,7 +39,7 @@ config = { "container_tags_table": "my_catalog.silver.container_tags", "channel_tags_table": "my_catalog.silver.channel_tags", }, - # no unity_sink -> sinkless: determine_report() computes, persist is a no-op + # no unity_sink -> sinkless: run() computes, nothing is written "query_engine": {"solver": "DefaultSolver", "data_type": "RLE"}, "measurement_dimensions": ["container_id", "vehicle_key", "start_ts", "stop_ts"], } @@ -65,12 +65,12 @@ page.add_aggregation(StatsAggregator( desc="Per-recording statistics", )) -report.determine_report() # computes in parallel across containers; no gold write needed +report.run(persist_results=False) # computes in parallel across containers; no gold write ``` ## 2. Read the computed statistics -After `determine_report()`, the results live on the report under `aggregation_dfs`, keyed by +After `run()`, the results live on the report under `aggregation_dfs`, keyed by aggregation type. `StatsAggregator` results are under `"STATS_AGGREGATOR"`, returned as a bundle keyed by whether the rows were freshly computed (`"changed"`) or reused from a prior incremental run (`"unchanged"`). In a fresh sinkless run take `"changed"`: @@ -133,4 +133,4 @@ ends once the feature matrix is built. - **`StatsAggregator` inputs must be `SampleSeries`** (channels or arithmetic over them), not comparisons. - To materialize the same statistics to gold instead of reading them inline (e.g. for a shared feature - store), add a `unity_sink` and call `persist_results()` — that is `impulse-reporting`. + store), add a `unity_sink` and call `run()` — that is `impulse-reporting`. diff --git a/skills/impulse-reporting/SKILL.md b/skills/impulse-reporting/SKILL.md index fe3030c3..9bcc063c 100644 --- a/skills/impulse-reporting/SKILL.md +++ b/skills/impulse-reporting/SKILL.md @@ -5,8 +5,7 @@ description: > star schema. Use when the user wants to "build an Impulse report", compute histograms/statistics across all recordings and write them to Delta tables for a dashboard or scheduled job, or run incrementally so only new/changed data is reprocessed. Covers the Report/Page lifecycle - (`add_event`, `add_page`, `determine_report`, `persist_results`), incremental processing, and - sinkless mode. + (`add_event`, `add_page`, `run`), incremental processing, and sinkless mode. --- # Impulse — reporting pipeline @@ -46,8 +45,9 @@ Exactly one of `config` / `config_path` must be provided. Useful methods: `get_db()` → the `MeasurementDB` for signal selection; `get_solver()` → the configured solver; `get_sink_config()` → the resolved sink; `add_event(event)`; `add_page(page)`; -`add_calculated_channel(channel)` (see `impulse-channels`); `determine_report(is_incremental=None)`; -`persist_results(cleanup_temp_tables=None)` (pass `True`/`False` to override the +`add_calculated_channel(channel)` (see `impulse-channels`); +`run(is_incremental=None, persist_results=True, cleanup_temp_tables=None)` — the single entry point +that computes and writes the report (`cleanup_temp_tables=True`/`False` overrides the `unity_sink.cleanup_temp_tables` config flag for dropping `__impulse_temp_*` tables after a successful write). @@ -83,34 +83,36 @@ report.add_calculated_channel(CalculatedChannel( identity={"channel_name": "speed_kmh", "data_key": "CALC"}, )) -# 5. Compute everything (in parallel across containers) -report.determine_report() - -# 6. Write the gold star schema -report.persist_results() +# 5. Compute everything (in parallel across containers) and write the gold star schema +report.run() ``` -`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. The aggregation must reference the registered object itself (a +`run()` 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. 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. +After `run()`, computed results are also available on the report object — `report.aggregation_dfs[...]` +(used by `impulse-ml`) — which is what makes sinkless mode useful. + +`determine_report()` + `persist_results()` are deprecated (they warn) — use `run()`. The +`persist_results=` flag of `run()` is unrelated to the deprecated method of the same name. ## Sinkless mode (compute without writing) -Omit `unity_sink` from the config. `determine_report()` still computes events, aggregations, and -dimensions and exposes them on the report; `persist_results()` becomes a no-op. Good for notebooks, -tests, and feeding ML (see `impulse-ml`). Reading a DataFrame straight out of the query engine without -any report at all is `impulse-analyze`. +Omit `unity_sink` from the config. `run()` still computes events, aggregations, and dimensions and +exposes them on the report, but writes nothing. To compute without writing even when a sink is +configured, call `run(persist_results=False)`. Good for notebooks, tests, and feeding ML (see +`impulse-ml`). Reading a DataFrame straight out of the query engine without any report at all is +`impulse-analyze`. ## Incremental processing Reuses prior results for unchanged definitions and reprocesses only containers that are new or updated in silver. Turn it on via `incremental.enabled` in config (see `impulse-config`) or pass -`is_incremental=True` to `determine_report()`. +`is_incremental=True` to `run()`. With `query_engine.max_containers_per_batch` set, `run()` commits at +most that many upserted containers per iteration and loops until all are processed. **Mode resolution** (in order): diff --git a/skills/impulse/SKILL.md b/skills/impulse/SKILL.md index 9b41ad56..0b5064b6 100644 --- a/skills/impulse/SKILL.md +++ b/skills/impulse/SKILL.md @@ -129,8 +129,7 @@ page.add_aggregation(HistogramDuration( )) report.add_page(page) -report.determine_report() # compute -report.persist_results() # write the gold star schema +report.run() # compute and write the gold star schema ``` ## Where to go next diff --git a/src/impulse_reporting/config/config_parser.py b/src/impulse_reporting/config/config_parser.py index 9bbde633..451fd5cb 100644 --- a/src/impulse_reporting/config/config_parser.py +++ b/src/impulse_reporting/config/config_parser.py @@ -174,7 +174,7 @@ class UnitySink(BaseModel): Prefix to use for generated output table names. cleanup_temp_tables : bool When ``True``, the intermediate ``__impulse_temp_*`` tables written to this - sink during batch solving are dropped after ``persist_results()`` completes + sink during batch solving are dropped after ``Report.run()`` persists successfully. Defaults to ``False`` (temp tables are retained for inspection and only cleared at the start of the next report run). @@ -634,7 +634,7 @@ class FullRecalculation(BaseModel): containers anyway. Names that do not match any registered entity are rejected at the start of - ``Report.determine_report`` (fail fast on typos / stale names). + ``Report.run`` (fail fast on typos / stale names). Attributes ---------- diff --git a/src/impulse_reporting/core/report.py b/src/impulse_reporting/core/report.py index 1a6b6f9f..b609694a 100644 --- a/src/impulse_reporting/core/report.py +++ b/src/impulse_reporting/core/report.py @@ -1,5 +1,6 @@ import json import logging +import warnings import zlib from typing import Any from databricks.sdk import WorkspaceClient @@ -69,6 +70,17 @@ logger = logging.getLogger(__name__) +_DETERMINE_REPORT_DEPRECATION_MSG = ( + "Report.determine_report() is deprecated; use Report.run() instead " + "(run(persist_results=False) computes without writing). " + "determine_report() will be removed in a future release." +) +_PERSIST_RESULTS_DEPRECATION_MSG = ( + "Report.persist_results() is deprecated; use Report.run() instead, which determines and " + "persists in one call and drains capped incremental batches. " + "persist_results() will be removed in a future release." +) + class Report: """Represents a report containing pages, events, and configurations for data processing and persistence.""" @@ -617,10 +629,25 @@ def _validate_full_recalculation_scope(self) -> None: @telemetry_logger("report", "persist_results") def persist_results(self, cleanup_temp_tables: bool | None = None): + """Deprecated: use :meth:`run`, which determines and persists in one call. + + Persists the results of a prior :meth:`determine_report` call; behavior is + unchanged. Unlike :meth:`run`, it does not drain capped incremental batches. + + Parameters + ---------- + cleanup_temp_tables : bool | None, optional + Same as the :meth:`run` argument. + """ + # stacklevel=3 skips the telemetry_logger wrapper so the warning points at the caller. + warnings.warn(_PERSIST_RESULTS_DEPRECATION_MSG, FutureWarning, stacklevel=3) + self._persist_results(cleanup_temp_tables) + + def _persist_results(self, cleanup_temp_tables: bool | None = None): """ Persist report results using appropriate strategy based on definition changes. - Uses tracked state from determine_report() to decide persistence strategy: + Uses tracked state from :meth:`_determine_report` to decide persistence strategy: - Changed definitions: replaceWhere (atomic delete + insert) - Unchanged definitions: MERGE (upsert) @@ -640,7 +667,7 @@ def persist_results(self, cleanup_temp_tables: bool | None = None): if not self._has_sink: return - # Use tracked state from determine_report + # Use tracked state from _determine_report changed_aggregation_ids = getattr(self, "_changed_aggregation_ids", {}) changed_event_ids = getattr(self, "_changed_event_ids", {}) changed_channel_ids = getattr(self, "_changed_channel_ids", {}) @@ -660,7 +687,7 @@ def _resolve_cleanup_temp_tables(self, cleanup: bool | None) -> bool: """Resolve whether to drop temp tables: explicit arg wins, else config flag. ``self.config.unity_sink`` is guaranteed non-None when called, since - ``persist_results`` returns early unless a sink is configured. + ``_persist_results`` returns early unless a sink is configured. """ if cleanup is not None: return cleanup @@ -713,7 +740,6 @@ def _persist_full(self): uri = writer.get_output_uri() writer.write(self.channel_mapping_resolution_dimension_df, uri=uri) - @telemetry_logger("report", "determine_report") def _persist_incremental( self, changed_aggregation_ids: dict[str, list[int]], @@ -988,24 +1014,41 @@ def run( ): """Determine and persist a report, draining all container batches. - Wraps :meth:`determine_report` + :meth:`persist_results` (both remain usable - standalone). With ``max_containers_per_batch`` set on an incremental run, ``run()`` - loops: each iteration commits at most that many upserted containers, and the loop - continues until a run observes at most that many remaining (the last batch clears - the rest). After the first iteration all definition hashes are current, so later - iterations recompute only unchanged entities over the next batch of new containers. - Without a cap (or in full mode) it is a single determine+persist pass. + The entry point for executing a report. Computes events, aggregations, calculated + channels, and container dimensions, then writes them to the gold layer. Computed + results stay accessible on the report's attributes (e.g. ``aggregation_dfs``, + ``event_dfs``), so ``run(persist_results=False)`` is the compute-only path. Without + a configured sink nothing is written. + + Supports incremental processing with definition-hash-based optimization: + - Changed definitions trigger full reprocessing (all containers) + - Unchanged definitions use incremental processing (only new/updated containers) + + With ``max_containers_per_batch`` set on an incremental run, ``run()`` loops: each + iteration commits at most that many upserted containers, and the loop continues + until a run observes at most that many remaining (the last batch clears the rest). + After the first iteration all definition hashes are current, so later iterations + recompute only unchanged entities over the next batch of new containers. Without a + cap (or in full mode) it is a single determine+persist pass. Parameters ---------- is_incremental : bool, optional - Forwarded to :meth:`determine_report` (config still overrides it). + Hint for processing mode. Overwritten by config when incremental + config is provided. + - True: Request incremental processing (if gold layer exists) + - False: Force full processing + - None: Use config value (default: full processing) persist_results : bool, optional When True (default), persist after determining and drain the batches. When - False, run :meth:`determine_report` once without persisting (no iteration — - nothing commits, so the batch cannot advance). + False, determine once without persisting (no iteration — nothing commits, so + the batch cannot advance). cleanup_temp_tables : bool | None, optional - Forwarded to :meth:`persist_results`. + Whether to drop the batch-solving ``__impulse_temp_*`` tables from the + sink schema after persistence completes successfully. + - True/False: use this value, overriding the config flag. + - None (default): fall back to ``config.unity_sink.cleanup_temp_tables`` + (which itself defaults to False). Notes ----- @@ -1017,10 +1060,10 @@ def run( """ previous_batch_ids = None while True: - self.determine_report(is_incremental) + self._determine_report(is_incremental) if not persist_results: return - self.persist_results(cleanup_temp_tables) + self._persist_results(cleanup_temp_tables) # Stop once a run processed the remainder (uncapped upserted <= cap); also # covers no cap / full mode, where the flag stays False. if not self._more_batches_pending: @@ -1044,22 +1087,29 @@ def run( @telemetry_logger("report", "determine_report") def determine_report(self, is_incremental: bool = None): + """Deprecated: use :meth:`run`, or ``run(persist_results=False)`` to compute only. + + Computes the report without persisting; results are accessible in the report's + attributes and behavior is unchanged. + + Parameters + ---------- + is_incremental : bool, optional + Same as the :meth:`run` argument. """ - Determine and process events, aggregations, and container dimensions for the report. - Results are accessible in the report's attributes. + # stacklevel=3 skips the telemetry_logger wrapper so the warning points at the caller. + warnings.warn(_DETERMINE_REPORT_DEPRECATION_MSG, FutureWarning, stacklevel=3) + self._determine_report(is_incremental) - Supports incremental processing with definition-hash-based optimization: - - Changed definitions trigger full reprocessing (all containers) - - Unchanged definitions use incremental processing (only new/updated containers) + def _determine_report(self, is_incremental: bool = None): + """ + Determine and process events, aggregations, and container dimensions for the report. + Results are accessible in the report's attributes; :meth:`run` documents the modes. Parameters ---------- is_incremental : bool, optional - Hint for processing mode. Overwritten by config when incremental - config is provided. - - True: Request incremental processing (if gold layer exists) - - False: Force full processing - - None: Use config value (default: full processing) + Processing-mode hint, see :meth:`run`. Returns ------- diff --git a/tests/impulse_reporting/integration/batched_report_pipeline_test.py b/tests/impulse_reporting/integration/batched_report_pipeline_test.py index 2b3eef53..affdcd24 100644 --- a/tests/impulse_reporting/integration/batched_report_pipeline_test.py +++ b/tests/impulse_reporting/integration/batched_report_pipeline_test.py @@ -272,7 +272,7 @@ def _stats_wrapper( "SELECT 1 AS stale_id" ) - report.determine_report() + report.run(persist_results=False) temp_tables = { row.tableName @@ -325,13 +325,13 @@ def _temp_tables(spark: SparkSession) -> set[str]: def test_persist_results_retains_temp_tables_by_default(spark, cleanup_gold): - """Default behavior: the current run's temp tables survive persist_results().""" + """Default behavior: the current run's temp tables survive persistence.""" report, _ = _build_batched_report(spark) - report.determine_report() + report._determine_report() assert _temp_tables(spark) # temp tables created during solving - report.persist_results() + report._persist_results() # Gold facts written with real values, and temp tables still present. assert ( @@ -341,13 +341,13 @@ def test_persist_results_retains_temp_tables_by_default(spark, cleanup_gold): def test_persist_results_cleans_temp_tables_when_param_true(spark, cleanup_gold): - """persist_results(cleanup_temp_tables=True) drops temp tables after a successful write.""" + """cleanup_temp_tables=True drops temp tables after a successful write.""" report, _ = _build_batched_report(spark) - report.determine_report() + report._determine_report() assert _temp_tables(spark) - report.persist_results(cleanup_temp_tables=True) + report._persist_results(cleanup_temp_tables=True) # Assert on the persisted gold fact table (not the report's lazy DataFrames, which # read from the now-dropped temp tables). Persistence completed before cleanup ran. @@ -356,7 +356,7 @@ def test_persist_results_cleans_temp_tables_when_param_true(spark, cleanup_gold) assert not _temp_tables(spark) -def test_persist_results_param_overrides_config_flag(spark, cleanup_gold): +def test_run_cleanup_param_overrides_config_flag(spark, cleanup_gold): """Explicit param wins over config: config True but param False retains temp tables.""" config = ImpulseConfig( source=Source( @@ -375,13 +375,12 @@ def test_persist_results_param_overrides_config_flag(spark, cleanup_gold): report, _ = _build_batched_report(spark) report.config = config - report.determine_report() - report.persist_results(cleanup_temp_tables=False) + report.run(cleanup_temp_tables=False) assert _temp_tables(spark) -def test_persist_results_config_flag_cleans_temp_tables(spark, cleanup_gold): +def test_run_config_flag_cleans_temp_tables(spark, cleanup_gold): """When param is None, the config flag drives cleanup.""" config = ImpulseConfig( source=Source( @@ -400,7 +399,6 @@ def test_persist_results_config_flag_cleans_temp_tables(spark, cleanup_gold): report, _ = _build_batched_report(spark) report.config = config - report.determine_report() - report.persist_results() + report.run() assert not _temp_tables(spark) diff --git a/tests/impulse_reporting/integration/calculated_channel_test.py b/tests/impulse_reporting/integration/calculated_channel_test.py index 481f7dea..5a426e8e 100644 --- a/tests/impulse_reporting/integration/calculated_channel_test.py +++ b/tests/impulse_reporting/integration/calculated_channel_test.py @@ -2,7 +2,7 @@ """Integration tests: Report orchestration of calculated channels. Builds a Report against the autouse ``spark_catalog.silver`` fixtures, adds a -calculated channel, runs determine_report + persist_results, and asserts real +calculated channel, runs ``Report.run()``, and asserts real values land in the gold fact/dimension tables — then verifies incremental re-run behavior (idempotent unchanged upsert; changed-definition replace). """ @@ -89,8 +89,7 @@ def test_persist_calculated_channel_full(spark): # gold fact rows must be identical to the silver channel rows (see below). ch_pass = _add_channel(report, factor=1.0, name="speed_raw") - report.determine_report() - report.persist_results() + report.run() assert spark.catalog.tableExists(_FACT) assert spark.catalog.tableExists(_DIM) @@ -180,8 +179,7 @@ def test_batched_calculated_channels_union(spark): report.add_calculated_channel(ch_speed) report.add_calculated_channel(ch_rpm) - report.determine_report() - report.persist_results() + report.run() # Union across batches is complete: both channels' rows land in the one fact. fact = spark.read.table(_FACT) @@ -207,8 +205,7 @@ def test_incremental_unchanged_is_idempotent(spark): config=dict(_config(is_enabled=False)), ) _add_channel(r1, factor=3.6) - r1.determine_report() - r1.persist_results() + r1.run() count_before = spark.read.table(_FACT).count() r2 = Report( @@ -218,8 +215,7 @@ def test_incremental_unchanged_is_idempotent(spark): config=dict(_config(is_enabled=True)), ) _add_channel(r2, factor=3.6) - r2.determine_report() - r2.persist_results() + r2.run() # Idempotent: unchanged definition + unchanged silver → row count stable. assert spark.read.table(_FACT).count() == count_before @@ -243,8 +239,7 @@ def test_incremental_modified_container_fewer_rows_deletes_stale(spark): config=dict(_config(is_enabled=False)), ) ch = _add_channel(r1, factor=3.6) - r1.determine_report() - r1.persist_results() + r1.run() fact_pre = spark.read.table(_FACT).filter(F.col("channel_id") == ch.get_id()) c1_before = fact_pre.filter(F.col("container_id") == 1).count() @@ -295,8 +290,7 @@ def test_incremental_modified_container_fewer_rows_deletes_stale(spark): ) ch2 = _add_channel(r2, factor=3.6) assert ch2.get_id() == ch.get_id() # unchanged definition - r2.determine_report() - r2.persist_results() + r2.run() fact_post = spark.read.table(_FACT).filter(F.col("channel_id") == ch.get_id()) # Container 1 shrank to exactly 5 rows — no stale orphans survive. @@ -327,8 +321,7 @@ def test_incremental_changed_definition_replaces(spark): config=dict(_config(is_enabled=False, calculated_channels=cc)), ) ch1 = _add_channel(r1, factor=3.6) - r1.determine_report() - r1.persist_results() + r1.run() hash_before = ( spark.read.table(_DIM) .filter(F.col("channel_id") == ch1.get_id()) @@ -352,8 +345,7 @@ def test_incremental_changed_definition_replaces(spark): ) ch2 = _add_channel(r2, factor=3.7) assert ch2.get_id() == ch1.get_id() # same identity → same entity id - r2.determine_report() - r2.persist_results() + r2.run() dim = spark.read.table(_DIM) hash_after = ( @@ -383,8 +375,7 @@ def test_incremental_identity_reorder_does_not_reprocess(spark): config=dict(_config(is_enabled=False)), ) ch1 = _add_channel(r1, factor=3.6, identity={"channel_name": "speed_kmh", "data_key": "CALC"}) - r1.determine_report() - r1.persist_results() + r1.run() count_before = spark.read.table(_FACT).count() hash_before = ( spark.read.table(_DIM) @@ -403,8 +394,7 @@ def test_incremental_identity_reorder_does_not_reprocess(spark): ) ch2 = _add_channel(r2, factor=3.6, identity={"data_key": "CALC", "channel_name": "speed_kmh"}) assert ch2.get_id() == ch1.get_id() - r2.determine_report() - r2.persist_results() + r2.run() # Stable hash → classified unchanged → idempotent upsert, no row growth. hash_after = ( @@ -467,8 +457,7 @@ def test_channel_metrics_not_emitted_by_default(spark): config=dict(_config(is_enabled=False)), ) _add_channel(report, factor=3.6) - report.determine_report() - report.persist_results() + report.run() assert spark.catalog.tableExists(_FACT) assert not spark.catalog.tableExists(_METRICS) @@ -498,8 +487,7 @@ def test_channel_metrics_emitted_and_usable_as_impulse_source(spark): ) report.add_calculated_channel(ch) - report.determine_report() - report.persist_results() + report.run() assert spark.catalog.tableExists(_METRICS) metrics = spark.read.table(_METRICS) @@ -586,8 +574,7 @@ def test_channel_metrics_incremental_is_idempotent(spark): config=dict(_config(is_enabled=False, calculated_channels=cc)), ) _add_channel(r1, factor=3.6) - r1.determine_report() - r1.persist_results() + r1.run() count_before = spark.read.table(_METRICS).count() assert count_before > 0 @@ -598,8 +585,7 @@ def test_channel_metrics_incremental_is_idempotent(spark): config=dict(_config(is_enabled=True, calculated_channels=cc)), ) _add_channel(r2, factor=3.6) - r2.determine_report() - r2.persist_results() + r2.run() assert spark.read.table(_METRICS).count() == count_before @@ -620,8 +606,7 @@ def test_channel_metrics_custom_kpis(spark): ), ) _add_channel(report, factor=3.6) - report.determine_report() - report.persist_results() + report.run() metrics = spark.read.table(_METRICS) # Only the selected KPIs are present; the dropped defaults are absent. diff --git a/tests/impulse_reporting/integration/channel_mapping_resolution_dimension_test.py b/tests/impulse_reporting/integration/channel_mapping_resolution_dimension_test.py index 4467a0bf..613995ed 100644 --- a/tests/impulse_reporting/integration/channel_mapping_resolution_dimension_test.py +++ b/tests/impulse_reporting/integration/channel_mapping_resolution_dimension_test.py @@ -1,8 +1,8 @@ """Integration test for the channel_mapping_resolution_dimension gold table. Exercises the end-to-end flow: a Report configured with -``channel_mapping_table``, aggregations that use ``channel_with_alias``, -``determine_report()`` and ``persist_results()`` writing the new +``channel_mapping_table``, aggregations that use ``channel_with_alias``, and +``run()`` writing the new ``channel_mapping_resolution_dimension`` Delta table to the gold schema. """ @@ -26,8 +26,7 @@ def test_alias_report_writes_channel_mapping_resolution_dimension( weights=channels["weights"], ) - report.determine_report() - report.persist_results() + report.run() gold = spark.read.table("spark_catalog.gold.alias_int_channel_mapping_resolution_dimension") @@ -106,8 +105,7 @@ def test_incremental_added_alias_resolves_over_all_containers( HistogramDuration(name="rpm_hist", base_expr=channels1["engine_speed"], bins=rpm_bins) ) report1.add_page(page1) - report1.determine_report() - report1.persist_results() + report1.run() after_run1 = {(r.container_id, r.channel_alias) for r in spark.read.table(table).collect()} assert after_run1 == { @@ -127,8 +125,7 @@ def test_incremental_added_alias_resolves_over_all_containers( HistogramDuration(name="speed_hist", base_expr=channels2["vehicle_speed"], bins=speed_bins) ) report2.add_page(page2) - report2.determine_report() - report2.persist_results() + report2.run() gold = spark.read.table(table) after_run2 = {(r.container_id, r.channel_alias) for r in gold.collect()} diff --git a/tests/impulse_reporting/integration/container_and_sequence_event_test.py b/tests/impulse_reporting/integration/container_and_sequence_event_test.py index 1b443d38..4ec096d5 100644 --- a/tests/impulse_reporting/integration/container_and_sequence_event_test.py +++ b/tests/impulse_reporting/integration/container_and_sequence_event_test.py @@ -74,7 +74,7 @@ def test_container_event_in_report(spark, basic_narrow_db): my_report.add_event(container_evt) # Determine report content - my_report.determine_report() + my_report.run(persist_results=False) # Access event DataFrames event_dfs = my_report.event_dfs @@ -187,8 +187,7 @@ def test_container_event_with_basic_event(spark, basic_narrow_db): my_report.add_event(basic_evt) # Determine report content - my_report.determine_report() - my_report.persist_results() + my_report.run() event_dfs = my_report.event_dfs # Both event types should exist @@ -270,7 +269,7 @@ def test_sequence_of_events_without_max_overlap_in_report(spark, basic_narrow_db ) my_report.add_event(sequence_evt) - my_report.determine_report() + my_report.run(persist_results=False) event_dfs = my_report.event_dfs diff --git a/tests/impulse_reporting/integration/container_event_test.py b/tests/impulse_reporting/integration/container_event_test.py index 8c89e55b..022bd737 100644 --- a/tests/impulse_reporting/integration/container_event_test.py +++ b/tests/impulse_reporting/integration/container_event_test.py @@ -74,7 +74,7 @@ def test_container_event_in_report(spark, basic_narrow_db): my_report.add_event(container_evt) # Determine report content - my_report.determine_report() + my_report.run(persist_results=False) # Access event DataFrames event_dfs = my_report.event_dfs @@ -185,8 +185,7 @@ def test_container_event_with_basic_event(spark, basic_narrow_db): my_report.add_event(basic_evt) # Determine report content - my_report.determine_report() - my_report.persist_results() + my_report.run() event_dfs = my_report.event_dfs # Both event types should exist @@ -330,7 +329,7 @@ def test_container_event_with_remapped_silver_timestamps(spark, basic_narrow_db) config=dict(impulse_config), ) my_report.add_event(ContainerEvent(name="full_measurement")) - my_report.determine_report() + my_report.run(persist_results=False) event_rows = my_report.event_dfs["CONTAINER_EVENT"]["changed"].collect() assert len(event_rows) == 3 diff --git a/tests/impulse_reporting/integration/custom_statistics_test.py b/tests/impulse_reporting/integration/custom_statistics_test.py index d48adbec..893b6005 100644 --- a/tests/impulse_reporting/integration/custom_statistics_test.py +++ b/tests/impulse_reporting/integration/custom_statistics_test.py @@ -100,7 +100,7 @@ def test_custom_statistics_report(spark): ) page.add_aggregation(stats) - my_report.determine_report() + my_report.run(persist_results=False) assert "STATS_AGGREGATOR" in my_report.aggregation_dfs stats_df = my_report.aggregation_dfs["STATS_AGGREGATOR"]["changed"] @@ -198,7 +198,7 @@ def test_multi_output_custom_statistics_report(spark): ) page.add_aggregation(stats) - my_report.determine_report() + my_report.run(persist_results=False) stats_df = my_report.aggregation_dfs["STATS_AGGREGATOR"]["changed"] rows = stats_df.collect() diff --git a/tests/impulse_reporting/integration/incremental_container_id_mapping_test.py b/tests/impulse_reporting/integration/incremental_container_id_mapping_test.py index 221f6c5d..d371c4e8 100644 --- a/tests/impulse_reporting/integration/incremental_container_id_mapping_test.py +++ b/tests/impulse_reporting/integration/incremental_container_id_mapping_test.py @@ -71,8 +71,7 @@ def _run(spark, container_metrics_table, *, is_enabled): # noqa: F811 config=_config(container_metrics_table, is_enabled=is_enabled), ) add_aggs_to_report(report) - report.determine_report() - report.persist_results() + report.run() def _container_ids(spark): # noqa: F811 diff --git a/tests/impulse_reporting/integration/incremental_report_test.py b/tests/impulse_reporting/integration/incremental_report_test.py index 6aa2a900..2c1cdf13 100644 --- a/tests/impulse_reporting/integration/incremental_report_test.py +++ b/tests/impulse_reporting/integration/incremental_report_test.py @@ -62,9 +62,7 @@ def test_incremental_in_report_case_1(spark): add_aggs_to_report(my_report) # Determine content of all pages - my_report.determine_report() - - my_report.persist_results() + my_report.run() meas_df_pre = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") @@ -80,9 +78,7 @@ def test_incremental_in_report_case_1(spark): add_aggs_to_report(my_report) # Determine content of all pages - my_report.determine_report() - - my_report.persist_results() + my_report.run() meas_df_post_case_1 = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") @@ -104,9 +100,7 @@ def test_incremental_in_report_case_2(spark): add_aggs_to_report(my_report) # # Determine content of all pages - my_report.determine_report() - - my_report.persist_results() + my_report.run() meas_df_post_case_2 = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") @@ -124,8 +118,7 @@ def test_incremental_in_report_case_3(spark): ) add_aggs_to_report(my_report) - my_report.determine_report() - my_report.persist_results() + my_report.run() # Save state before case 3 for comparison meas_df_pre_case3 = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") @@ -172,8 +165,7 @@ def test_incremental_in_report_case_3(spark): config=dict(set_config("container_metrics_inc_1_2_modified", True)), ) add_aggs_to_report(my_report) - my_report.determine_report() - my_report.persist_results() + my_report.run() meas_df_post_case_3 = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") assert meas_df_post_case_3.count() == 2 @@ -226,9 +218,7 @@ def test_incremental_in_report_case_4(spark): add_aggs_to_report(my_report) # # Determine content of all pages - my_report.determine_report() - - my_report.persist_results() + my_report.run() # Save histogram dimension and facts before the definition change hist_dim_pre_case4 = spark.read.table("spark_catalog.gold.evaluation_histogram_dimension") @@ -288,8 +278,7 @@ def test_incremental_in_report_case_4(spark): ) add_aggs_to_report_changed_bins(my_report) - my_report.determine_report() - my_report.persist_results() + my_report.run() meas_df_post_case_4 = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") assert meas_df_post_case_4.count() == 2 # No container changes @@ -356,8 +345,7 @@ def test_incremental_in_report_case_5(spark): add_aggs_to_report(my_report) - my_report.determine_report() - my_report.persist_results() + my_report.run() meas_df_pre_case_5 = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") assert meas_df_pre_case_5.count() == 2 @@ -408,8 +396,7 @@ def test_incremental_in_report_case_5(spark): ) add_aggs_to_report_changed_bins(my_report) - my_report.determine_report() - my_report.persist_results() + my_report.run() meas_df_post_case_5 = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") assert meas_df_post_case_5.count() == 3 # Container 3 added @@ -492,8 +479,7 @@ def test_incremental_scoped_full_recalculation(spark): config=dict(set_config("container_metrics_inc_1_2", False)), ) add_aggs_to_report(report_1) - report_1.determine_report() - report_1.persist_results() + report_1.run() hist_dim = spark.read.table("spark_catalog.gold.evaluation_histogram_dimension") rpm_vid = hist_dim.where(F.col("name") == "rpm_hist_p1").select("visual_id").collect()[0][0] @@ -564,8 +550,7 @@ def _rpm_sum(df, container_id): config=dict(config_2), ) add_aggs_to_report(report_2) - report_2.determine_report() - report_2.persist_results() + report_2.run() hist_fact_post = spark.read.table("spark_catalog.gold.evaluation_histogram_fact") @@ -745,8 +730,7 @@ def test_incremental_cross_type_event_definition_change(spark): ) page.add_aggregation(hist) - report_1.determine_report() - report_1.persist_results() + report_1.run() event_fact_run1 = spark.read.table("spark_catalog.gold.evaluation_event_instance_fact") basic_event_id = basic_event.get_id() @@ -785,8 +769,7 @@ def test_incremental_cross_type_event_definition_change(spark): ) page2.add_aggregation(hist2) - report_2.determine_report() - report_2.persist_results() + report_2.run() event_fact_run2 = spark.read.table("spark_catalog.gold.evaluation_event_instance_fact") @@ -831,8 +814,7 @@ def test_incremental_long_seeded_event_fact_accepts_double_event(spark): event=pit_event, ) ) - report_1.determine_report() - report_1.persist_results() + report_1.run() event_fact_run1 = spark.read.table("spark_catalog.gold.evaluation_event_instance_fact") pit_event_id = pit_event.get_id() @@ -877,8 +859,7 @@ def test_incremental_long_seeded_event_fact_accepts_double_event(spark): event=basic_event, ) ) - report_2.determine_report() - report_2.persist_results() + report_2.run() event_fact_run2 = spark.read.table("spark_catalog.gold.evaluation_event_instance_fact") basic_event_id = basic_event.get_id() @@ -1008,8 +989,7 @@ def test_incremental_modified_container_multi_family_shrink_deletes_stale(spark) ), ) calc, event, stats = _add_multi_family(r1) - r1.determine_report() - r1.persist_results() + r1.run() # Baseline container-1 counts (exact) + byte-identical snapshots of 2 & 3. assert _rows(spark, _CC_FACT, "channel_id", calc.get_id(), 1) == 50 @@ -1057,8 +1037,7 @@ def test_incremental_modified_container_multi_family_shrink_deletes_stale(spark) assert calc2.get_id() == calc.get_id() assert event2.get_id() == event.get_id() assert stats2.get_id() == stats.get_id() - r2.determine_report() - r2.persist_results() + r2.run() # Container 1 shrank on every fact table — no stale orphans survive. assert _rows(spark, _CC_FACT, "channel_id", calc.get_id(), 1) == 5 diff --git a/tests/impulse_reporting/integration/max_containers_per_batch_test.py b/tests/impulse_reporting/integration/max_containers_per_batch_test.py index 173eb428..50c88905 100644 --- a/tests/impulse_reporting/integration/max_containers_per_batch_test.py +++ b/tests/impulse_reporting/integration/max_containers_per_batch_test.py @@ -130,7 +130,7 @@ def _make_report( def _run(spark, silver_table, prefix, *, max_containers_per_batch=None, add_aggs=_add_light_aggs): - """Run ONE determine+persist pass (single batch) — the granular API, no run() loop.""" + """Run ONE determine+persist pass (single batch) via the private steps, no run() loop.""" report = _make_report( spark, silver_table, @@ -138,8 +138,8 @@ def _run(spark, silver_table, prefix, *, max_containers_per_batch=None, add_aggs max_containers_per_batch=max_containers_per_batch, add_aggs=add_aggs, ) - report.determine_report() - report.persist_results() + report._determine_report() + report._persist_results() def _container_ids(spark, prefix): @@ -269,8 +269,7 @@ def test_full_mode_container_chunked_solve_matches_uncapped(spark, uncapped_base config=dict(config), ) _add_light_aggs(report) - report.determine_report() - report.persist_results() + report.run() assert _container_ids(spark, "fullcap") == [ 1, @@ -302,7 +301,7 @@ def _sinkless_hist(spark, *, max_containers_per_batch): ), ) _add_light_aggs(report) - report.determine_report() + report.run(persist_results=False) return report.aggregation_dfs["HISTOGRAM"]["changed"] diff --git a/tests/impulse_reporting/integration/point_value_aggregator_test.py b/tests/impulse_reporting/integration/point_value_aggregator_test.py index 3f8e17b4..b50ae063 100644 --- a/tests/impulse_reporting/integration/point_value_aggregator_test.py +++ b/tests/impulse_reporting/integration/point_value_aggregator_test.py @@ -70,14 +70,12 @@ def test_persist_point_value_aggregator(spark): ) page.add_aggregation(pva) - my_report.determine_report() + my_report.run() assert "POINT_VALUE_AGGREGATOR" in my_report.aggregation_dfs assert my_report.aggregation_dfs["POINT_VALUE_AGGREGATOR"]["changed"].count() > 0 assert my_report.aggregation_metadata_dfs["POINT_VALUE_AGGREGATOR"].count() == 1 - my_report.persist_results() - fact = spark.read.table("spark_catalog.gold.evaluation_stats_aggregator_fact") dim = spark.read.table("spark_catalog.gold.evaluation_stats_aggregator_dimension") event_fact = spark.read.table("spark_catalog.gold.evaluation_event_instance_fact") @@ -158,8 +156,7 @@ def test_stats_and_point_value_share_fact_table_without_clobber(spark): page.add_aggregation(stats) page.add_aggregation(pva) - my_report.determine_report() - my_report.persist_results() + my_report.run() fact = spark.read.table("spark_catalog.gold.evaluation_stats_aggregator_fact") dim = spark.read.table("spark_catalog.gold.evaluation_stats_aggregator_dimension") @@ -245,8 +242,7 @@ def test_incremental_stats_and_point_value_shared_fact_changed_defs(spark): config=dict(_incremental_config(is_enabled=False)), ) stats_1, pva_1 = _add_stats_and_pva(report_1, statistics=["min", "max"], pva_scale=1.0) - report_1.determine_report() - report_1.persist_results() + report_1.run() fact_run1 = spark.read.table("spark_catalog.gold.evaluation_stats_aggregator_fact") assert fact_run1.where(F.col("visual_id") == stats_1.get_id()).count() > 0 @@ -265,8 +261,7 @@ def test_incremental_stats_and_point_value_shared_fact_changed_defs(spark): assert stats_2.get_id() == stats_1.get_id() assert pva_2.get_id() == pva_1.get_id() - report_2.determine_report() - report_2.persist_results() + report_2.run() fact_run2 = spark.read.table("spark_catalog.gold.evaluation_stats_aggregator_fact") stats_rows = fact_run2.where(F.col("visual_id") == stats_2.get_id()) diff --git a/tests/impulse_reporting/integration/points_in_time_event_test.py b/tests/impulse_reporting/integration/points_in_time_event_test.py index 1ee64e74..77a675b9 100644 --- a/tests/impulse_reporting/integration/points_in_time_event_test.py +++ b/tests/impulse_reporting/integration/points_in_time_event_test.py @@ -30,15 +30,13 @@ def test_persist_points_in_time_event(spark): ) my_report.add_event(pit_event) - my_report.determine_report() + my_report.run() event_dfs = my_report.event_dfs event_metadata_dfs = my_report.event_metadata_dfs assert event_dfs["POINTS_IN_TIME_EVENT"]["changed"].count() > 0 assert event_metadata_dfs["POINTS_IN_TIME_EVENT"].count() > 0 - my_report.persist_results() - assert spark.catalog.tableExists("spark_catalog.gold.evaluation_event_instance_fact") assert spark.catalog.tableExists("spark_catalog.gold.evaluation_event_dimension") diff --git a/tests/impulse_reporting/integration/raw_encoder_test.py b/tests/impulse_reporting/integration/raw_encoder_test.py index 712997d1..29c5704c 100644 --- a/tests/impulse_reporting/integration/raw_encoder_test.py +++ b/tests/impulse_reporting/integration/raw_encoder_test.py @@ -89,7 +89,7 @@ def _run_histogram_report( ) ) - report.determine_report() + report.run(persist_results=False) return report.aggregation_dfs["HISTOGRAM"]["changed"] diff --git a/tests/impulse_reporting/integration/simple_report_test.py b/tests/impulse_reporting/integration/simple_report_test.py index 68e5e0af..d65bc433 100644 --- a/tests/impulse_reporting/integration/simple_report_test.py +++ b/tests/impulse_reporting/integration/simple_report_test.py @@ -152,7 +152,7 @@ def test_simple_report1(spark, setup_narrow_db): my_first_page.add_aggregation(hist5) # Determine content of all pages - my_report.determine_report() + my_report.run(persist_results=False) visual_dfs = my_report.aggregation_dfs assert visual_dfs["HISTOGRAM"]["changed"].filter(F.col("hist_value") > 0).count() > 0 assert visual_dfs["HISTOGRAM2D"]["changed"].filter(F.col("hist_value") > 0).count() > 0 @@ -254,7 +254,7 @@ def test_basic_narrow_report_no_uuts(spark): my_first_page.add_aggregation(hist5) # Determine content of all pages - my_report.determine_report() + my_report.run(persist_results=False) agg_dfs = my_report.aggregation_dfs assert agg_dfs["HISTOGRAM"]["changed"].filter(F.col("hist_value") > 0).count() > 0 assert agg_dfs["HISTOGRAM2D"]["changed"].filter(F.col("hist_value") > 0).count() > 0 @@ -345,7 +345,7 @@ def test_basic_narrow_report(spark): my_first_page.add_aggregation(hist5) # Determine content of all pages - my_report.determine_report() + my_report.run(persist_results=False) agg_dfs = my_report.aggregation_dfs assert agg_dfs["HISTOGRAM"]["changed"].filter(F.col("hist_value") > 0).count() > 0 assert agg_dfs["HISTOGRAM2D"]["changed"].filter(F.col("hist_value") > 0).count() > 0 @@ -450,7 +450,7 @@ def test_report_with_events(spark): my_first_page.add_aggregation(hist5) # Determine content of all pages - my_report.determine_report() + my_report.run(persist_results=False) agg_dfs = my_report.aggregation_dfs agg_metadata_dfs = my_report.aggregation_metadata_dfs event_dfs = my_report.event_dfs @@ -552,11 +552,8 @@ def test_persist_report(spark): ) my_first_page.add_aggregation(hist5) - # Determine content of all pages - my_report.determine_report() - - # Persist the report - my_report.persist_results() + # Determine and persist the report + my_report.run() assert spark.catalog.tableExists("spark_catalog.gold.evaluation_histogram_fact") assert spark.catalog.tableExists("spark_catalog.gold.evaluation_histogram2d_fact") @@ -689,8 +686,8 @@ def test_persist_report_with_events(spark): ) my_first_page.add_aggregation(hist5) - # Determine content of all pages - my_report.determine_report() + # Determine and persist the report + my_report.run() agg_dfs = my_report.aggregation_dfs agg_metadata_dfs = my_report.aggregation_metadata_dfs @@ -706,8 +703,6 @@ def test_persist_report_with_events(spark): assert event_dfs["BASIC_EVENT"]["changed"].count() > 0 assert event_metadata_dfs["BASIC_EVENT"].count() > 0 - my_report.persist_results() - assert spark.catalog.tableExists("spark_catalog.gold.evaluation_event_dimension") assert spark.catalog.tableExists("spark_catalog.gold.evaluation_event_instance_fact") assert spark.catalog.tableExists("spark_catalog.gold.evaluation_histogram_dimension") @@ -819,7 +814,7 @@ def test_simple_report_key_value_store(spark, key_value_store_db): my_first_page.add_aggregation(hist2d) # --- Determine content of all pages --- - my_report.determine_report() + my_report.run(persist_results=False) visual_dfs = my_report.aggregation_dfs assert visual_dfs["HISTOGRAM"]["changed"].filter(F.col("hist_value") > 0).count() > 0 assert visual_dfs["HISTOGRAM2D"]["changed"].filter(F.col("hist_value") > 0).count() > 0 diff --git a/tests/impulse_reporting/integration/statistics_test.py b/tests/impulse_reporting/integration/statistics_test.py index 024af3fb..a58673c2 100644 --- a/tests/impulse_reporting/integration/statistics_test.py +++ b/tests/impulse_reporting/integration/statistics_test.py @@ -63,7 +63,7 @@ def test_statistics_basic_report(spark): my_first_page.add_aggregation(stats2) # Determine content of all pages - my_report.determine_report() + my_report.run(persist_results=False) agg_dfs = my_report.aggregation_dfs agg_metadata_dfs = my_report.aggregation_metadata_dfs @@ -183,7 +183,7 @@ def test_statistics_with_multiple_selections(spark): my_first_page.add_aggregation(multi_stats) # Determine content of all pages - my_report.determine_report() + my_report.run(persist_results=False) agg_dfs = my_report.aggregation_dfs agg_metadata_dfs = my_report.aggregation_metadata_dfs @@ -250,7 +250,7 @@ def test_statistics_mixed_with_histogram(spark): my_first_page.add_aggregation(rpm_hist) # Determine content of all pages - my_report.determine_report() + my_report.run(persist_results=False) agg_dfs = my_report.aggregation_dfs agg_metadata_dfs = my_report.aggregation_metadata_dfs @@ -305,11 +305,8 @@ def test_persist_statistics_report(spark): ) my_first_page.add_aggregation(stats) - # Determine content of all pages - my_report.determine_report() - - # Persist the report - my_report.persist_results() + # Determine and persist the report + my_report.run() # Verify tables exist assert spark.catalog.tableExists("spark_catalog.gold.evaluation_stats_aggregator_fact") @@ -377,11 +374,8 @@ def test_persist_statistics_with_events(spark): ) my_first_page.add_aggregation(stats) - # Determine content of all pages - my_report.determine_report() - - # Persist the report - my_report.persist_results() + # Determine and persist the report + my_report.run() # Verify all expected tables exist assert spark.catalog.tableExists("spark_catalog.gold.evaluation_stats_aggregator_fact") diff --git a/tests/impulse_reporting/integration/string_container_id_report_test.py b/tests/impulse_reporting/integration/string_container_id_report_test.py index b473e034..28133d91 100644 --- a/tests/impulse_reporting/integration/string_container_id_report_test.py +++ b/tests/impulse_reporting/integration/string_container_id_report_test.py @@ -90,10 +90,9 @@ def test_persist_report_string_container_id(spark, setup_string_cid_db): ) ) - report.determine_report() - report.persist_results() + report.run() # Second persist merges string-keyed rows into the existing gold tables (Delta MERGE). - report.persist_results() + report._persist_results() histogram_fact = spark.read.table("spark_catalog.gold.evaluation_histogram_fact") measurement_dimension = spark.read.table("spark_catalog.gold.evaluation_measurement_dimension") diff --git a/tests/impulse_reporting/integration/time_window_event_test.py b/tests/impulse_reporting/integration/time_window_event_test.py index cee850e5..ac7ce8c5 100644 --- a/tests/impulse_reporting/integration/time_window_event_test.py +++ b/tests/impulse_reporting/integration/time_window_event_test.py @@ -97,7 +97,7 @@ def test_time_window_event_in_report(spark, basic_narrow_db): ) my_report.add_event(window_evt) - my_report.determine_report() + my_report.run(persist_results=False) event_dfs = my_report.event_dfs assert "TIME_WINDOW_EVENT" in event_dfs @@ -430,8 +430,7 @@ def test_time_window_event_aggregation_join(spark, setup_tw_aligned_db): my_report.add_page(page) page.add_aggregation(_rpm_stats(my_report, window_evt)) - my_report.determine_report() - my_report.persist_results() + my_report.run() _assert_ids_join(spark, table_prefix) _assert_window_stats_match_samples(spark, schema, table_prefix) @@ -490,8 +489,7 @@ def test_time_window_event_aggregation_join_raw(spark, setup_tw_aligned_db, raw_ my_report.add_page(page) page.add_aggregation(_rpm_stats(my_report, window_evt)) - my_report.determine_report() - my_report.persist_results() + my_report.run() _assert_ids_join(spark, table_prefix) _assert_window_stats_match_samples(spark, schema, table_prefix, channels_table) @@ -553,8 +551,7 @@ def test_multiple_time_window_events_coexist(spark, setup_tw_aligned_db): page.add_aggregation(_rpm_stats(my_report, evt_short, name="rpm_stats_ten_min")) page.add_aggregation(_rpm_stats(my_report, evt_long, name="rpm_stats_thirty_min")) - my_report.determine_report() - my_report.persist_results() + my_report.run() event_fact = spark.read.table(f"spark_catalog.gold.{table_prefix}_event_instance_fact") event_rows = event_fact.collect() @@ -643,7 +640,7 @@ def test_time_window_event_covers_containers_without_aggregated_channel( my_report.add_page(page) page.add_aggregation(_rpm_stats(my_report, window_evt)) - my_report.determine_report() + my_report.run(persist_results=False) rows = my_report.event_dfs["TIME_WINDOW_EVENT"]["changed"].collect() _assert_windows_for_all_containers(rows) @@ -667,7 +664,7 @@ def test_standalone_time_window_event_covers_all_containers(spark, basic_narrow_ ) my_report.add_event(TimeWindowEvent(name="ten_sec", window_length=WINDOW_LENGTH)) - my_report.determine_report() + my_report.run(persist_results=False) rows = my_report.event_dfs["TIME_WINDOW_EVENT"]["changed"].collect() _assert_windows_for_all_containers(rows) @@ -713,8 +710,7 @@ def _run(cm_table: str, is_incremental: bool, statistics) -> None: page = Page(page_number=1) report.add_page(page) page.add_aggregation(_rpm_stats(report, window_evt, statistics)) - report.determine_report() - report.persist_results() + report.run() _run(cm_run_1, is_incremental=False, statistics=("min", "max", "mean")) @@ -762,7 +758,7 @@ def test_time_window_event_timestamp_boundaries_require_channel_time_unit( my_report.add_event(TimeWindowEvent(name="ten_min", window_length=window_length)) with pytest.raises(ValueError, match=r"TimeWindowEvent.*channel_time_unit"): - my_report.determine_report() + my_report.run(persist_results=False) @pytest.mark.parametrize( @@ -786,8 +782,7 @@ def test_container_event_timestamp_boundaries(spark, setup_tw_aligned_db, channe config=_aligned_config(schema, table_prefix, channel_time=channel_time), ) my_report.add_event(ContainerEvent(name="full_container")) - my_report.determine_report() - my_report.persist_results() + my_report.run() expected = { r.container_id: (r.s, r.e) @@ -854,8 +849,7 @@ def _run(cm_table: str, unit: str, is_incremental: bool): report.add_page(page) stats = _rpm_stats(report, window_evt) page.add_aggregation(stats) - report.determine_report() - report.persist_results() + report.run() return report, window_evt, container_evt, stats def _event_rows(event_id: int): @@ -941,8 +935,7 @@ def test_udf_reads_absolute_start_ts_next_to_relative_time_window_event( event=window_evt, ) page.add_aggregation(absolute_start) - report.determine_report() - report.persist_results() + report.run() finally: spark.conf.set("spark.sql.session.timeZone", previous_tz) diff --git a/tests/impulse_reporting/unit/core/report_incremental_test.py b/tests/impulse_reporting/unit/core/report_incremental_test.py index f3ca4663..fe89c2b0 100644 --- a/tests/impulse_reporting/unit/core/report_incremental_test.py +++ b/tests/impulse_reporting/unit/core/report_incremental_test.py @@ -1,11 +1,13 @@ """ -Unit tests for Report.determine_report() and Report.persist_results() -with is_incremental=True and is_incremental=False. +Unit tests for Report._determine_report() and Report._persist_results() +with is_incremental=True and is_incremental=False, plus the deprecated public +determine_report() / persist_results() wrappers. Tests use mocking to isolate the Report class logic from downstream Spark processing (event/aggregation determination, persistence). """ +import warnings from unittest.mock import MagicMock, create_autospec, patch import pyspark.sql.functions as F @@ -85,10 +87,10 @@ def _build_report(spark, config_dict=None): # ============================================================================ -# Tests: determine_report with is_incremental=False +# Tests: _determine_report with is_incremental=False # ============================================================================ class TestDetermineReportFullMode: - """Tests for determine_report with is_incremental=False (full processing).""" + """Tests for _determine_report with is_incremental=False (full processing).""" @pytest.fixture(autouse=True) def mock_gold_exists(self): @@ -109,7 +111,7 @@ def test_full_mode_does_not_call_container_detection(self, spark): return_value=None, ), ): - report.determine_report(is_incremental=False) + report._determine_report(is_incremental=False) mock_detect.assert_not_called() @@ -126,12 +128,12 @@ def test_full_mode_does_not_use_hash_comparator(self, spark): ), patch("impulse_reporting.core.report.DefinitionHashComparator") as mock_hash_cls, ): - report.determine_report(is_incremental=False) + report._determine_report(is_incremental=False) mock_hash_cls.assert_called() def test_full_mode_sets_is_incremental_false(self, spark): - """After determine_report(is_incremental=False), _is_incremental is False.""" + """After _determine_report(is_incremental=False), _is_incremental is False.""" report = _build_report(spark) with ( @@ -142,7 +144,7 @@ def test_full_mode_sets_is_incremental_false(self, spark): return_value=None, ), ): - report.determine_report(is_incremental=False) + report._determine_report(is_incremental=False) assert report._is_incremental is False assert report._changed_event_ids == {} @@ -161,16 +163,16 @@ def test_full_mode_uses_config_default_when_none(self, spark): return_value=None, ), ): - report.determine_report(is_incremental=None) + report._determine_report(is_incremental=None) mock_detect.assert_not_called() # ============================================================================ -# Tests: determine_report with is_incremental=True +# Tests: _determine_report with is_incremental=True # ============================================================================ class TestDetermineReportIncrementalMode: - """Tests for determine_report with is_incremental=True.""" + """Tests for _determine_report with is_incremental=True.""" @pytest.fixture(autouse=True) def mock_gold_exists(self): @@ -195,12 +197,12 @@ def test_is_incremental_calls_container_detection(self, spark): ), patch("impulse_reporting.core.report.DefinitionHashComparator"), ): - report.determine_report(is_incremental=True) + report._determine_report(is_incremental=True) mock_detect.assert_called_once() def test_incremental_sets_is_incremental_true(self, spark): - """After determine_report(is_incremental=True), _is_incremental is True.""" + """After _determine_report(is_incremental=True), _is_incremental is True.""" report = _build_report(spark, INCREMENTAL_CONFIG) with ( @@ -217,7 +219,7 @@ def test_incremental_sets_is_incremental_true(self, spark): ), patch("impulse_reporting.core.report.DefinitionHashComparator"), ): - report.determine_report(is_incremental=True) + report._determine_report(is_incremental=True) assert report._is_incremental is True def test_incremental_with_hash_optimization_creates_comparator(self, spark): @@ -235,7 +237,7 @@ def test_incremental_with_hash_optimization_creates_comparator(self, spark): ), patch("impulse_reporting.core.report.DefinitionHashComparator") as mock_hash_cls, ): - report.determine_report(is_incremental=True) + report._determine_report(is_incremental=True) mock_hash_cls.assert_called_once_with(spark) @@ -257,12 +259,12 @@ def test_incremental_uses_config_when_mode_is_none(self, spark): ), patch("impulse_reporting.core.report.DefinitionHashComparator"), ): - report.determine_report(is_incremental=None) + report._determine_report(is_incremental=None) mock_detect.assert_called_once() def test_persist_full_calls_persist_full_method(self, spark): - """persist_results() calls _persist_full() when _is_incremental=False.""" + """_persist_results() calls _persist_full() when _is_incremental=False.""" report = _build_report(spark) report._is_incremental = False @@ -270,34 +272,34 @@ def test_persist_full_calls_persist_full_method(self, spark): patch.object(report, "_persist_full") as mock_full, patch.object(report, "_persist_incremental") as mock_incr, ): - report.persist_results() + report._persist_results() mock_full.assert_called_once() mock_incr.assert_not_called() def test_persist_full_default_mode(self, spark): - """persist_results() without tracked state defaults to _persist_full().""" + """_persist_results() without tracked state defaults to _persist_full().""" report = _build_report(spark) # report._is_incremental= None with ( patch.object(report, "_persist_full") as mock_full, patch.object(report, "_persist_incremental") as mock_incr, ): - # report.determine_report() - report.persist_results() + # report._determine_report() + report._persist_results() mock_full.assert_called_once() mock_incr.assert_not_called() # ============================================================================ -# Tests: persist_results with is_incremental=True +# Tests: _persist_results with is_incremental=True # ============================================================================ class TestPersistResultsIncrementalMode: - """Tests for persist_results when _is_incremental is True.""" + """Tests for _persist_results when _is_incremental is True.""" def test_persist_incremental_calls_persist_incremental_method(self, spark): - """persist_results() calls _persist_incremental() when _is_incremental=True.""" + """_persist_results() calls _persist_incremental() when _is_incremental=True.""" report = _build_report(spark) report._is_incremental = True report._changed_aggregation_ids = {"HISTOGRAM": [1]} @@ -308,7 +310,7 @@ def test_persist_incremental_calls_persist_incremental_method(self, spark): patch.object(report, "_persist_full") as mock_full, patch.object(report, "_persist_incremental") as mock_incr, ): - report.persist_results() + report._persist_results() mock_incr.assert_called_once_with( {"HISTOGRAM": [1]}, {"BASIC_EVENT": [2]}, {"CALCULATED_CHANNEL": [3]} @@ -316,10 +318,10 @@ def test_persist_incremental_calls_persist_incremental_method(self, spark): mock_full.assert_not_called() def test_persist_uses_tracked_state_from_determine_report(self, spark): - """persist_results() uses _is_incremental and changed IDs from determine_report().""" + """_persist_results() uses _is_incremental and changed IDs from _determine_report().""" report = _build_report(spark) - # Simulate state set by determine_report + # Simulate state set by _determine_report report._is_incremental = True report._changed_aggregation_ids = {"HISTOGRAM": [100]} report._changed_event_ids = {"BASIC_EVENT": [200]} @@ -330,7 +332,7 @@ def test_persist_uses_tracked_state_from_determine_report(self, spark): patch.object(report, "_persist_incremental") as mock_incr, ): # Call with defaults - should use tracked state - report.persist_results() + report._persist_results() mock_incr.assert_called_once_with( {"HISTOGRAM": [100]}, {"BASIC_EVENT": [200]}, {"CALCULATED_CHANNEL": [300]} @@ -339,10 +341,10 @@ def test_persist_uses_tracked_state_from_determine_report(self, spark): # ============================================================================ -# Tests: determine_report uses _resolve_is_incremental +# Tests: _determine_report uses _resolve_is_incremental # ============================================================================ class TestDetermineReportProcessingMode: - """Tests for determine_report respecting config and gold layer.""" + """Tests for _determine_report respecting config and gold layer.""" def test_determine_report_config_enabled_no_gold(self, spark): """Config enabled=True but no gold → _is_incremental=False.""" @@ -357,7 +359,7 @@ def test_determine_report_config_enabled_no_gold(self, spark): return_value=None, ), ): - report.determine_report() + report._determine_report() assert report._is_incremental is False @@ -374,7 +376,7 @@ def test_determine_report_config_disabled(self, spark): return_value=None, ), ): - report.determine_report(is_incremental=True) + report._determine_report(is_incremental=True) assert report._is_incremental is False @@ -392,7 +394,7 @@ def test_config_overrides_signature_true(self, spark): return_value=None, ), ): - report.determine_report(is_incremental=False) + report._determine_report(is_incremental=False) assert report._is_incremental is True @@ -409,7 +411,7 @@ def test_config_overrides_signature_false(self, spark): return_value=None, ), ): - report.determine_report(is_incremental=True) + report._determine_report(is_incremental=True) assert report._is_incremental is False @@ -427,7 +429,7 @@ def test_no_config_signature_true_gold_exists(self, spark): return_value=None, ), ): - report.determine_report(is_incremental=True) + report._determine_report(is_incremental=True) assert report._is_incremental is True @@ -444,7 +446,7 @@ def test_no_config_signature_false(self, spark): return_value=None, ), ): - report.determine_report(is_incremental=False) + report._determine_report(is_incremental=False) assert report._is_incremental is False @@ -461,7 +463,7 @@ def test_no_config_no_signature_defaults_to_full(self, spark): return_value=None, ), ): - report.determine_report() + report._determine_report() assert report._is_incremental is False @@ -479,19 +481,19 @@ def test_detect_upserted_returns_none_falls_back_to_full(self, spark): return_value=None, ), ): - report.determine_report() + report._determine_report() assert report._is_incremental is True # ============================================================================ -# Tests: persist_results with config-driven processing mode +# Tests: _persist_results with config-driven processing mode # ============================================================================ class TestPersistResultsProcessingMode: - """Tests for persist_results respecting config set by determine_report.""" + """Tests for _persist_results respecting config set by _determine_report.""" def test_persist_full_when_config_disabled(self, spark): - """persist_results calls _persist_full when config enabled=False.""" + """_persist_results calls _persist_full when config enabled=False.""" report = _build_report(spark, config_dict=INCREMENTAL_DISABLED_CONFIG) with ( @@ -503,19 +505,19 @@ def test_persist_full_when_config_disabled(self, spark): return_value=None, ), ): - report.determine_report() + report._determine_report() with ( patch.object(report, "_persist_full") as mock_full, patch.object(report, "_persist_incremental") as mock_incr, ): - report.persist_results() + report._persist_results() mock_full.assert_called_once() mock_incr.assert_not_called() def test_persist_incremental_when_config_enabled(self, spark): - """persist_results calls _persist_incremental when config enabled=True and gold exists.""" + """_persist_results calls _persist_incremental when config enabled=True and gold exists.""" report = _build_report(spark, config_dict=INCREMENTAL_CONFIG) with ( @@ -528,19 +530,19 @@ def test_persist_incremental_when_config_enabled(self, spark): return_value=None, ), ): - report.determine_report() + report._determine_report() with ( patch.object(report, "_persist_full") as mock_full, patch.object(report, "_persist_incremental") as mock_incr, ): - report.persist_results() + report._persist_results() mock_incr.assert_called_once() mock_full.assert_not_called() def test_persist_full_when_no_gold_layer(self, spark): - """persist_results calls _persist_full when no gold layer exists.""" + """_persist_results calls _persist_full when no gold layer exists.""" report = _build_report(spark, config_dict=INCREMENTAL_CONFIG) with ( @@ -552,19 +554,19 @@ def test_persist_full_when_no_gold_layer(self, spark): return_value=None, ), ): - report.determine_report() + report._determine_report() with ( patch.object(report, "_persist_full") as mock_full, patch.object(report, "_persist_incremental") as mock_incr, ): - report.persist_results() + report._persist_results() mock_full.assert_called_once() mock_incr.assert_not_called() def test_persist_full_when_no_config(self, spark): - """persist_results calls _persist_full when no incremental config.""" + """_persist_results calls _persist_full when no incremental config.""" report = _build_report(spark, config_dict=DUMMY_CONFIG) with ( @@ -576,13 +578,13 @@ def test_persist_full_when_no_config(self, spark): return_value=None, ), ): - report.determine_report() + report._determine_report() with ( patch.object(report, "_persist_full") as mock_full, patch.object(report, "_persist_incremental") as mock_incr, ): - report.persist_results() + report._persist_results() mock_full.assert_called_once() mock_incr.assert_not_called() @@ -943,7 +945,7 @@ def test_persist_full_skips_when_all_dfs_none(self, spark): # Tests: duplicate calculated-channel identity rejection # ============================================================================ class TestDuplicateCalculatedChannelIdentities: - """determine_report must reject two calculated channels sharing an identity.""" + """_determine_report must reject two calculated channels sharing an identity.""" @staticmethod def _channel(name, identity): @@ -973,7 +975,7 @@ def test_duplicate_identity_raises_from_determine_report(self, spark): ), pytest.raises(ValueError, match="unique identities"), ): - report.determine_report(is_incremental=False) + report._determine_report(is_incremental=False) def test_distinct_identities_do_not_raise(self, spark): report = _build_report(spark) @@ -983,3 +985,64 @@ def test_distinct_identities_do_not_raise(self, spark): ] # Should not raise (validation passes); call the guard directly. report._validate_unique_calculated_channels() + + +# ============================================================================ +# Tests: deprecated determine_report() / persist_results() wrappers +# ============================================================================ +class TestDeprecatedEntryPoints: + """The public two-step API warns, names run() as the replacement, and delegates.""" + + @staticmethod + def _run_deprecation(record): + """The run()-pointing deprecation warnings among everything pytest recorded.""" + return [w for w in record if "Report.run()" in str(w.message)] + + def test_determine_report_warns_and_delegates(self, spark): + report = _build_report(spark) + + with ( + patch.object(report, "_determine_report") as mock_determine, + pytest.warns( + FutureWarning, match=r"Report\.determine_report\(\) is deprecated" + ) as record, + ): + report.determine_report(is_incremental=True) + + mock_determine.assert_called_once_with(True) + deprecations = self._run_deprecation(record) + assert len(deprecations) == 1 + # Attributed to the caller, not the telemetry_logger wrapper. + assert deprecations[0].filename == __file__ + + def test_persist_results_warns_and_delegates(self, spark): + report = _build_report(spark) + + with ( + patch.object(report, "_persist_results") as mock_persist, + pytest.warns( + FutureWarning, match=r"Report\.persist_results\(\) is deprecated" + ) as record, + ): + report.persist_results(cleanup_temp_tables=True) + + mock_persist.assert_called_once_with(True) + deprecations = self._run_deprecation(record) + assert len(deprecations) == 1 + assert deprecations[0].filename == __file__ + + def test_run_does_not_warn(self, spark): + report = _build_report(spark) + # Normally set by _determine_report; preset since it is mocked here. + report._more_batches_pending = False + + with ( + patch.object(report, "_determine_report") as mock_determine, + patch.object(report, "_persist_results") as mock_persist, + warnings.catch_warnings(), + ): + warnings.simplefilter("error", FutureWarning) + report.run(is_incremental=False, cleanup_temp_tables=False) + + mock_determine.assert_called_once_with(False) + mock_persist.assert_called_once_with(False) diff --git a/tests/impulse_reporting/unit/core/report_utils_test.py b/tests/impulse_reporting/unit/core/report_utils_test.py index 61faab4c..b50e6a37 100644 --- a/tests/impulse_reporting/unit/core/report_utils_test.py +++ b/tests/impulse_reporting/unit/core/report_utils_test.py @@ -600,7 +600,7 @@ def test_empty_event_list_for_type_is_skipped(self): assert "BASIC_EVENT" not in event_dfs def test_metadata_not_computed_in_dispatch(self): - """dispatch_events does not compute metadata (done separately in determine_report).""" + """dispatch_events does not compute metadata (done separately in _determine_report).""" class FakeContainerBase: pass From b2526e7b0f8c46c4c762bfc491042fe70b4e0935 Mon Sep 17 00:00:00 2001 From: "tom.bonfert" Date: Wed, 14 Oct 2026 13:00:47 +0200 Subject: [PATCH 2/3] docs(reporting): clarify `run(persist_results=False)` semantics and fix doc typos - Clarify that `run(persist_results=False)` leaves lazy DataFrames on the report object, which are unsafe to read after a persisting run because incremental results re-detect against updated gold, capped runs keep only the last batch, and `cleanup_temp_tables` drops their backing temp tables. - Update API reference, report reference, and skill docs to steer users toward reading gold tables after persistence. - Fix misplaced admonition text in the report reference and tighten deprecated-method docstrings. - Adjust integration tests to use internal `_determine_report()`/`_persist_results()` where the test subject is persistence behavior, not the deprecated public API. --- .../api/impulse_reporting/core/report.md | 19 ++++++++----- docs/impulse/docs/references/report/index.md | 8 +++--- skills/impulse-ml/SKILL.md | 2 +- skills/impulse-reporting/SKILL.md | 9 ++++--- src/impulse_reporting/core/report.py | 27 ++++++++++--------- .../batched_report_pipeline_test.py | 4 +-- .../point_value_aggregator_test.py | 4 ++- .../integration/points_in_time_event_test.py | 4 ++- .../integration/simple_report_test.py | 6 +++-- 9 files changed, 49 insertions(+), 34 deletions(-) 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 207ca616..085f49cd 100644 --- a/docs/impulse/docs/references/api/impulse_reporting/core/report.md +++ b/docs/impulse/docs/references/api/impulse_reporting/core/report.md @@ -307,8 +307,8 @@ def persist_results(cleanup_temp_tables: bool | None = None) Deprecated: use :meth:`run`, which determines and persists in one call. -Persists the results of a prior :meth:`determine_report` call; behavior is -unchanged. Unlike :meth:`run`, it does not drain capped incremental batches. +Persists the results of a prior :meth:`determine_report` call. Unlike :meth:`run`, +it does not drain capped incremental batches. **Arguments**: @@ -325,10 +325,15 @@ def run(is_incremental: bool = None, Determine and persist a report, draining all container batches. The entry point for executing a report. Computes events, aggregations, calculated -channels, and container dimensions, then writes them to the gold layer. Computed -results stay accessible on the report's attributes (e.g. ``aggregation_dfs``, -``event_dfs``), so ``run(persist_results=False)`` is the compute-only path. Without -a configured sink nothing is written. +channels, and container dimensions, then writes them to the gold layer. Without a +configured sink nothing is written. + +``run(persist_results=False)`` is the compute-only path: results are left on the +report's attributes (e.g. ``aggregation_dfs``, ``event_dfs``) for inspection. These +are lazy DataFrames, so after a persisting run read the gold tables instead: an +incremental run's attributes re-detect containers against the updated gold, a capped +run's hold only the last batch, and ``cleanup_temp_tables`` drops the temp tables +they read from. Supports incremental processing with definition-hash-based optimization: - Changed definitions trigger full reprocessing (all containers) @@ -366,7 +371,7 @@ def determine_report(is_incremental: bool = None) Deprecated: use :meth:`run`, or ``run(persist_results=False)`` to compute only. Computes the report without persisting; results are accessible in the report's -attributes and behavior is unchanged. +attributes. **Arguments**: diff --git a/docs/impulse/docs/references/report/index.md b/docs/impulse/docs/references/report/index.md index 47ebe84b..de77f8b8 100644 --- a/docs/impulse/docs/references/report/index.md +++ b/docs/impulse/docs/references/report/index.md @@ -43,7 +43,7 @@ Either `config` or `config_path` must be provided. | `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. | -- | -| `run(is_incremental, persist_results, cleanup_temp_tables)` | Computes all events, aggregations, calculated channels, and container dimensions and writes them (fact and dimension tables) to the configured Gold layer sink. Results are also stored on the report object. On a capped incremental run it loops until all containers are processed. | `is_incremental`: `bool` or `None`. Mode hint; overridden by `config.incremental` when present. See [Incremental processing](#incremental-processing). `persist_results`: `bool`, default `True`. `False` computes without writing. `cleanup_temp_tables`: `bool` or `None`. When truthy, drops the batch-solving `__impulse_temp_*` tables after a successful write; `None` (default) falls back to `config.unity_sink.cleanup_temp_tables`. | +| `run(is_incremental, persist_results, cleanup_temp_tables)` | Computes all events, aggregations, calculated channels, and container dimensions and writes them (fact and dimension tables) to the configured Gold layer sink. With `persist_results=False`, results are kept on the report object instead. On a capped incremental run it loops until all containers are processed, stopping early with a warning if a batch makes no forward progress. | `is_incremental`: `bool` or `None`. Mode hint; overridden by `config.incremental` when present. See [Incremental processing](#incremental-processing). `persist_results`: `bool`, default `True`. `False` computes without writing. `cleanup_temp_tables`: `bool` or `None`. When truthy, drops the batch-solving `__impulse_temp_*` tables after a successful write; `None` (default) falls back to `config.unity_sink.cleanup_temp_tables`. | ### Execution workflow @@ -58,7 +58,8 @@ my_report.run() ``` `run()` 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. :::note Deprecated: `determine_report()` / `persist_results()` The two-step `determine_report()` + `persist_results()` still works but emits a `FutureWarning` and will be @@ -66,8 +67,7 @@ removed in a future release. Replace the pair with `run()`, and a standalone `de `run(persist_results=False)`. Unlike `run()`, a single `determine_report()` + `persist_results()` pass does not drain a capped incremental run. The `persist_results` argument of `run()` is a flag and is unrelated to the deprecated method of the same name. -::: 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-ml/SKILL.md b/skills/impulse-ml/SKILL.md index 5ef0a800..83449a8f 100644 --- a/skills/impulse-ml/SKILL.md +++ b/skills/impulse-ml/SKILL.md @@ -70,7 +70,7 @@ report.run(persist_results=False) # computes in parallel across containers; no ## 2. Read the computed statistics -After `run()`, the results live on the report under `aggregation_dfs`, keyed by +After `run(persist_results=False)`, the results live on the report under `aggregation_dfs`, keyed by aggregation type. `StatsAggregator` results are under `"STATS_AGGREGATOR"`, returned as a bundle keyed by whether the rows were freshly computed (`"changed"`) or reused from a prior incremental run (`"unchanged"`). In a fresh sinkless run take `"changed"`: diff --git a/skills/impulse-reporting/SKILL.md b/skills/impulse-reporting/SKILL.md index 9bcc063c..597c39b6 100644 --- a/skills/impulse-reporting/SKILL.md +++ b/skills/impulse-reporting/SKILL.md @@ -93,8 +93,10 @@ the only thing that uses them. The aggregation must reference the registered obj re-created event with the same name raises too), and `add_event()` rejects duplicate event names, ignoring case. -After `run()`, computed results are also available on the report object — `report.aggregation_dfs[...]` -(used by `impulse-ml`) — which is what makes sinkless mode useful. +With `run(persist_results=False)` (or no sink), computed results stay on the report object — +`report.aggregation_dfs[...]` (used by `impulse-ml`) — which is what makes sinkless mode useful. They +are lazy DataFrames: after a persisting run, read the gold tables instead (incremental results re-detect +against the updated gold, and a capped run keeps only its last batch). `determine_report()` + `persist_results()` are deprecated (they warn) — use `run()`. The `persist_results=` flag of `run()` is unrelated to the deprecated method of the same name. @@ -112,7 +114,8 @@ configured, call `run(persist_results=False)`. Good for notebooks, tests, and fe Reuses prior results for unchanged definitions and reprocesses only containers that are new or updated in silver. Turn it on via `incremental.enabled` in config (see `impulse-config`) or pass `is_incremental=True` to `run()`. With `query_engine.max_containers_per_batch` set, `run()` commits at -most that many upserted containers per iteration and loops until all are processed. +most that many upserted containers per iteration and loops until all are processed (if a batch makes +no forward progress it logs a warning and stops early). **Mode resolution** (in order): diff --git a/src/impulse_reporting/core/report.py b/src/impulse_reporting/core/report.py index b609694a..3f4fc828 100644 --- a/src/impulse_reporting/core/report.py +++ b/src/impulse_reporting/core/report.py @@ -631,8 +631,8 @@ def _validate_full_recalculation_scope(self) -> None: def persist_results(self, cleanup_temp_tables: bool | None = None): """Deprecated: use :meth:`run`, which determines and persists in one call. - Persists the results of a prior :meth:`determine_report` call; behavior is - unchanged. Unlike :meth:`run`, it does not drain capped incremental batches. + Persists the results of a prior :meth:`determine_report` call. Unlike :meth:`run`, + it does not drain capped incremental batches. Parameters ---------- @@ -653,12 +653,8 @@ def _persist_results(self, cleanup_temp_tables: bool | None = None): Parameters ---------- - cleanup_temp_tables : bool, optional - Whether to drop the batch-solving ``__impulse_temp_*`` tables from the - sink schema after persistence completes successfully. - - True/False: use this value, overriding the config flag. - - None (default): fall back to ``config.unity_sink.cleanup_temp_tables`` - (which itself defaults to False). + cleanup_temp_tables : bool | None, optional + Temp-table cleanup override, see :meth:`run`. Returns ------- @@ -1015,10 +1011,15 @@ def run( """Determine and persist a report, draining all container batches. The entry point for executing a report. Computes events, aggregations, calculated - channels, and container dimensions, then writes them to the gold layer. Computed - results stay accessible on the report's attributes (e.g. ``aggregation_dfs``, - ``event_dfs``), so ``run(persist_results=False)`` is the compute-only path. Without - a configured sink nothing is written. + channels, and container dimensions, then writes them to the gold layer. Without a + configured sink nothing is written. + + ``run(persist_results=False)`` is the compute-only path: results are left on the + report's attributes (e.g. ``aggregation_dfs``, ``event_dfs``) for inspection. These + are lazy DataFrames, so after a persisting run read the gold tables instead: an + incremental run's attributes re-detect containers against the updated gold, a capped + run's hold only the last batch, and ``cleanup_temp_tables`` drops the temp tables + they read from. Supports incremental processing with definition-hash-based optimization: - Changed definitions trigger full reprocessing (all containers) @@ -1090,7 +1091,7 @@ def determine_report(self, is_incremental: bool = None): """Deprecated: use :meth:`run`, or ``run(persist_results=False)`` to compute only. Computes the report without persisting; results are accessible in the report's - attributes and behavior is unchanged. + attributes. Parameters ---------- diff --git a/tests/impulse_reporting/integration/batched_report_pipeline_test.py b/tests/impulse_reporting/integration/batched_report_pipeline_test.py index affdcd24..a949b3b1 100644 --- a/tests/impulse_reporting/integration/batched_report_pipeline_test.py +++ b/tests/impulse_reporting/integration/batched_report_pipeline_test.py @@ -324,7 +324,7 @@ def _temp_tables(spark: SparkSession) -> set[str]: } -def test_persist_results_retains_temp_tables_by_default(spark, cleanup_gold): +def test_persist_retains_temp_tables_by_default(spark, cleanup_gold): """Default behavior: the current run's temp tables survive persistence.""" report, _ = _build_batched_report(spark) @@ -340,7 +340,7 @@ def test_persist_results_retains_temp_tables_by_default(spark, cleanup_gold): assert _temp_tables(spark) -def test_persist_results_cleans_temp_tables_when_param_true(spark, cleanup_gold): +def test_persist_cleans_temp_tables_when_param_true(spark, cleanup_gold): """cleanup_temp_tables=True drops temp tables after a successful write.""" report, _ = _build_batched_report(spark) diff --git a/tests/impulse_reporting/integration/point_value_aggregator_test.py b/tests/impulse_reporting/integration/point_value_aggregator_test.py index b50ae063..acabd3af 100644 --- a/tests/impulse_reporting/integration/point_value_aggregator_test.py +++ b/tests/impulse_reporting/integration/point_value_aggregator_test.py @@ -70,12 +70,14 @@ def test_persist_point_value_aggregator(spark): ) page.add_aggregation(pva) - my_report.run() + my_report._determine_report() assert "POINT_VALUE_AGGREGATOR" in my_report.aggregation_dfs assert my_report.aggregation_dfs["POINT_VALUE_AGGREGATOR"]["changed"].count() > 0 assert my_report.aggregation_metadata_dfs["POINT_VALUE_AGGREGATOR"].count() == 1 + my_report._persist_results() + fact = spark.read.table("spark_catalog.gold.evaluation_stats_aggregator_fact") dim = spark.read.table("spark_catalog.gold.evaluation_stats_aggregator_dimension") event_fact = spark.read.table("spark_catalog.gold.evaluation_event_instance_fact") diff --git a/tests/impulse_reporting/integration/points_in_time_event_test.py b/tests/impulse_reporting/integration/points_in_time_event_test.py index 77a675b9..dc6ffffd 100644 --- a/tests/impulse_reporting/integration/points_in_time_event_test.py +++ b/tests/impulse_reporting/integration/points_in_time_event_test.py @@ -30,13 +30,15 @@ def test_persist_points_in_time_event(spark): ) my_report.add_event(pit_event) - my_report.run() + my_report._determine_report() event_dfs = my_report.event_dfs event_metadata_dfs = my_report.event_metadata_dfs assert event_dfs["POINTS_IN_TIME_EVENT"]["changed"].count() > 0 assert event_metadata_dfs["POINTS_IN_TIME_EVENT"].count() > 0 + my_report._persist_results() + assert spark.catalog.tableExists("spark_catalog.gold.evaluation_event_instance_fact") assert spark.catalog.tableExists("spark_catalog.gold.evaluation_event_dimension") diff --git a/tests/impulse_reporting/integration/simple_report_test.py b/tests/impulse_reporting/integration/simple_report_test.py index d65bc433..acca60fc 100644 --- a/tests/impulse_reporting/integration/simple_report_test.py +++ b/tests/impulse_reporting/integration/simple_report_test.py @@ -686,8 +686,8 @@ def test_persist_report_with_events(spark): ) my_first_page.add_aggregation(hist5) - # Determine and persist the report - my_report.run() + # Determine content of all pages + my_report._determine_report() agg_dfs = my_report.aggregation_dfs agg_metadata_dfs = my_report.aggregation_metadata_dfs @@ -703,6 +703,8 @@ def test_persist_report_with_events(spark): assert event_dfs["BASIC_EVENT"]["changed"].count() > 0 assert event_metadata_dfs["BASIC_EVENT"].count() > 0 + my_report._persist_results() + assert spark.catalog.tableExists("spark_catalog.gold.evaluation_event_dimension") assert spark.catalog.tableExists("spark_catalog.gold.evaluation_event_instance_fact") assert spark.catalog.tableExists("spark_catalog.gold.evaluation_histogram_dimension") From a490f0eb247d53da286d454f1773bc25b6a3dbd1 Mon Sep 17 00:00:00 2001 From: "tom.bonfert" Date: Wed, 14 Oct 2026 13:04:56 +0200 Subject: [PATCH 3/3] fix(reporting): move telemetry decorator to private helpers and add run() telemetry test - Shift `@telemetry_logger` from deprecated public `persist_results()`/`determine_report()` to their private `_persist_results()`/`_determine_report()` counterparts so `run()` logs per-step tags correctly. - Adjust deprecation warning `stacklevel` from 3 to 2 now that the wrappers no longer sit between the caller and the warning. - Add `TestRunTelemetry` verifying that `run()` emits `report/run`, `report/determine_report`, and `report/persist_results` telemetry tags. --- src/impulse_reporting/core/report.py | 10 +++--- .../unit/core/report_incremental_test.py | 31 +++++++++++++++++++ 2 files changed, 35 insertions(+), 6 deletions(-) diff --git a/src/impulse_reporting/core/report.py b/src/impulse_reporting/core/report.py index 3f4fc828..003c1814 100644 --- a/src/impulse_reporting/core/report.py +++ b/src/impulse_reporting/core/report.py @@ -627,7 +627,6 @@ def _validate_full_recalculation_scope(self) -> None: }, ) - @telemetry_logger("report", "persist_results") def persist_results(self, cleanup_temp_tables: bool | None = None): """Deprecated: use :meth:`run`, which determines and persists in one call. @@ -639,10 +638,10 @@ def persist_results(self, cleanup_temp_tables: bool | None = None): cleanup_temp_tables : bool | None, optional Same as the :meth:`run` argument. """ - # stacklevel=3 skips the telemetry_logger wrapper so the warning points at the caller. - warnings.warn(_PERSIST_RESULTS_DEPRECATION_MSG, FutureWarning, stacklevel=3) + warnings.warn(_PERSIST_RESULTS_DEPRECATION_MSG, FutureWarning, stacklevel=2) self._persist_results(cleanup_temp_tables) + @telemetry_logger("report", "persist_results") def _persist_results(self, cleanup_temp_tables: bool | None = None): """ Persist report results using appropriate strategy based on definition changes. @@ -1086,7 +1085,6 @@ def run( return previous_batch_ids = current_batch_ids - @telemetry_logger("report", "determine_report") def determine_report(self, is_incremental: bool = None): """Deprecated: use :meth:`run`, or ``run(persist_results=False)`` to compute only. @@ -1098,10 +1096,10 @@ def determine_report(self, is_incremental: bool = None): is_incremental : bool, optional Same as the :meth:`run` argument. """ - # stacklevel=3 skips the telemetry_logger wrapper so the warning points at the caller. - warnings.warn(_DETERMINE_REPORT_DEPRECATION_MSG, FutureWarning, stacklevel=3) + warnings.warn(_DETERMINE_REPORT_DEPRECATION_MSG, FutureWarning, stacklevel=2) self._determine_report(is_incremental) + @telemetry_logger("report", "determine_report") def _determine_report(self, is_incremental: bool = None): """ Determine and process events, aggregations, and container dimensions for the report. diff --git a/tests/impulse_reporting/unit/core/report_incremental_test.py b/tests/impulse_reporting/unit/core/report_incremental_test.py index fe89c2b0..80f0c6f1 100644 --- a/tests/impulse_reporting/unit/core/report_incremental_test.py +++ b/tests/impulse_reporting/unit/core/report_incremental_test.py @@ -1046,3 +1046,34 @@ def test_run_does_not_warn(self, spark): mock_determine.assert_called_once_with(False) mock_persist.assert_called_once_with(False) + + +# ============================================================================ +# Tests: run() telemetry +# ============================================================================ +class TestRunTelemetry: + """run() logs its own tag plus the per-step tags carried by the private methods.""" + + def test_run_logs_run_and_step_tags(self, spark): + report = _build_report(spark) + + with ( + patch.object(report, "_gold_layer_exists", return_value=True), + patch.object(report, "_group_events_by_type", return_value={}), + patch.object(report, "_group_aggregations_by_type", return_value={}), + patch( + "impulse_reporting.core.report.ContainerDimension.get_dimension", + return_value=None, + ), + patch.object(report, "_persist_full") as mock_full, + patch("impulse_query_engine.telemetry.log_telemetry") as mock_log, + ): + report.run(is_incremental=False) + + mock_full.assert_called_once() + tags = [(c.args[1], c.args[2]) for c in mock_log.call_args_list] + assert tags == [ + ("report", "run"), + ("report", "determine_report"), + ("report", "persist_results"), + ]