Core: Make psycopg (v3) the default synchronous Postgres driver

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

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

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
 ------------------------------

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'

airflow-core/docs/howto/set-up-database.rst

@@ -197,12 +197,34 @@ in the Postgres documentation to learn more.
 
    Details in the `SQLAlchemy Changelog <https://docs.sqlalchemy.org/en/20/changelog/changelog_14.html#change-3687655465c25a39b968b4f5f6e9170b>`_.
 
-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+<driver>://<user>:<password>@<host>/<db>
 
+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://<user>:<password>@<host>/<db>
+
+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://<user>:<password>@<host>/<db>
+
+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.

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).

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(

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)
 

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

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")

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 {}
             )
         )
     }

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

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():

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")

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:<version>`` 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