fix sync for map task in dynamic #3141

troychiu · 2025-02-19T07:23:23Z

Why are the changes needed?

Currently if you put map task inside dynamic and sync the execution, you will get an error.

Example

from flytekit import workflow, task, map_task, dynamic

threshold = 11

@task(retries=2)
def detect_anomalies(data_point: int) -> bool:
    return data_point > threshold


@dynamic
def detect_anomalies_wrapper(data: list[int]):
    map_task(detect_anomalies)(data_point=data)

@task
def return_data(i: int) -> list[int]:
    return [x for x in range(i)]


@workflow
def wf(i: int = 13):
    data = return_data(i=i)
    detect_anomalies_wrapper(data=data)

r = FlyteRemote(config=Config.for_endpoint(endpoint=endpoint))

execs = r.recent_executions(project=project, domain=domain, limit=limit)
a = r.sync(execs[0], sync_nodes=True)

FlyteEntityNotExistException: USER:EntityNotExist: error=None, cause=<_InactiveRpcError of RPC that terminated with:
        status = StatusCode.NOT_FOUND
        details = "missing entity of type TASK with identifier project:"troy"  domain:"development"
name:"dynamic_map.map_detect_anomalies_6b3bd0353da5de6e84d7982921ead2b3-arraynode"  version:"Cp3YAs4ZP6F-RMF18m5ioA""
        debug_error_string = "UNKNOWN:Error received from peer  {created_time:"2025-02-19T13:51:10.932195-08:00", grpc_status:5, grpc_message:"missing entity of type TASK
with identifier project:\"troy\"  domain:\"development\"  name:\"dynamic_map.map_detect_anomalies_6b3bd0353da5de6e84d7982921ead2b3-arraynode\"
version:\"Cp3YAs4ZP6F-RMF18m5ioA\""}"

This is because we try to fetch task but we don't register task inside a dynamic.

What changes were proposed in this pull request?

Store flyte_task_node in array_node so that we can use later

How was this patch tested?

Use the example above and I am able to get the correct data.
Integration test

Check all the applicable boxes

I updated the documentation accordingly.
All new and existing tests passed.
All commits are signed-off.

Related PRs

Docs link

Summary by Bito

This PR fixes a critical synchronization bug in map tasks within dynamic workflows where task references were missing. The changes enhance FlyteArrayNode class by adding type hints and storage for flyte_task_node references, improving error handling, and updating execution synchronization logic to use correct node access paths. The implementation eliminates the need for fetching unregistered tasks and includes proper validation of workflow execution phases.

Unit tests added: True

Estimated effort to review (1-5, lower is better): 2

Signed-off-by: Troy Chiu <[email protected]>

flyte-bot · 2025-02-19T07:23:37Z

Code Review Agent Run Status

Limitations and other issues: ❌ Failure - The AI Code Review Agent skipped reviewing this change because it is configured to exclude certain pull requests based on the source/target branch or the pull request status. You can change the settings here, or contact the agent instance creator at [email protected].

codecov · 2025-02-19T07:44:26Z

Codecov Report

Attention: Patch coverage is 84.61538% with 2 lines in your changes missing coverage. Please review.

Project coverage is 58.26%. Comparing base (66d4aed) to head (32e4142).
Report is 11 commits behind head on master.

Files with missing lines	Patch %	Lines
flytekit/remote/entities.py	83.33%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##           master    #3141       +/-   ##
===========================================
- Coverage   76.85%   58.26%   -18.60%     
===========================================
  Files         206      206               
  Lines       21851    21788       -63     
  Branches     2837     2837               
===========================================
- Hits        16794    12695     -4099     
- Misses       4269     8310     +4041     
+ Partials      788      783        -5

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Signed-off-by: Troy Chiu <[email protected]>

flyte-bot · 2025-02-19T19:20:54Z

Code Review Agent Run Status

Limitations and other issues: ❌ Failure - The AI Code Review Agent skipped reviewing this change because it is configured to exclude certain pull requests based on the source/target branch or the pull request status. You can change the settings here, or contact the agent instance creator at [email protected].

Signed-off-by: Troy Chiu <[email protected]>

flyte-bot · 2025-02-19T22:50:55Z

Code Review Agent Run #462f2a

Actionable Suggestions - 4

