fix: tighten MV TO regex; add unit tests for ClickHouse lineage rewrite

Author: Joseph Thomas Stalin

ingestion/src/metadata/ingestion/lineage/parser.py

@@ -513,17 +513,23 @@ def clean_raw_query(cls, raw_query: str) -> Optional[str]:
 
         clean_query = clean_query.replace("\\n", "\n")
 
-        # Convert Clickhouse MATERIALIZED VIEW ... TO ... AS SELECT ...
-        # into CREATE TABLE ... AS SELECT ... so sqllineage parses the correct target
+        # Convert ClickHouse MATERIALIZED VIEW ... TO <target> AS SELECT ...
+        # into CREATE TABLE <target> AS SELECT ... so sqllineage identifies the
+        # correct downstream target instead of the view name.
+        # Handles optional ON CLUSTER and ENGINE/POPULATE/SETTINGS clauses.
         if insensitive_match(clean_query, r"^\s*CREATE\s+MATERIALIZED\s+VIEW\s+"):
             mv_to_match = re.search(
-                r"^\s*CREATE\s+MATERIALIZED\s+VIEW\s+(?:IF\s+NOT\s+EXISTS\s+)?(.*?)\s+TO\s+(.*?)\s+AS\s+(SELECT.*)",
+                r"^\s*CREATE\s+MATERIALIZED\s+VIEW\s+"
+                r"(?:IF\s+NOT\s+EXISTS\s+)?\S+\s+"  # skip MV name
+                r"(?:ON\s+CLUSTER\s+\S+\s+)?"       # optional ON CLUSTER
+                r"TO\s+([\w`\.\[\]\"]+)"              # target table (stop at whitespace)
+                r".*?AS\s+(SELECT.*)",               # skip ENGINE/SETTINGS, capture SELECT
                 clean_query,
-                re.IGNORECASE | re.DOTALL
+                re.IGNORECASE | re.DOTALL,
             )
             if mv_to_match:
-                target_table = mv_to_match.group(2)
-                select_query = mv_to_match.group(3)
+                target_table = mv_to_match.group(1).strip()
+                select_query = mv_to_match.group(2)
                 clean_query = f"CREATE TABLE {target_table} AS {select_query}"
 
         if insensitive_match(

ingestion/tests/unit/lineage/test_sql_lineage.py

@@ -459,3 +459,105 @@ def test_stage_lineage_with_file_format_options(self):
                 len(parser.target_tables) > 0,
                 f"Expected target tables for query: {query}",
             )
+
+
+class ClickhouseMaterializedViewLineageTest(TestCase):
+    """
+    Tests for ClickHouse MATERIALIZED VIEW TO <target> lineage rewriting
+    in LineageParser.clean_raw_query.
+    """
+
+    def test_simple_materialized_view_to(self):
+        """
+        Basic MATERIALIZED VIEW ... TO target AS SELECT is rewritten so
+        the target table — not the MV — appears as the write table.
+        """
+        query = (
+            "CREATE MATERIALIZED VIEW default.my_mv "
+            "TO default.my_target "
+            "AS SELECT * FROM default.my_source"
+        )
+        cleaned = LineageParser.clean_raw_query(query)
+        self.assertIsNotNone(cleaned)
+        self.assertIn("CREATE TABLE default.my_target", cleaned)
+        self.assertNotIn("my_mv", cleaned)
+
+    def test_materialized_view_with_engine_clause(self):
+        """
+        ENGINE clause between TO and AS must not be captured as the target name.
+        """
+        query = (
+            "CREATE MATERIALIZED VIEW db.mv_name "
+            "TO db.target_table "
+            "ENGINE = MergeTree() "
+            "AS SELECT id, name FROM db.source_table"
+        )
+        cleaned = LineageParser.clean_raw_query(query)
+        self.assertIsNotNone(cleaned)
+        self.assertIn("CREATE TABLE db.target_table", cleaned)
+        self.assertNotIn("ENGINE", cleaned.split("AS SELECT")[0])
+
+    def test_materialized_view_if_not_exists(self):
+        """
+        IF NOT EXISTS variant should still resolve the correct target table.
+        """
+        query = (
+            "CREATE MATERIALIZED VIEW IF NOT EXISTS analytics.mv_clicks "
+            "TO analytics.clicks_agg "
+            "AS SELECT page, count(*) as cnt FROM analytics.raw_clicks GROUP BY page"
+        )
+        cleaned = LineageParser.clean_raw_query(query)
+        self.assertIsNotNone(cleaned)
+        self.assertIn("CREATE TABLE analytics.clicks_agg", cleaned)
+
+    def test_materialized_view_on_cluster(self):
+        """
+        ON CLUSTER clause should be skipped and not interfere with TO parsing.
+        """
+        query = (
+            "CREATE MATERIALIZED VIEW cluster_db.mv_events "
+            "ON CLUSTER my_cluster "
+            "TO cluster_db.events_summary "
+            "AS SELECT event_type, count() FROM cluster_db.events GROUP BY event_type"
+        )
+        cleaned = LineageParser.clean_raw_query(query)
+        self.assertIsNotNone(cleaned)
+        self.assertIn("CREATE TABLE cluster_db.events_summary", cleaned)
+
+    def test_materialized_view_without_to_clause_unchanged(self):
+        """
+        A MATERIALIZED VIEW without a TO clause should not be transformed —
+        the query is returned as-is (or filtered by the caller).
+        """
+        query = (
+            "CREATE MATERIALIZED VIEW my_mv AS SELECT * FROM source_table"
+        )
+        cleaned = LineageParser.clean_raw_query(query)
+        # Without TO, no transformation; the query should pass through unchanged
+        self.assertIsNotNone(cleaned)
+        self.assertNotIn("CREATE TABLE", cleaned)
+
+    def test_clean_query_produces_correct_lineage(self):
+        """
+        End-to-end: after clean_raw_query the LineageParser should identify
+        the TO target as the write table, not the MV name.
+        """
+        query = (
+            "CREATE MATERIALIZED VIEW default.mv_orders "
+            "TO default.orders_agg "
+            "AS SELECT customer_id, sum(total) FROM default.orders GROUP BY customer_id"
+        )
+        cleaned = LineageParser.clean_raw_query(query)
+        self.assertIsNotNone(cleaned)
+        parser = LineageParser(cleaned)
+        target_names = [str(t) for t in parser.target_tables]
+        source_names = [str(t) for t in parser.source_tables]
+        self.assertTrue(
+            any("orders_agg" in t for t in target_names),
+            f"Expected orders_agg in targets, got: {target_names}",
+        )
+        self.assertTrue(
+            any("orders" in s for s in source_names),
+            f"Expected orders in sources, got: {source_names}",
+        )
+