fix: SQL Targets ignore collation when evaluating column data types

singer_sdk/connectors/sql.py

@@ -943,6 +943,41 @@ def get_column_alter_ddl(
             },
         )
 
+    @staticmethod
+    def remove_collation(
+        column_type: sqlalchemy.types.TypeEngine,
+    ) -> tuple[sqlalchemy.types.TypeEngine, str | None]:
+        """Removes collation for the given column TypeEngine instance.
+
+        Args:
+            column_type: Column SQLAlchemy type.
+
+        Returns:
+            A tuple containing the revised column type and the removed collation.
+        """
+        if hasattr(column_type, "collation") and column_type.collation:
+            column_type_collation = column_type.collation
+            setattr(column_type, "collation", None)
+            return column_type, column_type_collation
+        return column_type, None
+
+    @staticmethod
+    def update_collation(
+        column_type: sqlalchemy.types.TypeEngine, collation: str | None
+    ) -> sqlalchemy.types.TypeEngine:
+        """Sets column collation if column type has a collation attribute.
+
+        Args:
+            column_type: Column SQLAlchemy type.
+            collation: The colation
+
+        Returns:
+            The revised column type with passed collation or unchanged column type.
+        """
+        if hasattr(column_type, "collation") and collation:
+            setattr(column_type, "collation", collation)
+        return column_type
+
     def _adapt_column_type(
         self,
         full_table_name: str,
@@ -959,10 +994,14 @@ def _adapt_column_type(
         Raises:
             NotImplementedError: if altering columns is not supported.
         """
+        # collation_removed = False
         current_type: sqlalchemy.types.TypeEngine = self._get_column_type(
             full_table_name, column_name
         )
 
+        # remove collation if present and save it
+        current_type, 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 +1016,12 @@ def _adapt_column_type(
             # Nothing to do
             return
 
+        # Put the collation level back before altering the column
+        if current_type_collation:
+            compatible_sql_type = 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, connector):
+        remove_collation = getattr(connector, "remove_collation")
+        test_collation = "SQL_Latin1_General_CP1_CI_AS"
+        current_type = sqlalchemy.types.Text(collation=test_collation)
+        current_type, current_type_collation = remove_collation(current_type)
+        # Check collation was set to None by the function
+        assert current_type.collation == 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, connector):
+        remove_collation = getattr(connector, "remove_collation")
+        current_type = sqlalchemy.types.Integer()
+        current_type, current_type_collation = remove_collation(current_type)
+        # Check there is not a collation attribute
+        assert hasattr(current_type, "collation") == False
+        # Check that we get the same type we put in
+        assert str(current_type) == "INTEGER"
+        # Check that this variable is missing
+        assert not current_type_collation
+
+    def test_update_collation_text_type(self, connector):
+        update_collation = getattr(connector, "update_collation")
+        test_collation = "SQL_Latin1_General_CP1_CI_AS"
+        compatible_type = sqlalchemy.types.Text(collation=None)
+        compatible_type = 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, connector):
+        update_collation = getattr(connector, "update_collation")
+        test_collation = "SQL_Latin1_General_CP1_CI_AS"
+        compatible_type = sqlalchemy.types.Integer()
+        compatible_type = update_collation(compatible_type, test_collation)
+        # Check there is not a collation attribute
+        assert hasattr(compatible_type, "collation") == False
+        # Check that we get the same type we put in
+        assert str(compatible_type) == "INTEGER"