Hi, i got an error while executing with argo-workflow the attached flow, it consist of branching using foreach and nested inside of it is regular branching. i'm getting this error while having more than 6 branches in the nested branching: (less than 6 - works)

Data missing: Some input datastores are missing. Expected: 7 Actual: 0

Is that something configurable or a limitation?

from metaflow import FlowSpec, step, kubernetes, environment, timeout, retry, catch, conda, card, project


class test_branching(FlowSpec):

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def start(self):
        self.arr = [1,2,3]
        self.next(self.first_step, foreach='arr')

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def first_step(self):
        self.first_step_output = self.input
        self.next(self.a,
                  self.b,
                  self.c,
                  self.d,
                  self.e,
                  self.f,
                  self.g)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def a(self):
        self.out_put_branch = ()
        self.next(self.join_all_branches)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def b(self):
        self.out_put_branch = ()
        self.next(self.join_all_branches)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def c(self):
        self.out_put_branch = ()
        self.next(self.join_all_branches)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def d(self):
        self.out_put_branch = ()
        self.next(self.join_all_branches)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def e(self):
        self.out_put_branch = ()
        self.next(self.join_all_branches)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def f(self):
        self.out_put_branch = ()
        self.next(self.join_all_branches)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def g(self):
        self.out_put_branch = ()
        self.next(self.join_all_branches)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @retry(times=1, minutes_between_retries=0)
    @step
    def join_all_branches(self, inputs):
        print("Done join all RUN artifacts!")
        self.next(self.join_all_results)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def join_all_results(self, inputs):
        print("Done join all RUN artifacts!")
        self.next(self.end)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def end(self):
        print("Done everything!")


if __name__ == "__main__":
    test_branching()

Interesting. We will reproduce and fix it. Thanks for the helpful example!

I had a go with the provided example and the flows I ran finished without any issues. Some questions for trying to reproduce the issues: • which version of metaflow was used to run the flow? • what step are you encountering the error at? • is it limited to static splits inside a foreach, or will nested foreach splits also fail? • is this limited to running the flow on argo-workflows, or reproducible with running on kubernetes as well? nested foreach example I tried out as well:

from metaflow import FlowSpec, step, kubernetes, environment, timeout, retry, catch, conda, card, project


class BranchingTest(FlowSpec):

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def start(self):
        self.arr = [1,2,3]
        self.next(self.first_step, foreach='arr')

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def first_step(self):
        self.first_step_output = self.input
        self.nested_arr = list(range(0,8))
        self.next(self.a, foreach='nested_arr')

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def a(self):
        self.out_put_branch = ()
        self.next(self.join_all_branches)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @retry(times=1, minutes_between_retries=0)
    @step
    def join_all_branches(self, inputs):
        print("Done join all RUN artifacts!")
        self.next(self.join_all_results)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def join_all_results(self, inputs):
        print("Done join all RUN artifacts!")
        self.next(self.end)

    @kubernetes(image="<http://docker.io/python:3.9|docker.io/python:3.9>")
    @step
    def end(self):
        print("Done everything!")


if __name__ == "__main__":
    BranchingTest()

1. Im using metaflow==2.8.5. 2. the error raises from the join step of the static branching. 3. good question, I haven’t tried cause steps a,b,c.. have business logic that depends on a docker image, so I cannot use foreach branching 😕 (because of the k8s decorator) 4. haven’t tried running on k8s - i will try that 👍 Actually I have a question here that might be related, i read that Argo is storing the flow as k8s resource in etcd, since it’s limited with 1M could it be related here in some way? (I’m planning on offloading the flow into Postgres so if that will succeed I’ll update here)