Batch calculated-channel solve into temp tables, mirroring aggregations/events - #80
Merged
Merged
Conversation
…ow solved_df - Move the narrow solve for calculated channels from each channel type into `Report._solve_calculated_channels_batched`, mirroring the existing batched expression solve path. - Add `solve_calculated_channels_batched` in `report_utils` to solve channels in configurable batches, persist each batch as a `__impulse_temp_*` Delta table or temp view, and combine results with `unionByName`. - Update `CalculatedChannel.determine_calculated_channels` and `dispatch_calculated_channels` to accept the pre-solved narrow DataFrame and only filter by `channel_id` and project to the fact schema. - Update unit and integration tests to exercise the new batched path, including multi-batch union, sink vs. sinkless persistence, and RAW-mode shaping.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #80 +/- ##
==========================================
+ Coverage 88.83% 88.89% +0.05%
==========================================
Files 61 61
Lines 5170 5194 +24
Branches 621 625 +4
==========================================
+ Hits 4593 4617 +24
Misses 466 466
Partials 111 111
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Calculated channels are now solved in batches, following the same pattern
already used for aggregations and events. The batching is driven from the
Reportclass: the channels are partitioned by
query_engine.batch_size, each batch issolved and persisted as a Unity Catalog temp table
(
__impulse_temp_{run_id}_{batch_idx}), and the batches are unioned into the finalcalculated_channel_fact.Why
Previously all calculated channels were solved in a single
query.select(*channels).solve_calculated_channels(...)call, unlike aggregationsand events which batch by
batch_sizeand persist intermediate results as temptables. Large channel sets therefore built one big lazy plan with no intermediate
materialization. This change gives calculated channels the same bounded,
batch-at-a-time execution and a consistent orchestration shape across all entity
types.
How
solve_calculated_channels_batched(report_utils.py): the narrow,row-append counterpart to
solve_expressions_batched. It reusesbuild_batches(selector-aware,
batch_size= max unique selectors per batch), solves each batchvia
query.select(*batch).solve_calculated_channels(...), persists each as a tempDelta table (or a Spark temp view when sinkless), and combines them with
unionByName. This is the one deliberate difference from the wide solver, whosebatches are one-row-per-container and combined with a
container_idjoin.Report: newReport._solve_calculated_channels_batched(parallel to
_solve_expressions_batched);determine_reportcollects thechanged/unchanged channel expressions, batch-solves each into a narrow
solved_df, and dispatches those.dispatch_calculated_channelsnow takessolved_df(droppingquery/solver), mirroringdispatch_aggregations.CalculatedChannel.determine_calculated_channelsno longer solves; it shapesthe already-solved
solved_df(filters bychannel_id, projects to the factschema), like
determine_aggregations.__impulse_temp_*prefix, so the start-of-runcleanup and the
unity_sink.cleanup_temp_tablesflag cover them automatically. Nonew config; reuses
query_engine.batch_size. Theadd_calculated_channelAPI isunchanged.
Tests
TestSolveCalculatedChannelsBatched(sinkless temp view, sink Delta table,unionByNamenot join, per-batch select, unique run_id) and the reworkedTestDetermineCalculatedChannels(shaping: filter by channel_id + project).test_batched_calculated_channels_unionrunsbatch_size=1with twodistinct-selector channels and asserts both land in the union plus two temp tables
are created; the RAW-mode test is updated to the new solve path.
report_utilsunit suite and the batched-pipelineintegration test confirm the shared batching/cleanup path still works for
aggregations and events.
Docs
Updated the channel reference, the
impulse-channelsskill, and thebatch_sizedescriptions in the config doc/skill; regenerated the pydoc-markdown API reference.
Test Plan
Checklist