tests/flytekit/integration/remote/workflows/basic/dynamic_array_map.py - 1
- Consider more descriptive function name · Line 9-11
tests/flytekit/integration/remote/test_remote.py - 1
- Consider validating execution structure before access · Line 630-630
flytekit/remote/entities.py - 2
- Consider adding type hints to parameters · Line 351-351
- Consider extracting task validation logic · Line 493-504

Review Details

Files reviewed - 4 · Commit Range: 8ba253c..19b1562
- flytekit/remote/entities.py
- flytekit/remote/remote.py
- tests/flytekit/integration/remote/test_remote.py
- tests/flytekit/integration/remote/workflows/basic/dynamic_array_map.py
Files skipped - 0
Tools
- Whispers (Secret Scanner) - ✔︎ Successful
- Detect-secrets (Secret Scanner) - ✔︎ Successful
- MyPy (Static Code Analysis) - ✔︎ Successful
- Astral Ruff (Static Code Analysis) - ✔︎ Successful

AI Code Review powered by

flyte-bot · 2025-02-19T22:56:10Z

Changelist by Bito

This pull request implements the following key changes.

Key Change	Files Impacted
Bug Fix - Fix Map Task Synchronization in Dynamic Workflows	- `entities.py` - Enhanced FlyteArrayNode to store and handle flyte_node references properly - `remote.py` - Updated task fetching logic to use stored flyte_task reference instead of fetching
Testing - Integration Tests for Map Task in Dynamic Workflows	- `test_remote.py` - Added test case for map task execution within dynamic workflow - `dynamic_array_map.py` - Created test workflow demonstrating map task usage in dynamic context

tests/flytekit/integration/remote/workflows/basic/dynamic_array_map.py

tests/flytekit/integration/remote/test_remote.py

flytekit/remote/entities.py

Signed-off-by: Troy Chiu <[email protected]>

flyte-bot · 2025-02-20T00:20:56Z

Code Review Agent Run #944657

Actionable Suggestions - 1

tests/flytekit/integration/remote/test_remote.py - 1
- Consider validating execution status before assertions · Line 630-632

Review Details

Files reviewed - 2 · Commit Range: 19b1562..f395792
- flytekit/remote/entities.py
- tests/flytekit/integration/remote/test_remote.py
Files skipped - 0
Tools
- Whispers (Secret Scanner) - ✔︎ Successful
- Detect-secrets (Secret Scanner) - ✔︎ Successful
- MyPy (Static Code Analysis) - ✔︎ Successful
- Astral Ruff (Static Code Analysis) - ✔︎ Successful

AI Code Review powered by

tests/flytekit/integration/remote/test_remote.py

wild-endeavor · 2025-02-20T02:24:28Z

flytekit/remote/entities.py

+        parallelism: int,
+        min_successes: int,
+        min_success_ratio: float,
+        flyte_task_node: Optional[FlyteTaskNode] = None,


my only question is, how is this flyte_task_node different from the task node that's inside the inner node object on line 353? I think @pvditt should take a look btw, he definitely knows more about the inner workings of map and where it's going than i do.

The types are different. The node object on line 353 (array_node.node.task_node) is TaskNode type, while this flyte_task_node is FlyteTaskNode type. Or they are fundamentally identical?

yeah FlyteTaskNode inherits from TaskNode - if it's possible i'd put it there instead, just to avoid confusion? what do you think?

Good callout. I used FlyteTaskNode to replace the original node object. Mind taking a look? TY

Signed-off-by: Troy Chiu <[email protected]>

flyte-bot · 2025-02-20T22:40:56Z

Code Review Agent Run #ff09ed

Actionable Suggestions - 3

flytekit/remote/entities.py - 3
- Consider removing redundant instance variable assignment · Line 358-359
- Consider impact of method name change · Line 362-363
- Consider keeping parameter optional for compatibility · Line 369-369

Review Details

Files reviewed - 2 · Commit Range: f395792..32e4142
- flytekit/remote/entities.py
- flytekit/remote/remote.py
Files skipped - 0
Tools
- Whispers (Secret Scanner) - ✔︎ Successful
- Detect-secrets (Secret Scanner) - ✔︎ Successful
- MyPy (Static Code Analysis) - ✔︎ Successful
- Astral Ruff (Static Code Analysis) - ✔︎ Successful

AI Code Review powered by

flyte-bot · 2025-02-20T22:44:40Z

flytekit/remote/entities.py

