Repository: meltano/sdk (python)

Issue:
[Bug]: `merge_sql_type()` does not handle varchar or nvarchar lengths less than `max`
### Singer SDK Version

0.13.1

### Python Version

3.9

### Bug scope

Targets (data type handling, batching, SQL object generation, etc.)

### Operating System

Windows

### Description

The `merge_sql_types` function only handles instances in which one of the character columns has `null` or `0` lengths.  Both values are converted to `max` by SQLAlchemy so the function is set to prefer them.  If there is no `max` present for example only `64` or `64` the function throws and error letting you know it can't merge the two types.

`ValueError: Unable to merge sql types: VARCHAR(64) COLLATE "SQL_Latin1_General_CP1_CI_AS", VARCHAR(64)`

This was first noticed here https://github.com/meltano/sdk/issues/1125#issuecomment-1302781265 

### Code

```python
assert len(sql_types) == 2
        # Get the generic type class
        for opt in sql_types:
            # Get the length
            opt_len: int = getattr(opt, "length", 0)
            generic_type = type(opt.as_generic())

            if isinstance(generic_type, type):
                if issubclass(
                    generic_type,
                    (sqlalchemy.types.String, sqlalchemy.types.Unicode),
                ):
                    # If length None or 0 then is varchar max ?
                    if (opt_len is None) or (opt_len == 0):
                        return opt
                elif isinstance(
                    generic_type,
                    (sqlalchemy.types.String, sqlalchemy.types.Unicode),
                ):
                    # If length None or 0 then is varchar max ?
                    if (opt_len is None) or (opt_len == 0):
                        return opt
                # If best conversion class is equal to current type
                # return the best conversion class
                elif str(opt) == str(current_type):
                    return opt

        raise ValueError(
            f"Unable to merge sql types: {', '.join([str(t) for t in sql_types])}"
        )


2022-11-11T15:26:46.648150Z [info     ] Traceback (most recent call last): cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.648150Z [info     ]   File "C:\Program Files\Python39\lib\runpy.py", line 197, in _run_module_as_main cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.648150Z [info     ]     return _run_code(code, main_globals, None, cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.648150Z [info     ]   File "C:\Program Files\Python39\lib\runpy.py", line 87, in _run_code cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.648150Z [info     ]     exec(code, run_globals)    cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.648150Z [info     ]   File "C:\development\projects\test-mssql\.meltano\loaders\target-mssql\venv\Scripts\target-mssql.exe\__main__.py", line 7, in <module> cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.663779Z [info     ]   File "C:\development\projects\test-mssql\.meltano\loaders\target-mssql\venv\lib\site-packages\click\core.py", line 1130, in __call__ cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.663779Z [info     ]     return self.main(*args, **kwargs) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.663779Z [info     ]   File "C:\development\projects\test-mssql\.meltano\loaders\target-mssql\venv\lib\site-packages\click\core.py", line 1055, in main cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.663779Z [info     ]     rv = self.invoke(ctx)      cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.663779Z [info     ]   File "C:\development\projects\test-mssql\.meltano\loaders\target-mssql\venv\lib\site-packages\click\core.py", line 1404, in invoke cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.663779Z [info     ]     return ctx.invoke(self.callback, **ctx.params) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.679404Z [info     ]   File "C:\development\projects\test-mssql\.meltano\loaders\target-mssql\venv\lib\site-packages\click\core.py", line 760, in invoke cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.679404Z [info     ]     return __callback(*args, **kwargs) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.679404Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\target_base.py", line 566, in cli cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.679404Z [info     ]     target.listen(file_input)  cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.679404Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\io_base.py", line 34, in listen cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.695027Z [info     ]     self._process_lines(file_input) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.695027Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\target_base.py", line 267, in _process_lines cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.695027Z [info     ]     counter = super()._process_lines(file_input) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.695027Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\io_base.py", line 78, in _process_lines cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.695027Z [info     ]     self._process_schema_message(line_dict) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.695027Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\target_base.py", line 366, in _process_schema_message cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.710652Z [info     ]     _ = self.get_sink(         cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.710652Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\target_base.py", line 160, in get_sink cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.710652Z [info     ]     return self.add_sink(stream_name, schema, key_properties) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.710652Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\target_base.py", line 235, in add_sink cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.710652Z [info     ]     sink.setup()               cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.726273Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\sinks\sql.py", line 215, in setup cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.726273Z [info     ]     self.connector.prepare_table( cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.726273Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\connectors\sql.py", line 685, in prepare_table cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.726273Z [info     ]     self.prepare_column(       cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.726273Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\connectors\sql.py", line 710, in prepare_column cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.741894Z [info     ]     self._adapt_column_type(   cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.741894Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\connectors\sql.py", line 981, in _adapt_column_type cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql      
2022-11-11T15:26:46.741894Z [info     ]     compatible_sql_type = self.merge_sql_types([current_type, sql_type]) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.741894Z [info     ]   File "C:\development\MeltanoContribute\sdk\singer_sdk\connectors\sql.py", line 803, in merge_sql_types cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.741894Z [info     ]     raise ValueError(          cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.757517Z [info     ] ValueError: Unable to merge sql types: VARCHAR(64) COLLATE "SQL_Latin1_General_CP1_CI_AS", VARCHAR(64) cmd_type=elb consumer=True name=target-mssql producer=False stdio=stderr string_id=target-mssql
2022-11-11T15:26:46.788768Z [error    ] Loader failed
```


