Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 1 addition & 2 deletions demos/getting_started.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -91,8 +91,7 @@
")\n",
"report.add_page(page)\n",
"\n",
"report.determine_report()\n",
"report.persist_results()\n"
"report.run()\n"
],
"id": "3c9aea0cf4a3ecc"
},
Expand Down
4 changes: 2 additions & 2 deletions demos/ml_feature_engineering.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -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."
Expand All @@ -553,7 +553,7 @@
},
"outputs": [],
"source": [
"report.determine_report()\n",
"report.run(persist_results=False)\n",
"print(\"Statistics computed\")"
]
},
Expand Down
6 changes: 2 additions & 4 deletions demos/reporting_pipeline.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -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."
]
},
{
Expand All @@ -800,8 +799,7 @@
},
"outputs": [],
"source": [
"report.determine_report()\n",
"report.persist_results()\n",
"report.run()\n",
"print(f\"Report persisted to {pfx}_*\")"
]
},
Expand Down
8 changes: 4 additions & 4 deletions docs/impulse/docs/config/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -80,17 +80,17 @@ 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`).

### 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.

---
Expand Down
5 changes: 2 additions & 3 deletions docs/impulse/docs/motivation.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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).

Expand Down Expand Up @@ -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**:

Expand Down
80 changes: 41 additions & 39 deletions docs/impulse/docs/references/api/impulse_reporting/core/report.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -333,45 +324,56 @@ 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
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.

2 changes: 1 addition & 1 deletion docs/impulse/docs/references/report/channel.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion docs/impulse/docs/references/report/event.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
31 changes: 18 additions & 13 deletions docs/impulse/docs/references/report/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.

---

Expand All @@ -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)),
Expand All @@ -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.
Expand Down
13 changes: 6 additions & 7 deletions docs/impulse/docs/tutorial/demo.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)`.

---

Expand Down
6 changes: 3 additions & 3 deletions skills/impulse-config/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
8 changes: 4 additions & 4 deletions skills/impulse-ml/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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"],
}
Expand All @@ -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"`:
Expand Down Expand Up @@ -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`.
Loading
Loading