AIP-51 - Single Threaded Executors

[o-nikolas]

Overview

Some Executors, currently a subset of the local Executors, run in a single threaded fashion and have certain limitations and requirements, many of which are hardcoded. To add a new single threaded Executor would require changes to core Airflow code.
Note: This coupling often shows up with SQLite compatibility checks since it does not support multiple connections.

Examples

• 2a) SQLite check done in configuration.py:

  airflow/airflow/configuration.py

  Lines 412 to 419 in 26f94c5

  	is_executor_without_sqlite_support = self.get("core", "executor") not in (
  	'DebugExecutor',
  	'SequentialExecutor',
  	)
  	is_sqlite = "sqlite" in self.get('database', 'sql_alchemy_conn')
  	if is_sqlite and is_executor_without_sqlite_support:
  	raise AirflowConfigException(f"error: cannot use sqlite with the {self.get('core', 'executor')}")
  	if is_sqlite:

• 2b) When running in standalone mode SQLite compatibility is checked:

  airflow/airflow/cli/commands/standalone_command.py

  Lines 160 to 165 in 26f94c5

  	if "sqlite" in conf.get("database", "sql_alchemy_conn"):
  	self.print_output("standalone", "Forcing executor to SequentialExecutor")
  	env["AIRFLOW__CORE__EXECUTOR"] = executor_constants.SEQUENTIAL_EXECUTOR
  	else:
  	self.print_output("standalone", "Forcing executor to LocalExecutor")
  	env["AIRFLOW__CORE__EXECUTOR"] = executor_constants.LOCAL_EXECUTOR

• 2c) Sensors in poke mode can block execution of DAGs when running with single process Executors, currently hardcoded to DebugExecutor (although should also include SequentialExecutor):

  airflow/airflow/sensors/base.py

  Line 243 in 27e2101

  if conf.get('core', 'executor') == "DebugExecutor":

Proposal

A static method or attribute on the Executor class which can be checked by core code.

There is a precedent already set with the supports_ad_hoc_ti_run attribute, see:

airflow/airflow/executors/kubernetes_executor.py

Line 435 in fb741fd

supports_ad_hoc_ti_run: bool = True

airflow/airflow/www/views.py

Lines 1735 to 1737 in 26f94c5

	if not getattr(executor, "supports_ad_hoc_ti_run", False):
	msg = "Only works with the Celery, CeleryKubernetes or Kubernetes executors"
	return redirect_or_json(origin, msg, "error", 400)

[utkarsharma2]

@o-nikolas Interested :)

[o-nikolas]

Awesome, I've assigned it to you @utkarsharma2!

[utkarsharma2]

@o-nikolas What's the intent of supports_ad_hoc_ti_run is it directly related to an executor being single-threaded? if so, I can use this field. WDYT?
Nevermind got the answer :)

[utkarsharma2]

@o-nikolas I tried using the below code to get the Executor, which leads to a circular dependency error.

executor_cls, _ = ExecutorLoader.import_executor_cls(conf.get("core", "executor"))
is_executor_without_sqlite_support = executor_cls.supports_ad_hoc_ti_run

Error:

/Users/utkarsharma/sandbox/airflow-sandbox/env3.8/bin/python "/Applications/PyCharm CE.app/Contents/plugins/python-ce/helpers/pycharm/_jb_pytest_runner.py" --target test_configuration.py::TestDeprecatedConf.test_deprecated_env_vars_upgraded_and_removed
Testing started at 10:41 AM ...
Launching pytest with arguments test_configuration.py::TestDeprecatedConf::test_deprecated_env_vars_upgraded_and_removed --no-header --no-summary -q in /Users/utkarsharma/sandbox/airflow-sandbox/airflow/tests/core

ImportError while loading conftest '/Users/utkarsharma/sandbox/airflow-sandbox/airflow/tests/conftest.py'.
../conftest.py:41: in <module>
    from airflow import settings  # noqa: E402
