Skip to content

Implement lazy subscription for Coalesce function #1322

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
shsms merged 26 commits into from
Dec 16, 2025
Merged

Implement lazy subscription for Coalesce function #1322

shsms merged 26 commits into from
Dec 16, 2025

Conversation

@shsms
Copy link
Contributor

@shsms shsms commented Dec 11, 2025

No description provided.

@shsms shsms requested a review from a team as a code owner December 11, 2025 11:06
@shsms shsms requested review from daniel-zullo-frequenz and removed request for a team December 11, 2025 11:06
@github-actions github-actions bot added part:docs Affects the documentation part:tests Affects the unit, integration and performance (benchmarks) tests part:tooling Affects the development tooling (CI, deployment, dependency management, etc.) part:data-pipeline Affects the data pipeline part:core Affects the SDK core components (data structures, etc.) part:microgrid Affects the interactions with the microgrid labels Dec 11, 2025
@shsms
Copy link
Contributor Author

shsms commented Dec 11, 2025

Based on #1295

@shsms shsms mentioned this pull request Dec 11, 2025
Merged
shsms added 17 commits December 15, 2025 11:49
@shsms
Use keyword-only arguments for AST node dataclasses
c1a4668 
Also use `None` as a default value for span, to avoid having to type
it when constructing higher level formulas.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Extract Node base class to separate module
a7dc25c 
Also rename it to AstNode

This way, nodes can be accessed by the _function.py module, without
circular imports.  That would enable functions to handle their own
arguments.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Update functions to fetch their own arguments
5e320a9 
This would enable special forms like `coalesce` to fetch only what's
necessary.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Propagate Sample types through AST evaluation
97510c7 
This changes `AstNode.evaluate()` to return
`Sample[QuantityT] | QuantityT | None` instead of `float | None`.

This makes timestamps available to coalesce node, so it knows how to
synchronize newly started fallback streams with the primary streams.

This also requires a `create_method` to be passed to TelemetryStream,
for accurate typing, so the `Quantity` types from the resampler don't
get sent out as they are, in case of simple formulas, because there is
no top-level re-wrapping of the values in the formula evaluator
anymore.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Make AST evaluate methods async
f888367 
Also update `Function.__call__()` to async.

This would make it possible for `coalesce` to start async streams as
necessary during evaluation.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Allow TelemetryStream AST nodes to subscribe to data on demand
2075ad3 
Earlier we were subscribing to all streams, even if they were going to
be discarded in a coalesce.

This commit makes further progress towards implementing lazy coalesce.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Return Sample after dividing QuantityT by float
a37a75a 
This is where we start to see the limitations of the string formula
spec.

For a power formula, we treat the constant in:
- `frequenz-floss#10 + 100.0` as a power value in watts.
- `frequenz-floss#10 / 100.0` as a float value.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Add subscribe method to AST nodes for stream initialization
187cf0e 
With this subscription can be delegated to individual nodes, allowing
coalesce to subscribe to only the interesting stuff.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Enable resend_latest on resampled stream channels
720748b 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
add NodeSynchronizer for coordinating AST node evaluation
c376dec 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Use NodeSynchronizer for concurrent node evaluation
93e5f33 
And drop the old centralized `synchronize_receivers` method,
delegating synchronization to individual nodes.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Update formula tests to not expect channels to be created in order
95c61ec 
Now that individual nodes control telemetry subscriptions, the order
in which subscriptions happen is not deterministic.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Add an unsubscribe method to AST nodes
d272263 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Handle missing timestamps gracefully in formula evaluator
7c2cf3c 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Update NodeSynchronizer to support syncing to the first node
298df08 
This will be needed for ongoing operation of `coalesce` nodes.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
move FunCall class from _ast to _functions module
631e5d2 
This enables us to import _ast from _functions, without it being a
circular dependency.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Implement lazy subscription for Coalesce function
a06702b 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
shsms added 6 commits December 15, 2025 11:49
@shsms
remove unused graph traversal methods
a08a925 
Most of these are provided by the rust component graph now.  The
`find_first_descendant_component` function needs to be replaced by the
coalesce formulas of the component graph, but not done yet.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Remove unused types-networkx from dev-mypy dependencies
5b1a7cf 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Remove unnecessary pylint exception in ComponentGraph instantiation
1ce6bcf 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Improve formula syntax docs for from_string method
bb9423c 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Make metric_fetcher a required argument for TelemetryStream
71b3215 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Add docstrings for class attributes in AST nodes
1cf3d4b 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms shsms added the cmd:skip-release-notes It is not necessary to update release notes for this PR label Dec 15, 2025
@shsms
Copy link
Contributor Author

