feat(sql): extract table-to-table lineage from INSERT/CTAS/MERGE/UPDATE (#1572)

graphify/detect.py

@@ -27,7 +27,7 @@ class FileType(str, Enum):
 
 _MANIFEST_PATH = str(out_path("manifest.json"))
 
-CODE_EXTENSIONS = {'.py', '.ts', '.tsx', '.mts', '.cts', '.js', '.jsx', '.mjs', '.ejs', '.ets', '.go', '.rs', '.java', '.groovy', '.gradle', '.cpp', '.cc', '.cxx', '.c', '.h', '.hpp', '.cu', '.cuh', '.metal', '.rb', '.swift', '.kt', '.kts', '.cs', '.scala', '.php', '.lua', '.luau', '.toc', '.zig', '.ps1', '.psm1', '.psd1', '.ex', '.exs', '.m', '.mm', '.jl', '.vue', '.svelte', '.astro', '.dart', '.v', '.sv', '.svh', '.sql', '.r', '.f', '.F', '.f90', '.F90', '.f95', '.F95', '.f03', '.F03', '.f08', '.F08', '.pas', '.pp', '.dpr', '.dpk', '.lpr', '.inc', '.dfm', '.lfm', '.lpk', '.sh', '.bash', '.json', '.tf', '.tfvars', '.hcl', '.dm', '.dme', '.dmi', '.dmm', '.dmf', '.sln', '.slnx', '.csproj', '.fsproj', '.vbproj', '.xaml', '.razor', '.cshtml', '.cls', '.trigger'}
+CODE_EXTENSIONS = {'.py', '.ts', '.tsx', '.mts', '.cts', '.js', '.jsx', '.mjs', '.ejs', '.ets', '.go', '.rs', '.java', '.groovy', '.gradle', '.cpp', '.cc', '.cxx', '.c', '.h', '.hpp', '.cu', '.cuh', '.metal', '.rb', '.swift', '.kt', '.kts', '.cs', '.scala', '.php', '.lua', '.luau', '.toc', '.zig', '.ps1', '.psm1', '.psd1', '.ex', '.exs', '.m', '.mm', '.jl', '.vue', '.svelte', '.astro', '.dart', '.v', '.sv', '.svh', '.sql', '.ddl', '.r', '.f', '.F', '.f90', '.F90', '.f95', '.F95', '.f03', '.F03', '.f08', '.F08', '.pas', '.pp', '.dpr', '.dpk', '.lpr', '.inc', '.dfm', '.lfm', '.lpk', '.sh', '.bash', '.json', '.tf', '.tfvars', '.hcl', '.dm', '.dme', '.dmi', '.dmm', '.dmf', '.sln', '.slnx', '.csproj', '.fsproj', '.vbproj', '.xaml', '.razor', '.cshtml', '.cls', '.trigger'}
 DOC_EXTENSIONS = {'.md', '.mdx', '.qmd', '.txt', '.rst', '.html', '.yaml', '.yml'}
 PAPER_EXTENSIONS = {'.pdf'}
 IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'}

graphify/extract.py

@@ -1741,7 +1741,7 @@ def _canonicalize_csharp_namespace_nodes(all_nodes: list[dict], all_edges: list[
 # env var are distinct) and folding manufactures false edges / super-hubs (#1581).
 _CASE_INSENSITIVE_EXTS = frozenset({
     ".php", ".phtml", ".php3", ".php4", ".php5", ".php7", ".phps",  # PHP fns/classes
-    ".sql",                                                          # SQL identifiers
+    ".sql", ".ddl",                                                  # SQL identifiers
     ".nim", ".nims", ".nimble",                                      # Nim (style-insensitive)
 })
 
@@ -3798,6 +3798,7 @@ def add_existing_edge(edge: dict) -> None:
     ".sv": extract_verilog,
     ".svh": extract_verilog,
     ".sql": extract_sql,
+    ".ddl": extract_sql,
     ".md": extract_markdown,
     ".mdx": extract_markdown,
     ".qmd": extract_markdown,
@@ -3840,6 +3841,7 @@ def add_existing_edge(edge: dict) -> None:
 # extract() to tell the user which extra restores the language.
 _EXTRA_FOR_EXTENSION = {
     ".sql": "sql",
+    ".ddl": "sql",
     ".tf": "terraform",
     ".tfvars": "terraform",
     ".hcl": "terraform",

graphify/extractors/sql.py

@@ -61,6 +61,19 @@ def _add_edge(src: str, tgt: str, relation: str, line: int) -> None:
                        "confidence": "EXTRACTED", "source_file": str_path,
                        "source_location": f"L{line}", "weight": 1.0})
 
+    def _ensure_table(name: str, line: int) -> str:
+        """Return the nid for a table name, creating its node on first sight.
+
+        DML write targets (INSERT/UPDATE/MERGE) are often defined in another
+        file (or not in the corpus at all), so the node may not exist yet —
+        mirror the alter_table behavior and materialize it."""
+        nid = table_nids.get(name.lower())
+        if not nid:
+            nid = _make_id(stem, name)
+            _add_node(nid, name, line)
+            table_nids[name.lower()] = nid
+        return nid
+
     def walk(node) -> None:
         t = node.type
         line = node.start_point[0] + 1
@@ -119,6 +132,11 @@ def walk(node) -> None:
                                     ref_nid = table_nids.get(ref_name.lower()) or _make_id(stem, ref_name)
                                     _add_edge(nid, ref_nid, "references", line)
                                     seen_refs.add(ref_name.lower())
+                # CTAS: CREATE TABLE <tgt> AS SELECT ... FROM/JOIN <src> — the
+                # SELECT lives in a create_query child; walk it for lineage
+                # edges just like create_view does (#1572).
+                if any(c.type == "create_query" for c in node.children):
+                    _walk_from_refs(node, nid, line)
 
         elif t == "create_view":
             name = _obj_name(node)
@@ -170,6 +188,35 @@ def walk(node) -> None:
                                     ref_nid = _make_id(stem, ref_name)
                                 _add_edge(src_nid, ref_nid, "references", line)
 
+        elif t == "insert":
+            # INSERT INTO <tgt> ... SELECT ... FROM/JOIN <src> — ETL lineage
+            # (#1572). The target is the first object_reference child; VALUES
+            # inserts have no `from` node, so _walk_from_refs emits nothing.
+            tgt_name = None
+            for c in node.children:
+                if c.type == "object_reference":
+                    tgt_name = _read(c)
+                    break
+            if tgt_name:
+                tgt_nid = _ensure_table(tgt_name, line)
+                _walk_from_refs(node, tgt_nid, line)
+
+        elif t == "update":
+            # UPDATE <tgt> SET ... FROM <src> — the target is the relation
+            # directly under update; sources live in the `from` child, so
+            # _walk_from_refs never sees the target itself (#1572).
+            tgt_name = None
+            for c in node.children:
+                if c.type == "relation":
+                    for cc in c.children:
+                        if cc.type == "object_reference":
+                            tgt_name = _read(cc)
+                            break
+                    break
+            if tgt_name:
+                tgt_nid = _ensure_table(tgt_name, line)
+                _walk_from_refs(node, tgt_nid, line)
+
         elif t == "create_trigger":
             trig_name: str | None = None
             tbl_name: str | None = None
@@ -240,13 +287,50 @@ def _walk_from_refs(node, caller_nid: str, line: int) -> None:
                         if cc.type == "object_reference":
                             tbl = _read(cc)
                             tbl_nid = _make_id(stem, tbl)
-                            _add_edge(caller_nid, tbl_nid, "reads_from",
-                                      c.start_point[0] + 1)
+                            # Skip self-loops: INSERT INTO t SELECT ... FROM t
+                            # (dedup/backfill patterns) reads its own target.
+                            if tbl_nid != caller_nid:
+                                _add_edge(caller_nid, tbl_nid, "reads_from",
+                                          c.start_point[0] + 1)
         for child in node.children:
             _walk_from_refs(child, caller_nid, line)
 
+    def _handle_merge(stmt_node) -> None:
+        """MERGE INTO <tgt> USING <src> — lineage edge tgt reads_from src (#1572).
+
+        tree-sitter-sql has no dedicated merge node: the keywords sit directly
+        under `statement`, so this runs on the statement itself. Only direct
+        children are scanned for the target/source object_references — nested
+        ones (ON-clause fields, WHEN-clause assignments) never surface here.
+        """
+        line = stmt_node.start_point[0] + 1
+        tgt_name: str | None = None
+        src_names: list[str] = []
+        after_using = False
+        for c in stmt_node.children:
+            if c.type == "keyword_using":
+                after_using = True
+            elif c.type == "object_reference":
+                if after_using:
+                    src_names.append(_read(c))
+                    after_using = False
+                elif tgt_name is None:
+                    tgt_name = _read(c)
+        if not tgt_name:
+            return
+        tgt_nid = _ensure_table(tgt_name, line)
+        for src_name in src_names:
+            src_nid = _ensure_table(src_name, line)
+            if src_nid != tgt_nid:
+                _add_edge(tgt_nid, src_nid, "reads_from", line)
+        # USING (SELECT ... FROM <src>) subqueries carry their sources in a
+        # nested `from` node instead of a direct object_reference.
+        _walk_from_refs(stmt_node, tgt_nid, line)
+
     for stmt in root.children:
         if stmt.type == "statement":
+            if any(c.type == "keyword_merge" for c in stmt.children):
+                _handle_merge(stmt)
             for child in stmt.children:
                 walk(child)
         elif stmt.type in ("fb_proc_or_trigger", "set_term", "declare_external_function"):

tests/fixtures/sample_dml_lineage.sql

@@ -0,0 +1,32 @@
+-- ETL-style DML lineage fixture (#1572)
+CREATE TABLE raw_events (id INT, user_id INT, amount INT);
+CREATE TABLE dim_users (id INT, name TEXT);
+
+-- INSERT ... SELECT with a JOIN: fct_daily reads from raw_events and dim_users
+INSERT INTO fct_daily (user_id, name, total)
+SELECT e.user_id, u.name, SUM(e.amount)
+FROM raw_events e
+JOIN dim_users u ON u.id = e.user_id
+GROUP BY e.user_id, u.name;
+
+-- CTAS: agg_monthly reads from fct_daily
+CREATE TABLE agg_monthly AS
+SELECT user_id, SUM(total) AS total
+FROM fct_daily
+GROUP BY user_id;
+
+-- MERGE: dim_users reads from staging_users
+MERGE INTO dim_users d
+USING staging_users s ON d.id = s.id
+WHEN MATCHED THEN UPDATE SET name = s.name;
+
+-- UPDATE ... FROM: fct_daily reads from dim_users
+UPDATE fct_daily SET name = u.name
+FROM dim_users u
+WHERE fct_daily.user_id = u.id;
+
+-- plain VALUES insert: audit_log node exists, but no reads_from edge
+INSERT INTO audit_log (msg) VALUES ('refreshed');
+
+-- self-referential backfill: no self-loop edge
+INSERT INTO raw_events SELECT * FROM raw_events WHERE id > 100;

tests/test_multilang.py

@@ -513,3 +513,70 @@ def test_sql_schema_qualified_alter_fk():
     for e in fk_edges:
         assert e["source"] in node_ids, f"dangling source: {e['source']}"
         assert e["target"] in node_ids, f"dangling target: {e['target']}"
+
+
+def _dml_lineage_pairs():
+    """(source_label, target_label) pairs of reads_from edges in the DML fixture."""
+    r = _extract_sql_or_skip("sample_dml_lineage.sql")
+    by_id = {n["id"]: n["label"] for n in r["nodes"]}
+    return r, {
+        (by_id[e["source"]], by_id[e["target"]])
+        for e in r["edges"] if e["relation"] == "reads_from"
+    }
+
+
+def test_sql_insert_select_lineage():
+    """#1572: INSERT INTO tgt SELECT ... FROM/JOIN src emits tgt reads_from src."""
+    _, pairs = _dml_lineage_pairs()
+    assert ("fct_daily", "raw_events") in pairs
+    assert ("fct_daily", "dim_users") in pairs
+
+
+def test_sql_ctas_lineage():
+    """#1572: CREATE TABLE tgt AS SELECT ... FROM src emits tgt reads_from src."""
+    _, pairs = _dml_lineage_pairs()
+    assert ("agg_monthly", "fct_daily") in pairs
+
+
+def test_sql_merge_lineage():
+    """#1572: MERGE INTO tgt USING src emits tgt reads_from src."""
+    _, pairs = _dml_lineage_pairs()
+    assert ("dim_users", "staging_users") in pairs
+
+
+def test_sql_update_from_lineage():
+    """#1572: UPDATE tgt ... FROM src emits tgt reads_from src."""
+    _, pairs = _dml_lineage_pairs()
+    assert ("fct_daily", "dim_users") in pairs
+
+
+def test_sql_insert_values_no_false_lineage():
+    """A plain VALUES insert materializes the target node but no reads_from edge."""
+    r, pairs = _dml_lineage_pairs()
+    labels = {n["label"] for n in r["nodes"]}
+    assert "audit_log" in labels
+    assert not any(s == "audit_log" for s, _ in pairs)
+
+
+def test_sql_self_insert_no_self_loop():
+    """INSERT INTO t SELECT ... FROM t (backfill) must not emit a self-loop."""
+    _, pairs = _dml_lineage_pairs()
+    assert ("raw_events", "raw_events") not in pairs
+
+
+def test_sql_dml_lineage_no_dangling_edges():
+    """Every lineage edge endpoint is a real node (targets are materialized)."""
+    r, _ = _dml_lineage_pairs()
+    node_ids = {n["id"] for n in r["nodes"]}
+    for e in r["edges"]:
+        assert e["source"] in node_ids, f"dangling source: {e['source']}"
+        assert e["target"] in node_ids, f"dangling target: {e['target']}"
+
+
+def test_ddl_extension_routes_to_sql_extractor():
+    """#1572 nice-to-have: .ddl files are classified as code and parsed as SQL."""
+    pytest.importorskip("tree_sitter_sql")
+    from graphify.detect import CODE_EXTENSIONS
+    from graphify.extract import _DISPATCH, extract_sql as _es
+    assert ".ddl" in CODE_EXTENSIONS
+    assert _DISPATCH[".ddl"] is _es