fix: SQL Targets ignore collation when evaluating column data types (#…

…1385)

Co-authored-by: Edgar R. M <[email protected]>
fix #1125

singer_sdk/connectors/sql.py

@@ -943,6 +943,37 @@ def get_column_alter_ddl(
             },
         )
 
+    @staticmethod
+    def remove_collation(
+        column_type: sqlalchemy.types.TypeEngine,
+    ) -> str | None:
+        """Removes collation for the given column TypeEngine instance.
+
+        Args:
+            column_type: Column SQLAlchemy type.
+
+        Returns:
+             The removed collation as a string.
+        """
+        if hasattr(column_type, "collation") and column_type.collation:
+            column_type_collation: str = column_type.collation
+            setattr(column_type, "collation", None)
+            return column_type_collation
+        return None
+
+    @staticmethod
+    def update_collation(
+        column_type: sqlalchemy.types.TypeEngine, collation: str | None
+    ) -> None:
+        """Sets column collation if column type has a collation attribute.
+
+        Args:
+            column_type: Column SQLAlchemy type.
+            collation: The colation
+        """
+        if hasattr(column_type, "collation") and collation:
+            setattr(column_type, "collation", collation)
+
     def _adapt_column_type(
         self,
         full_table_name: str,
@@ -963,6 +994,9 @@ def _adapt_column_type(
             full_table_name, column_name
         )
 
+        # remove collation if present and save it
+        current_type_collation = self.remove_collation(current_type)
+
         # Check if the existing column type and the sql type are the same
         if str(sql_type) == str(current_type):
             # The current column and sql type are the same
@@ -977,6 +1011,10 @@ def _adapt_column_type(
             # Nothing to do
             return
 
+        # Put the collation level back before altering the column
+        if current_type_collation:
+            self.update_collation(compatible_sql_type, current_type_collation)
+
         if not self.allow_column_alter:
             raise NotImplementedError(
                 "Altering columns is not supported. "

tests/core/test_connector_sql.py

@@ -91,3 +91,42 @@ def test_get_column_ddl(
             )
         )
         assert statement == rendered_statement
+
+    def test_remove_collation_text_type(self):
+        remove_collation = SQLConnector.remove_collation
+        test_collation = "SQL_Latin1_General_CP1_CI_AS"
+        current_type = sqlalchemy.types.Text(collation=test_collation)
+        current_type_collation = remove_collation(current_type)
+        # Check collation was set to None by the function
+        assert current_type.collation is None
+        # Check that we get the same collation we put in back out
+        assert current_type_collation == test_collation
+
+    def test_remove_collation_non_text_type(self):
+        remove_collation = SQLConnector.remove_collation
+        current_type = sqlalchemy.types.Integer()
+        current_type_collation = remove_collation(current_type)
+        # Check there is not a collation attribute
+        assert not hasattr(current_type, "collation")
+        # Check that we get the same type we put in
+        assert str(current_type) == "INTEGER"
+        # Check that this variable is missing
+        assert current_type_collation is None
+
+    def test_update_collation_text_type(self):
+        update_collation = SQLConnector.update_collation
+        test_collation = "SQL_Latin1_General_CP1_CI_AS"
+        compatible_type = sqlalchemy.types.Text(collation=None)
+        update_collation(compatible_type, test_collation)
+        # Check collation was set to the value we put in
+        assert compatible_type.collation == test_collation
+
+    def test_update_collation_non_text_type(self):
+        update_collation = SQLConnector.update_collation
+        test_collation = "SQL_Latin1_General_CP1_CI_AS"
+        compatible_type = sqlalchemy.types.Integer()
+        update_collation(compatible_type, test_collation)
+        # Check there is not a collation attribute
+        assert not hasattr(compatible_type, "collation")
+        # Check that we get the same type we put in
+        assert str(compatible_type) == "INTEGER"