test(ingest): test case statements with sql parser (#8437)

Author: hsheth2

metadata-ingestion/tests/unit/sql_parsing/goldens/test_snowflake_case_statement.json

@@ -0,0 +1,37 @@
+{
+    "query_type": "SELECT",
+    "in_tables": [
+        "urn:li:dataset:(urn:li:dataPlatform:snowflake,snowflake_sample_data.tpch_sf1.orders,PROD)"
+    ],
+    "out_tables": [],
+    "column_lineage": [
+        {
+            "downstream": {
+                "table": null,
+                "column": "TOTAL_PRICE_CATEGORY"
+            },
+            "upstreams": [
+                {
+                    "table": "urn:li:dataset:(urn:li:dataPlatform:snowflake,snowflake_sample_data.tpch_sf1.orders,PROD)",
+                    "column": "totalprice"
+                }
+            ]
+        },
+        {
+            "downstream": {
+                "table": null,
+                "column": "TOTAL_PRICE_SUCCESS"
+            },
+            "upstreams": [
+                {
+                    "table": "urn:li:dataset:(urn:li:dataPlatform:snowflake,snowflake_sample_data.tpch_sf1.orders,PROD)",
+                    "column": "is_payment_successful"
+                },
+                {
+                    "table": "urn:li:dataset:(urn:li:dataPlatform:snowflake,snowflake_sample_data.tpch_sf1.orders,PROD)",
+                    "column": "totalprice"
+                }
+            ]
+        }
+    ]
+}

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

@@ -287,6 +287,34 @@ def test_snowflake_column_normalization():
     )
 
 
+def test_snowflake_case_statement():
+    assert_sql_result(
+        """
+SELECT
+    CASE
+        WHEN o."totalprice" > 1000 THEN 'high'
+        WHEN o."totalprice" > 100 THEN 'medium'
+        ELSE 'low'
+    END as total_price_category,
+    -- Also add a case where the column is in the THEN clause.
+    CASE
+        WHEN o."is_payment_successful" THEN o."totalprice"
+        ELSE 0
+    END as total_price_success
+FROM snowflake_sample_data.tpch_sf1.orders o
+""",
+        dialect="snowflake",
+        schemas={
+            "urn:li:dataset:(urn:li:dataPlatform:snowflake,snowflake_sample_data.tpch_sf1.orders,PROD)": {
+                "orderkey": "NUMBER",
+                "totalprice": "FLOAT",
+                "is_payment_successful": "BOOLEAN",
+            },
+        },
+        expected_file=RESOURCE_DIR / "test_snowflake_case_statement.json",
+    )
+
+
 @pytest.mark.skip(reason="We don't handle the unnest lineage correctly")
 def test_bigquery_unnest_columns():
     assert_sql_result(