diff --git a/docs/impulse/docs/config/configuration.md b/docs/impulse/docs/config/configuration.md index 55f10185..310daf50 100644 --- a/docs/impulse/docs/config/configuration.md +++ b/docs/impulse/docs/config/configuration.md @@ -76,9 +76,10 @@ Defines where gold-layer tables are written. | Field | Type | Required | Description | |----------------|-------|----------|---------------------------------------| -| `catalog` | `str` | Yes | Target catalog name. | -| `schema` | `str` | Yes | Target schema name. | -| `table_prefix` | `str` | Yes | Prefix for all generated table names. | +| `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=...)`. | Output tables are named `{table_prefix}_{entity}` (e.g. `my_report_histogram_fact`). diff --git a/docs/impulse/docs/references/report/index.md b/docs/impulse/docs/references/report/index.md index 82700fdf..f4236572 100644 --- a/docs/impulse/docs/references/report/index.md +++ b/docs/impulse/docs/references/report/index.md @@ -44,7 +44,7 @@ Either `config` or `config_path` must be provided. | `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()` | Writes all computed results (fact and dimension tables) to the configured Gold layer sink. | -- | +| `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`. | ### Execution workflow diff --git a/skills/impulse-config/SKILL.md b/skills/impulse-config/SKILL.md index 912a7be5..c8fff84d 100644 --- a/skills/impulse-config/SKILL.md +++ b/skills/impulse-config/SKILL.md @@ -62,11 +62,12 @@ Maps the silver-layer input tables. Values are full Unity Catalog paths (`catalo Where gold tables are written. Output tables are named `{table_prefix}_{entity}`. -| Field | Required | Description | -|----------------|----------|--------------------| -| `catalog` | Yes | Target catalog. | -| `schema` | Yes | Target schema. | -| `table_prefix` | Yes | Prefix for tables. | +| Field | Required | Description | +|-----------------------|----------|--------------------| +| `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=...)`. | **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, diff --git a/skills/impulse-reporting/SKILL.md b/skills/impulse-reporting/SKILL.md index d4b32d3f..6bd0d58a 100644 --- a/skills/impulse-reporting/SKILL.md +++ b/skills/impulse-reporting/SKILL.md @@ -47,7 +47,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()`. +`persist_results(cleanup_temp_tables=None)` (pass `True`/`False` to override the +`unity_sink.cleanup_temp_tables` config flag for dropping `__impulse_temp_*` tables after a +successful write). ## The lifecycle diff --git a/src/impulse_reporting/config/config_parser.py b/src/impulse_reporting/config/config_parser.py index 6bff951e..bef26e36 100644 --- a/src/impulse_reporting/config/config_parser.py +++ b/src/impulse_reporting/config/config_parser.py @@ -163,6 +163,11 @@ class UnitySink(BaseModel): 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 + successfully. Defaults to ``False`` (temp tables are retained for inspection + and only cleared at the start of the next report run). Notes ----- @@ -175,6 +180,7 @@ class UnitySink(BaseModel): str, AfterValidator(lambda v: v if v == "" else is_valid_unity_entity_name(v)), ] + cleanup_temp_tables: bool = False class Comparator(str, Enum): diff --git a/src/impulse_reporting/core/report.py b/src/impulse_reporting/core/report.py index a0e4b6f2..f0e9876d 100644 --- a/src/impulse_reporting/core/report.py +++ b/src/impulse_reporting/core/report.py @@ -546,7 +546,7 @@ def _validate_aggregation_events(self) -> None: raise ValueError(error_message) @telemetry_logger("report", "persist_results") - def persist_results(self): + def persist_results(self, cleanup_temp_tables: bool | None = None): """ Persist report results using appropriate strategy based on definition changes. @@ -554,6 +554,15 @@ def persist_results(self): - 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). + Returns ------- None @@ -573,6 +582,20 @@ def persist_results(self): else: self._persist_full() + # Only drop the current run's temp tables once persistence has succeeded. + if self._resolve_cleanup_temp_tables(cleanup_temp_tables): + self._cleanup_temp_tables() + + 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. + """ + if cleanup is not None: + return cleanup + return bool(self.config.unity_sink.cleanup_temp_tables) + def _persist_full(self): """ Persist results using full overwrite strategy. diff --git a/tests/impulse_reporting/integration/batched_report_pipeline_test.py b/tests/impulse_reporting/integration/batched_report_pipeline_test.py index fb135ced..2d8314cf 100644 --- a/tests/impulse_reporting/integration/batched_report_pipeline_test.py +++ b/tests/impulse_reporting/integration/batched_report_pipeline_test.py @@ -315,3 +315,92 @@ def _stats_wrapper( assert stats_df.count() > 0 assert report.aggregation_metadata_dfs["HISTOGRAM"].count() == 1 assert report.aggregation_metadata_dfs["STATS_AGGREGATOR"].count() == 1 + + +def _temp_tables(spark: SparkSession) -> set[str]: + return { + row.tableName + for row in spark.sql("SHOW TABLES IN spark_catalog.gold LIKE '__impulse_temp_*'").collect() + } + + +def test_persist_results_retains_temp_tables_by_default(spark, cleanup_gold): + """Default behavior: the current run's temp tables survive persist_results().""" + report, _ = _build_batched_report(spark) + + report.determine_report() + assert _temp_tables(spark) # temp tables created during solving + + report.persist_results() + + # Gold facts written with real values, and temp tables still present. + assert ( + report.aggregation_dfs["HISTOGRAM"]["changed"].filter(F.col("hist_value") > 0).count() > 0 + ) + 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.""" + report, _ = _build_batched_report(spark) + + report.determine_report() + assert _temp_tables(spark) + + 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. + hist_fact = spark.read.table("spark_catalog.gold.evaluation_histogram_fact") + assert hist_fact.filter(F.col("hist_value") > 0).count() > 0 + assert not _temp_tables(spark) + + +def test_persist_results_param_overrides_config_flag(spark, cleanup_gold): + """Explicit param wins over config: config True but param False retains temp tables.""" + config = ImpulseConfig( + source=Source( + container_metrics_table="spark_catalog.silver.container_metrics", + channel_metrics_table="spark_catalog.silver.channel_metrics", + channels_uri="spark_catalog.silver.channels", + ), + unity_sink=UnitySink( + catalog="spark_catalog", + schema="gold", + table_prefix="evaluation", + cleanup_temp_tables=True, + ), + query_engine=QueryEngine(solver=Solvers.KEY_VALUE_STORE_SOLVER, batch_size=1), + ) + report, _ = _build_batched_report(spark) + report.config = config + + report.determine_report() + report.persist_results(cleanup_temp_tables=False) + + assert _temp_tables(spark) + + +def test_persist_results_config_flag_cleans_temp_tables(spark, cleanup_gold): + """When param is None, the config flag drives cleanup.""" + config = ImpulseConfig( + source=Source( + container_metrics_table="spark_catalog.silver.container_metrics", + channel_metrics_table="spark_catalog.silver.channel_metrics", + channels_uri="spark_catalog.silver.channels", + ), + unity_sink=UnitySink( + catalog="spark_catalog", + schema="gold", + table_prefix="evaluation", + cleanup_temp_tables=True, + ), + query_engine=QueryEngine(solver=Solvers.KEY_VALUE_STORE_SOLVER, batch_size=1), + ) + report, _ = _build_batched_report(spark) + report.config = config + + report.determine_report() + report.persist_results() + + assert not _temp_tables(spark) diff --git a/tests/impulse_reporting/unit/config/config_parser_test.py b/tests/impulse_reporting/unit/config/config_parser_test.py index ec9936b7..28aa3e7c 100644 --- a/tests/impulse_reporting/unit/config/config_parser_test.py +++ b/tests/impulse_reporting/unit/config/config_parser_test.py @@ -131,6 +131,25 @@ def test_impulse_config_drop_implausible_data_rejects_rle(): ImpulseConfig.model_validate(config_json) +def test_impulse_config_cleanup_temp_tables_defaults_to_false(): + config = ImpulseConfig.model_validate(impulse_config_JSON.copy()) + assert config.unity_sink.cleanup_temp_tables is False + + +def test_impulse_config_cleanup_temp_tables_enabled(): + config_json = { + **impulse_config_JSON, + "unity_sink": { + "catalog": "test_catalog", + "schema": "test_schema", + "table_prefix": "test_prefix", + "cleanup_temp_tables": True, + }, + } + config = ImpulseConfig.model_validate(config_json) + assert config.unity_sink.cleanup_temp_tables is True + + def test_impulse_config_raw_encoder_interval(): config_json = { **impulse_config_JSON,