Fix caching writes #469

PhilippeMoussalli · 2023-09-27T15:42:35Z

PR that fixes issues related to caching subsequent components.

Since cached manifests contain the same run id as the pipeline where they originated from, they can sometimes override existing data if the component needs to run again (e.g. cached first component loaded into second component which has different arguments and thus needs to execute). This PR makes sure that:

The run-id of the loaded input manifest matches the one from the current pipeline run, this ensures that all new subsets from the component being executed are written under the corresponding path of that pipeline run
The run-id of the output manifest is that of the cached manifest (if it applies). This ensures that subsequent components are able to tell whether the previous component execution was cached or not link to code

RobbeSneyders · 2023-09-27T17:01:00Z

So If I'm understanding correctly, the logic should be as follows:

If the component is cached, it outputs the cached manifest with the run_id of the previous run
If the component is not cached, it outputs a new manifest with the run_id of the current run
A component checks if the previous component was cached by comparing the run_id in the manifest it receives with the run_id of the current run (from the metadata).

But the problem is that the component propagates the run_id from the input manifest to the output manifest.

If that is correct, can we change it where it's propagated instead? So in either:

I believe a change there would be simpler and easier to understand.

RobbeSneyders · 2023-09-29T09:04:38Z

I think it can still be further simplified. Just pass the run_id to .evolve() and set it in the metadata at the start. That run_id is then used for the paths automatically as well, and then we don't need to update the metadata separately.

PhilippeMoussalli · 2023-09-29T10:11:34Z

I think it can still be further simplified. Just pass the run_id to .evolve() and set it in the metadata at the start. That run_id is then used for the paths automatically as well, and then we don't need to update the metadata separately.

Thanks for the feedback Robbe! I implemented the suggestion. Please double check to see if that's what you were implying.

Also added other small fixes mainly related to propogating the cache arguments from the op to the pipeline spec

RobbeSneyders

Thanks @PhilippeMoussalli!

Almost there 🙂 I would make the argument optional and rename it to just run_id. See my suggestions.

RobbeSneyders · 2023-09-29T10:42:50Z

src/fondant/manifest.py

+
+        Args:
+            component_spec: the component spec
+            write_run_id: the run id used to define the write subset path.


Suggested change

write_run_id: the run id used to define the write subset path.

run_id: the run id to include in the evolved manifest. If no run id is provided, the run id from the original manifest is propagated.

RobbeSneyders · 2023-09-29T10:44:39Z

src/fondant/manifest.py

