Qualify server version call in PostgreSQL

Fixes: #6912
Closes: #6920
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/6920
Pull-request-sha: 79af75dfddef25435afd9623698354d280d7c879

Change-Id: Ib6b472452f978378d9f511d17a26988323a89459

3 files changed, 55 insertions, 1 deletions

diff --git a/doc/build/changelog/unreleased_14/6912.rst b/doc/build/changelog/unreleased_14/6912.rst
new file mode 100644
index 000000000..4a8be5e2e
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6912.rst
@@ -0,0 +1,6 @@
+.. change::
+    :tags: bug, postgresql
+    :tickets: 6912
+
+    Qualify ``version()`` call to avoid shadowing issues if a different
+    search path is configured by the user.
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index d8e4d5d20..82b848ece 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -3425,7 +3425,7 @@ class PGDialect(default.DefaultDialect):
         return bool(cursor.scalar())
 
     def _get_server_version_info(self, connection):
-        v = connection.exec_driver_sql("select version()").scalar()
+        v = connection.exec_driver_sql("select pg_catalog.version()").scalar()
         m = re.match(
             r".*(?:PostgreSQL|EnterpriseDB) "
             r"(\d+)\.?(\d+)?(?:\.(\d+))?(?:\.\d+)?(?:devel|beta)?",
diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index 371a17819..a86da8ce7 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -9,6 +9,7 @@ from sqlalchemy import bindparam
 from sqlalchemy import cast
 from sqlalchemy import Column
 from sqlalchemy import DateTime
+from sqlalchemy import DDL
 from sqlalchemy import event
 from sqlalchemy import exc
 from sqlalchemy import extract
@@ -115,6 +116,53 @@ class DialectTest(fixtures.TestBase):
         ]:
             eq_(dialect._get_server_version_info(mock_conn(string)), version)
 
+    @testing.only_on("postgresql")
+    def test_ensure_version_is_qualified(
+        self, future_connection, testing_engine, metadata
+    ):
+
+        default_schema_name = future_connection.dialect.default_schema_name
+        event.listen(
+            metadata,
+            "after_create",
+            DDL(
+                """
+CREATE OR REPLACE FUNCTION %s.version() RETURNS integer AS $$
+BEGIN
+    return 0;
+END;
+$$ LANGUAGE plpgsql;"""
+                % (default_schema_name,)
+            ),
+        )
+        event.listen(
+            metadata,
+            "before_drop",
+            DDL("DROP FUNCTION %s.version" % (default_schema_name,)),
+        )
+
+        metadata.create_all(future_connection)
+        future_connection.commit()
+
+        e = testing_engine()
+
+        @event.listens_for(e, "do_connect")
+        def receive_do_connect(dialect, conn_rec, cargs, cparams):
+            conn = dialect.dbapi.connect(*cargs, **cparams)
+            cursor = conn.cursor()
+            cursor.execute(
+                "set search_path = %s,pg_catalog" % (default_schema_name,)
+            )
+            cursor.close()
+            return conn
+
+        with e.connect():
+            pass
+        eq_(
+            e.dialect.server_version_info,
+            future_connection.dialect.server_version_info,
+        )
+
     @testing.requires.python3
     @testing.requires.psycopg2_compatibility
     def test_pg_dialect_no_native_unicode_in_python3(self, testing_engine):