Add optional cleanup of batch-solving temp tables after successful persist

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`).

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
 

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,

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
 

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):

src/impulse_reporting/core/report.py

@@ -546,14 +546,23 @@ 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.
 
         Uses tracked state from 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).
+
         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.

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)

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,