@@ -257,21 +257,27 @@ def remove_subset(self, name: str) -> None:
    def evolve(  # noqa : PLR0912 (too many branches)
        self,
        component_spec: ComponentSpec,
+        write_run_id: str,


Suggested change

write_run_id: str,

*,

run_id: t.Optional[str],

RobbeSneyders · 2023-09-29T10:45:53Z

src/fondant/manifest.py

        """
        evolved_manifest = self.copy()

        # Update `component_id` of the metadata
        component_id = component_spec.name.lower().replace(" ", "_")
+        evolved_manifest.update_metadata(key="run_id", value=write_run_id)


Suggested change

evolved_manifest.update_metadata(key="run_id", value=write_run_id)

if run_id is not None:

evolved_manifest.update_metadata(key="run_id", value=write_run_id)

RobbeSneyders · 2023-09-29T10:46:18Z

src/fondant/manifest.py

        evolved_manifest.update_metadata(key="component_id", value=component_id)

        # Update index location as this is currently always rewritten
        evolved_manifest.index._specification[
            "location"
-        ] = f"/{self.pipeline_name}/{self.run_id}/{component_id}/index"
+        ] = f"/{self.pipeline_name}/{write_run_id}/{component_id}/index"


Suggested change

] = f"/{self.pipeline_name}/{write_run_id}/{component_id}/index"

] = f"/{self.pipeline_name}/{evolved_manifest.run_id}/{component_id}/index"

RobbeSneyders · 2023-09-29T10:46:30Z

src/fondant/manifest.py

@@ -322,7 +328,7 @@ def evolve(  # noqa : PLR0912 (too many branches)
                # Update subset location as this is currently always rewritten
                evolved_manifest.subsets[subset_name]._specification[
                    "location"
-                ] = f"/{self.pipeline_name}/{self.run_id}/{component_id}/{subset_name}"
+                ] = f"/{self.pipeline_name}/{write_run_id}/{component_id}/{subset_name}"


Suggested change

] = f"/{self.pipeline_name}/{write_run_id}/{component_id}/{subset_name}"

] = f"/{self.pipeline_name}/{evolved_manifest.run_id}/{component_id}/{subset_name}"

RobbeSneyders · 2023-09-29T10:46:44Z

src/fondant/pipeline.py

@@ -444,7 +444,7 @@ def _validate_pipeline_definition(self, run_id: str):
                            raise InvalidPipelineDefinition(
                                msg,
                            )
-            manifest = manifest.evolve(component_spec)
+            manifest = manifest.evolve(component_spec, write_run_id=run_id)


Suggested change

manifest = manifest.evolve(component_spec, write_run_id=run_id)

manifest = manifest.evolve(component_spec, run_id=run_id)

RobbeSneyders · 2023-09-29T10:48:57Z

tests/test_manifest_evolution.py

-    evolved_manifest = manifest.evolve(component_spec=component_spec)
+    evolved_manifest = manifest.evolve(
+        component_spec=component_spec,
+        write_run_id=manifest.run_id,


Can we update this test to provide a different run_id and validate that it's set correctly?

RobbeSneyders · 2023-09-29T10:49:05Z

tests/test_manifest_evolution.py

-    evolved_manifest = manifest.evolve(component_spec=component_spec)
+    evolved_manifest = manifest.evolve(
+        component_spec=component_spec,
+        write_run_id=manifest.run_id,


This change can be reverted.

RobbeSneyders

Thanks @PhilippeMoussalli!

PR that fixes issues related to caching subsequent components. Since cached manifests contain the same run id as the pipeline where they originated from, they can sometimes override existing data if the component needs to run again (e.g. cached first component loaded into second component which has different arguments and thus needs to execute). This PR makes sure that: * The run-id of the loaded **input manifest** matches the one from the **current pipeline run**, this ensures that all new subsets from the component being executed are written under the corresponding path of that pipeline run * The run-id of the **output manifest** is that of the **cached manifest** (if it applies). This ensures that subsequent components are able to tell whether the previous component execution was cached or not [link to code](https://github.com/ml6team/fondant/blob/39436e7faaa3e5acdd5eb347ce965ad33d669f73/src/fondant/executor.py#L294)

PhilippeMoussalli added 5 commits September 27, 2023 16:30

improve caching

209e205

further improve

b2a91ce

logging

a63d225

fixes

22b903d

remove print statements

033a496

PhilippeMoussalli requested review from RobbeSneyders, GeorgesLorre and mrchtr and removed request for RobbeSneyders September 27, 2023 15:44

Merge branch 'main' into improve-caching-writing

62b6658

implement PR feedback

dbb5706

PhilippeMoussalli added 3 commits September 29, 2023 12:08

Implement PR feedback

a9fff3f

Additional small fixes

f54d979

Merge branch 'main' into improve-caching-writing

196a02a

RobbeSneyders reviewed Sep 29, 2023

View reviewed changes

implement PR feedback

c5c28ac

RobbeSneyders approved these changes Oct 2, 2023

View reviewed changes

RobbeSneyders merged commit 645ab7a into main Oct 2, 2023
5 checks passed

RobbeSneyders deleted the improve-caching-writing branch October 2, 2023 13:25

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix caching writes #469

Fix caching writes #469

PhilippeMoussalli commented Sep 27, 2023

RobbeSneyders commented Sep 27, 2023

RobbeSneyders commented Sep 29, 2023

PhilippeMoussalli commented Sep 29, 2023

RobbeSneyders left a comment

RobbeSneyders Sep 29, 2023

RobbeSneyders Sep 29, 2023

RobbeSneyders Sep 29, 2023

RobbeSneyders Sep 29, 2023

RobbeSneyders Sep 29, 2023

RobbeSneyders Sep 29, 2023

RobbeSneyders Sep 29, 2023

RobbeSneyders Sep 29, 2023

RobbeSneyders left a comment

	write_run_id: the run id used to define the write subset path.
	run_id: the run id to include in the evolved manifest. If no run id is provided, the run id from the original manifest is propagated.

	evolved_manifest.update_metadata(key="run_id", value=write_run_id)
	if run_id is not None:
	evolved_manifest.update_metadata(key="run_id", value=write_run_id)

	] = f"/{self.pipeline_name}/{write_run_id}/{component_id}/index"
	] = f"/{self.pipeline_name}/{evolved_manifest.run_id}/{component_id}/index"

	manifest = manifest.evolve(component_spec, write_run_id=run_id)
	manifest = manifest.evolve(component_spec, run_id=run_id)

Fix caching writes #469

Fix caching writes #469

Conversation

PhilippeMoussalli commented Sep 27, 2023

RobbeSneyders commented Sep 27, 2023

RobbeSneyders commented Sep 29, 2023

PhilippeMoussalli commented Sep 29, 2023

RobbeSneyders left a comment

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

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

Choose a reason for hiding this comment

RobbeSneyders left a comment

Choose a reason for hiding this comment