Case sensitive/insensitive table validation (#3580)

closes #3568 

Added case sensitive flag for metadata comparison. To consider/ignore
column name case.

Author: FastLee

src/databricks/labs/ucx/recon/base.py

@@ -1,5 +1,6 @@
 import dataclasses
 from abc import ABC, abstractmethod
+from collections.abc import Callable
 from dataclasses import dataclass
 
 
@@ -82,7 +83,9 @@ def as_dict(self):
 
 class TableMetadataRetriever(ABC):
     @abstractmethod
-    def get_metadata(self, entity: TableIdentifier) -> TableMetadata:
+    def get_metadata(
+        self, entity: TableIdentifier, *, column_name_transformer: Callable[[str], str] = str
+    ) -> TableMetadata:
         """
         Get metadata for a given table
         """

src/databricks/labs/ucx/recon/metadata_retriever.py

@@ -1,4 +1,4 @@
-from collections.abc import Iterator
+from collections.abc import Iterator, Callable
 
 from databricks.labs.lsql.backends import SqlBackend
 from databricks.labs.lsql.core import Row
@@ -10,7 +10,9 @@ class DatabricksTableMetadataRetriever(TableMetadataRetriever):
     def __init__(self, sql_backend: SqlBackend):
         self._sql_backend = sql_backend
 
-    def get_metadata(self, entity: TableIdentifier) -> TableMetadata:
+    def get_metadata(
+        self, entity: TableIdentifier, *, column_name_transformer: Callable[[str], str] = str
+    ) -> TableMetadata:
         """
         This method retrieves the metadata for a given table. It takes a TableIdentifier object as input,
         which represents the table for which the metadata is to be retrieved.
@@ -24,11 +26,11 @@ def get_metadata(self, entity: TableIdentifier) -> TableMetadata:
         # Partition information are typically prefixed with a # symbol,
         # so any column name starting with # is excluded from the final set of column metadata.
         # The column metadata objects are sorted by column name to ensure a consistent order.
-        columns = {
-            ColumnMetadata(str(row["col_name"]), str(row["data_type"]))
-            for row in query_result
-            if not str(row["col_name"]).startswith("#")
-        }
+        columns = set()
+        for row in query_result:
+            if str(row["col_name"]).startswith("#"):
+                continue
+            columns.add(ColumnMetadata(column_name_transformer(str(row["col_name"])), str(row["data_type"])))
         return TableMetadata(entity, sorted(columns, key=lambda x: x.name))
 
     @classmethod
@@ -38,7 +40,7 @@ def _build_metadata_query(cls, entity: TableIdentifier) -> str:
 
         query = f"""
             SELECT
-                LOWER(column_name) AS col_name,
+                column_name AS col_name,
                 full_data_type AS data_type
             FROM
                 {entity.catalog_escaped}.information_schema.columns

src/databricks/labs/ucx/recon/schema_comparator.py

@@ -1,3 +1,5 @@
+from collections.abc import Callable
+
 from .base import (
     SchemaComparator,
     SchemaComparisonEntry,
@@ -9,8 +11,14 @@
 
 
 class StandardSchemaComparator(SchemaComparator):
-    def __init__(self, metadata_retriever: TableMetadataRetriever):
+    def __init__(self, metadata_retriever: TableMetadataRetriever, *, case_sensitive: bool = False):
         self._metadata_retriever = metadata_retriever
+        self._case_sensitive = case_sensitive
+
+    def _column_name_transformer(self) -> Callable[[str], str]:
+        if self._case_sensitive:
+            return lambda _: _
+        return str.lower
 
     def compare_schema(self, source: TableIdentifier, target: TableIdentifier) -> SchemaComparisonResult:
         """
@@ -26,8 +34,12 @@ def compare_schema(self, source: TableIdentifier, target: TableIdentifier) -> Sc
         return SchemaComparisonResult(is_matching, comparison_result)
 
     def _eval_schema_diffs(self, source: TableIdentifier, target: TableIdentifier) -> list[SchemaComparisonEntry]:
-        source_metadata = self._metadata_retriever.get_metadata(source)
-        target_metadata = self._metadata_retriever.get_metadata(target)
+        source_metadata = self._metadata_retriever.get_metadata(
+            source, column_name_transformer=self._column_name_transformer()
+        )
+        target_metadata = self._metadata_retriever.get_metadata(
+            target, column_name_transformer=self._column_name_transformer()
+        )
         # Combine the sets of column names for both the source and target tables
         # to create a set of all unique column names from both tables.
         source_column_names = {column.name for column in source_metadata.columns}

tests/unit/recon/test_schema_comparator.py

@@ -1,3 +1,4 @@
+import pytest
 from databricks.labs.lsql.backends import MockBackend
 
 from databricks.labs.ucx.recon.base import TableIdentifier, SchemaComparisonResult, SchemaComparisonEntry
@@ -100,3 +101,36 @@ def test_schema_comparison_failure(metadata_row_factory):
     schema_comparator = StandardSchemaComparator(metadata_retriever)
     actual_comparison_result = schema_comparator.compare_schema(source, target)
     assert actual_comparison_result == expected_comparison_result
+
+
+@pytest.mark.parametrize(
+    "source_column, target_column, case_sensitive, expected_pass",
+    [
+        ("column1", "columnx", True, False),
+        ("column1", "column1", True, True),
+        ("column1", "Column1", True, False),
+        ("column1", "Column1", False, True),
+        ("CoLuMn1", "cOlUmN1", True, False),
+        ("CoLuMn1", "cOlUmN1", False, True),
+    ],
+)
+def test_schema_comparison_case(metadata_row_factory, source_column, target_column, case_sensitive, expected_pass):
+    source = TableIdentifier("hive_metastore", "db1", "table1")
+    target = TableIdentifier("catalog1", "schema1", "table1")
+    sql_backend = MockBackend(
+        rows={
+            "DESCRIBE TABLE": metadata_row_factory[
+                (source_column, "int"),
+                ("column2", "string"),
+            ],
+            f"{target.catalog_escaped}\\.information_schema\\.columns": metadata_row_factory[
+                (target_column, "int"),
+                ("column2", "string"),
+            ],
+        }
+    )
+
+    metadata_retriever = DatabricksTableMetadataRetriever(sql_backend)
+    schema_comparator = StandardSchemaComparator(metadata_retriever, case_sensitive=case_sensitive)
+    actual_comparison_result = schema_comparator.compare_schema(source, target)
+    assert actual_comparison_result.is_matching == expected_pass