Pull request description:
fix: Handle merging of SQL types when character column lengths are less than the max
This is an attempt to fix `merge_sql_type` to handle instances when all passed character column lengths are less than max.  SDK 0.13.1 needs PR #1168 present to reach this fix.  

fixes #1170 

<!-- readthedocs-preview meltano-sdk start -->
----
:books: Documentation preview :books:: https://meltano-sdk--1172.org.readthedocs.build/en/1172/

<!-- readthedocs-preview meltano-sdk end -->

Patch:
diff --git a/poetry.lock b/poetry.lock
index bf5b6db5..cbd1fea7 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.4.0 and should not be changed by hand.
 
 [[package]]
 name = "alabaster"
@@ -2335,7 +2335,7 @@ files = [
 ]
 
 [package.dependencies]
-greenlet = {version = "!=0.4.17", markers = "python_version >= \"3\" and (platform_machine == \"aarch64\" or platform_machine == \"ppc64le\" or platform_machine == \"x86_64\" or platform_machine == \"amd64\" or platform_machine == \"AMD64\" or platform_machine == \"win32\" or platform_machine == \"WIN32\")"}
+greenlet = {version = "!=0.4.17", markers = "python_version >= \"3\" and platform_machine == \"aarch64\" or python_version >= \"3\" and platform_machine == \"ppc64le\" or python_version >= \"3\" and platform_machine == \"x86_64\" or python_version >= \"3\" and platform_machine == \"amd64\" or python_version >= \"3\" and platform_machine == \"AMD64\" or python_version >= \"3\" and platform_machine == \"win32\" or python_version >= \"3\" and platform_machine == \"WIN32\""}
 importlib-metadata = {version = "*", markers = "python_version < \"3.8\""}
 
 [package.extras]
@@ -2688,7 +2688,7 @@ docs = ["furo", "jaraco.packaging (>=9)", "jaraco.tidelift (>=1.4)", "rst.linker
 testing = ["flake8 (<5)", "func-timeout", "jaraco.functools", "jaraco.itertools", "more-itertools", "pytest (>=6)", "pytest-black (>=0.3.7)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=1.3)", "pytest-flake8", "pytest-mypy (>=0.9.1)"]
 
 [extras]
-docs = ["sphinx", "furo", "sphinx-copybutton", "myst-parser", "sphinx-autobuild", "sphinx-reredirects"]
+docs = ["furo", "myst-parser", "sphinx", "sphinx-autobuild", "sphinx-copybutton", "sphinx-reredirects"]
 s3 = ["fs-s3fs"]
 testing = ["pytest", "pytest-durations"]
 
diff --git a/singer_sdk/connectors/sql.py b/singer_sdk/connectors/sql.py
index 42f432c1..1a307e6c 100644
--- a/singer_sdk/connectors/sql.py
+++ b/singer_sdk/connectors/sql.py
@@ -834,11 +834,7 @@ class SQLConnector:
         # 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 @@ class SQLConnector:
                     (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


Test patch:
diff --git a/tests/core/test_connector_sql.py b/tests/core/test_connector_sql.py
index b1ca69ba..dc16264e 100644
--- a/tests/core/test_connector_sql.py
+++ b/tests/core/test_connector_sql.py
@@ -198,6 +198,34 @@ class TestConnectorSQL:
             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 @@ class TestConnectorSQL:
             ),
         ],
     )
-    def test_merge_sql_types(
+    def test_merge_generic_sql_types(
         self,
         connector: SQLConnector,
         types: list[sqlalchemy.types.TypeEngine],