code: def filter_artifacts_by_span(
      artifacts: List[types.Artifact],
      span_descending: bool,
      n: int = 1,
      skip_last_n: int = 0,
      keep_all_versions: bool = False,
      min_span: Optional[int] = None,
  ) -> List[types.Artifact]:
      """Filters artifacts by their "span" PROPERTY.
  
      This should only be used a shared utility for LatestSpan and ConsecutiveSpans.
  
      Args:
        artifacts: The list of Artifacts to filter.
        span_descending: If true, then the artifacts will be sorted by span in
          descending order.  Else, they will be sorted in ascending order by span.
          Set to true for LatestSpan, and set to false for ConsecutiveSpans.
        n: The number of spans to return. If n <= 0, then n is set to the total
          number of unique spans.
        skip_last_n: Number of largest spans to skip. For example, if the spans are
          [1, 2, 3] and skip_last_n=1, then only spans [1, 2] will be considered.
        keep_all_versions: If true, all versions of the n spans are returned. Else,
          only the latest version is returned.
        min_span: Minimum span before which no span will be considered.
  
      Returns:
        The filtered artifacts.
      """
      if not artifacts:
          return []
  
      # Only keep artifacts with spans >= min_span and account for skip_last_n
      spans = sorted({a.span for a in artifacts})
      if min_span is not None:
          spans = [s for s in spans if s >= min_span]
      if skip_last_n:
          spans = spans[:-skip_last_n]
  
      # Sort spans in descending order, if specified.
      if span_descending:
          spans = spans[::-1]
  
      # Keep n spans, if n is positive.
      if n > 0:
          spans = spans[:n]
  
      if not spans:
          return []
  
      artifacts_by_span = {}
      for artifact in artifacts:
          artifacts_by_span.setdefault(artifact.span, []).append(
              artifact
          )
  
      result = []
      version_time_and_id = lambda a: (  # pylint: disable=g-long-lambda
          a.version,
          a.mlmd_artifact.create_time_since_epoch,
          a.id,
      )
      for span in sorted(spans):
          if keep_all_versions:
              # span_descending only applies to sorting by span, but version should
              # always be sorted in ascending order.
              result.extend(
                  sorted(
                      artifacts_by_span[span], key=version_time_and_id
                  )
              )
          else:
              # Latest version is defined as the largest version. Ties broken by id.
              result.append(
                  max(artifacts_by_span[span], key=version_time_and_id)
              )
  
      return result
  
file_path: tfx/tfx/dsl/input_resolution/ops/ops_utils.py
imports: from typing import Dict, List, Optional, Sequence, Set
  from tfx import types
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.orchestration.portable.mlmd import filter_query_builder
  from tfx.utils import typing_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 5 for calling a global function with example usage: filter_artifacts_by_span(artifacts, span_descending, n, skip_last_n, keep_all_versions, min_span) and returns: result, list, list