diff --git a/poetry.lock b/poetry.lock
index bf5b6db5a..cbd1fea77 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 42f432c1f..1a307e6cb 100644
--- a/singer_sdk/connectors/sql.py
+++ b/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
diff --git a/tests/core/test_connector_sql.py b/tests/core/test_connector_sql.py
index b1ca69ba1..dc16264e1 100644
--- a/tests/core/test_connector_sql.py
+++ b/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],