diff --git a/airflow-core/docs/administration-and-deployment/logging-monitoring/logging-tasks.rst b/airflow-core/docs/administration-and-deployment/logging-monitoring/logging-tasks.rst index 3f2a70e49c971..e947fb5c9d8fe 100644 --- a/airflow-core/docs/administration-and-deployment/logging-monitoring/logging-tasks.rst +++ b/airflow-core/docs/administration-and-deployment/logging-monitoring/logging-tasks.rst @@ -150,7 +150,7 @@ the example below. version | 2.9.0.dev0 executor | LocalExecutor task_logging_handler | airflow.utils.log.file_task_handler.FileTaskHandler - sql_alchemy_conn | postgresql+psycopg2://postgres:airflow@postgres/airflow + sql_alchemy_conn | postgresql+psycopg://postgres:airflow@postgres/airflow dags_folder | /files/dags plugins_folder | /root/airflow/plugins base_log_folder | /root/airflow/logs diff --git a/airflow-core/docs/administration-and-deployment/modules_management.rst b/airflow-core/docs/administration-and-deployment/modules_management.rst index 4ea73ea6fd77b..62f3430da22e5 100644 --- a/airflow-core/docs/administration-and-deployment/modules_management.rst +++ b/airflow-core/docs/administration-and-deployment/modules_management.rst @@ -293,7 +293,7 @@ Below is the sample output of the ``airflow info`` command: Config info executor | LocalExecutor task_logging_handler | airflow.utils.log.file_task_handler.FileTaskHandler - sql_alchemy_conn | postgresql+psycopg2://postgres:airflow@postgres/airflow + sql_alchemy_conn | postgresql+psycopg://postgres:airflow@postgres/airflow dags_folder | /files/dags plugins_folder | /root/airflow/plugins base_log_folder | /root/airflow/logs diff --git a/airflow-core/docs/core-concepts/multi-team.rst b/airflow-core/docs/core-concepts/multi-team.rst index 6a9d8a2ec63e7..b45b0379b7ff6 100644 --- a/airflow-core/docs/core-concepts/multi-team.rst +++ b/airflow-core/docs/core-concepts/multi-team.rst @@ -486,7 +486,7 @@ is uppercase. export AIRFLOW__TEAM_B___CELERY__BROKER_URL="redis://team-b-redis:6379/0" # team_b's Celery result backend - export AIRFLOW__TEAM_B___CELERY__RESULT_BACKEND="db+postgresql+psycopg2://team-b-db/celery_results" + export AIRFLOW__TEAM_B___CELERY__RESULT_BACKEND="db+postgresql+psycopg://team-b-db/celery_results" Via Config File """"""""""""""" @@ -499,17 +499,17 @@ name followed by an equals sign: # Global celery settings (used by the global executor, NOT as a fallback for teams) [celery] broker_url = redis://default-redis:6379/0 - result_backend = db+postgresql+psycopg2://default-db/celery_results + result_backend = db+postgresql+psycopg://default-db/celery_results # team_a overrides [team_a=celery] broker_url = redis://team-a-redis:6379/0 - result_backend = db+postgresql+psycopg2://team-a-db/celery_results + result_backend = db+postgresql+psycopg://team-a-db/celery_results # team_b overrides [team_b=celery] broker_url = redis://team-b-redis:6379/0 - result_backend = db+postgresql+psycopg2://team-b-db/celery_results + result_backend = db+postgresql+psycopg://team-b-db/celery_results Dag Bundle to Team Association ------------------------------ diff --git a/airflow-core/docs/howto/docker-compose/docker-compose.yaml b/airflow-core/docs/howto/docker-compose/docker-compose.yaml index 6a8891f3f58cc..b68f1afd3cc9f 100644 --- a/airflow-core/docs/howto/docker-compose/docker-compose.yaml +++ b/airflow-core/docs/howto/docker-compose/docker-compose.yaml @@ -57,8 +57,8 @@ x-airflow-common: &airflow-common-env AIRFLOW__CORE__EXECUTOR: CeleryExecutor AIRFLOW__CORE__AUTH_MANAGER: airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow - AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql+psycopg2://airflow:airflow@postgres/airflow + AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg://airflow:airflow@postgres/airflow + AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql+psycopg://airflow:airflow@postgres/airflow AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0 AIRFLOW__CORE__FERNET_KEY: ${FERNET_KEY} AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true' diff --git a/airflow-core/docs/howto/set-up-database.rst b/airflow-core/docs/howto/set-up-database.rst index ec9604f2ccbeb..03e0a7a627d7f 100644 --- a/airflow-core/docs/howto/set-up-database.rst +++ b/airflow-core/docs/howto/set-up-database.rst @@ -197,12 +197,34 @@ in the Postgres documentation to learn more. Details in the `SQLAlchemy Changelog `_. -We recommend specifying a driver, such as ``psycopg2``, explicitly in your SQLAlchemy connection string: +We recommend specifying a driver, such as ``psycopg``, explicitly in your SQLAlchemy connection string: .. code-block:: text postgresql+://:@/ +Sync SQLAlchemy engine and the sync driver +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +``psycopg`` (psycopg3) is the default synchronous driver for the metadata database. When ``sql_alchemy_conn`` omits a driver +(a bare ``postgresql://`` scheme) or still uses the legacy ``postgres://`` / ``postgres+psycopg2://`` schemes, Airflow +rewrites it to: + +.. code-block:: text + + postgresql+psycopg://:@/ + +If you need the old ``psycopg2`` driver, install the ``apache-airflow-providers-postgres[psycopg2]`` extra and set the +connection string explicitly: + +.. code-block:: ini + + [database] + sql_alchemy_conn = postgresql+psycopg2://:@/ + +An explicit ``postgresql+psycopg2://`` URL is never rewritten. If ``psycopg2`` is not installed while such a URL is +configured, Airflow fails loudly at connection time rather than silently falling back to a different driver. + Async SQLAlchemy engine and the async driver ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ @@ -387,7 +409,7 @@ For instance, you can specify a database schema where Airflow will create its re .. code-block:: bash - export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg2://postgres@localhost:5432/my_database?options=-csearch_path%3Dairflow" + export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg://postgres@localhost:5432/my_database?options=-csearch_path%3Dairflow" export AIRFLOW__DATABASE__SQL_ALCHEMY_SCHEMA="airflow" Note the ``search_path`` at the end of the ``SQL_ALCHEMY_CONN`` database URL. diff --git a/airflow-core/newsfragments/69469.improvement.rst b/airflow-core/newsfragments/69469.improvement.rst new file mode 100644 index 0000000000000..2260dc1287d00 --- /dev/null +++ b/airflow-core/newsfragments/69469.improvement.rst @@ -0,0 +1 @@ +``psycopg`` (psycopg3) is now the default synchronous Postgres driver for the metadata database, mirroring the existing async default; a bare ``postgresql://`` or legacy ``postgres://``/``postgres+psycopg2://`` ``sql_alchemy_conn`` is now rewritten to ``postgresql+psycopg://`` instead of ``postgresql+psycopg2://``, while an explicit ``postgresql+psycopg2://`` connection string is left untouched (install the ``apache-airflow-providers-postgres[psycopg2]`` extra to keep using psycopg2). diff --git a/airflow-core/src/airflow/configuration.py b/airflow-core/src/airflow/configuration.py index d43eb715d320f..4eb92054ceab9 100644 --- a/airflow-core/src/airflow/configuration.py +++ b/airflow-core/src/airflow/configuration.py @@ -28,6 +28,7 @@ from base64 import b64encode from collections.abc import Callable from configparser import ConfigParser +from importlib.util import find_spec from inspect import ismodule from io import StringIO from re import Pattern @@ -372,13 +373,18 @@ def _upgrade_postgres_metastore_conn(self): Upgrade SQL schemas. As of SQLAlchemy 1.4, schemes `postgres+psycopg2` and `postgres` - must be replaced with `postgresql+psycopg2`. The bare `postgresql` - scheme is also upgraded to make the psycopg2 driver explicit. + must be replaced with `postgresql+psycopg` if the psycopg (v3) driver + is installed, or `postgresql+psycopg2` otherwise. The bare `postgresql` + scheme is upgraded the same way to make the driver explicit. """ section, key = "database", "sql_alchemy_conn" old_value = self.get(section, key, _extra_stacklevel=1) bad_schemes = ["postgres+psycopg2", "postgres", "postgresql"] - good_scheme = "postgresql+psycopg2" + # The provider-side hooks (common.sql / postgres / amazon) also gate psycopg (v3) on an + # ``_is_sqlalchemy_2()`` check, because they support Airflow 2.11 on SQLAlchemy 1.4, which has + # no native ``postgresql+psycopg`` dialect. airflow-core pins ``sqlalchemy>=2.0``, so that + # dialect is always present when the package is importable — ``find_spec`` alone suffices here. + good_scheme = "postgresql+psycopg" if find_spec("psycopg") is not None else "postgresql+psycopg2" parsed = urlsplit(old_value) if parsed.scheme in bad_schemes: warnings.warn( diff --git a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py index fa24916df6faa..64531a957b302 100644 --- a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py +++ b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py @@ -110,7 +110,7 @@ def upgrade(): batch_op.create_unique_constraint(batch_op.f("connection_conn_id_uq"), ["conn_id"]) max_cons = sa.table("dag", sa.column("max_consecutive_failed_dag_runs")) - op.execute(max_cons.update().values(max_consecutive_failed_dag_runs=literal("0"))) + op.execute(max_cons.update().values(max_consecutive_failed_dag_runs=literal(0))) with op.batch_alter_table("dag") as batch_op: batch_op.alter_column("max_consecutive_failed_dag_runs", existing_type=sa.Integer(), nullable=False) diff --git a/airflow-core/src/airflow/migrations/versions/0094_3_2_0_replace_deadline_inline_callback_with_fkey.py b/airflow-core/src/airflow/migrations/versions/0094_3_2_0_replace_deadline_inline_callback_with_fkey.py index 0cab3d57c252f..bf1f66510920d 100644 --- a/airflow-core/src/airflow/migrations/versions/0094_3_2_0_replace_deadline_inline_callback_with_fkey.py +++ b/airflow-core/src/airflow/migrations/versions/0094_3_2_0_replace_deadline_inline_callback_with_fkey.py @@ -92,7 +92,7 @@ def _upgrade_postgresql(conn, batch_size): '__var', json_build_object( 'path', b.cb_path, 'kwargs', b.cb_kwargs, - 'prefix', :prefix, + 'prefix', CAST(:prefix AS text), 'dag_id', b.dag_id ), '__type', 'dict' @@ -325,7 +325,7 @@ def _downgrade_postgresql(conn, batch_size): 'path', c.data::jsonb->'__var'->>'path', 'kwargs', c.data::jsonb->'__var'->'kwargs' ), - '__classname__', :classname, + '__classname__', CAST(:classname AS text), '__version__', 0 )::json, callback_state = CASE @@ -360,7 +360,7 @@ def _downgrade_postgresql(conn, batch_size): 'path', c.data::jsonb->'__var'->>'path', 'kwargs', c.data::jsonb->'__var'->'kwargs' ), - '__classname__', :classname, + '__classname__', CAST(:classname AS text), '__version__', 0 )::json, callback_state = CASE diff --git a/airflow-core/src/airflow/models/serialized_dag.py b/airflow-core/src/airflow/models/serialized_dag.py index fbff450a9939b..9dda74d42b5fc 100644 --- a/airflow-core/src/airflow/models/serialized_dag.py +++ b/airflow-core/src/airflow/models/serialized_dag.py @@ -28,7 +28,7 @@ import uuid6 from sqlalchemy import JSON, ForeignKey, LargeBinary, String, Uuid, exists, select, tuple_, update -from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.dialects.postgresql import ARRAY, JSONB from sqlalchemy.orm import Mapped, backref, foreign, mapped_column, relationship from sqlalchemy.sql.expression import func, literal @@ -955,7 +955,9 @@ def load_json(deps_data): elif dialect == "postgresql": # Use #> operator which works for both JSON and JSONB types # Returns the JSON sub-object at the specified path - data_col_to_select = cls._data.op("#>")(literal('{"dag","dag_dependencies"}')) + data_col_to_select = cls._data.op("#>")( + literal(["dag", "dag_dependencies"], type_=ARRAY(String)) + ) load_json = lambda x: x else: data_col_to_select = func.json_extract_path(cls._data, "dag", "dag_dependencies") diff --git a/airflow-core/src/airflow/settings.py b/airflow-core/src/airflow/settings.py index 54150dd8bfd5b..0a0e1c6619726 100644 --- a/airflow-core/src/airflow/settings.py +++ b/airflow-core/src/airflow/settings.py @@ -524,15 +524,16 @@ def _is_sqlite_in_memory(url: str) -> bool: def prepare_engine_args(disable_connection_pool=False, pool_class=None): """Prepare SQLAlchemy engine args.""" + use_psycopg2_tuning = SQL_ALCHEMY_CONN.startswith("postgresql+psycopg2") DEFAULT_ENGINE_ARGS: dict[str, dict[str, Any]] = { "postgresql": ( { "insertmanyvalues_page_size": 10000, } | ( - {} - if _USE_PSYCOPG3 - else {"executemany_mode": "values_plus_batch", "executemany_batch_page_size": 2000} + {"executemany_mode": "values_plus_batch", "executemany_batch_page_size": 2000} + if use_psycopg2_tuning + else {} ) ) } diff --git a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py index 118f0b1a5d794..cbd17ca0c7aed 100644 --- a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py +++ b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py @@ -277,7 +277,22 @@ def test_handle_single_column_unique_constraint_error_with_stacktrace( exeinfo_response_error.value.detail.pop("message", None) # type: ignore[attr-defined] assert exeinfo_response_error.value.status_code == expected_exception.status_code - assert exeinfo_response_error.value.detail == expected_exception.detail + + # The exact rendered SQL (e.g. explicit driver-added type casts) and trailing whitespace in + # the driver's own error text are driver-specific rendering details, not meaningful content — + # match on the statement's target table and normalize trailing newlines instead of an exact + # string compare. + response_detail = exeinfo_response_error.value.detail + expected_detail = expected_exception.detail + actual_statement = response_detail.pop("statement", None) # type: ignore[attr-defined] + expected_statement = expected_detail.pop("statement", None) + response_detail["orig_error"] = response_detail["orig_error"].rstrip("\n") # type: ignore[index] + expected_detail["orig_error"] = expected_detail["orig_error"].rstrip("\n") + + assert response_detail == expected_detail + assert expected_statement is not None + assert actual_statement is not None + assert f"INSERT INTO {expected_statement.split()[2]}" in actual_statement @patch("airflow.api_fastapi.common.exceptions.get_random_string", return_value=MOCKED_ID) @conf_vars({("api", "expose_stacktrace"): "False"}) @@ -391,6 +406,9 @@ def test_handle_multiple_columns_unique_constraint_error_with_stacktrace( # Removes the stacktrace from response to remove during comparison. response_detail.pop("message", None) # type: ignore[attr-defined] + # Trailing whitespace in the driver's own error text is a driver-specific rendering detail. + response_detail["orig_error"] = response_detail["orig_error"].rstrip("\n") # type: ignore[index] + expected_detail["orig_error"] = expected_detail["orig_error"].rstrip("\n") assert response_detail == expected_detail assert "INSERT INTO dag_run" in actual_statement assert exeinfo_response_error.value.detail == expected_exception.detail diff --git a/airflow-core/tests/unit/core/test_configuration.py b/airflow-core/tests/unit/core/test_configuration.py index fc44c398e8e84..11d1809fe9f82 100644 --- a/airflow-core/tests/unit/core/test_configuration.py +++ b/airflow-core/tests/unit/core/test_configuration.py @@ -1124,16 +1124,16 @@ def test_write_handles_multiline_non_json_string(self): ("input_scheme", "expected_scheme"), [ pytest.param( - "postgres://user:pass@host/db", "postgresql+psycopg2://user:pass@host/db", id="postgres" + "postgres://user:pass@host/db", "postgresql+psycopg://user:pass@host/db", id="postgres" ), pytest.param( "postgres+psycopg2://user:pass@host/db", - "postgresql+psycopg2://user:pass@host/db", + "postgresql+psycopg://user:pass@host/db", id="postgres+psycopg2", ), pytest.param( "postgresql://user:pass@host/db", - "postgresql+psycopg2://user:pass@host/db", + "postgresql+psycopg://user:pass@host/db", id="postgresql-bare", ), pytest.param( @@ -1145,7 +1145,37 @@ def test_write_handles_multiline_non_json_string(self): ], ) @mock.patch.dict("os.environ", {}, clear=False) - def test_upgrade_postgres_metastore_conn(self, input_scheme, expected_scheme): + @mock.patch("airflow.configuration.find_spec", return_value=object()) + def test_upgrade_postgres_metastore_conn(self, mock_find_spec, input_scheme, expected_scheme): + """Assumes psycopg (v3) is installed; see the sibling ``_without_psycopg3`` test for the + fallback, since whether psycopg3 is actually present varies across CI environments + (e.g. a lowest-dependencies job that doesn't install it).""" + os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = input_scheme + test_conf = AirflowConfigParser() + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + test_conf._upgrade_postgres_metastore_conn() + assert test_conf.get("database", "sql_alchemy_conn") == expected_scheme + + @pytest.mark.parametrize( + ("input_scheme", "expected_scheme"), + [ + pytest.param( + "postgres://user:pass@host/db", "postgresql+psycopg2://user:pass@host/db", id="postgres" + ), + pytest.param( + "postgresql+psycopg2://user:pass@host/db", + "postgresql+psycopg2://user:pass@host/db", + id="postgresql+psycopg2-noop", + ), + ], + ) + @mock.patch.dict("os.environ", {}, clear=False) + @mock.patch("airflow.configuration.find_spec", return_value=None) + def test_upgrade_postgres_metastore_conn_without_psycopg3( + self, mock_find_spec, input_scheme, expected_scheme + ): + """Falls back to psycopg2 when the psycopg (v3) package isn't installed.""" os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = input_scheme test_conf = AirflowConfigParser() with warnings.catch_warnings(): diff --git a/airflow-core/tests/unit/core/test_sqlalchemy_config.py b/airflow-core/tests/unit/core/test_sqlalchemy_config.py index db211bab2c5af..ab364dd9a7b5b 100644 --- a/airflow-core/tests/unit/core/test_sqlalchemy_config.py +++ b/airflow-core/tests/unit/core/test_sqlalchemy_config.py @@ -121,6 +121,30 @@ def test_prepare_engine_args_sqlite_in_memory_skips_pool_settings(self, conn_str assert "pool_recycle" not in engine_args assert "pool_pre_ping" not in engine_args + @pytest.mark.parametrize( + ("conn_str", "use_psycopg3", "expect_psycopg2_tuning"), + [ + ("postgresql+psycopg2://user:pass@host/db", True, True), + ("postgresql+psycopg2://user:pass@host/db", False, True), + ("postgresql+psycopg://user:pass@host/db", True, False), + ("postgresql+psycopg://user:pass@host/db", False, False), + ], + ) + def test_prepare_engine_args_psycopg2_tuning_follows_configured_scheme( + self, conn_str, use_psycopg3, expect_psycopg2_tuning, monkeypatch + ): + """executemany tuning must follow the configured scheme, not whether psycopg happens to be importable.""" + monkeypatch.setattr(settings, "SQL_ALCHEMY_CONN", conn_str) + monkeypatch.setattr(settings, "_USE_PSYCOPG3", use_psycopg3) + engine_args = settings.prepare_engine_args() + assert engine_args["insertmanyvalues_page_size"] == 10000 + if expect_psycopg2_tuning: + assert engine_args["executemany_mode"] == "values_plus_batch" + assert engine_args["executemany_batch_page_size"] == 2000 + else: + assert "executemany_mode" not in engine_args + assert "executemany_batch_page_size" not in engine_args + @patch("airflow.settings.setup_event_handlers") @patch("airflow.settings.scoped_session") @patch("airflow.settings.sessionmaker") diff --git a/airflow-e2e-tests/docker/openlineage-compat-db.yml b/airflow-e2e-tests/docker/openlineage-compat-db.yml new file mode 100644 index 0000000000000..01c9e68e2fd4c --- /dev/null +++ b/airflow-e2e-tests/docker/openlineage-compat-db.yml @@ -0,0 +1,48 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# Docker Compose override for the openlineage E2E compat matrix (``--airflow-version``). +# +# docker-compose.yaml points sql_alchemy_conn/result_backend at the ``psycopg`` (v3) driver, matching +# the current default. The compat matrix runs unmodified released ``apache/airflow:`` images +# for versions that only ever bundled psycopg2 by default (no psycopg3 provider dependency existed +# yet), and does not reinstall the postgres provider on top of them — so the psycopg3 dialect can't +# load there. Pin the explicit psycopg2 scheme these images actually support. +--- +x-airflow-common-compat-db-env: &airflow-common-compat-db-env + AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow + AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql+psycopg2://airflow:airflow@postgres/airflow + +services: + airflow-apiserver: + environment: + <<: *airflow-common-compat-db-env + airflow-scheduler: + environment: + <<: *airflow-common-compat-db-env + airflow-dag-processor: + environment: + <<: *airflow-common-compat-db-env + airflow-triggerer: + environment: + <<: *airflow-common-compat-db-env + airflow-worker: + environment: + <<: *airflow-common-compat-db-env + airflow-init: + environment: + <<: *airflow-common-compat-db-env diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py index b97add595fd11..5b45974f053cb 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py @@ -56,6 +56,7 @@ LOCALSTACK_PATH, LOGS_FOLDER, NODE_IMAGE, + OPENLINEAGE_COMPAT_DB_COMPOSE_PATH, OPENLINEAGE_COMPOSE_PATH, OPENSEARCH_PATH, PROVIDERS_MOUNT_CONTAINER_PATH, @@ -630,7 +631,7 @@ def _setup_go_sdk_integration(dot_env_file, tmp_dir): os.environ["ENV_FILE_PATH"] = str(dot_env_file) -def _setup_openlineage_integration(dot_env_file, tmp_dir): +def _setup_openlineage_integration(dot_env_file, tmp_dir, compose_file_names): """Set up the openlineage E2E test mode. The OpenLineage system-test DAGs are the single source of truth; ``prepare_dags`` copies them @@ -644,6 +645,10 @@ def _setup_openlineage_integration(dot_env_file, tmp_dir): prepare_dags(tmp_dir / "dags") copyfile(OPENLINEAGE_COMPOSE_PATH, tmp_dir / "openlineage.yml") + if os.environ.get("E2E_TARGET_AIRFLOW_VERSION", "").strip(): + copyfile(OPENLINEAGE_COMPAT_DB_COMPOSE_PATH, tmp_dir / "openlineage-compat-db.yml") + compose_file_names.append("openlineage-compat-db.yml") + def _build_ts_sdk_example_bundle(*, native=False): build_commands = ( @@ -784,7 +789,7 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory): _setup_go_sdk_integration(dot_env_file, tmp_dir) elif E2E_TEST_MODE == "openlineage": compose_file_names.append("openlineage.yml") - _setup_openlineage_integration(dot_env_file, tmp_dir) + _setup_openlineage_integration(dot_env_file, tmp_dir, compose_file_names) elif E2E_TEST_MODE == "ts_sdk": compose_file_names.append("ts.yml") _setup_ts_sdk_integration(dot_env_file, tmp_dir) diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py index e389a8bc6b9bf..cbba1dde33f25 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py @@ -53,6 +53,11 @@ # OpenLineage E2E test paths. The DAGs are sourced from the provider system tests at runtime by # openlineage_tests/prepare_dags.py; the overlay carries the OpenLineage-specific env + dag_doc mount. OPENLINEAGE_COMPOSE_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "openlineage.yml" +# Pins the metadata-DB driver back to psycopg2 for the compat matrix (--airflow-version), whose +# released base images predate the psycopg3 default; see openlineage-compat-db.yml for details. +OPENLINEAGE_COMPAT_DB_COMPOSE_PATH = ( + AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "openlineage-compat-db.yml" +) # CI sets this (the same switch the lang-SDK k8s job uses) to build the lang-SDK # artifacts with the host toolchain instead of ephemeral toolchain containers. diff --git a/chart/values.yaml b/chart/values.yaml index 890f9976d970c..e322a0d85ae29 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -442,12 +442,14 @@ data: # connection: base64_encoded_connection_string # # The 'connection' key is base64-encoded SQLAlchemy connection string, e.g.: - # postgresql+psycopg2://airflow:password@postgres/airflow + # postgresql+psycopg://airflow:password@postgres/airflow (Airflow >=3.4.0) + # postgresql+psycopg2://airflow:password@postgres/airflow (Airflow <3.4.0, or with the [psycopg2] extra) metadataSecretName: ~ # If not set, falls back to metadataSecretName. The secret must contain 'connection' key which is # a base64-encoded connection string, e.g.: - # postgresql+psycopg2://user:password@host/db + # postgresql+psycopg://user:password@host/db (Airflow >=3.4.0) + # postgresql+psycopg2://user:password@host/db (Airflow <3.4.0, or with the [psycopg2] extra) resultBackendSecretName: ~ brokerUrlSecretName: ~ diff --git a/scripts/ci/docker-compose/backend-postgres.yml b/scripts/ci/docker-compose/backend-postgres.yml index d2971bc1e628e..b1a64f16a01a5 100644 --- a/scripts/ci/docker-compose/backend-postgres.yml +++ b/scripts/ci/docker-compose/backend-postgres.yml @@ -19,8 +19,8 @@ services: airflow: environment: - BACKEND=postgres - - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://postgres:airflow@postgres/airflow - - AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql+psycopg2://postgres:airflow@postgres/airflow + - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql://postgres:airflow@postgres/airflow + - AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql+psycopg://postgres:airflow@postgres/airflow depends_on: postgres: condition: service_healthy diff --git a/task-sdk-integration-tests/docker-compose.yaml b/task-sdk-integration-tests/docker-compose.yaml index 82143e98b5e4a..b71de00d27ea9 100644 --- a/task-sdk-integration-tests/docker-compose.yaml +++ b/task-sdk-integration-tests/docker-compose.yaml @@ -27,7 +27,7 @@ x-airflow-common: # yamllint disable rule:line-length AIRFLOW__CORE__AUTH_MANAGER: 'airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager' AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: 'true' - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow + AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg://airflow:airflow@postgres/airflow AIRFLOW__CORE__FERNET_KEY: ${FERNET_KEY} AIRFLOW__CORE__LOAD_EXAMPLES: 'false' AIRFLOW__CORE__DAGS_FOLDER: '/opt/airflow/dags'