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
8 changes: 7 additions & 1 deletion docs/impulse/docs/config/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -142,11 +142,17 @@ Two independent filter families:
| `max_channels_per_batch` | `int` | `500` | Caps how many distinct channels are read in one solve batch, to bound per-solve memory. Counts channel selections (each distinct `channel(...)` / `poi_channel(...)` in your expressions). Applies to events, aggregations, and calculated channels; for calculated channels it counts the *input* channels they read, not the number of calculated channels. The cap is a packing target, not a hard limit: an expression that on its own reads more channels than the cap still runs, in a batch of its own. Must be `>= 1`. The old name `batch_size` is a **deprecated alias** (still accepted, warns, removed in a future release); setting both raises a validation error. |
| `max_containers_per_batch` | `int` | `null` | Caps how many containers are solved at once, to bound per-solve memory. `null` disables the cap; when set it must be `>= 1`. On an **incremental** run, `Report.run()` commits at most this many upserted containers per iteration and loops until the population is drained (see [incremental](#incremental-optional)). On a **full** run, and when a changed definition recomputes over all containers, the population is split into chunks of about this size and solved chunk by chunk. Chunk sizes are approximate (the cap is a memory heuristic, not an exact bound). |
| `solver_config` | `SolverConfig` | `null` | Per-table column mappings, per-table equality filters, and project scoping. Set `project_id` to scope reads by project — it is applied to `container_tags` (if configured), `container_metrics`, and `channel_mapping` (if configured), so it works in both narrow EAV and wide-only data models. Omit it when you don't need project scoping. See [Solver column mappings and filters](#solver-column-mappings-and-filters). |
| `measurement_db` | `str` | `"MeasurementDB"` | Registered name of the MeasurementDB implementation to use. The default is the built-in reader; a custom implementation can reshape a different source schema at read time. See [Custom measurement DB](../extensibility/custom_measurement_db.md). |
| `measurement_db_config` | `dict` | `null` | Keyword arguments for that implementation's config class, merged with the `source` tables (e.g. extra source tables, prefilter keys). |

If `query_engine` is omitted, the default is `DefaultSolver` with
`data_type = "RLE"`.

---
Unknown keys under `query_engine` are rejected when the config is loaded: a misspelled key such as
`"measurment_db"` fails with `Extra inputs are not permitted` instead of being silently ignored.

For a source schema that needs reshaping beyond column renames, select your own read seam — see
[Custom measurement DB](../extensibility/custom_measurement_db.md).

## Solver column mappings and filters

Expand Down
11 changes: 11 additions & 0 deletions docs/impulse/docs/extensibility/_category_.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
{
"label": "Extensibility",
"position": 5,
"link": {
"type": "generated-index",
"title": "Extensibility",
"description": "Extension points for adapting Impulse to data or behaviour that the built-in configuration cannot express.",
"slug": "/extensibility"
},
"collapsed": true
}
140 changes: 140 additions & 0 deletions docs/impulse/docs/extensibility/custom_measurement_db.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
---
sidebar_position: 1
title: Custom measurement DB
---

# Custom measurement DB

The **MeasurementDB** is the layer between your stored tables and everything that reads them — the
solve, incremental container detection, and ad-hoc analysis all read through it. Impulse ships a
built-in MeasurementDB that reads the [silver model](../data_model/silver_layer_schema.md) directly.
When your raw tables need more than column renames to match that model — unions, joins, a grain
change, an EAV unpivot, or a row filter that must run *before* a reshape — you can register your own
MeasurementDB and select it by name in the report config. The stock solver then runs unchanged, and
every read path sees the same silver-shaped data.

This is an additive, opt-in extension point: with nothing configured, the built-in MeasurementDB is
used and behaviour is unchanged.

## When to use it

| Your data differs from the silver model by… | Use |
|---|---|
| Column names, or narrow-EAV vs. wide layout | `DefaultSolver` + [`solver_config`](../config/configuration.md#solver-column-mappings-and-filters) — no code |
| Different tables, unions, joins, grain changes, EAV unpivots, or a pre-reshape row filter | **A custom MeasurementDB** (this page) |
| Query planning / execution the config cannot express | Implementation of a custom solver (see [Custom solvers](../data_model/ingestion.md#custom-solvers)) |

## Select it in the report config

Two optional keys under `query_engine` choose the implementation and pass it settings:

```json
"query_engine": {
"measurement_db": "MyMeasurementDB",
"measurement_db_config": { "session_table": "catalog.raw.sessions" }
}
```

- `measurement_db` — the **registered name** of the implementation (default `"MeasurementDB"`, the built-in reader).
- `measurement_db_config` — keyword arguments for that implementation's config class, merged with the
`source` tables (e.g. extra source tables, prefilter keys).

In Python the same keys go on `QueryEngine`:
`QueryEngine(measurement_db="MyMeasurementDB", measurement_db_config={...})`.

## Implement one

A custom MeasurementDB is a small subclass of the built-in `MeasurementDB`, plus a config subclass of
`MeasurementDBConfig`, registered under a name:

```python
from impulse_query_engine.measurement_db import MeasurementDB, MeasurementDBConfig
from impulse_query_engine.measurement_db_registry import register_measurement_db


class MyDBConfig(MeasurementDBConfig):
def __init__(self, *, session_table: str, **base_kwargs): # required: no default
super().__init__(**base_kwargs) # keeps the source tables, table_locations, and pinning state
self.session_table = session_table

def configured_table_uris(self) -> list[str]:
# Declare extra tables so they are pinned to one Delta snapshot per run (see below).
return super().configured_table_uris() + [self.session_table]


@register_measurement_db("MyMeasurementDB", MyDBConfig)
class MyMeasurementDB(MeasurementDB):
def container_metrics(self, spark):
raw = super().container_metrics(spark) # raw read, pinned
sessions = self._read_table(spark, self.config.session_table) # extra read, pinned
return ... # reshape into the silver container_metrics shape
```

Override one method per silver table you need to reshape; the rest fall back to the base read. Each
override follows the same shape: call `super().<table>(spark)` for the raw, pinned read, then
join / union / reshape into the silver frame the solver expects.

### Optional tables must also be set in `source`

Overriding the read of an optional table only takes effect if you also set that table in the
report config's `source`. Otherwise the solver doesn't use the table, and your override is never
called:

- `container_tags()` needs `source.container_tags_table`
- `channel_tags()` needs `source.channel_tags_table`
- `channel_mapping()` needs `source.channel_mapping_table`
- `unit_conversion()` needs `source.unit_conversion_table`
- `poi_channels()` needs `has_poi_channels()` to return `True`; the base implementation checks
`source.poi_channels_uri`

Set each entry to the raw table your override reshapes; `super().<table>(spark)` reads it. For
channel tags, unit conversion and POI, a missing entry doesn't raise an error: the feature just
isn't used.

## Rules

- **Import the registering package before the report config is parsed.** Selection is by name only
(config data never triggers an import), so the package whose import runs `@register_measurement_db`
must be imported first — including in any tool that only *generates* or *edits* a config.
- **Subclass `MeasurementDBConfig` and call `super().__init__(**base_kwargs)`.** The base initializer
sets the source tables, `table_locations`, and the per-run pinning state the reads rely on.
- **Keep the config `__init__` free of side effects** — it is constructed once to validate the config
at parse time and once more to build the DB.
- **Keep the DB constructor signature `__init__(self, config, ws)`** — Impulse builds it as `db_cls(config, ws)`.
- **If you override the DB `__init__`, call `super().__init__(config, ws)`.** The base initializer
stores the config and verifies the workspace client; reading `db.ws` (which `Report` and `db.query`
do) raises a `TypeError` if it was skipped. Don't assign `self.ws` yourself.
- **Your DB owns correctness.** Impulse does not validate the frames a custom DB returns; it must
provide the columns, types, row grain, and time unit the solver expects. The built-in `MeasurementDB`
and the schemas in `impulse_query_engine/schema.py` are the reference.

## Parse-time validation

The config is checked when it is loaded (`ImpulseConfig.model_validate`), so these fail fast as a
validation error rather than at run time:

- an unregistered `measurement_db` name (the message lists the registered names);
- a missing required argument of the config class;
- an unknown or misspelled key;
- a key present in both `source` and `measurement_db_config`.

## Snapshot pinning (extra tables)

At the start of each run Impulse pins every configured table to one Delta version so all lazy reads
observe a consistent snapshot. The built-in pinning iterates `configured_table_uris()`, so any extra
table your DB reads must be:

1. **declared** — override `configured_table_uris()` to add its URI; and
2. **read through `self._read_table(spark, uri)`** (never `spark.read.table`), using the **same URI
string** you declared.

Views and non-Delta tables cannot be time-travelled: they are skipped with a warning and read at the
latest version.

## Incremental reporting

Because every reader goes through the MeasurementDB, incremental container detection reads the same
reshaped `container_metrics` as the solve, so the silver↔gold join stays aligned by construction. If
your reshape renames the physical container-id column, the rename is applied in exactly one place and
both the solve and detection see it. See [incremental](../config/configuration.md#incremental-optional)
for how the mode is resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,9 @@ scoping. Key sub-fields:

When omitted, all default column names are used and no
project/toolbox filtering is applied.
- `measurement_db` (`str, default="MeasurementDB"`): Registered name of the MeasurementDB implementation to use.
- `measurement_db_config` (`dict`): Keyword arguments for that implementation's config class,
merged with the ``source`` tables.

#### validate\_drop\_implausible\_data\_requires\_raw

Expand All @@ -221,6 +224,32 @@ The filter is applied inside the RAW -> interval conversion path by the
selected ``raw_encoder`` (``RleEncoder`` / ``IntervalEncoder``).


#### reject\_channels\_filter\_on\_is\_plausible\_in\_raw

```python
def reject_channels_filter_on_is_plausible_in_raw()
```

Reject an is_plausible channels filter in RAW mode (delegates to SolverConfig).

The shared invariant lives on ``SolverConfig`` so direct query-engine use
(``DefaultSolver``) enforces the same rule; here we surface it at config-parse time.


#### warn\_channels\_filters\_bridge\_in\_raw

```python
def warn_channels_filters_bridge_in_raw()
```

Warn (not reject) on any non-plausibility channels filter in RAW mode.

Such filters run before raw encoding and bridge intervals across dropped
samples. That is fine for whole-channel scoping but corrupts per-sample cleaning,
and the two cannot be told apart statically, so we warn. ``is_plausible`` is
hard-rejected above as the one unambiguously per-sample case.


#### default\_raw\_encoder\_for\_raw\_data

```python
Expand Down Expand Up @@ -294,6 +323,22 @@ Names that do not match any registered entity are rejected at the start of
- `events` (`list of str, default=[]`): Names of events to fully recalculate.
- `calculated_channels` (`list of str, default=[]`): Names of calculated channels to fully recalculate.

#### build\_measurement\_db\_config

```python
def build_measurement_db_config(
source_dict: dict[str, Any], query_engine: QueryEngine
) -> tuple[type[AbstractMeasurementDB], MeasurementDBConfig]
```

Resolve the selected MeasurementDB implementation and build its config class from

``measurement_db_config`` merged with the ``source`` tables in *source_dict*.

Raises ``KeyError`` for an unregistered ``measurement_db`` and ``TypeError`` for a missing,
unknown or duplicated config key.


## ImpulseConfig

```python
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,14 +52,14 @@ Returns a unique identifier for the report.
#### get\_db

```python
def get_db() -> MeasurementDB
def get_db() -> AbstractMeasurementDB
```

Get the measurement database associated with this report.

**Returns**:

`MeasurementDB`: The measurement database instance.
`AbstractMeasurementDB`: The measurement database instance.

#### get\_solver

Expand Down Expand Up @@ -109,11 +109,13 @@ Load Impulse configuration from a dictionary.

```python
def create_measurement_db(config: ImpulseConfig,
ws: WorkspaceClient) -> MeasurementDB
ws: WorkspaceClient) -> AbstractMeasurementDB
```

Create a measurement database based on the provided configuration.
Create the measurement database selected by ``query_engine.measurement_db``.

Builds the registered config class from the ``source`` tables plus
``query_engine.measurement_db_config``, then the registered DB class.
Maps the optional ``container_tags`` field from the Source config
to the ``container_tags_table`` parameter expected by
``MeasurementDBConfig``.
Expand All @@ -125,12 +127,12 @@ to the ``container_tags_table`` parameter expected by

**Returns**:

`MeasurementDB`: The measurement database instance.
`AbstractMeasurementDB`: The measurement database instance.

#### create\_query\_builder

```python
def create_query_builder(db: MeasurementDB,
def create_query_builder(db: AbstractMeasurementDB,
config: ImpulseConfig) -> QueryBuilder
```

Expand All @@ -143,7 +145,7 @@ supports tag and metric filters, but tag filters require the narrow

**Arguments**:

- `db` (`MeasurementDB`): The measurement database instance.
- `db` (`AbstractMeasurementDB`): The measurement database instance.
- `config` (`ImpulseConfig`): The Impulse configuration.

**Raises**:
Expand Down
26 changes: 25 additions & 1 deletion skills/impulse-config/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@ description: >
"configure an Impulse report", set the source/sink tables, filter which containers are processed,
choose RLE vs RAW, turn on incremental processing, run without writing (sinkless), remap column
names, or scope by project. Covers source, unity_sink, container_filters, query_engine, solver_config,
incremental, full_recalculation, measurement_dimensions, and calculated_channels, all validated by Pydantic.
custom measurement DBs, incremental, full_recalculation, measurement_dimensions, and calculated_channels,
all validated by Pydantic.
---

# Impulse — configuration
Expand Down Expand Up @@ -103,6 +104,8 @@ Omit entirely for `DefaultSolver` + `data_type="RLE"`.
| `drop_implausible_data` | `false` | Drop `channels` rows where `is_plausible = false`. **Requires `data_type="RAW"`** (RLE raises). |
| `max_channels_per_batch` | `500` | Max distinct channels (`channel(...)` / `poi_channel(...)` selections) read per solve batch. Applies to events, aggregations, and calculated channels (counts their *input* channels). `batch_size` is a **deprecated alias** (warns; removed in a future release). |
| `solver_config` | `null` | Per-table column mappings, per-table filters, and project scoping (below). |
| `measurement_db` | `"MeasurementDB"` | Registered name of the MeasurementDB (the read seam). The default reads the silver tables directly; a custom one reshapes a different source schema at read time. See [Custom measurement DB](#custom-measurement-db). |
| `measurement_db_config` | `null` | Keyword args for the selected MeasurementDB's config class, merged with `source` (e.g. extra tables, prefilter keys). |

## solver_config (optional) — adapt to an existing layout

Expand Down Expand Up @@ -176,6 +179,27 @@ to `channel_mapping` so aliased selectors auto-convert during `solve()`. Constan
an aliased selector must then be in the target unit. Direct `channel(...)` selectors are never
converted — conversion is a property of the alias.

## Custom measurement DB

`solver_config` only renames and filters columns. When the raw tables need *reshaping* to reach the
silver model (unions, joins, a grain change, an EAV unpivot), register a custom MeasurementDB and select
it by name — the stock solver then runs unchanged:

```python
"query_engine": {
"measurement_db": "MyMeasurementDB",
"measurement_db_config": {"session_table": "catalog.raw.sessions"},
}
```

- `measurement_db` is a **registered name**; the package that registers it must be imported before the
config is parsed. Validation runs at load time — an unregistered name, a missing required argument, an
unknown key, or a key already present in `source` all raise.
- `measurement_db_config` holds that implementation's own constructor kwargs, merged with `source`.
- Implementing one (subclass `MeasurementDB` + `MeasurementDBConfig`, decorate with
`@register_measurement_db`, and declare any extra tables in `configured_table_uris()` so they are
pinned per run) is covered in the Impulse docs' *Custom measurement DB* section.

## incremental (optional)

Reuses prior results for unchanged definitions and reprocesses only new/updated containers. See
Expand Down
17 changes: 4 additions & 13 deletions src/impulse_query_engine/measurement_db.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,9 @@
import warnings

from databricks.sdk import WorkspaceClient
from pyspark.sql import DataFrame

from impulse_query_engine import __version__
from impulse_query_engine.telemetry import verify_workspace_client
from .analyze.query.query_builder import QueryBuilder
from .measurement_db_abstract import AbstractMeasurementDB
from .measurement_db_registry import register_measurement_db


class MeasurementDBConfig:
Expand Down Expand Up @@ -100,15 +98,8 @@ def configured_table_uris(self) -> list[str]:
]


class MeasurementDB:
def __init__(self, config: MeasurementDBConfig, ws: WorkspaceClient):
self.config = config
self.ws = verify_workspace_client(ws, "databricks-impulse", __version__)

@property
def query(self):
return QueryBuilder(db=self)

@register_measurement_db("MeasurementDB", MeasurementDBConfig)
class MeasurementDB(AbstractMeasurementDB):
def _current_delta_version(self, spark, uri: str) -> int | None:
"""Latest committed Delta version of ``uri``, or ``None`` if unpinnable.

Expand Down
Loading
Loading