fix: Handle merging of SQL types when character column lengths are less than the max

singer_sdk/connectors/sql.py

@@ -834,11 +834,7 @@ def merge_sql_types(
         # Gathering Type to match variables
         # sent in _adapt_column_type
         current_type = sql_types[0]
-
-        # Getting the length of each type
-        sql_type_len: int = getattr(sql_types[1], "length", 0)
-        if sql_type_len is None:
-            sql_type_len = 0
+        cur_len: int = getattr(current_type, "length", 0)
 
         # Convert the two types given into a sorted list
         # containing the best conversion classes
@@ -866,7 +862,11 @@ def merge_sql_types(
                     (sqlalchemy.types.String, sqlalchemy.types.Unicode),
                 ):
                     # If length None or 0 then is varchar max ?
-                    if (opt_len is None) or (opt_len == 0):
+                    if (
+                        (opt_len is None)
+                        or (opt_len == 0)
+                        or (cur_len and (opt_len >= cur_len))
+                    ):
                         return opt
                 # If best conversion class is equal to current type
                 # return the best conversion class

tests/core/test_connector_sql.py

@@ -197,3 +197,35 @@ def test_dialect_uses_engine(self, connector):
         with mock.patch.object(attached_engine, "dialect") as _:
             res = connector._dialect
             assert res == attached_engine.dialect
+
+    def test_merge_sql_types_text_current_max(self):
+        cls = SQLConnector()
+        current_type = sqlalchemy.types.VARCHAR(length=None)
+        sql_type = sqlalchemy.types.VARCHAR(length=255)
+        compatible_sql_type = cls.merge_sql_types([current_type, sql_type])
+        # Check that the current VARCHAR(MAX) type is kept
+        assert compatible_sql_type is current_type
+
+    def test_merge_sql_types_text_current_greater_than(self):
+        cls = SQLConnector()
+        current_type = sqlalchemy.types.VARCHAR(length=255)
+        sql_type = sqlalchemy.types.VARCHAR(length=64)
+        compatible_sql_type = cls.merge_sql_types([current_type, sql_type])
+        # Check the current greater VARCHAR(255) is kept
+        assert compatible_sql_type is current_type
+
+    def test_merge_sql_types_text_proposed_max(self):
+        cls = SQLConnector()
+        current_type = sqlalchemy.types.VARCHAR(length=64)
+        sql_type = sqlalchemy.types.VARCHAR(length=None)
+        compatible_sql_type = cls.merge_sql_types([current_type, sql_type])
+        # Check the current VARCHAR(64) is chosen over default varcahr(max)
+        assert compatible_sql_type is current_type
+
+    def test_merge_sql_types_text_current_less_than(self):
+        cls = SQLConnector()
+        current_type = sqlalchemy.types.VARCHAR(length=64)
+        sql_type = sqlalchemy.types.VARCHAR(length=255)
+        compatible_sql_type = cls.merge_sql_types([current_type, sql_type])
+        # Check that VARCHAR(255) is chosen over the lesser current VARCHAR(64)
+        assert compatible_sql_type is sql_type