shsms commented Dec 15, 2025

No release notes because this is an improvement to an unreleased feature, which will stop being a new feature, after its api is changed to be compatible with the old one.

llucax
llucax reviewed Dec 15, 2025
View reviewed changes
Copy link
Contributor

@llucax llucax left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM in general, just a few minor comments.

tests/timeseries/_formulas/test_formulas.py
io_pairs: list[tuple[list[float | None], float | None]],
) -> None:
"""Run a formula test."""
_logger.debug("TESTING FORMULA: %s", formula_str)
Copy link
Contributor

@llucax llucax Dec 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You'll get this for free when using parametrize 😉

@llucax llucax requested a review from Copilot December 15, 2025 13:13
Copilot AI reviewed Dec 15, 2025
View reviewed changes
Copy link

Copilot AI left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR implements lazy subscription for the Coalesce function in the formula engine. The key change is that Coalesce now subscribes to its parameters progressively: it starts by subscribing only to the first parameter, and if that returns None, it subscribes to the next parameter, and so on. This optimizes performance by avoiding unnecessary subscriptions when earlier parameters provide non-None values.

Key changes:

  • Converted formula evaluation from synchronous to asynchronous throughout the codebase
  • Implemented lazy subscription mechanism for Coalesce function
  • Refactored AST nodes to support async evaluation and subscription management
  • Changed ResampledStreamFetcher.fetch_stream() to be async and subscribe immediately rather than batching subscriptions

Reviewed changes

Copilot reviewed 17 out of 17 changed files in this pull request and generated 14 comments.

Show a summary per file
File Description
tests/timeseries/_formulas/utils.py Made get_resampled_stream async to match new async API
tests/timeseries/_formulas/test_formulas_3_phase.py Updated tests for async behavior, added event loop policy fixture
tests/timeseries/_formulas/test_formulas.py Enhanced tests with component IDs list, added debug logging, updated for async
tests/timeseries/_formulas/test_formula_composition.py Made test helper async, added extra receives for lazy subscription behavior
tests/microgrid/test_grid.py Made get_resampled_stream calls async
src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py Updated documentation for formula syntax
src/frequenz/sdk/timeseries/formulas/_token.py Added docstring for Component.id field
src/frequenz/sdk/timeseries/formulas/_resampled_stream_fetcher.py Made fetch_stream async, removed batch subscription logic
src/frequenz/sdk/timeseries/formulas/_parser.py Updated to create async metric fetchers, changed constant creation
src/frequenz/sdk/timeseries/formulas/_functions.py Implemented lazy subscription in Coalesce, made all functions async
src/frequenz/sdk/timeseries/formulas/_formula_evaluator.py Simplified evaluator to use async AST node evaluation
src/frequenz/sdk/timeseries/formulas/_formula_3_phase_evaluator.py Updated for async evaluation with synchronizer
src/frequenz/sdk/timeseries/formulas/_formula.py Added metric_fetcher helper, updated FormulaBuilder for async
src/frequenz/sdk/timeseries/formulas/_base_ast_node.py New file with AstNode base class and NodeSynchronizer
src/frequenz/sdk/timeseries/formulas/_ast.py Complete rewrite to support async evaluation and subscriptions
src/frequenz/sdk/_internal/_graph_traversal.py Removed unused helper functions
pyproject.toml Removed unused types-networkx dependency

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

