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
@@ -865,7 +861,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

@@ -198,6 +198,34 @@ def test_dialect_uses_engine(self, connector):
             res = connector._dialect
             assert res == attached_engine.dialect
 
+    def test_merge_sql_types_text_current_max(self, connector):
+        current_type = sqlalchemy.types.VARCHAR(length=None)
+        sql_type = sqlalchemy.types.VARCHAR(length=255)
+        compatible_sql_type = connector.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, connector):
+        current_type = sqlalchemy.types.VARCHAR(length=255)
+        sql_type = sqlalchemy.types.VARCHAR(length=64)
+        compatible_sql_type = connector.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, connector):
+        current_type = sqlalchemy.types.VARCHAR(length=64)
+        sql_type = sqlalchemy.types.VARCHAR(length=None)
+        compatible_sql_type = connector.merge_sql_types([current_type, sql_type])
+        # Check the current VARCHAR(64) is chosen over default VARCHAR(max)
+        assert compatible_sql_type is current_type
+
+    def test_merge_sql_types_text_current_less_than(self, connector):
+        current_type = sqlalchemy.types.VARCHAR(length=64)
+        sql_type = sqlalchemy.types.VARCHAR(length=255)
+        compatible_sql_type = connector.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
+
     @pytest.mark.parametrize(
         "types,expected_type",
         [
@@ -222,7 +250,7 @@ def test_dialect_uses_engine(self, connector):
             ),
         ],
     )
-    def test_merge_sql_types(
+    def test_merge_generic_sql_types(
         self,
         connector: SQLConnector,
         types: list[sqlalchemy.types.TypeEngine],