Skip to content

Reference Airflow and Dagster pipelines, tested against a server (INT-10, #1715) - #1926

Merged
Kalyanpulikanti-01 merged 5 commits into
samyama-ai:devfrom
Raghuvarma905:int-10/1715-orchestration-dags
Oct 22, 2026
Merged

Kalyanpulikanti-01 merged 5 commits into
samyama-ai:devfrom
Raghuvarma905:int-10/1715-orchestration-dags

Conversation

@Raghuvarma905

Copy link
Copy Markdown
Collaborator

Part of #1715 (INT-10). It delivers what this repository can: one reference pipeline per tool (Airflow and Dagster), a how-to page for each, and tests that run both against a real server in CI. The scorecard row flips only when the CH-GOVERNANCE check in samyama-graph-competitor-benchmarks is changed to look for these, and this PR does not touch that repository. So it says "Part of", not "Closes".

It also fixes an engine bug found while doing this, in its own commit. No issue exists for it. Without the fix the reference pipeline loads the wrong data.

1. Engine fix: relationship MERGE ignored values taken from the row (commit 1)

UNWIND $rows AS r
MATCH (s:Account {id: r.src}), (d:Account {id: r.dst})
MERGE (s)-[t:TRANSFER {id: r.id}]->(d)

This is the standard bulk edge-upsert pattern, and it lost data with no error. A MERGE between two already-bound nodes goes to MatchMergeEdgeOperator, which was given only the literal part of the relationship's property map. {id: r.id} lives in property_exprs, and that part was dropped. So:

  • the operator matched any :TRANSFER between the two nodes, ignoring id;
  • when it did create an edge, it created it without id;
  • two rows for the same pair produced one edge, and the statement still reported both as written.

On the sample data, 60 transfers became 55 edges. #642 fixed this for nodes and for whole-pattern MERGE; this is the bound-nodes relationship case it missed.

How it's fixed: each row's expressions are now resolved with the same resolver node MERGE already uses (MergeOperator::resolved_props). The one resolved map is used both to look for a match and to create the edge. EXPLAIN now shows the expressions too. The expressions are attached through a new builder (with_property_exprs), so the operator's constructor and its existing callers are unchanged.

Tests: six new cases in tests/merge_row_properties.rs:

  • distinct keys between one pair
  • an existing edge with a different key is left alone
  • a key taken from another bound node
  • literal and row properties on the same edge
  • ON CREATE / ON MATCH
  • an undirected pattern

All six fail without the fix and pass with it. The five existing cases pass either way.

2. INT-10: reference pipelines (commit 2)

examples/orchestration/:

File
samyama_pipeline.py The load steps both pipelines run: schema → accounts → transfers → validate. Standard library only.
samyama_airflow_dag.py Airflow 3 DAG. Server and token come from an Airflow Connection; files and batch size are DAG params.
samyama_dagster.py Dagster assets with blocking asset checks. Server and token come from SAMYAMA_URL / SAMYAMA_TOKEN via EnvVar.
sample/ Synthetic data: 24 accounts, 60 transfers. Not in a data/ folder, because .gitignore ignores every folder with that name.
tests/ 52 tests. Each starts its own release server on spare ports with a fresh data directory.

Docs:

  • docs/ORCHESTRATION.md covers the design, the guarantees and the limits.
  • ORCHESTRATION-AIRFLOW.md and ORCHESTRATION-DAGSTER.md are the per-tool how-tos.
  • docs/README.md and examples/README.md index all of them.

CI: a new orchestration job in ci.yml runs the tests against the release binary the test job already builds. It reads the Airflow version for the constraints file from requirements.txt instead of restating it.

The guarantees, and how each is met:

  • A retry is safe. Every write is a MERGE keyed on the source id, behind a unique constraint created with IF NOT EXISTS. Both suites run each pipeline twice and compare the graph with the files.
  • Values are bound, never pasted into the query. They travel as params (Query parameters are parsed and executable but no remote surface can supply them #1463). The Python SDK's query() takes no parameters, which is why the pipelines call the HTTP API. A test loads quotes, Cypher-looking text and Unicode, and reads them back unchanged.
  • Only retryable failures are retried. That means no connection, a timeout, 5xx, 408 or 429. Everything else fails at once (AirflowFailException / Failure(allow_retries=False)): a refused statement or token, bad input (the message names the file and line), or a failed validation. The tests cover each status through a stub server.
  • No silent drops. A transfer that names an unknown account is refused before anything is written. Otherwise the MATCH would find nothing and the row would be skipped in silence.
  • Validation compares the graph with the files. It re-reads the files and asks the graph directly; it never relies on the load step's own report. It also flags a transfer whose account changed between runs.

Two engine behaviours that shaped the Cypher, documented in ORCHESTRATION.md. I opened no issue for either:

  • Two MATCH clauses after UNWIND plan as a cartesian product. With 1,000 accounts and 200 rows, the single comma-separated MATCH took 4.5 ms; the two-clause form timed out after 281 s. That was one run each, on a laptop, on a release build of dev at 23a96ec. The pipeline uses the comma form.
  • A non-literal property map on an anonymous node is refused in MATCH. For example, (:Account {id: r.src}) is rejected; the pipeline names the node instead.

Verified

Check Result
cargo test --workspace --no-fail-fast (debug, as CI runs it), before the rebase 7,419 passed, 0 failed
Re-run on the rebased tree: all merge_* tests, load_csv_merge, docs_index_lists_every_document, ci_gate_enumeration 14 test files, 82 passed, 0 failed
openCypher TCK, --min-pass 3700 3,843 pass, 0 wrong, 4 errored, after the rebase. main CI shows 3,843 passing with the same 4 failures (Temporal10 ×2, Literals7, Literals8).
doc_check (CH-DOC-EXEC) 0 failures, 31 statements parsed, after the rebase
check_lint_ratchet.sh 796 783 warnings, none on changed lines
examples/orchestration/tests against the rebased release build 52 passed
Every command in both how-to pages, run as written against a local server with an auth file Passed. Doing this caught that the pages didn't say to copy sample/; now fixed.

Not verified / not covered

  • The orchestration CI job itself hasn't run yet. This PR is its first run.
  • It's not one of the checks ci-ok waits for. It's listed in UNGATED in tests/ci_gate_enumeration.rs, the same treatment gpml gets. Whether a suite that installs Airflow from PyPI should block merges is a separate decision for you.
  • Only in-process runs are tested: dag.test() and materialize. A deployed scheduler with separate workers isn't.
  • Only the pinned versions are tested: Airflow 3.3.2 and Dagster 1.13.26. Airflow 2 can't import the DAG at all.
  • dbt isn't covered. It compiles SQL for a SQL warehouse, and no dbt adapter for Cypher exists.
  • A pipeline run isn't one transaction. Each batch commits on its own, so a failed run leaves a partial load, which the next run completes. The doc explains why /api/tx wasn't used: it holds the store's write lock for the whole load.
  • Rows deleted from a file stay in the graph.

    UNWIND $rows AS r
    MATCH (s:Account {id: r.src}), (d:Account {id: r.dst})
    MERGE (s)-[t:TRANSFER {id: r.id}]->(d)

is the bulk edge-upsert idiom, and it lost data without an error. The
planner hands a MERGE between already-bound endpoints to
MatchMergeEdgeOperator, and it handed over only the literal half of the
relationship's property map; `{id: r.id}` lives in `property_exprs`,
which was dropped. So the operator matched any :TRANSFER between the pair
and, when it created one, created it without `id`. Two rows for one pair
wrote one edge, and the statement reported both as written.

samyama-ai#642 fixed the same defect for nodes and for whole-pattern MERGE; this is
the relationship-between-bound-nodes half it left. Found loading the
INT-10 reference pipeline (samyama-ai#1715): 60 transfers became 55 edges.

The fix resolves the expressions against each row with the resolver node
MERGE already uses, and uses the one resolved map both to match and to
create -- a MERGE that searched on one set of values and wrote another
would make a second edge on every re-run. EXPLAIN now shows the
expressions too.

tests/merge_row_properties.rs gains six cases: distinct keys between one
pair, an existing edge with a different key left alone, a key from another
bound node, literal and row properties together, ON CREATE / ON MATCH, and
an undirected pattern. All six fail without the change and pass with it;
the five existing cases pass either way. The openCypher TCK reads the same
as main (4 failures, the same 4).
…INT-10, samyama-ai#1715)

INT-10 asks for a documented, tested integration with Airflow/Dagster-style
orchestration: one reference pipeline per tool. Nothing in the repository
mentioned either.

examples/orchestration/ holds both, sharing one module for what a step
does, so the two cannot drift:

  samyama_pipeline.py      the load: constraint, accounts, transfers, and
                           validation against the source files; standard
                           library only
  samyama_airflow_dag.py   Airflow 3 DAG; server and token from an Airflow
                           Connection, files and batch size as DAG params
  samyama_dagster.py       Dagster assets with blocking asset checks; server
                           from SAMYAMA_URL / SAMYAMA_TOKEN via EnvVar
  sample/                  synthetic accounts and transfers (not data/,
                           which .gitignore swallows)

What the load guarantees, and how:

- A retry is safe: every write is a MERGE keyed on the source id, behind a
  unique constraint. Both suites run each pipeline twice.
- Values are bound in `params`, never interpolated. The Python SDK's
  query() takes no parameters, which is why this uses HTTP.
- Only an unavailable server (no connection, timeout, 5xx, 408, 429) is
  retried. Refused statements, a refused token, wrong input and a failed
  validation fail at once (AirflowFailException / Failure(allow_retries=
  False)).
- A transfer naming an account the graph does not hold is refused before
  anything is written; otherwise the MATCH finds nothing and the row is
  skipped in silence.

Two engine behaviours shaped the Cypher and are written down in
docs/ORCHESTRATION.md: two MATCH clauses after UNWIND plan as a cartesian
product (4.5 ms vs a 281 s timeout at 1,000 accounts, so the pipeline uses
one comma-separated MATCH), and a non-literal property map on an anonymous
node is refused in MATCH.

Tests (examples/orchestration/tests/, 52): each test starts its own release
server on spare ports with a fresh data directory. The Airflow DAG runs
through dag.test(), the Dagster assets through materialize. A missing
binary is an error, not a skip. A new `orchestration` job in ci.yml runs
them against the binary `test` already builds; it is listed in UNGATED in
tests/ci_gate_enumeration.rs rather than in ci-ok's needs, because making a
PyPI-installing suite block every merge is a separate decision.

Docs: ORCHESTRATION.md (design, guarantees, limits), and a how-to per tool.
Every command in the how-tos was run as written against a local server
with an auth file; doing so found that they omitted copying sample/, now
fixed. test_docs.py checks each Cypher block shown is one the pipeline
runs.

@Kalyanpulikanti-01 Kalyanpulikanti-01 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Merge is held: Merge conflict: manual fix required.

@Kalyanpulikanti-01 Kalyanpulikanti-01 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reviewed all 22 changed files. The Dagster integration drops credentials in a supported deployment scenario and does not retry transient failures in asset checks.

  • P2 examples/orchestration/samyama_dagster.py:133 — The environment check runs when the code location builds its definitions. If SAMYAMA_TOKEN is supplied only to execution workers, this permanently configures token=None, even though the token exists when the run starts. Authenticated runs then fail with 401. Resolve the optional token in the execution environment instead of conditionally attaching EnvVar during code loading. Add validation using definitions built before the worker token is set; the current tests rebuild definitions after setting it.

  • P2 examples/orchestration/samyama_dagster.py:108 — Neither asset check has a retry policy. Asset checks execute in separate steps and do not inherit the assets' RETRY policy, so a timeout, 429, or 503 during validation fails the run immediately. The wrapper merely rethrows SamyamaUnavailable; it does not schedule a retry. Attach retry_policy=RETRY to both checks and test a transient check failure followed by success.

@Kalyanpulikanti-01 Kalyanpulikanti-01 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reviewed all 22 changed files. Found a credential disclosure risk in HTTP redirects and a false-positive account validation case.

  • P2 examples/orchestration/samyama_pipeline.py:119 — The default urllib redirect handler forwards the explicitly supplied Authorization header when following a redirect to another origin, including an HTTPS-to-HTTP redirect. A redirected authenticated query can therefore disclose the pipeline's write token to another server. Reject redirects for this API client, or explicitly prevent credential forwarding across origins and protocol downgrades; validate this with a two-server redirect test.

  • P2 examples/orchestration/samyama_pipeline.py:329 — This total count does not establish that each account ID occurs exactly once. For a source containing A and B, a graph containing two identical, correct copies of A and no B makes both account checks return 2, so validation passes. This affects the explicitly supported validation of graphs without the schema constraint and can let the blocking Dagster check approve incorrect accounts. Check cardinality per source ID and add a test combining a missing account with a duplicate account.

@Kalyanpulikanti-01 Kalyanpulikanti-01 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Inspected all 22 changed files. No concrete defect established from the supplied diff. The engine change needs additional context: MergeOperator::resolved_props and the omitted relationship matching and creation code are necessary to verify expression handling consistently. Test-pass claims were not treated as verification.

@Kalyanpulikanti-01 Kalyanpulikanti-01 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reviewed the omitted MERGE expression resolver, planner collection and complete relationship matching/creation paths. Expression maps stay aligned with their edges, and the same resolved properties govern matching and creation while preserving ordered SET behavior. The 43 Rust regression tests passed on the unchanged engine code; all 56 current local Dagster, pipeline and documentation tests also passed. The credential, retry, redirect and account-validation findings are addressed. Merge remains gated on the complete current-head CI, including Airflow, Dagster, GPML and SDK compatibility.

@Kalyanpulikanti-01
Kalyanpulikanti-01 merged commit 12fb1cb into samyama-ai:dev Oct 22, 2026
14 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants