dont render VARCHAR length for PG casts

Fixed critical regression in PostgreSQL dialects such as asyncpg which rely
upon explicit casts in SQL in order for datatypes to be passed to the
driver correctly, where a :class:`.String` datatype would be cast along
with the exact column length being compared, leading to implicit truncation
when comparing a ``VARCHAR`` of a smaller length to a string of greater
length regardless of operator in use (e.g. LIKE, MATCH, etc.). The
PostgreSQL dialect now omits the length from ``VARCHAR`` when rendering
these casts.

Fixes: #9511
Change-Id: If094146d8cfd989a0b780872f38e86fd41ebfec2

6 files changed, 96 insertions, 40 deletions

diff --git a/doc/build/changelog/unreleased_20/9511.rst b/doc/build/changelog/unreleased_20/9511.rst
new file mode 100644
index 000000000..2b2720b61
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/9511.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, postgresql
+    :tickets: 9511
+
+    Fixed critical regression in PostgreSQL dialects such as asyncpg which rely
+    upon explicit casts in SQL in order for datatypes to be passed to the
+    driver correctly, where a :class:`.String` datatype would be cast along
+    with the exact column length being compared, leading to implicit truncation
+    when comparing a ``VARCHAR`` of a smaller length to a string of greater
+    length regardless of operator in use (e.g. LIKE, MATCH, etc.). The
+    PostgreSQL dialect now omits the length from ``VARCHAR`` when rendering
+    these casts.
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 47a516a5a..18f31ce47 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1717,6 +1717,10 @@ class PGCompiler(compiler.SQLCompiler):
         return f"{element.name}{self.function_argspec(element, **kw)}"
 
     def render_bind_cast(self, type_, dbapi_type, sqltext):
+        if dbapi_type._type_affinity is sqltypes.String:
+            # use VARCHAR with no length for VARCHAR cast.
+            # see #9511
+            dbapi_type = sqltypes.STRINGTYPE
         return f"""{sqltext}::{
                 self.dialect.type_compiler_instance.process(
                     dbapi_type, identifier_preparer=self.preparer
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index bc2885341..44d174086 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -397,6 +397,25 @@ class StringTest(_LiteralRoundTripFixture, fixtures.TestBase):
     def test_literal_non_ascii(self, literal_round_trip):
         literal_round_trip(String(40), ["réve🐍 illé"], ["réve🐍 illé"])
 
+    @testing.combinations(
+        ("%B%", ["AB", "BC"]),
+        ("A%C", ["AC"]),
+        ("A%C%Z", []),
+        argnames="expr, expected",
+    )
+    def test_dont_truncate_rightside(
+        self, metadata, connection, expr, expected
+    ):
+        t = Table("t", metadata, Column("x", String(2)))
+        t.create(connection)
+
+        connection.execute(t.insert(), [{"x": "AB"}, {"x": "BC"}, {"x": "AC"}])
+
+        eq_(
+            connection.scalars(select(t.c.x).where(t.c.x.like(expr))).all(),
+            expected,
+        )
+
     def test_literal_quoting(self, literal_round_trip):
         data = """some 'text' hey "hi there" that's text"""
         literal_round_trip(String(40), [data], [data])
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py
index 7ef7033b2..8b8a80053 100644
--- a/test/dialect/postgresql/test_query.py
+++ b/test/dialect/postgresql/test_query.py
@@ -213,30 +213,30 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             asserter.assert_(
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 30, "data": "d1"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 1, "data": "d2"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 31, "data": "d3"}, {"id": 32, "data": "d4"}],
                 ),
                 DialectSQL(
-                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR(30))",
+                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR)",
                     [{"data": "d5"}, {"data": "d6"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 33, "data": "d7"}],
                 ),
                 DialectSQL(
-                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR(30))",
+                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR)",
                     [{"data": "d8"}],
                 ),
             )
@@ -311,30 +311,30 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             asserter.assert_(
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 30, "data": "d1"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 5, "data": "d2"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 31, "data": "d3"}, {"id": 32, "data": "d4"}],
                 ),
                 DialectSQL(
-                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR(30))",
+                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR)",
                     [{"data": "d5"}, {"data": "d6"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 33, "data": "d7"}],
                 ),
                 DialectSQL(
-                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR(30))",
+                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR)",
                     [{"data": "d8"}],
                 ),
             )
@@ -425,31 +425,31 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             asserter.assert_(
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 30, "data": "d1"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (data) VALUES "
-                    "(:data::VARCHAR(30)) RETURNING "
+                    "(:data::VARCHAR) RETURNING "
                     "testtable.id",
                     {"data": "d2"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 31, "data": "d3"}, {"id": 32, "data": "d4"}],
                 ),
                 DialectSQL(
-                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR(30))",
+                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR)",
                     [{"data": "d5"}, {"data": "d6"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 33, "data": "d7"}],
                 ),
                 DialectSQL(
-                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR(30))",
+                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR)",
                     [{"data": "d8"}],
                 ),
             )
