docs: def _data_scan(self, data_id=None, executor='resolwe.flow.executors.local', **kwargs):
          """Scan for new Data objects and execute them.
  
          :param data_id: Optional id of Data object which (+ its
              children) should be scanned. If it is not given, all
              resolving objects are processed.
          :param executor: The fully qualified name of the executor to use
              for all :class:`~resolwe.flow.models.Data` objects
              discovered in this pass.
          """
          def process_data_object(data):
              """Process a single data object."""
              # Lock for update. Note that we want this transaction to be as short as possible in
              # order to reduce contention and avoid deadlocks. This is why we do not lock all
              # resolving objects for update, but instead only lock one object at a time. This
              # allows managers running in parallel to process different objects.
              data = Data.objects.select_for_update().get(pk=data.pk)
              if data.status != Data.STATUS_RESOLVING:
                  # The object might have already been processed while waiting for the lock to be
                  # obtained. In this case, skip the object.
                  return
  
              dep_status = dependency_status(data)
  
              if dep_status == Data.STATUS_ERROR:
                  data.status = Data.STATUS_ERROR
                  data.process_error.append("One or more inputs have status ERROR")
                  data.process_rc = 1
                  data.save()
                  return
  
              elif dep_status != Data.STATUS_DONE:
                  return
  
              if data.process.run:
                  try:
                      execution_engine = data.process.run.get('language', None)
                      # Evaluation by the execution engine may spawn additional data objects and
                      # perform other queries on the database. Queries of all possible execution
                      # engines need to be audited for possibilities of deadlocks in case any
                      # additional locks are introduced. Currently, we only take an explicit lock on
                      # the currently processing object.
                      program = self.get_execution_engine(execution_engine).evaluate(data)
                  except (ExecutionError, InvalidEngineError) as error:
                      data.status = Data.STATUS_ERROR
                      data.process_error.append("Error in process script: {}".format(error))
                      data.save()
                      return
  
                  # Set allocated resources:
                  resource_limits = data.process.get_resource_limits()
                  data.process_memory = resource_limits['memory']
                  data.process_cores = resource_limits['cores']
              else:
                  # If there is no run section, then we should not try to run anything. But the
                  # program must not be set to None as then the process will be stuck in waiting
                  # state.
                  program = ''
  
              if data.status != Data.STATUS_DONE:
                  # The data object may already be marked as done by the execution engine. In this
                  # case we must not revert the status to STATUS_WAITING.
                  data.status = Data.STATUS_WAITING
              data.save(render_name=True)
  
              # Actually run the object only if there was nothing with the transaction.
              transaction.on_commit(
                  # Make sure the closure gets the right values here, since they're
                  # changed in the loop.
                  lambda d=data, p=program: self._data_execute(d, p, executor)
              )
  
          logger.debug(__("Manager processing communicate command triggered by Data with id {}.", data_id))
  
          if is_testing():
              # NOTE: This is a work-around for Django issue #10827
              # (https://code.djangoproject.com/ticket/10827), same as in
              # TestCaseHelpers._pre_setup(). Because the worker is running
              # independently, it must clear the cache on its own.
              ContentType.objects.clear_cache()
  
              # Ensure settings overrides apply
              self.discover_engines(executor=executor)
  
          try:
              queryset = Data.objects.filter(status=Data.STATUS_RESOLVING)
              if data_id is not None:
                  # Scan only given data object and its children.
                  queryset = queryset.filter(Q(parents=data_id) | Q(id=data_id)).distinct()
  
              for data in queryset:
                  try:
                      with transaction.atomic():
                          process_data_object(data)
  
                          # All data objects created by the execution engine are commited after this
                          # point and may be processed by other managers running in parallel. At the
                          # same time, the lock for the current data object is released.
                  except Exception as error:  # pylint: disable=broad-except
                      logger.exception(__(
                          "Unhandled exception in _data_scan while processing data object {}.",
                          data.pk
                      ))
  
                      # Unhandled error while processing a data object. We must set its
                      # status to STATUS_ERROR to prevent the object from being retried
                      # on next _data_scan run. We must perform this operation without
                      # using the Django ORM as using the ORM may be the reason the error
                      # occurred in the first place.
                      error_msg = "Internal error: {}".format(error)
                      process_error_field = Data._meta.get_field('process_error')  # pylint: disable=protected-access
                      max_length = process_error_field.base_field.max_length
                      if len(error_msg) > max_length:
                          error_msg = error_msg[:max_length - 3] + '...'
  
                      try:
                          with connection.cursor() as cursor:
                              cursor.execute(
                                  """
                                      UPDATE {table}
                                      SET
                                          status = %(status)s,
                                          process_error = process_error || (%(error)s)::varchar[]
                                      WHERE id = %(id)s
                                  """.format(
                                      table=Data._meta.db_table  # pylint: disable=protected-access
                                  ),
                                  {
                                      'status': Data.STATUS_ERROR,
                                      'error': [error_msg],
                                      'id': data.pk
                                  }
                              )
                      except Exception as error:  # pylint: disable=broad-except
                          # If object's state cannot be changed due to some database-related
                          # issue, at least skip the object for this run.
                          logger.exception(__(
                              "Unhandled exception in _data_scan while trying to emit error for {}.",
                              data.pk
                          ))
  
          except IntegrityError as exp:
              logger.error(__("IntegrityError in manager {}", exp))
              return