class_docstr: MetaflowTask prepares a Flow instance for execution of a single step.
class_docstr_tok: ["MetaflowTask","prepares","a","Flow","instance","for","execution","of","a","single","step","."]
code: def _init_foreach(self, step_name, join_type, inputs, split_index):
      # these variables are only set by the split step in the output
      # data. They don't need to be accessible in the flow.
      self.flow._foreach_var = None
      self.flow._foreach_num_splits = None
  
      # There are three cases that can alter the foreach state:
      # 1) start - initialize an empty foreach stack
      # 2) join - pop the topmost frame from the stack
      # 3) step following a split - push a new frame in the stack
  
      # case 1) - reset the stack
      if step_name == "start":
          self.flow._foreach_stack = []
  
      # case 2) - this is a join step
      elif join_type:
          # assert the lineage of incoming branches
          def lineage():
              for i in inputs:
                  if join_type == "foreach":
                      top = i["_foreach_stack"][-1]
                      bottom = i["_foreach_stack"][:-1]
                      # the topmost indices in the stack are all
                      # different naturally, so ignore them in the
                      # assertion
                      yield bottom + [top._replace(index=0)]
                  else:
                      yield i["_foreach_stack"]
  
          if not all_equal(lineage()):
              raise MetaflowInternalError(
                  "Step *%s* tried to join branches "
                  "whose lineages don't match." % step_name
              )
  
          # assert that none of the inputs are splits - we don't
          # allow empty `foreach`s (joins immediately following splits)
          if any(not i.is_none("_foreach_var") for i in inputs):
              raise MetaflowInternalError(
                  "Step *%s* tries to join a foreach "
                  "split with no intermediate steps." % step_name
              )
  
          inp = inputs[0]
          if join_type == "foreach":
              # Make sure that the join got all splits as its inputs.
              # Datastore.resolve() leaves out all undone tasks, so if
              # something strange happened upstream, the inputs list
              # may not contain all inputs which should raise an exception
              stack = inp["_foreach_stack"]
              if (
                  stack[-1].num_splits
                  and len(inputs) != stack[-1].num_splits
              ):
                  raise MetaflowDataMissing(
                      "Foreach join *%s* expected %d "
                      "splits but only %d inputs were "
                      "found"
                      % (step_name, stack[-1].num_splits, len(inputs))
                  )
              # foreach-join pops the topmost frame from the stack
              self.flow._foreach_stack = stack[:-1]
          else:
              # a non-foreach join doesn't change the stack
              self.flow._foreach_stack = inp["_foreach_stack"]
  
      # case 3) - our parent was a split. Initialize a new foreach frame.
      elif not inputs[0].is_none("_foreach_var"):
          if len(inputs) != 1:
              raise MetaflowInternalError(
                  "Step *%s* got multiple inputs "
                  "although it follows a split step." % step_name
              )
  
          if self.ubf_context != UBF_CONTROL and split_index is None:
              raise MetaflowInternalError(
                  "Step *%s* follows a split step "
                  "but no split_index is "
                  "specified." % step_name
              )
  
          # push a new index after a split to the stack
          frame = ForeachFrame(
              step_name,
              inputs[0]["_foreach_var"],
              inputs[0]["_foreach_num_splits"],
              split_index,
          )
  
          stack = inputs[0]["_foreach_stack"]
          stack.append(frame)
          self.flow._foreach_stack = stack
  
imports: from __future__ import print_function
  from io import BytesIO
  import math
  import sys
  import os
  import time
  from types import MethodType, FunctionType
  from metaflow.datastore.exceptions import DataException
  from .metaflow_config import MAX_ATTEMPTS
  from .metadata import MetaDatum
  from .mflog import TASK_LOG_SOURCE
  from .datastore import Inputs, TaskDataStoreSet
  from .exception import MetaflowInternalError, MetaflowDataMissing, MetaflowExceptionWrapper
  from .unbounded_foreach import UBF_CONTROL
  from .util import all_equal, get_username, resolve_identity, unicode_type
  from .current import current
  from metaflow.tracing import get_trace_id
  from collections import namedtuple
label_desc: Use image node_id 5 for calling the MetaflowTask obj's underlying member method code with example usage: obj._init_foreach(step_name, join_type, inputs, split_index) without return types