fix(ingestion/lookml): resolve CLL issue caused by column name casing.

metadata-ingestion/src/datahub/ingestion/source/looker/view_upstream.py

@@ -25,14 +25,14 @@
     LookMLSourceReport,
 )
 from datahub.ingestion.source.looker.urn_functions import get_qualified_table_name
+from datahub.sql_parsing.schema_resolver import match_columns_to_schema
 from datahub.sql_parsing.sqlglot_lineage import (
     ColumnLineageInfo,
     ColumnRef,
     SqlParsingResult,
     Urn,
     create_and_cache_schema_resolver,
     create_lineage_sql_parsed_result,
-    match_columns_to_schema,
 )
 
 logger = logging.getLogger(__name__)

metadata-ingestion/src/datahub/sql_parsing/schema_resolver.py

@@ -300,3 +300,19 @@ def _convert_schema_field_list_to_info(
 
 def _convert_schema_aspect_to_info(schema_metadata: SchemaMetadataClass) -> SchemaInfo:
     return _convert_schema_field_list_to_info(schema_metadata.fields)
+
+
+def match_columns_to_schema(
+    schema_info: SchemaInfo, input_columns: List[str]
+) -> List[str]:
+    column_from_gms: List[str] = list(schema_info.keys())  # list() to silent lint
+
+    gms_column_map: Dict[str, str] = {
+        column.lower(): column for column in column_from_gms
+    }
+
+    output_columns: List[str] = [
+        gms_column_map.get(column.lower(), column) for column in input_columns
+    ]
+
+    return output_columns

metadata-ingestion/src/datahub/sql_parsing/sqlglot_lineage.py

@@ -1275,20 +1275,6 @@ def infer_output_schema(result: SqlParsingResult) -> Optional[List[SchemaFieldCl
     return output_schema
 
 
-def match_columns_to_schema(
-    schema_info: SchemaInfo, input_columns: List[str]
-) -> List[str]:
-    column_from_gms: List[str] = list(schema_info.keys())  # list() to silent lint
-
-    output_columns: List[str] = [
-        column
-        for column in column_from_gms
-        if column.lower() in map(str.lower, input_columns)
-    ]
-
-    return output_columns
-
-
 def view_definition_lineage_helper(
     result: SqlParsingResult, view_urn: str
 ) -> SqlParsingResult:

metadata-ingestion/tests/unit/sql_parsing/test_schemaresolver.py

@@ -1,26 +1,64 @@
-from datahub.sql_parsing.schema_resolver import SchemaResolver, _TableName
+from datahub.sql_parsing.schema_resolver import (
+    SchemaInfo,
+    SchemaResolver,
+    _TableName,
+    match_columns_to_schema,
+)
 
 
-def test_basic_schema_resolver():
+def create_default_schema_resolver(urn: str) -> SchemaResolver:
     schema_resolver = SchemaResolver(
         platform="redshift",
         env="PROD",
         graph=None,
     )
 
     schema_resolver.add_raw_schema_info(
-        urn="urn:li:dataset:(urn:li:dataPlatform:redshift,my_db.public.test_table,PROD)",
+        urn=urn,
         schema_info={"name": "STRING"},
     )
 
+    return schema_resolver
+
+
+def test_basic_schema_resolver():
+    input_urn = (
+        "urn:li:dataset:(urn:li:dataPlatform:redshift,my_db.public.test_table,PROD)"
+    )
+
+    schema_resolver = create_default_schema_resolver(urn=input_urn)
+
     urn, schema = schema_resolver.resolve_table(
         _TableName(database="my_db", db_schema="public", table="test_table")
     )
-    assert (
-        urn
-        == "urn:li:dataset:(urn:li:dataPlatform:redshift,my_db.public.test_table,PROD)"
+
+    assert urn == input_urn
+
+    assert schema
+
+    assert schema["name"]
+
+    assert schema_resolver.schema_count() == 1
+
+
+def test_resolve_urn():
+    input_urn: str = (
+        "urn:li:dataset:(urn:li:dataPlatform:redshift,my_db.public.test_table,PROD)"
+    )
+
+    schema_resolver = create_default_schema_resolver(urn=input_urn)
+
+    schema_resolver.add_raw_schema_info(
+        urn=input_urn,
+        schema_info={"name": "STRING"},
     )
+
+    urn, schema = schema_resolver.resolve_urn(urn=input_urn)
+
+    assert urn == input_urn
+
     assert schema
+
     assert schema["name"]
 
     assert schema_resolver.schema_count() == 1
@@ -62,3 +100,29 @@ def test_get_urn_for_table_not_lower_should_keep_capital_letters():
         == "urn:li:dataset:(urn:li:dataPlatform:mssql,Uppercased-Instance.Database.DataSet.Table,PROD)"
     )
     assert schema_resolver.schema_count() == 0
+
+
+def test_match_columns_to_schema():
+    input_urn = (
+        "urn:li:dataset:(urn:li:dataPlatform:redshift,my_db.public.test_table,PROD)"
+    )
+
+    schema_resolver = create_default_schema_resolver(urn=input_urn)
+
+    schema_resolver.add_raw_schema_info(
+        urn=input_urn,
+        schema_info={"address": "STRING"},
+    )
+
+    schema_resolver.add_raw_schema_info(
+        urn=input_urn,
+        schema_info={"Id": "STRING"},
+    )
+
+    schema_info: SchemaInfo = {"id": "string", "Name": "string", "Address": "string"}
+
+    output_columns = match_columns_to_schema(
+        schema_info, input_columns=["Id", "name", "address", "weight"]
+    )
+
+    assert output_columns == ["id", "Name", "Address", "weight"]