Repository navigation
Feature/configurable db - #109
Conversation
430b928 to
43e228a
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #109 +/- ##
==========================================
+ Coverage 90.21% 90.29% +0.08%
==========================================
Files 66 68 +2
Lines 6009 6060 +51
Branches 757 758 +1
==========================================
+ Hits 5421 5472 +51
Misses 466 466
Partials 122 122
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
tombonfert
left a comment
There was a problem hiding this comment.
Thanks, the registry and config-class split reads cleanly. Three things I'd like addressed before merge:
1. Docs: optional tables are enabled by their source slot (custom_measurement_db.md)
The solver only reads container_tags, channel_tags, channel_mapping and unit_conversion when the matching source.*_table is set (e.g. default_solver.py:486, :669). If the slot is empty, an override of that method is never called. For channel_tags and unit_conversion this happens without an error: the solver falls back to wide mode or skips unit conversion. The current wording ("Override one method per silver table you need to reshape") suggests the override alone is enough. Could you document that the source slot turns the optional table on and the override reshapes what that slot reads?
2. Telemetry relies on custom DBs calling super().__init__ (measurement_db_abstract.py:25)
Report and QueryBuilder take their workspace client from db.ws (report.py:133, query_builder.py:38). If a custom DB overrides __init__ without calling super().__init__, or sets self.ws itself, the product tag and the connectivity check are lost without any error. A read-only property turns this into a clear error on both the Report and the standalone path:
def __init__(self, config: "MeasurementDBConfig", ws: WorkspaceClient):
self.config = config
self._ws = verify_workspace_client(ws, "databricks-impulse", __version__)
@property
def ws(self) -> WorkspaceClient:
try:
return self._ws
except AttributeError:
raise TypeError(
f"{type(self).__name__}.__init__ must call super().__init__(config, ws)"
) from NonePlease also add "call super().__init__(config, ws)" to the Rules section of the docs.
3. Misspelled query_engine keys are silently ignored (config_parser.py:362)
QueryEngine doesn't forbid extra keys, so a typo like "measurment_db" falls back to the built-in DB. The report then runs on the raw tables and writes wrong gold data. The docs promise a parse-time failure (custom_measurement_db.md:98). Setting extra="forbid" on QueryEngine fixes this. Since configs with stray keys will start failing, it's worth a changelog note.
…n impl of measurement_db.py
13c1451 to
61f6477
Compare
Customer‑overridable
MeasurementDB: read any source schema with the stock solverSummary
Impulse's solver expects data in its silver model. Customers whose data is stored
differently have no supported way to plug in their layout:
SolverConfigcovers renames andnarrow‑vs‑wide, but nothing beyond that. This PR lets a customer select their own
MeasurementDBimplementation by name in the report config. It reshapes their tables into the silver model at
read time; the stock
DefaultSolverthen runs unchanged, and every read path (solve,incremental detection, ad‑hoc) sees the same data. The change is additive: if nothing is
configured, the built‑in
MeasurementDBis used. The one breaking change is stricter parsing ofquery_engine(below).Breaking change
query_enginenow rejects unknown keys. A config with a misspelled or leftover key underquery_engine(for example"measurment_db") fails at load time withExtra inputs are not permittedinstead of being silently ignored. The deprecatedbatch_sizealias still works.Changes
AbstractMeasurementDB(impulse_query_engine/measurement_db_abstract.py)with a read-only
wsproperty that raises aTypeErrorif a subclass skipssuper().__init__(config, ws). The built-inMeasurementDBsubclasses it and registers itself.register_measurement_db/resolve_measurement_db(
impulse_query_engine/measurement_db_registry.py).query_engine.measurement_dbandquery_engine.measurement_db_config, validated whenthe config is loaded;
QueryEngineforbids unknown keys (impulse_reporting/config/config_parser.py).Report.create_measurement_dbbuilds the selected implementation(
impulse_reporting/core/report.py).query_enginerows on the configurationpage, regenerated API reference, updated
impulse-configskill.Review feedback addressed
its override, when the matching
sourceentry is set.ws: read-only property that raises ifsuper().__init__(config, ws)is skipped, plus thematching rule in the docs.
query_enginekeys:extra="forbid"onQueryEngine(see Breaking change).their docs.
Test Plan
AbstractMeasurementDBcontract and registry (including thewsproperty),extra-table pinning,
build_measurement_db_config, parse-time rejection of misspelledquery_enginekeys and of badmeasurement_db_configkeys, andReportfailing with a clear errorfor a DB that skips
super().__init__custom_measurement_db_report_test.py): the selected DB serves all reads, itsconfig reaches the DB, incremental detection reads through it, and a declared extra table is pinned
query_enginekeys)Checklist
Design
Throughout this doc, Acme is an example customer whose raw layer does not match the silver model.
Contents
1. Why this is necessary
1.1 Customers don't store data in the silver model
Impulse's query engine assumes silver tables such as
container_metrics,channel_metrics,channels,poi_channelsand the EAV*_tags. Real customers may store data in a different way.Acme's raw layer, for example, keys containers by a string session id and channels by a 3‑part
string signal name, stores chunked statistics instead of one row per channel, keeps wide reference
tables instead of EAV tags, and stores raw points instead of intervals.
Renames and narrow‑vs‑wide shape are already handled by
DefaultSolver+SolverConfig. Acme needsmore than that: unions, joins, a grain collapse and an EAV unpivot.
Example: a row filter that must run before the reshape. Acme's metadata and statistics tables
carry a marker column
data_type, and only rows withdata_type = 3may be used: they are the onerepresentative row per session. Without that filter, joining these rows to the recording sessions
fans out and duplicates sessions. Acme's app already applies it before that join.
Stock Impulse cannot express this.
SolverConfig.<table>.filtersrun after theMeasurementDBhas returned its frame (
QuerySolver.scoped_container_metrics,query_solver.py:79-80). After areshape, that frame is already joined and collapsed: the
data_typecolumn is gone and theduplicates have already been produced. The filter has to be applied to the raw rows, before the
join and the aggregation, which is only possible inside a custom
MeasurementDB.1.2 Performance depends on the customer's clustering keys
Customers' existing pipelines are fast because they read narrowly, filtering on their tables'
clustering keys (for Acme: date, recording session and signal). The stock Impulse read path does not
filter on those keys, so on the same tables it reads far more data. To reach comparable performance,
reads must filter on the customer's clustering keys at the raw table (§3.2).
1.3 There is no extension point today
The only way to support Acme's layout today is to change Impulse itself.
Report.create_measurement_db(report.py:216-241) always builds the built‑inMeasurementDB, andReport.create_solver(report.py:317) is a closedmatchover theSolversenum that onlyreturns
DefaultSolver.2. The concept
2.1 In one picture
The
MeasurementDBis the layer between the stored tables and everything that reads them. This PRmakes it replaceable by name: the customer ships a small
MeasurementDBsubclass (plus a configclass) in their own package, and the report config selects it.
flowchart LR RAW["Customer raw tables<br/>(any schema)"] -->|"prefilter on raw<br/>clustering keys"| CDB["Customer MeasurementDB<br/>(reshape → silver)"] CDB --> S["Stock DefaultSolver<br/>solve()"] CDB --> I["Incremental detection"] CDB --> A["Ad‑hoc / exports"] CFG["Report config<br/>query_engine.measurement_db = 'AcmeMeasurementDB'"] -.->|"resolved by name"| REG["measurement_db registry"] REG -.-> CDBEvery consumer reads through the same customer DB, so they all see identical silver‑shaped data.
In the report config, only two optional keys are new:
2.2 What it offers customers with different schemas
reference tables, raw points: all are reshaped when the tables are read.
unchanged. Histograms, events, statistics and POI all run on the stock solver.
container_metricsas thesolve, so the silver↔gold join aligns (§4.4).
the silver tables (§4.5).
skips files outside the selected dates, sessions and signals (§3.2).
of Impulse core.
MeasurementDB.2.3 Which extension point to use
DefaultSolver+SolverConfig(no code)MeasurementDB(this PR)query.solve(solver=…); reports only accept the built-in solvers (see the "Custom solvers" docs)2.4 Alternatives considered
The obvious alternative is to leave Impulse unchanged and have the customer materialize the
silver model upstream: a scheduled job that writes
container_metrics,channel_metrics,channels,poi_channelsand the tag tables, or SQL views over the raw layer.We want customers to be able to adopt Impulse smoothly, including those who don't want to build
and run another set of tables and a pipeline just to feed it. A custom
MeasurementDBgives them theflexibility to define their own reshape in code, on top of the data they already have:
MeasurementDB(this PR)Materializing silver upstream remains fully supported: a customer who prefers it keeps using the
built‑in
MeasurementDBand needs none of this.A pluggable solver that reshapes inside its read stages is the other alternative. It is the
wrong layer, because the solver is not the only reader of silver data: incremental detection and
changed‑entity scoping read
container_metricsoutsidesolve(). The details are in §4.4.3. The Acme example: workflow, data schema and clustering keys
3.1 Acme's raw layer vs. the silver model
Acme raw layer
erDiagram SESSION ||--o{ SIGNAL : "session_id = recording_session_id" SESSION ||--o{ SIGNAL_STATISTICS : "session_id = recording_session_id" SESSION ||--o{ POI : "session_id = recording_session_id" SIGNAL }o--|| SIGNAL_METADATA : "signal_name" SIGNAL_STATISTICS }o--o| REF_TEST : "test_id" REF_TEST }o--|| REF_VEHICLE : "vehicle_id" REF_VEHICLE }o--|| REF_FLEET : "fleet_id" SIGNAL { string recording_session_id FK string signal_name double value_double string value long timestamp_abs date dt } SIGNAL_STATISTICS { string recording_session_id FK string signal_name int count float min float max float mean long start_ts long end_ts int test_id date dt } POI { string recording_session_id FK string poi_type string value long timestamp_abs date dt } SESSION { string session_id PK timestamp start_datetime timestamp end_datetime } REF_TEST { int test_id PK int vehicle_id FK string test_name } REF_VEHICLE { int vehicle_id PK int fleet_id FK string vin } REF_FLEET { int fleet_id PK string fleet_name } SIGNAL_METADATA { string signal_name PK string unit }Impulse silver model (what the solver requires), shown with Impulse's reference id types
(
long container_id,int channel_id). Acme's ids are strings; see the id contract in §4.3.erDiagram CONTAINER_METRICS ||--o{ CONTAINER_TAGS : "container_id" CONTAINER_METRICS ||--o{ CHANNEL_METRICS : "container_id" CONTAINER_METRICS ||--o{ POI_CHANNELS : "container_id" CHANNEL_METRICS ||--o{ CHANNEL_TAGS : "container_id, channel_id" CHANNEL_METRICS ||--o{ CHANNELS : "container_id, channel_id" CONTAINER_METRICS { long container_id PK timestamp start_dt timestamp stop_dt int num_channels long odo_start long odo_stop array dtc_error_codes } CHANNEL_METRICS { long container_id FK int channel_id int sample_count float min float max float mean } CHANNELS { long container_id FK int channel_id FK long tstart long tend double value } POI_CHANNELS { long container_id FK int channel_id long timestamp double value_double string value_string } CONTAINER_TAGS { long container_id FK string key string value } CHANNEL_TAGS { long container_id FK int channel_id FK string key string value }3.2 Clustering keys and the prefilter
Acme's raw tables are liquid‑clustered on
dt,recording_session_id,signal_name: thepartition date, the recording‑session identity and the physical signal identity. Acme's existing
pipeline is fast because it reads narrowly on exactly those keys; if Impulse reads the same tables
without that scope, it full‑scans them. So the custom DB pushes a clustering‑key prefilter at the
raw read:
measurement_db_config)recording_session_id: list[str]recording_session_id IN (…)signal_name: list[str]signal_name IN (…)start_dt/stop_dtdt >= …/dt <= …, no cast (a single day sets both to the same date)Which clustering keys each read pushes
dtrecording_session_idsignal_namechannels()signalchannel_metrics()signal_statisticschannel_tags()signal_metadatacontainer_metrics(),container_tags()signal_statisticsnum_channels,odo_*)poi_channels()poiRule: apply the prefilter on the raw physical columns, before the reshape, and compare
dtto a same‑type literal without a cast. Renaming alone does not stop pruning (Catalyst pushes
filters through aliases). What does stop it is a
CAST/function on the clustering column, a typemismatch, or an optimization barrier between filter and scan (
groupBy, join,checkpoint). Areshape contains all of those, so a filter applied after it cannot prune the source read.
Known gap (customer package, owned by the customer team): the 5‑minute variant tables
(
signal_statistics_5m,signal_metadata_5m) are unioned in after the prefilter, so today theyare read without any clustering‑key filter. The fix is to apply the same prefilter to them before
the union (see §4.6).
3.3 Typical Acme workflow
An Acme analyst builds a report in the web app in three selection steps. Each step becomes one
clustering‑key prefilter, so the report run only touches the files for that date range, those
recording sessions and those signals.
flowchart TD D["1. Pick date range<br/>start / stop"] --> C["2. Narrow candidate recording sessions<br/>fleet · vehicle (VIN) · geo area · vehicle attributes ·<br/>POI · datalogger · distance / odometer"] C --> P["Pick recording sessions<br/>(picker shows run id · test id · date)"] P --> S["3. Pick logical signals<br/>e.g. ENGINE_SPEED, VEHICLE_SPEED, ODOMETER"] S --> R["Resolved to physical signal names<br/>ENGINE_SPEED:12:A1, …"] R --> A["4. Define events, histograms, statistics<br/>on the logical signals"] A --> G["App generates the report config<br/>measurement_db = AcmeMeasurementDB + prefilter"] G --> RUN["5. Report run"] RUN --> PIN["pin_versions<br/>(silver + extra tables)"] PIN --> RD["Prefiltered raw reads<br/>dt · recording_session_id · signal_name"] RD --> RS["Reshape → silver"] RS --> SOL["Stock DefaultSolver → gold"]measurement_db_configstart_ts/stop_ts)start_dt/stop_dt(date part); for a single day both are the same datedtrecording_session_id: [...](the picked ids)recording_session_idsignal_name: [...]signal_nameThe filter criteria in step 2 (fleet, VIN, geo, attributes, POI, odometer…) only build the
candidate list in the app. The report itself filters on the picked recording session ids,
which is why they can be pushed down as a clustering key.
3.4 How
AcmeMeasurementDBmaps Acme to silverchannels()signalvalue;value_double→value,timestamp_abs→timestamp(µs), idschannel_metrics()signal_statistics(+ 5‑min variant)(container, channel); fold POI channel‑metrics stubcontainer_metrics()signal_statistics(+ 5‑min variant)num_channels,odo_*, bounds via session); folddtc_error_codes[]container_tags()signal_statistics+ref_*test→vehicle→fleet; unpivottest_name/vin/fleet_nameto EAVchannel_tags()signal_metadata(+ 5‑min variant)unit+channel_nameto EAVpoi_channels()poipoi_type→channel_id,value→value_stringchannel_mapping()signal_metadatasignal_name; if several physical names share an alias, the lexically first winsA second customer with a different layout writes its own subclass and config class, registers it
under its own name, and reuses the same framework mechanism.
3.5 Acme's report config
{ "source": { "container_metrics_table": "acme.raw.signal_statistics", "container_tags_table": "acme.raw.signal_statistics", "channel_metrics_table": "acme.raw.signal_statistics", "channel_tags_table": "acme.raw.signal_metadata", "channel_mapping_table": "acme.raw.signal_metadata", "channels_uri": "acme.raw.signal", "poi_channels_uri": "acme.raw.poi" }, "query_engine": { "solver": "DefaultSolver", "data_type": "RAW", "measurement_db": "AcmeMeasurementDB", "measurement_db_config": { "recording_session_table": "acme.raw.recording_session", "start_dt": "2026-04-01", "stop_dt": "2026-04-30", "recording_session_id": ["<session-id-1>"], "column_name_mapping": { "channels": { "recording_session_id": "container_id", "signal_name": "channel_id" } }, "…": "further extra tables and prefilter keys" } } }sourcekeeps its shape: one table per silver role. With a custom DB these entries point atthe customer's raw tables, which the base read methods (
super().<table>(spark)) read and pin.query_engine.measurement_db_configholds everythingsourcecannot express: extra tables,column mappings, prefilter keys, customer‑specific flags.
query_engine.solver_configkeeps solver behaviour only (e.g.channel_mapping.join_keys).Column renames belong in
measurement_db_config, because the DB already returns internal names.4. Technical details
4.1 Framework changes (Impulse side, additive)
An abstract contract.
AbstractMeasurementDB(ABC)(
impulse_query_engine/measurement_db_abstract.py, re‑exported frommeasurement_db.py).It implements
__init__(config, ws)(including the telemetryverify_workspace_client), aread-only
wsproperty that raises aTypeErrorif a subclass skippedsuper().__init__(config, ws), and the.queryproperty, and declares as abstractcontainer_metrics,channel_metrics,channels,container_tags,channel_tags,poi_channels,has_poi_channels,channel_mapping,unit_conversion,channel_uriandpin_versions(called byReportat run start, §4.5).MeasurementDBsubclasses it;Report.get_db/create_measurement_dbare typed with it.A name registry.
register_measurement_db(name, config_cls)andresolve_measurement_db(name)(
impulse_query_engine/measurement_db_registry.py), storing a frozenMeasurementDatabaseRegistration(db_cls, config_cls). The decorator raisesTypeErrorunless theclass subclasses
AbstractMeasurementDB;config_clsis aMeasurementDBConfigsubclass. Thebuilt‑in registers itself as
("MeasurementDB", MeasurementDBConfig). Re‑registering a namereplaces the previous entry.
A config selector.
query_engine.measurement_db: str = "MeasurementDB"andquery_engine.measurement_db_config: dict | Noneinconfig_parser.py, next tosolver/solver_config.QueryEnginenow forbids unknown keys (extra="forbid"), so a misspelled keyfails when the config is loaded instead of silently selecting the built-in DB (see Breaking change).
query_engine.measurement_dbstr"MeasurementDB"query_engine.measurement_db_configdictnullsourcetablesFactory resolution.
build_measurement_db_config(source_dict, query_engine)inconfig_parser.pyresolves the registration and builds one config object,config_cls(**source_dict, **measurement_db_config, table_locations="unity_catalog").Report.create_measurement_dbkeeps its existingsource_dictpreparation and then calls it;the result is
db_cls(config, ws). With no selector this is exactly the previousMeasurementDBConfig(**source_dict)plus the built‑inMeasurementDB.No change to the pinning logic in
MeasurementDB.4.2 How the customer's config reaches the DB
flowchart LR S["source<br/>(silver-role tables)"] --> B["build_measurement_db_config<br/>config_cls(**source, **measurement_db_config)"] J["query_engine.measurement_db_config<br/>(customer settings)"] --> B N["query_engine.measurement_db<br/>'AcmeMeasurementDB'"] -->|"resolve_measurement_db"| R["registration<br/>db_cls · config_cls = AcmeDBConfig"] R --> B B --> C["AcmeDBConfig instance"] C --> F["AcmeMeasurementDB(config, ws)"]ImpulseConfigcallsbuild_measurement_db_configonce as acheck, so these fail as a Pydantic
ValidationErrorwhen the config is loaded:measurement_db_configor anywhere underquery_engine;__init__performs.Report.create_measurement_dbcallsbuild_measurement_db_configagain andreturns
db_cls(config, ws).The config class is therefore constructed twice, so its
__init__must not have side effects.4.3 What a customer implements
Read‑method pipeline. Every override follows the same steps:
super().<table>(spark)(raw read) → prefilter → optional union/join of extra tables →column_name_mapping(physical → internal) → per‑table reshape → silver‑shaped frame with internal ids.Rules for the customer:
MeasurementDBConfigand callsuper().__init__(**base_kwargs). The baseinitializer sets
table_locations,debug_tablesandpinned_versions, which pinning and_read_tablerely on.configured_table_uris()so it is pinned (§4.5).generate or edit configs. An unregistered name fails with
Unknown measurement DB 'AcmeMeasurementDB'; registered: ['MeasurementDB']. Import the package that registers it before parsing the report config.as missing; a field with a default like
=Noneis silently accepted when left out.__init__free of side effects; it runs once for the check and once for thebuild.
__init__(self, config, ws); Impulse builds the DB asdb_cls(config, ws).__init__, callsuper().__init__(config, ws)and don't assignself.wsyourself. Reading
db.ws(whichReportanddb.querydo) raises aTypeErrorotherwise.sourceentry for every optional table you override (container_tags_table,channel_tags_table,channel_mapping_table,unit_conversion_table,poi_channels_uri).Otherwise the solver doesn't use the table and your override is never called; for channel tags,
unit conversion and POI this happens without an error.
MeasurementDBConfig.for_debug(...)returns the base class, so a customconfig provides its own debug constructor, and
debug_tablesmust include the extra tables.QueryEngine:QueryEngine(solver="DefaultSolver", measurement_db="AcmeMeasurementDB", measurement_db_config={...}).Supported for subclasses. A custom DB and its config may rely on these members, and changing
them is a breaking change for implementers:
MeasurementDBConfig.__init__keyword arguments,configured_table_uris()andtable_locationsMeasurementDB._read_table(spark, uri)super().<table>(spark)) as the raw, pinned read of each configured table4.4 Why the reshape belongs in the
MeasurementDB, not in the solverMeasurementDB(read layer)solve(), incremental container detection, changed‑entity scoping, ad‑hoc analysis, exportsQuerySolver(engine)solve()and the container pipeline, and each stage reads the DB itselfIncremental processing breaks with a solver‑side reshape. Incremental reporting reads
container_metricsoutsidesolve()and joins it against gold. The call chain:Reportstarts incremental mode withpre_filtered_containers_df = self._detect_upserted_containers()(
report.py:1068)._detect_upserted_containers(report.py:1428-1464) gets its inputs from_container_detection_args(report.py:1507-1527), which readssilver_containers = self._filtered_container_metrics()(report.py:1524)._filtered_container_metrics(report.py:1419-1426) callssolver.filter_container_tagsandsolver.filter_container_metrics, which end inQuerySolver.scoped_container_metrics(
query_solver.py:46-81) and the raw readquery.db.container_metrics(spark)(query_solver.py:73).ContainerUpsertDetector.detect_upserted_containers(container_detector.py:48-100) left‑anti‑joinsand inner‑joins that frame against gold
measurement_dimensiononcontainer_idand compares thelast_modifiedfreshness columns (container_detector.py:137-207).A reshape that lives in the solver breaks this chain in three places:
passed back into the solve as
pre_filtered_containers_df.scoped_container_metricsthen skipsthe DB read and uses the subset as‑is (
query_solver.py:70-71), but still re‑applies the columnmapping (
query_solver.py:74-76). That is harmless for renames (withColumnRenamedon a missingcolumn is a no‑op). It is not harmless for a real reshape: a solver that collapses chunk‑grain
statistics into per‑container rows would have to tell raw input (line 73) from already‑reshaped
input (line 71) inside the same method, or it reshapes twice. With the reshape in the DB, both
branches carry the same silver shape and the mapping is a no‑op.
report.py:1520-1523records the bug that already happened once: "Reading the raw table herewould break configs that remap the physical container‑id column (Incremental container detection ignores column_name_mapping and assumes a physical container_id column #99)." Incremental container detection ignores column_name_mapping and assumes a physical container_id column #99 fixed detection by
routing it through the solver pipeline.
_changed_scope(report.py:1480-1505) reuses the samesilver_containersframe and joins it against gold ids (report.py:1500-1505). Each new reader ofcontainer_metricsrepeats this risk. With the reshape in the DB, any reader, including a plaindb.container_metrics(spark), gets the reshaped frame by construction.container_idis its session id, renamed from the physical
recording_session_id. Goldmeasurement_dimensionstores the id the solve produced, and detection compares against it. If the rename is applied in
more than one code path and they drift, detection sees a different column or value than the solve
wrote. Every container then looks new on every run (silent full reprocessing), or updated
containers are missed. That is exactly the Incremental container detection ignores column_name_mapping and assumes a physical container_id column #99 bug. A DB‑side reshape applies the mapping in one place.
The solver reads the DB in many places, not one hook.
DefaultSolverstages read the DB directlyat
default_solver.py:316(container_tags),:503(channel_tags),:556and:652(
channel_metrics),:1142(channels) and:1209(poi_channels), plusquery_solver.py:73and:195. A solver‑side reshape would have to override every stage that touches one of these tables andfollow upstream changes to them. A DB override implements the read methods behind a stable, abstract
contract (§4.1).
Fewer plan workarounds. Reshaping inside solver stages means re‑reading the same raw tables within
one plan and joining across the copies. That trips
MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_APPEAR_IN_OPERATIONand needs.distinct()/localCheckpointworkarounds. The read layer is the natural single owner of those reads.4.5 Snapshot consistency (Delta version pinning)
At the start of each run
Reportcallsself.db.pin_versions(self.spark)(report.py:1039), so alllazy stages read one Delta snapshot (issue #87). How it works today:
MeasurementDB.pin_versionspins the tables inMeasurementDBConfig.configured_table_uris()(
measurement_db.py:85-100,:140-170)._read_table(spark, uri)looks upconfig.pinned_versions[uri]and addsversionAsOf(
measurement_db.py:172-181).(
measurement_db.py:112-138).A reshape reads extra tables that are not in that list (sessions, reference tables, 5‑minute
statistics), so they are not pinned automatically.
Decision: the implementer declares its extra tables; Impulse's inherited
pin_versionspins them.The pinning logic itself stays unchanged; the only addition is declaring
pin_versionson theabstract class.
AbstractMeasurementDBdeclarespin_versions(spark)as an abstract method.Reportalready calls it, so this only makes the existing de‑facto contract explicit.Otherwise a registered implementation that doesn't inherit
MeasurementDBwould pass registrationand then fail with
AttributeErroratreport.py:1039.configured_table_uris()toreturn
super().configured_table_uris()plus the extra URIs. The inheritedpin_versionsiterates exactly that list, so the extras are pinned together with the silver tables, including
the base handling of debug mode, views and non‑Delta tables.
configured_table_uris()is onlyused by
pin_versions(measurement_db.py:166), so extending it has no other effect.self._read_table(spark, uri), neverspark.read.table. Thisincludes transformers: they receive the reader or the frame from the DB and do not read tables
themselves.
debug_tables, since_read_tablelooksthem up by name there.
A DB that implements
AbstractMeasurementDBwithout inheritingMeasurementDBhas no inheritedpin_versionsand must implement pinning itself.4.6 Scope of this PR
measurement_db_abstract.py:AbstractMeasurementDB, withpin_versionson the contractand a read-only
wsproperty;measurement_db_registry.py:register_measurement_db,resolve_measurement_db,MeasurementDatabaseRegistration;measurement_db.py:MeasurementDBsubclasses the ABC and registers itself; its__init__andquerymoved to the ABC unchanged;config_parser.py:query_engine.measurement_db/measurement_db_config,build_measurement_db_config, the parse‑time validator onImpulseConfig, andextra="forbid"onQueryEngine;report.py: registry resolution inReport.create_measurement_db;query_enginerows indocs/impulse/docs/config/configuration.md, a newExtensibility → Custom measurement DB page
(
docs/impulse/docs/extensibility/custom_measurement_db.md), the regenerated API reference, and theimpulse-configskill.configured_table_uris()needs a
databricks-impulsethat has version pinning (issue Pin silver input tables to fixed Delta versions for a report run #87). Older releases don't pin at all.Report(db=…)instance‑injection escape hatch.Out of scope: owned by the customer team (separate repo).
AcmeMeasurementDBand itstransformers live in the customer package and are not part of this PR. To follow the implementer
rules in §4.3 and §4.5, that team:
MeasurementDBConfigsubclass and declares its extra tables(sessions, reference tables, 5‑minute statistics) in
configured_table_uris();spark.read.tablecalls in its transformers through the injected_read_table;"solver": "DefaultSolver"+"measurement_db": "AcmeMeasurementDB".This PR defines the contract they build against; Acme appears in this doc as the reference use case.
5. Accepted risks
This PR provides the option to plug in a custom
MeasurementDB. The customer who implements onetakes responsibility for it:
custom DB returns. The implementer must provide the columns, types, row grain and time unit the
solver expects; otherwise the report fails at solve time or silently computes wrong results. The
reference is the built‑in
MeasurementDBplus the silver schemas inimpulse_query_engine/schema.py(fordata_type: RAW,channels()returns point rows as inCHANNELS_SCHEMA_WITHOUT_RLE).configured_table_uris(), or read outside_read_table, is not pinned and reads the latest version.measurement_db.py:125-131).clustering‑key prefilter keeps that bounded (§3.2).
config is parsed, so every tool that parses a config (including ones that only generate or edit
configs) must import the package that registers the DB first.
MeasurementDBConfig.__init__andconfigured_table_uris(), andMeasurementDB._read_table(§4.3). Changing them is a breakingchange for implementers.
This pull request and its description were written by Isaac.