class_docstr: GroupByDisjointLineage operator.
  
  Let's say we have a lineage of artifacts (executions omitted for brevity):
  
  ```dot
  digraph {
      a1 -> b1 -> c1
      a2 -> b2 -> c2
      a2 -> b3 -> c3
      a3 -> b4 -> c4
  }
  ```
  
  Then `GroupByDisjointLineage` would group artifacts by each disjoint lineage
  where artifacts from different group is not reachable.
  
  ```python
  GroupByDisjointLineage({
      'a': [a1, a2, a3],
      'b': [b1, b2, b3, b4],
      'c': [c1, c2, c3, c4],
  }) == [
      {'a': [a1], 'b': [b1], 'c': [c1]},
      {'a': [a2], 'b': [b2, b3], 'c': [c2, c3]},
      {'a': [a3], 'b': [b4], 'c': [c4]},
  ]
  ```
  
  CAVEAT: Lineage is only searched for the 2-hop distances (i.e. artifact ->
  execution -> artifact), so in order to traverse for the deeper relationships,
  provide the intermediate artifacts as well so that there exists a chain of
  2-hop connections.
class_docstr_tok: ["GroupByDisjointLineage","operator",".","Let","'s","say","we","have","a","lineage","of","artifacts","(","executions","omitted","for","brevity",")",":","``","`","dot","digraph","{","a1","-",">","b1","-",">","c1","a2","-",">","b2","-",">","c2","a2","-",">","b3","-",">","c3","a3","-",">","b4","-",">","c4","}","``","`","Then","`","GroupByDisjointLineage","`","would","group","artifacts","by","each","disjoint","lineage","where","artifacts","from","different","group","is","not","reachable",".","``","`","python","GroupByDisjointLineage","(","{","'","a","'",":","[","a1",",","a2",",","a3","]",",","'","b","'",":","[","b1",",","b2",",","b3",",","b4","]",",","'","c","'",":","[","c1",",","c2",",","c3",",","c4","]",",","}",")","==","[","{","'","a","'",":","[","a1","]",",","'","b","'",":","[","b1","]",",","'","c","'",":","[","c1","]","}",",","{","'","a","'",":","[","a2","]",",","'","b","'",":","[","b2",",","b3","]",",","'","c","'",":","[","c2",",","c3","]","}",",","{","'","a","'",":","[","a3","]",",","'","b","'",":","[","b4","]",",","'","c","'",":","[","c4","]","}",",","]","``","`","CAVEAT",":","Lineage","is","only","searched","for","the","2-hop","distances","(","i.e",".","artifact","-",">","execution","-",">","artifact",")",",","so","in","order","to","traverse","for","the","deeper","relationships",",","provide","the","intermediate","artifacts","as","well","so","that","there","exists","a","chain","of","2-hop","connections","."]
code: def apply(
      self, artifact_map: typing_utils.ArtifactMultiMap
  ) -> List[typing_utils.ArtifactMultiDict]:
      artifacts_by_id = {}
      input_keys_by_id = collections.defaultdict(set)
      for input_key, artifacts in artifact_map.items():
          for a in artifacts:
              input_keys_by_id[a.id].add(input_key)
              artifacts_by_id[a.id] = a
  
      if not artifacts_by_id:
          return []
  
      events = self.context.store.get_events_by_artifact_ids(
          artifact_ids=artifacts_by_id
      )
  
      result = []
      for disjoint_set in _find_disjoint_sets(
          artifacts_by_id, _get_neighbor_artifact_pairs(events)
      ):
          result_item = {input_key: [] for input_key in artifact_map}
          for artifact_id in disjoint_set:
              for input_key in input_keys_by_id[artifact_id]:
                  result_item[input_key].append(
                      artifacts_by_id[artifact_id]
                  )
          if not self.require_all or all(result_item.values()):
              result.append(result_item)
      return result
  
file_path: tfx/tfx/dsl/input_resolution/ops/group_by_lineage_op.py
imports: import collections
  from typing import List, Iterable, Tuple
  from tfx.dsl.input_resolution import resolver_op
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.utils import typing_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 1 for calling the GroupByDisjointLineage obj's underlying member method code with example usage: obj.apply(artifact_map) and returns: result, list