../../airflow/__init__.py:35: in <module>
    from airflow import settings
../../airflow/settings.py:37: in <module>
    from airflow.configuration import AIRFLOW_HOME, WEBSERVER_CONFIG, conf  # NOQA F401
../../airflow/configuration.py:1752: in <module>
    conf.validate()
../../airflow/configuration.py:341: in validate
    self._validate_config_dependencies()
../../airflow/configuration.py:438: in _validate_config_dependencies
    executor_cls, _ = ExecutorLoader.import_executor_cls(conf.get("core", "executor"))
../../airflow/executors/executor_loader.py:128: in import_executor_cls
    return import_string(cls.executors[executor_name]), ConnectorSource.CORE
../../airflow/utils/module_loading.py:36: in import_string
    module = import_module(module_path)
../../airflow/executors/sequential_executor.py:30: in <module>
    from airflow.executors.base_executor import BaseExecutor, CommandType
../../airflow/executors/base_executor.py:29: in <module>
    from airflow.models.taskinstance import TaskInstance, TaskInstanceKey
../../airflow/models/taskinstance.py:70: in <module>
    from airflow.datasets.manager import dataset_manager
../../airflow/datasets/manager.py:27: in <module>
    from airflow.models.dataset import DatasetDagRunQueue, DatasetEvent, DatasetModel
../../airflow/models/dataset.py:40: in <module>
    from airflow.utils import timezone
../../airflow/utils/timezone.py:27: in <module>
    from airflow.settings import TIMEZONE
E   ImportError: cannot import name 'TIMEZONE' from partially initialized module 'airflow.settings' (most likely due to a circular import) (/Users/utkarsharma/sandbox/airflow-sandbox/airflow/airflow/settings.py)

Is there any other way we can access the supports_ad_hoc_ti_run of an executor in configuration.py ???

[o-nikolas]

I tried using the below code to get the Executor, which leads to a circular dependency error.

Hey @utkarsharma2!

Unfortunately it's not too surprising, conf and settings are both used in many places (the latter importing the former and using it quite a bit) and importing an executor is causing a bridge between the two. You can see settings imports conf which when it's loaded calls validate() on itself, during which an executor is now imported with your change (in this case the sequential executor) which then imports the BaseExecutor (of course) and that thing imports a bunch of stuff which all use settings in the end. So tl;dr, importing a module inside conf turns out to be dangerous stuff! 😆

You can try some things like locally importing modules only where they're needed, or limiting imports only if they're used for type checking. I played around quickly and didn't find a clean way out of the loop but I do encourage you to have a look as well now that you have more context.

The ultimate solution may be to move the executor config validation to runtime. i.e. the executor will check if it's incompatible with the DB being used during startup and throw if it's now (which can actually be pretty well encapsulated in the base executor actually).

[potiuk]

Yes. Unfortunately we have this circular behaviour. Which I hate personally.

I am not sure, maybe that could be a good time to start decoupling this as well. I believe we should be able to do it by extracting this as well?

However EVERY SINGLE TIME I tried to do it (I tried possibly three times already) things look easy to begin with but after enough number of refactoring you end up with some unsolvable entanglement in both directions. And eventually I abandoned the idea each time because there were more pressing things to do.

But I think we can solve it by applying a reasonable split between modules and possibly very little code duplication because we are trying to use generic conf in settings and settings in conf and this is very bad decision in general.

We should extract some common code use in both conf and setting to a separate module and use it from both places (but it's not super easy and it will lead to likely many changes and some potential back-compat issues.

[o-nikolas]

Hey @utkarsharma2

I have a proposal that you go ahead and write a PR to add the new singlethreaded flag/property to the base executor and write the requisite changes for 2b (standalone command) and 2c (DebugExecutor) and we can defer the resolution to the import cycle blocking 2a.

This is because 2c blocks #28528 right here specifically

WDYT?