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..085f49cd 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. 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,36 +324,28 @@ 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. - -**Arguments**: - -- `is_incremental` (`bool`): Forwarded to :meth:`determine_report` (config still overrides it). -- `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`. - -#### determine\_report - -```python -def determine_report(is_incremental: bool = None) -``` - -Determine and process events, aggregations, and container dimensions for the report. +The entry point for executing a report. Computes events, aggregations, calculated +channels, and container dimensions, then writes them to the gold layer. Without a +configured sink nothing is written. -Results are accessible in the report's attributes. +``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) - 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`): Hint for processing mode. Overwritten by config when incremental @@ -370,8 +353,27 @@ 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, 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). -**Returns**: +#### determine\_report -`None`: +```python +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. + +**Arguments**: + +- `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..de77f8b8 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. 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 @@ -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()` +`run()` 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 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 +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. +::: + ### 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..83449a8f 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(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"`: @@ -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..597c39b6 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,39 @@ 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. +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. ## 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 (if a batch makes +no forward progress it logs a warning and stops early). **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..003c1814 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.""" @@ -615,23 +627,33 @@ 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. Unlike :meth:`run`, + it does not drain capped incremental batches. + + Parameters + ---------- + cleanup_temp_tables : bool | None, optional + Same as the :meth:`run` argument. + """ + 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. - 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) 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 ------- @@ -640,7 +662,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 +682,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 +735,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 +1009,46 @@ 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. 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) + - 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: @@ -1042,24 +1085,30 @@ 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. + + Computes the report without persisting; results are accessible in the report's + attributes. + + 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. + warnings.warn(_DETERMINE_REPORT_DEPRECATION_MSG, FutureWarning, stacklevel=2) + 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) + @telemetry_logger("report", "determine_report") + 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..a949b3b1 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 @@ -324,14 +324,14 @@ 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().""" +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) - 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 ( @@ -340,14 +340,14 @@ 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): - """persist_results(cleanup_temp_tables=True) drops temp tables after a successful write.""" +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) - 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..acabd3af 100644 --- a/tests/impulse_reporting/integration/point_value_aggregator_test.py +++ b/tests/impulse_reporting/integration/point_value_aggregator_test.py @@ -70,13 +70,13 @@ def test_persist_point_value_aggregator(spark): ) page.add_aggregation(pva) - my_report.determine_report() + 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() + 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") @@ -158,8 +158,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 +244,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 +263,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..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,14 +30,14 @@ def test_persist_points_in_time_event(spark): ) my_report.add_event(pit_event) - my_report.determine_report() + 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() + 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..acca60fc 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") @@ -690,7 +687,7 @@ def test_persist_report_with_events(spark): my_first_page.add_aggregation(hist5) # Determine content of all pages - my_report.determine_report() + my_report._determine_report() agg_dfs = my_report.aggregation_dfs agg_metadata_dfs = my_report.aggregation_metadata_dfs @@ -706,7 +703,7 @@ 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() + 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") @@ -819,7 +816,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..80f0c6f1 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,95 @@ 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) + + +# ============================================================================ +# 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"), + ] 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