+        super().__init__(flyte_node, parallelism, min_successes, min_success_ratio)
+        self._flyte_node = flyte_node


Consider removing redundant instance variable assignment

Consider removing the redundant assignment of _flyte_node in line 359 since it's already being set in the parent class through super().__init__() call.

Code suggestion

Check the AI-generated fix before applying

Suggested change

super().__init__(flyte_node, parallelism, min_successes, min_success_ratio)

self._flyte_node = flyte_node

super().__init__(flyte_node, parallelism, min_successes, min_success_ratio)

Code Review Run #ff09ed

Should Bito avoid suggestions like this for future reviews? (Manage Rules)

Yes, avoid them

flyte-bot · 2025-02-20T22:44:41Z

flytekit/remote/entities.py

+    def flyte_node(self) -> FlyteNode:
+        return self._flyte_node


Consider impact of method name change

Consider keeping the method name consistent. The method name is being changed from flyte_task_node to flyte_node which could break backward compatibility. If this is an intentional API change, it might be worth documenting the change and ensuring all callers are updated.

Code suggestion

Check the AI-generated fix before applying

Suggested change

def flyte_node(self) -> FlyteNode:

return self._flyte_node

def flyte_task_node(self) -> FlyteNode:

return self._flyte_task_node

Code Review Run #ff09ed

Should Bito avoid suggestions like this for future reviews? (Manage Rules)

Yes, avoid them

flyte-bot · 2025-02-20T22:44:41Z

flytekit/remote/entities.py

+    def promote_from_model(
+        cls,
+        model: _workflow_model.ArrayNode,
+        flyte_node: FlyteNode,


Consider keeping parameter optional for compatibility

Consider making the flyte_node parameter optional with a default value of None. The parameter type change from Optional[FlyteTaskNode] to FlyteNode may break existing code that passes None.

Code suggestion

Check the AI-generated fix before applying

Suggested change

flyte_node: FlyteNode,

flyte_node: Optional[FlyteNode] = None,

Code Review Run #ff09ed

Should Bito avoid suggestions like this for future reviews? (Manage Rules)

Yes, avoid them

wild-endeavor · 2025-02-21T01:39:38Z

thank you troy

pvditt · 2025-02-21T02:52:31Z

flytekit/remote/entities.py

+        cls,
+        model: _workflow_model.ArrayNode,
+        flyte_node: FlyteNode,
+    ):


nit: could you add -> FlyteArrayNode for the return type annotation here

Let me add this in the following PR. Thank you!

pvditt · 2025-02-21T02:58:34Z

flytekit/remote/entities.py

        return cls(
-            node=model._node,
+            flyte_node=flyte_node,
            parallelism=model._parallelism,


this is outside the scope of this change, but I noticed we aren't setting a few fields here (

flytekit/flytekit/models/core/workflow.py

Lines 394 to 396 in 6d7c738

execution_mode=None,

is_original_sub_node_interface=False,

data_mode=None,

)

I wonder if there's a way we could error in places like here if the fields aren't updated when ArrayNode/any node's IDL changes.

I don't think it's easy to do since there are lots of models that don't match the IDL.

pvditt

changes look good. Thanks for fixing this

fix sync for map task in dynamic

8ba253c

Signed-off-by: Troy Chiu <[email protected]>

remove unnecessary

8bc2096

Signed-off-by: Troy Chiu <[email protected]>

troychiu added 3 commits February 19, 2025 13:48

remove raise error

afdd73a

Signed-off-by: Troy Chiu <[email protected]>

add integration tests

a3d7157

Signed-off-by: Troy Chiu <[email protected]>

lint

19b1562

Signed-off-by: Troy Chiu <[email protected]>

troychiu marked this pull request as ready for review February 19, 2025 22:25

troychiu requested review from wild-endeavor, kumare3, eapolinario, pingsutw, cosmicBboy, samhita-alla, thomasjpfan and Future-Outlier as code owners February 19, 2025 22:25

flyte-bot reviewed Feb 19, 2025

View reviewed changes

tests/flytekit/integration/remote/workflows/basic/dynamic_array_map.py Show resolved Hide resolved

flyte-bot reviewed Feb 19, 2025

View reviewed changes

tests/flytekit/integration/remote/test_remote.py Show resolved Hide resolved

flyte-bot reviewed Feb 19, 2025

View reviewed changes