@@ -527,31 +527,31 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             asserter.assert_(
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 30, "data": "d1"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (data) VALUES "
-                    "(:data::VARCHAR(30)) RETURNING "
+                    "(:data::VARCHAR) RETURNING "
                     "testtable.id",
                     {"data": "d2"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 31, "data": "d3"}, {"id": 32, "data": "d4"}],
                 ),
                 DialectSQL(
-                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR(30))",
+                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR)",
                     [{"data": "d5"}, {"data": "d6"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 33, "data": "d7"}],
                 ),
                 DialectSQL(
-                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR(30))",
+                    "INSERT INTO testtable (data) VALUES (:data::VARCHAR)",
                     [{"data": "d8"}],
                 ),
             )
@@ -623,33 +623,33 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             asserter.assert_(
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 30, "data": "d1"},
                 ),
                 CursorSQL("select nextval('my_seq')", consume_statement=False),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 1, "data": "d2"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 31, "data": "d3"}, {"id": 32, "data": "d4"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES (nextval('%s'), "
-                    ":data::VARCHAR(30))" % seqname,
+                    ":data::VARCHAR)" % seqname,
                     [{"data": "d5"}, {"data": "d6"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 33, "data": "d7"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES (nextval('%s'), "
-                    ":data::VARCHAR(30))" % seqname,
+                    ":data::VARCHAR)" % seqname,
                     [{"data": "d8"}],
                 ),
             )
@@ -722,33 +722,33 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             asserter.assert_(
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     {"id": 30, "data": "d1"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(nextval('my_seq'), :data::VARCHAR(30)) "
+                    "(nextval('my_seq'), :data::VARCHAR) "
                     "RETURNING testtable.id",
                     {"data": "d2"},
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 31, "data": "d3"}, {"id": 32, "data": "d4"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES (nextval('%s'), "
-                    ":data::VARCHAR(30))" % seqname,
+                    ":data::VARCHAR)" % seqname,
                     [{"data": "d5"}, {"data": "d6"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES "
-                    "(:id::INTEGER, :data::VARCHAR(30))",
+                    "(:id::INTEGER, :data::VARCHAR)",
                     [{"id": 33, "data": "d7"}],
                 ),
                 DialectSQL(
                     "INSERT INTO testtable (id, data) VALUES (nextval('%s'), "
-                    ":data::VARCHAR(30))" % seqname,
+                    ":data::VARCHAR)" % seqname,
                     [{"data": "d8"}],
                 ),
             )
@@ -982,8 +982,7 @@ class MatchTest(fixtures.TablesTest, AssertsCompiledSQL):
         if self._strs_render_bind_casts(connection):
             self.assert_compile(
                 matchtable.c.title.match("somstr"),
-                "matchtable.title @@ "
-                "plainto_tsquery(%(title_1)s::VARCHAR(200))",
+                "matchtable.title @@ " "plainto_tsquery(%(title_1)s::VARCHAR)",
             )
         else:
             self.assert_compile(
@@ -998,7 +997,7 @@ class MatchTest(fixtures.TablesTest, AssertsCompiledSQL):
         if self._strs_render_bind_casts(connection):
             self.assert_compile(
                 matchtable.c.title.match("somstr"),
-                "matchtable.title @@ plainto_tsquery($1::VARCHAR(200))",
+                "matchtable.title @@ plainto_tsquery($1::VARCHAR)",
             )
         else:
             self.assert_compile(
diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py
index 1ff9d785f..0ee909541 100644
--- a/test/dialect/postgresql/test_types.py
+++ b/test/dialect/postgresql/test_types.py
@@ -91,6 +91,28 @@ from sqlalchemy.types import UserDefinedType
 from ...engine.test_ddlevents import DDLEventWCreateHarness
 
 
+class MiscTypesTest(AssertsCompiledSQL, fixtures.TestBase):
+    @testing.combinations(
+        ("asyncpg", "x LIKE $1::VARCHAR"),
+        ("psycopg", "x LIKE %(x_1)s::VARCHAR"),
+        ("psycopg2", "x LIKE %(x_1)s"),
+        ("pg8000", "x LIKE %s::VARCHAR"),
+    )
+    def test_string_coercion_no_len(self, driver, expected):
+        """test #9511.
+
+        comparing to string does not include length in the cast for those
+        dialects that require a cast.
+
+        """
+
+        self.assert_compile(
+            column("x", String(2)).like("%a%"),
+            expected,
+            dialect=f"postgresql+{driver}",
+        )
+
+
 class FloatCoercionTest(fixtures.TablesTest, AssertsExecutionResults):
     __only_on__ = "postgresql"
     __dialect__ = postgresql.dialect()
diff --git a/test/sql/test_type_expressions.py b/test/sql/test_type_expressions.py
index c2ecd46e2..77502b488 100644
--- a/test/sql/test_type_expressions.py
+++ b/test/sql/test_type_expressions.py
@@ -93,7 +93,7 @@ class _ExprFixture:
     def _variant_fixture(self, inner_fixture):
         type_ = inner_fixture.c.y.type
 
-        variant = String().with_variant(type_, "default")
+        variant = String(30).with_variant(type_, "default")
         return self._test_table(variant)
 
     def _dialect_level_fixture(self):