llucax
llucax reviewed Dec 15, 2025
View reviewed changes
@llucax
Copy link
Contributor

llucax commented Dec 15, 2025

Oh, I forgot about a few minor comments on commit messages:

  1. Return Sample after dividing QuantityT by float says:

    This is where we start to see the limitations of the string formula
    spec.

    For a power formula, we treat the constant in:

    • #10 + 100.0 as a power value in watts.
    • #10 / 100.0 as a float value.

    Any idea on how to deal with this properly? Should we introduce quantities to formulas so for power #10 / 100.0 -> power but #10 / 100.0w -> float?

  2. Some commit messages are not properly capitalized:

    • add NodeSynchronizer for coordinating AST node evaluation
    • move FunCall class from _ast to _functions module
    • remove unused graph traversal methods

@shsms
Copy link
Contributor Author

shsms commented Dec 16, 2025

  1. Any idea on how to deal with this properly? Should we introduce quantities to formulas so for power #10 / 100.0 -> power but #10 / 100.0w -> float?

multiply and divide are not needed for the automatically generated formulas. I can't think of a usecase where people would need to divide by a quantity. So I think we don't need to worry about it yet.

@shsms
Apply suggestions from review comments
9b02348 
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
Copy link
Contributor Author

shsms commented Dec 16, 2025

I've put the review comment changes in a single commit. Don't want to change the commit ID, because I've pasted them in the previous PR, don't want to invalidate them. Will remember to not make lower case commit messages next time. 🤞🏽

@shsms shsms requested a review from llucax December 16, 2025 10:08
@shsms shsms mentioned this pull request Dec 16, 2025
Merged
@llucax
Copy link
Contributor

llucax commented Dec 16, 2025

I've put the review comment changes in a single commit. Don't want to change the commit ID, because I've pasted them in the previous PR, don't want to invalidate them. Will remember to not make lower case commit messages next time. 🤞🏽

Yeah, no problem, I agree at some point we need to relax as rebasing within big PRs it very costly. Balancing is good.

llucax
llucax approved these changes Dec 16, 2025
View reviewed changes
Copy link
Contributor

@llucax llucax left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎉

@github-project-automation github-project-automation bot moved this from To do to Review approved in Python SDK Roadmap Dec 16, 2025
@shsms shsms added this pull request to the merge queue Dec 16, 2025
Merged via the queue into frequenz-floss:v1.x.x with commit 6721560 Dec 16, 2025
5 checks passed
@shsms shsms deleted the lazy-coalesce branch December 16, 2025 13:20
@github-project-automation github-project-automation bot moved this from Review approved to Done in Python SDK Roadmap Dec 16, 2025
@llucax llucax added this to the v1.0.0-rc2300 milestone Dec 16, 2025
github-merge-queue bot pushed a commit that referenced this pull request Dec 16, 2025
@shsms
Revert unreleased interface changes (#1330)
f539389 
These changes were introduced in the PRs #1295 and #1322, and are being
reverted here.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

Copilot code review Copilot Copilot left review comments

@llucax llucax llucax approved these changes

@daniel-zullo-frequenz daniel-zullo-frequenz Awaiting requested review from daniel-zullo-frequenz daniel-zullo-frequenz is a code owner automatically assigned from frequenz-floss/python-sdk-team

Assignees

No one assigned

Labels

cmd:skip-release-notes It is not necessary to update release notes for this PR part:core Affects the SDK core components (data structures, etc.) part:data-pipeline Affects the data pipeline part:docs Affects the documentation part:microgrid Affects the interactions with the microgrid part:tests Affects the unit, integration and performance (benchmarks) tests part:tooling Affects the development tooling (CI, deployment, dependency management, etc.)

Projects

Python SDK Roadmap
Status: Done

Milestone

v1.0.0-rc2202

Development

Successfully merging this pull request may close these issues.

2 participants

@shsms @llucax