code: def test_get_executions_additional_filter(self):
      with self._mlmd_connection as m:
          for node in [n.pipeline_node for n in self._pipeline.nodes]:
              self.assertEmpty(task_gen_utils.get_executions(m, node))
  
      self._set_pipeline_context(
          self._pipeline, "pipeline", "my_pipeline1"
      )
  
      # Create three COMPLETE executions.
      otu.fake_example_gen_execution_with_state(
          self._mlmd_connection, self._example_gen, State.COMPLETE
      )
      otu.fake_example_gen_execution_with_state(
          self._mlmd_connection, self._example_gen, State.COMPLETE
      )
      otu.fake_example_gen_execution_with_state(
          self._mlmd_connection, self._example_gen, State.COMPLETE
      )
  
      # Get all ExampleGen executions across all pipeline contexts.
      with self._mlmd_connection as m:
          all_eg_execs = sorted(
              m.store.get_executions_by_type(
                  self._example_gen.node_info.type.name
              ),
              key=lambda e: e.create_time_since_epoch,
          )
  
          # Check that correct executions are returned.
          self.assertCountEqual(
              all_eg_execs[1:],
              task_gen_utils.get_executions(
                  m,
                  self._example_gen,
                  additional_filters=[
                      "create_time_since_epoch >="
                      f" {all_eg_execs[1].create_time_since_epoch}"
                  ],
              ),
          )
          self.assertCountEqual(
              all_eg_execs,
              task_gen_utils.get_executions(
                  m,
                  self._example_gen,
                  additional_filters=["create_time_since_epoch >= 0"],
              ),
          )
  
imports: import os
  import time
  from unittest import mock
  import uuid
  from absl.testing import parameterized
  import tensorflow
  from tfx import types
  from tfx import version
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_async_pipeline
  from tfx.orchestration.experimental.core.testing import test_dynamic_exec_properties_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import placeholder_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.utils import status
  from tfx.utils import test_case_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 6 for calling the TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_get_executions_additional_filter() without return types