feat(adf): save source graphs with dataset lineage at ADF discover

[matthewmoorcroft]

Intent

Put ADF discovery on the source-neutral discovery graph contract from #84. The ADF loader projects each pipeline into a SourceGraph, saves source_graphs.json first (versioned, hashed, motifs inside), and builds inventory.json through the shared emitter, which records the graphs' hash. ADF data lineage is derived conservatively: only provable physical values become dataset identities.

What changed

• Review fixes, round three (20 Oct): only a real boolean waitOnCompletion is recorded (null or a string leaves the wait unknown); an Expression override with an empty value no longer wipes the dataset identity; two Switch cases with the same value both keep their activities (the parser no longer replaces the first, the graph gives each case its own branch, and the bundle gives each its own condition task key), and the repeat is reported on the Switch node (duplicate_switch_cases) and logged; a bracket-quoted table such as [Sales (2024)] keeps its identity, because only ARM's own functions make an ARM expression; Blob and ADLS Gen2 datasets share one abfss:// identity on purpose, since both reach the same storage account; @dataset()?.x resolves like @dataset().x; the dataset lineage module is split into identity, store and signature parts (a pure move). refactor: add source-neutral discovery graph contract and shared lineage core #84's round three is merged in with a merge commit.
• Review fixes, round two (16-18 Oct): the signature rule below; non-map parameters are ignored instead of crashing discover; a null or non-name ExecutePipeline reference stays unresolved; a control edge read back without resolved: true is unresolved, and a wait flag that is not true or false reads back as unset; an ARM-expression linked service name never proves a store; Fabric Lakehouse and ADLS Gen1 locations are qualified by their plain linked service name, like SFTP; repeated motif detections are counted once, on the inventory's rule (same motif id and same activities, in any order), so the profile, the saved graph and the inventory agree; the --debug pipeline dump includes the pipeline's lineage block.
• Merged refactor: add source-neutral discovery graph contract and shared lineage core #84's final head with a merge commit (no rebase). This PR reads lineage through the shared ir_serde.lineage_from_dict, which now also carries refactor: add source-neutral discovery graph contract and shared lineage core #84's wait-flag rule.
• Review fixes (14-15 Oct): a linked service name must be a plain name before it qualifies an identity; the S3 / GCS bucketName now names the file store, so two buckets with the same folder layout no longer link; an ExecutePipeline edge is resolved only when its child pipeline is in the export, including under --pipeline, matched case-insensitively as ADF does and recorded under the exported spelling; a test checks that running discover twice gives byte-identical inventory.json with no generated_at. In the same style: an unevaluated ARM or {{ placeholder folderPath no longer anchors a path signature, an unevaluated ARM fileName is kept as written so different ones never link, SFTP, file server, FTP, Azure Files, HDFS and HTTP locations are qualified by their plain linked service name, a literal container or bucket whose store cannot be proven gets no signature, and the IR-side build_control_edges takes the same known-pipelines check.
• sources/adf/loader.py:
  • Attaches detected motifs to each graph's lineage.
  • Writes source_graphs.json, then inventory.json through build_source_inventory with source_graphs_sha256.
  • inventory.json deliberately has no top-level generated_at, so the same export, given the same --source-dir, always gives the same bytes. This is documented in skills/flowx-discover/sources/adf.md.
• sources/adf/discovery_mapping.py: maps the ADF AST to SourceGraph one to one.
  • Every declared branch is kept, including Switch cases with no activities.
  • A case whose value is literally default is keyed case:default (kept unique), so it never collides with the Switch's own default branch.
• sources/adf/dataset_lineage.py (with dataset_identity.py, dataset_store.py and path_signature.py, split in round three): dataset identities.
  • Tables are qualified by the linked service's literal server (and database). Paths run down to the literal file name.
  • @dataset().x location parts resolve against the call-site bindings.
  • Anything parameterised or unresolved gives no identity: ADF expressions, DAB placeholders, unevaluated ARM expressions, and parameterised linked services, databases or storage accounts. Bracket-quoted SQL names such as [dbo].[Orders] stay valid.
  • Path signature rule (round two): an unresolved path gets a weak signature only when the activity passes a folderPath with literal text at the call site; the dataset's own location is not enough on its own. The folder and file name are then read from the dataset's location with the call-site values filled in (the call-site value is used when the location sets none). Only provably literal text forms the path (quoted text, numbers, text outside @{...}); every other value is one <P> slot where it sits, and a function over a bound dataset() value is a slot keyed by that value (<P:tolower('orders')>). So nothing vanishes, no argument text leaks, and no link is made from a non-literal value. An interpolated call-site folder links like its @concat spelling. tests/unit/test_adf_path_signature_corpus.py guards the rule.
  • An unresolved path's weak signature names the dataset's store when it is known, so datasets on different stores never join on folder shape alone. The store is <container>@<account>, the linked service name in place of an account hidden in a secret (only for a linked service with no parameters), or the table store. An unknown store is never guessed; that signature stays unqualified.
  • Run-time override rule: a read or write whose activity overrides the dataset's physical source or target at run time gets no identity and no signature, so it never forms a data edge. That covers storeSettings wildcard, file-list or prefix paths; a source query, *Query or *StoredProcedureName; and a sink *StoredProcedureName.
• Groundwork for cross-pipeline deterministic conversion, not called yet (models/ir.py, lineage.py, ir_serde.py, bundler/dab_writer.py):
  • The IR lineage fields (Activity.data_reads, data_writes and motif_id, and Pipeline.lineage), the build_* and with_lineage functions, their IR serialisation, and package's read-back of them in dab_writer.py.
  • Nothing in Phase 1 populates or calls them yet, so output is unchanged.
  • Within that layer, a run-now of an existing job by ID is not a cross-workflow edge, and motif annotations carry source_type_hint.
  • The one shared lineage parser, ir_serde.lineage_from_dict, is used in production by discovery_serde.
• Discover console (rides along): discover prints a loud warning when it parses 0 pipelines (naming --adf-source-path), and labels the activity count "raw, incl. nested". This affects console output only; nothing on disk changes.
• Docs: AGENTS.md, the README output tree and the flowx-discover skill describe source_graphs.json and the inventory hash.

How it was tested

• Unit tests cover the loader, the mapping, dataset lineage and the lineage layer. Dataset lineage tests cover identity tiers, servers, file names, ARM expressions, bracket-quoted names, parameterised files and every override kind. The full unit and integration suites, ruff and mypy pass in CI.
• ADF and Airflow discover, convert and package on the fixture corpora: bundles and translation reports are byte-identical to the merge-base with main (b8d0766). A scratch merge of refactor: add source-neutral discovery graph contract and shared lineage core #84 and feat(adf): save source graphs with dataset lineage at ADF discover #94 onto current main (858550e) is byte-identical to main too, with the full unit suite passing. Only the ADF inventory changes, gaining the shared emitter's fields, source_graphs_sha256 and each activity's task_key.
• Codex cross-review, read-only, twice. Round 1 found 4 issues and round 2 found 3. All are fixed here, along with the review-found follow-ons: the Switch default label clash and the override-rule siblings. An independent commit-by-commit review then found one more, the cross-store signature edge, which is also fixed here. Every fix has a regression test that fails on the previous code.

Merge order

Merge after #84. This PR targets discovery/shared-ast-standard; retarget it to main once #84 has merged. The retarget needs a one-line import resolution in bundler/dab_writer.py: keep both from flowx.ir_serde import data_asset_from_dict, lineage_from_dict and main's from flowx.models.dab import DabNotebook, SecretInstruction, SetupTask. #95, #96 and #97 follow in order.

This pull request and its description were written by Isaac.

[matthewmoorcroft]

The changes from the second review are applied at commit 846b51a. Rather than one more patch per case, the path signature now follows one rule.

What was wrong with the old approach. The signature kept quoted text plus a fixed list of known run-time values (pipeline(), item(), variables(), dataset(), activity()) and silently dropped everything else. A dropped value doesn't make the signature weaker; it makes a run-specific folder look literal. So @concat('landing/', rand(1,100)), pipeline()?.TriggerTime or item()['zone'] read as plain landing or raw/zone and linked to folders they aren't. Each review round found the next missing value, because adding one value to the list fixes only that value.

The rule now. Only provably literal text forms the path: quoted text in an expression, or text outside the @{...} blocks of an interpolated value. Any other value is one <P> slot where it sits, whatever function or reference it is, so nothing vanishes and no argument text leaks into the path. A function over a dataset() value the call site bound to a literal, such as toLower(dataset().folderPath), becomes a slot keyed by that value (<P:toLower('orders')>). It is still never path text, but orders and customers stay apart. The folder and file name are read from the dataset's own location, with the passed-in values spliced in raw and the slashes normalised once.

A1. An interpolated call-site folder such as landing/@{item().run} now gets a signature and links exactly like its @concat('landing/', ...) spelling. Those are correct links, so we accept them. The earlier note that no reference gains a signature is corrected accordingly: these interpolated folders gain one, and nothing else does.

Accuracy, and the guard. On 17 real examples (the false links found in review, the links that must keep working, including the unknown-store watermark tier, and your two cases), the result is 16 right with no invented link. 9ef2ab0 got 8 right with 5 invented. The one miss is a folder made only of a formatted date, which no longer links on its format string. tests/unit/test_adf_path_signature_corpus.py guards the rule: it builds 216 expressions from literal and run-time pieces in four spellings, plus bound-value cases. It asserts that a run-time value never reads as a literal folder and that no argument text leaks into the path. A new construct gets added to the corpus, not patched per value.

Your eight points:

1. Folder signature ignored the dataset's folder: the shape is now read from the dataset's own location.folderPath with the passed-in values filled in, so bronze/orders and silver/orders no longer link. The call-site value is used only when the location sets no folder.
2. Slot positions lost: each run-time value stays in place as <P>, so landing/<P> and <P>/landing differ. The |slots=N count is gone.
3. parameters that are not a map, on the dataset or at the call site, are ignored instead of crashing discover.
4. "pipeline": null, or any reference that is not an object or a name, records an empty, unresolved child. Before, it could resolve to an exported pipeline called None.
5. _resolve_dataset_store now returns the store and whether the location names one, and _location_names_store is gone.
6. A missing file name is written FN[].
7. The doc says byte-identical inventory.json holds for the same --source-dir.
8. "or the dataset name" is dropped from the test docstring.

Each fix has a regression test that fails before it.

I also went through the whole diff against #84 with your review points in mind, and fixed what that turned up: a control edge read back without resolved: true is unresolved, and a wait flag that is not true or false reads back as unset; an ARM-expression linked service name never proves a store; Fabric Lakehouse and ADLS Gen1 locations are qualified by their linked service, like SFTP, so two lakehouses with the same folder layout no longer link; repeated motif detections are counted once, on the inventory's rule, so the profile and the inventory agree; and the --debug pipeline dump includes the lineage block. #84's final head is merged in with a merge commit, not a rebase.

[ghanse]

The ADF-to-source-graph work is solid and well tested overall. Fix the wait-flag handling and the empty-Expression case before merging.

• tests/unit/test_adf_discovery_mapping.py: Add tests for a null or non-boolean waitOnCompletion, and for a ForEach/Until with a Switch inside, so the bugs above can't come back unnoticed.

[ghanse]

Blocking: bool(...) turns a null wait flag into False and the string "false" into True. Only keep real booleans and leave the rest unset, which matches how the shared reader already works.

wait = type_properties.get("waitOnCompletion", True)
if isinstance(wait, bool):
    properties[INVOKES_WAIT_PROPERTY] = wait

[ghanse]

Blocking: An Expression object with an empty value ({"type": "Expression", "value": ""}) counts as an override and wipes out the dataset's identity, while a bare "" correctly doesn't. Unwrap the expression before checking whether it's empty.

[ghanse]

Requiring a literal call-site folderPath looks intentional (there's a test for it), but this docstring and the PR description make it sound like the dataset's own location is enough. Make the wording match the rule.

[ghanse]

Two Switch cases with the same value write to the same branch key, so the first case's activities get dropped without a warning. Rare, but worth a finding or a check that fails loudly.

[ghanse]

Every file identity is written as abfss://…dfs.core.windows.net/…, including Azure Blob. So a Blob dataset and an ADLS Gen2 dataset with the same path would get linked. Use a prefix that matches the connector.

[ghanse]

This regex treats [Sales (2024)] as an ARM expression, so that table loses its identity. Only [Order Details] and [dbo].[Orders] are tested.

[ghanse]

Suggestion: At about 1,000 lines, much of it docstrings, this file is hard to follow. Consider splitting identity, store and signature into separate parts, merging _DATASET_PARAM_RE and _DATASET_PARAMETER_RE, cutting the issue-history docs down to short summaries, and shortening any verbose docstrings.

[matthewmoorcroft]

Thanks, Greg. Round three is applied up to 337d2d6, one line per comment:

1. Wait flag (blocking): only a real boolean waitOnCompletion is recorded, using your snippet; null, "false", numbers and Expression objects leave the wait unknown, as the shared reader expects. A missing flag still takes the translator's default, True. Tests cover each value, plus a Switch inside a ForEach and an Until. (cb4ddbc)
2. Empty Expression override (blocking): the override check unwraps an Expression to its value first, so {"type": "Expression", "value": ""} overrides nothing, like a bare "", for a source query, a sink stored procedure and a storeSettings path. A non-empty Expression, or any other query object such as a Key Vault reference, still overrides. (a0d0931, 8f4f6bd)
3. Call-site folderPath wording: the _path_signature and module docstrings now say a signature needs both a call-site folderPath with literal text and a literal segment in the folder it resolves to. The PR description says the same. (2f03836)
4. Duplicate Switch case values: nothing is dropped now. Conversion already converted every case, so the parser now adds a repeated value's activities to the same list instead of replacing them, and the graph keeps each declared case as its own branch (full, then case:full). So the AST, the profile count, the graph and conversion all keep every activity. The Switch node lists the repeated value and its branch labels under duplicate_switch_cases, and discover logs a warning naming the pipeline. The bundle also gives a repeated value its own condition task key, which used to collide. Tested, including an empty last duplicate. (cb4ddbc, 38a0469, 337d2d6)
5. Blob vs ADLS Gen2 prefix: I kept one abfss:// identity for both, on purpose. A storage account name is unique across Azure, and the Blob and ADLS Gen2 connectors are two endpoints onto the same account, so the same container and path are the same files through either one. A Copy writing through the Blob connector and a notebook reading the same container through ADLS should link, and a wasbs:// prefix for Blob would break that link. The docstring says this now, and a test pins it: the same account and container link through either connector; a different account or container does not. If you'd still rather have wasbs:// for Blob, I'm happy to switch. (3ff83e4)
6. [Sales (2024)]: a bracketed value is read as ARM when it calls one of ARM's own template functions (any case or spacing), a list* function or a namespaced user function, or when its arguments hold a quoted string or another call, which only ARM's do. So [Sales (2024)] keeps its identity, tested alongside [Order Details] and [dbo].[Orders], while [ns.fn(...)] (missed before) and an ARM function missing from the list still read as ARM. (936db3b, 8f4f6bd)
7. File size: _DATASET_PARAM_RE and _DATASET_PARAMETER_RE are merged, so @dataset()?.x now resolves too (tested). The issue-history docs are cut to short summaries, and the longest docstrings are shortened (a519535). The module is then split into dataset_identity.py, dataset_store.py, path_signature.py and dataset_lineage.py, which puts them together. That commit only moves code: every definition's body is unchanged, checked by comparing each one's AST before and after (0145059).

The behaviour fixes (1, 2, 4, 6 and the ?. lookup) each have a regression test that fails before them; the tests for 5 pin the current behaviour. #84's round three is merged in with a merge commit (0813081), and the full unit suite, ruff and mypy pass.