flytekit/remote/entities.py Outdated Show resolved Hide resolved

flyte-bot reviewed Feb 19, 2025

View reviewed changes

flytekit/remote/entities.py Outdated Show resolved Hide resolved

type hint

f395792

Signed-off-by: Troy Chiu <[email protected]>

pingsutw previously approved these changes Feb 19, 2025

View reviewed changes

flyte-bot reviewed Feb 20, 2025

View reviewed changes

tests/flytekit/integration/remote/test_remote.py Show resolved Hide resolved

wild-endeavor reviewed Feb 20, 2025

View reviewed changes

use flyte node

e5d6dd2

Signed-off-by: Troy Chiu <[email protected]>

troychiu dismissed pingsutw’s stale review via e5d6dd2 February 20, 2025 21:32

lint

32e4142

Signed-off-by: Troy Chiu <[email protected]>

pingsutw approved these changes Feb 20, 2025

View reviewed changes

flyte-bot reviewed Feb 20, 2025

View reviewed changes

wild-endeavor approved these changes Feb 21, 2025

View reviewed changes

pvditt reviewed Feb 21, 2025

View reviewed changes

pvditt approved these changes Feb 21, 2025

View reviewed changes

troychiu merged commit d7a3cef into master Feb 21, 2025
111 of 112 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix sync for map task in dynamic #3141

fix sync for map task in dynamic #3141

troychiu commented Feb 19, 2025 •

edited by flyte-bot

Loading

flyte-bot commented Feb 19, 2025

codecov bot commented Feb 19, 2025 •

edited

Loading

flyte-bot commented Feb 19, 2025

flyte-bot commented Feb 19, 2025 •

edited

Loading

Code Review Agent Run #462f2a

flyte-bot commented Feb 19, 2025 •

edited

Loading

Changelist by Bito

flyte-bot commented Feb 20, 2025 •

edited

Loading

Code Review Agent Run #944657

wild-endeavor Feb 20, 2025

troychiu Feb 20, 2025

wild-endeavor Feb 20, 2025

troychiu Feb 20, 2025

flyte-bot commented Feb 20, 2025 •

edited

Loading

Code Review Agent Run #ff09ed

flyte-bot Feb 20, 2025

flyte-bot Feb 20, 2025

flyte-bot Feb 20, 2025

wild-endeavor commented Feb 21, 2025

pvditt Feb 21, 2025

troychiu Feb 21, 2025

pvditt Feb 21, 2025

troychiu Feb 21, 2025

pvditt left a comment •

edited

Loading

		super().__init__(flyte_node, parallelism, min_successes, min_success_ratio)
		self._flyte_node = flyte_node

	flyte_node: FlyteNode,
	flyte_node: Optional[FlyteNode] = None,

	execution_mode=None,
	is_original_sub_node_interface=False,
	data_mode=None,

fix sync for map task in dynamic #3141

fix sync for map task in dynamic #3141

Conversation

troychiu commented Feb 19, 2025 • edited by flyte-bot Loading

Why are the changes needed?

Example

What changes were proposed in this pull request?

How was this patch tested?

Check all the applicable boxes

Related PRs

Docs link

Summary by Bito

flyte-bot commented Feb 19, 2025

Code Review Agent Run Status

codecov bot commented Feb 19, 2025 • edited Loading

Codecov Report

flyte-bot commented Feb 19, 2025

Code Review Agent Run Status

flyte-bot commented Feb 19, 2025 • edited Loading

Code Review Agent Run #462f2a

flyte-bot commented Feb 19, 2025 • edited Loading

Changelist by Bito

flyte-bot commented Feb 20, 2025 • edited Loading

Code Review Agent Run #944657

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flyte-bot commented Feb 20, 2025 • edited Loading

Code Review Agent Run #ff09ed

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wild-endeavor commented Feb 21, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pvditt left a comment • edited Loading

Choose a reason for hiding this comment

troychiu commented Feb 19, 2025 •

edited by flyte-bot

Loading

codecov bot commented Feb 19, 2025 •

edited

Loading

flyte-bot commented Feb 19, 2025 •

edited

Loading

flyte-bot commented Feb 19, 2025 •

edited

Loading

flyte-bot commented Feb 20, 2025 •

edited

Loading

flyte-bot commented Feb 20, 2025 •

edited

Loading

pvditt left a comment •

edited

Loading