Correct for Variant + ARRAY cases in psycopg2

Fixed regression caused by :ticket:`6023` where the PostgreSQL cast
operator applied to elements within an :class:`_types.ARRAY` when using
psycopg2 would fail to use the correct type in the case that the datatype
were also embedded within an instance of the :class:`_types.Variant`
adapter.

Additionally, repairs support for the correct CREATE TYPE to be emitted
when using a ``Variant(ARRAY(some_schema_type))``.

Fixes: #6182
Change-Id: I1b9ba7c876980d4650715a0b0801b46bdc72860d

4 files changed, 28 insertions, 15 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 0854214d0..97eb07bdb 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1989,7 +1989,6 @@ class ENUM(sqltypes.NativeForEmulated, sqltypes.Enum):
             return False
 
     def _on_table_create(self, target, bind, checkfirst=False, **kw):
-
         if (
             checkfirst
             or (
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index 1969eb844..16f9ecefa 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -605,12 +605,17 @@ class PGCompiler_psycopg2(PGCompiler):
         )
         # note that if the type has a bind_expression(), we will get a
         # double compile here
-        if not skip_bind_expression and bindparam.type._is_array:
-            text += "::%s" % (
-                elements.TypeClause(bindparam.type)._compiler_dispatch(
-                    self, skip_bind_expression=skip_bind_expression, **kw
-                ),
-            )
+        if not skip_bind_expression and (
+            bindparam.type._is_array or bindparam.type._is_type_decorator
+        ):
+            typ = bindparam.type._unwrapped_dialect_impl(self.dialect)
+
+            if typ._is_array:
+                text += "::%s" % (
+                    elements.TypeClause(typ)._compiler_dispatch(
+                        self, skip_bind_expression=skip_bind_expression, **kw
+                    ),
+                )
         return text
 
 
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 367b2e203..7cc50d99c 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -1221,13 +1221,19 @@ class SchemaType(SchemaEventTarget):
         if variant_mapping is None:
             return True
 
-        if (
-            dialect.name in variant_mapping
-            and variant_mapping[dialect.name] is self
+        # since PostgreSQL is the only DB that has ARRAY this can only
+        # be integration tested by PG-specific tests
+        def _we_are_the_impl(typ):
+            return (
+                typ is self or isinstance(typ, ARRAY) and typ.item_type is self
+            )
+
+        if dialect.name in variant_mapping and _we_are_the_impl(
+            variant_mapping[dialect.name]
         ):
             return True
         elif dialect.name not in variant_mapping:
-            return variant_mapping["_default"] is self
+            return _we_are_the_impl(variant_mapping["_default"])
 
 
 class Enum(Emulated, String, SchemaType):
@@ -2857,16 +2863,16 @@ class ARRAY(SchemaEventTarget, Indexable, Concatenable, TypeEngine):
     def compare_values(self, x, y):
         return x == y
 
-    def _set_parent(self, column, **kw):
+    def _set_parent(self, column, outer=False, **kw):
         """Support SchemaEventTarget"""
 
-        if isinstance(self.item_type, SchemaEventTarget):
+        if not outer and isinstance(self.item_type, SchemaEventTarget):
             self.item_type._set_parent(column, **kw)
 
     def _set_parent_with_dispatch(self, parent):
         """Support SchemaEventTarget"""
 
-        super(ARRAY, self)._set_parent_with_dispatch(parent)
+        super(ARRAY, self)._set_parent_with_dispatch(parent, outer=True)
 
         if isinstance(self.item_type, SchemaEventTarget):
             self.item_type._set_parent_with_dispatch(parent)
diff --git a/lib/sqlalchemy/sql/type_api.py b/lib/sqlalchemy/sql/type_api.py
index bfce00cb5..69cd3c5ca 100644
--- a/lib/sqlalchemy/sql/type_api.py
+++ b/lib/sqlalchemy/sql/type_api.py
@@ -48,6 +48,7 @@ class TypeEngine(Traversible):
     _is_tuple_type = False
     _is_table_value = False
     _is_array = False
+    _is_type_decorator = False
 
     class Comparator(operators.ColumnOperators):
         """Base class for custom comparison operations defined at the
@@ -955,6 +956,8 @@ class TypeDecorator(SchemaEventTarget, TypeEngine):
 
     __visit_name__ = "type_decorator"
 
+    _is_type_decorator = True
+
     def __init__(self, *args, **kwargs):
         """Construct a :class:`.TypeDecorator`.
 
@@ -1497,7 +1500,7 @@ class Variant(TypeDecorator):
         else:
             return self.impl
 
-    def _set_parent(self, column, **kw):
+    def _set_parent(self, column, outer=False, **kw):
         """Support SchemaEventTarget"""
 
         if isinstance(self.impl, SchemaEventTarget):