diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9f2dafca..26ffc3bb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -897,6 +897,78 @@ jobs: ${{ runner.temp }}/gpml-conformance/suite/results/summary.json if-no-files-found: warn + # INT-10 asks for an orchestration integration that is documented AND tested: + # one reference pipeline per tool, run against a server. This runs both -- + # the Airflow DAG through `dag.test()` and the Dagster assets through + # `materialize` -- against the release binary `test` built, each test on its + # own server with a fresh data directory (examples/orchestration/tests/). + # + # A reference pipeline that is not run rots exactly like documentation does, + # only later: it keeps importing cleanly long after the HTTP API it calls has + # moved, and the first person to find out is the one who copied it. + # + # Not in `ci-ok`'s `needs`, for the reason `gpml` is not: making a new suite + # block every merge -- here, one that installs Airflow from PyPI -- is a policy + # decision, separate from adding the suite. tests/ci_gate_enumeration.rs + # records the exemption so it cannot be forgotten. + orchestration: + name: orchestration (Airflow and Dagster reference pipelines) + needs: [changes, test] + if: needs.changes.outputs.code == 'true' + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: {python-version: "3.12"} + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + + # The binary `test` already built. A missing artifact is not a failure: + # the next step builds the engine instead, as the gpml job does. + - name: Discard any cached release executable + run: rm -f target/release/samyama + + - name: Take the release binary from the test job + id: orchestration_release_binary + continue-on-error: true + uses: actions/download-artifact@v4 + with: + name: samyama-release + path: target/release + + - name: Use it if it arrived, otherwise build it + run: | + set -euo pipefail + if [ "${{ steps.orchestration_release_binary.outcome }}" = "success" ] && [ -f target/release/samyama ]; then + # An artifact download does not preserve the executable bit. + chmod +x target/release/samyama + echo "using the binary built by the test job" + else + echo "::notice::no release artifact; building the engine" + rm -f target/release/samyama + cargo build --release --bin samyama --locked + fi + + - name: Install Airflow and Dagster at the tested versions + working-directory: examples/orchestration + # Airflow's constraints file is per Airflow version and per Python + # version; both are read rather than restated, so bumping the pin in + # requirements.txt cannot leave this step installing against the old + # constraints. + run: | + set -euo pipefail + airflow=$(sed -n 's/^apache-airflow==//p' requirements.txt) + py=$(python -c 'import sys; print(f"{sys.version_info[0]}.{sys.version_info[1]}")') + pip install -r requirements.txt \ + -c "https://raw.githubusercontent.com/apache/airflow/constraints-${airflow}/constraints-${py}.txt" + + - name: Run both pipelines against the engine + working-directory: examples/orchestration + env: + SAMYAMA_BIN: ${{ github.workspace }}/target/release/samyama + run: python -m pytest -v tests/ + # API-09: the SDKs users already have keep working against this build. # # docs/VERSIONING.md promises that the previous minor's SDKs work against the diff --git a/docs/ORCHESTRATION-AIRFLOW.md b/docs/ORCHESTRATION-AIRFLOW.md new file mode 100644 index 00000000..e682875b --- /dev/null +++ b/docs/ORCHESTRATION-AIRFLOW.md @@ -0,0 +1,128 @@ +# Loading Samyama from Apache Airflow + +How to run the reference DAG, +[`examples/orchestration/samyama_airflow_dag.py`](../examples/orchestration/samyama_airflow_dag.py), +against your server and your files. What the pipeline does, the guarantees it +makes and what it does not do are in [ORCHESTRATION.md](./ORCHESTRATION.md). + +Tested with Airflow 3.3.2 on Python 3.12. Airflow 2 is not supported: the DAG +uses Airflow 3's `airflow.sdk`. + +--- + +## 1. A credential for the pipeline + +Skip this if your server runs without `--auth-file`. + +```bash +samyama auth-token pipeline +``` + +prints a line for the credential file and, once, the token. Add the line to +the file the server reads with `:roles=write` on the end -- a pipeline that +loads data has no business managing tenants or restoring snapshots: + +```text +pipeline::roles=write +``` + +Keep the token for step 3. The server stores only its digest. + +## 2. Install, and put the files in the DAGs folder + +```bash +pip install "apache-airflow==3.3.2" \ + -c https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt + +cp examples/orchestration/samyama_airflow_dag.py \ + examples/orchestration/samyama_pipeline.py \ + "$AIRFLOW_HOME/dags/" +cp -r examples/orchestration/sample "$AIRFLOW_HOME/dags/" +``` + +Both `.py` files: the DAG imports `samyama_pipeline`, and Airflow puts the DAGs +folder on `sys.path`. `sample/` is the default input, found next to +`samyama_pipeline.py`; leave it out if every run will name its own files, and a +run that does not fails at once with `no such file`. Nothing else is needed -- +`samyama_pipeline` is standard library only, and there is no Samyama provider +package to install. + +## 3. A Connection to the server + +The DAG reads the server from an Airflow Connection, `samyama_default` unless a +run says otherwise. `schema` is the protocol, the convention Airflow's own HTTP +hook uses; the token, if any, goes in `password`: + +```bash +airflow connections add samyama_default \ + --conn-type http \ + --conn-host graph.internal \ + --conn-port 8080 \ + --conn-schema https \ + --conn-password "$SAMYAMA_TOKEN" +``` + +Any secrets backend Airflow supports works the same way, because the DAG asks +Airflow for the Connection rather than reading a variable of its own. As an +environment variable: + +```bash +export AIRFLOW_CONN_SAMYAMA_DEFAULT='{"conn_type": "http", "host": "graph.internal", "port": 8080, "schema": "https", "password": "..."}' +``` + +Optional: `{"timeout_seconds": 60}` in the Connection's extras sets how long one +statement may take. The default is 30. + +## 4. Run it + +Once, without the scheduler, to see it work -- with the bundled sample: + +```bash +airflow dags test samyama_accounts_transfers +``` + +With your own files: + +```bash +airflow dags test samyama_accounts_transfers \ + -c '{"accounts_csv": "/data/accounts.csv", "transfers_csv": "/data/transfers.csv"}' +``` + +The same JSON is what you pass to `airflow dags trigger --conf`, or type into +the *Trigger DAG* form in the UI. + +| Param | Default | | +|---|---|---| +| `samyama_conn_id` | `samyama_default` | the Connection to use | +| `accounts_csv` | the bundled sample | columns `id, holder, country, opened` | +| `transfers_csv` | the bundled sample | columns `id, src, dst, amount, at` | +| `batch_size` | `500` | rows per statement | + +The DAG has `schedule=None`: it runs when triggered. Give it a schedule, or an +asset to wait on, when you deploy it -- that choice belongs to the deployment, +not to a reference. + +## 5. When a run fails + +The failing task's log names the cause. Two kinds: + +- **Retried, with backoff, up to three times**: the server could not be + reached, timed out, or answered 5xx. +- **Failed at once, no retry**: the server refused the statement or the token, + the input is wrong (the message names the file and line), or validation found + the graph disagrees with the files (the message names each check, what it + expected and what it found). + +A failed run can leave part of a load in the graph. Fix the cause and run it +again: every write is an upsert keyed on the source id, so the second run +completes the first rather than duplicating it. + +## How this page is tested + +`examples/orchestration/tests/test_airflow.py` runs the DAG through +`dag.test()` against a real server: a full load, a second run that must leave +the graph unchanged, a file naming an unknown account (one try, nothing +written), a token taken from the Connection, a missing token (one try), and a +Connection named by the `samyama_conn_id` param. The commands above were run as +written against a local server before this page was published; they are not +re-run by CI. diff --git a/docs/ORCHESTRATION-DAGSTER.md b/docs/ORCHESTRATION-DAGSTER.md new file mode 100644 index 00000000..1c382e77 --- /dev/null +++ b/docs/ORCHESTRATION-DAGSTER.md @@ -0,0 +1,114 @@ +# Loading Samyama from Dagster + +How to run the reference assets, +[`examples/orchestration/samyama_dagster.py`](../examples/orchestration/samyama_dagster.py), +against your server and your files. What the pipeline does, the guarantees it +makes and what it does not do are in [ORCHESTRATION.md](./ORCHESTRATION.md). + +Tested with Dagster 1.13.26 on Python 3.12. + +--- + +## 1. A credential for the pipeline + +Skip this if your server runs without `--auth-file`. + +```bash +samyama auth-token pipeline +``` + +prints a line for the credential file and, once, the token. Add the line to +the file the server reads with `:roles=write` on the end -- a pipeline that +loads data has no business managing tenants or restoring snapshots: + +```text +pipeline::roles=write +``` + +## 2. Install, and put the files in your code location + +```bash +pip install "dagster==1.13.26" "dagster-webserver==1.13.26" +cp examples/orchestration/samyama_dagster.py examples/orchestration/samyama_pipeline.py my_project/ +cp -r examples/orchestration/sample my_project/ +``` + +Both `.py` files: `samyama_dagster` imports `samyama_pipeline`, which is +standard library only. `sample/` is the default input, found next to +`samyama_pipeline.py`; leave it out if every run will name its own files. There +is no Samyama integration package to install. + +## 3. Point it at the server + +| Variable | | +|---|---| +| `SAMYAMA_URL` | required -- the server's HTTP endpoint, e.g. `https://graph.internal:8080` | +| `SAMYAMA_TOKEN` | the token from step 1; leave unset for a server without an auth file | + +The URL uses `dg.EnvVar`; the optional token is read when the resource creates +its client on the execution worker, so worker-only secrets are available. The +token value is not stored with the code location or shown in the UI. + +## 4. Run it + +```bash +export SAMYAMA_URL=http://localhost:8080 +cd my_project +dagster dev -f samyama_dagster.py +``` + +and materialize the `samyama` group from the UI, or without the UI: + +```bash +dagster asset materialize -f samyama_dagster.py --select '*' +``` + +The source files and batch size are the `source_files` resource. They default +to the bundled sample; point a run at your own in the launchpad's config: + +```yaml +resources: + source_files: + config: + accounts_csv: /data/accounts.csv + transfers_csv: /data/transfers.csv + batch_size: 500 +``` + +| Asset | Depends on | Check | +|---|---|---| +| `samyama_schema` | | | +| `accounts` | `samyama_schema` | `accounts_match_source` | +| `transfers` | `accounts` | `transfers_match_source` | + +The checks are **blocking**: if `accounts_match_source` fails, `transfers` is +not loaded on top of accounts that do not match their file. Each check's +metadata records, per condition, how many it expected and how many it found. + +No schedule or sensor is defined; add one when you deploy it -- that choice +belongs to the deployment, not to a reference. + +## 5. When a run fails + +- **Retried, with exponential backoff, up to three times**: the server could + not be reached, timed out, or answered 5xx. +- **Failed at once, no retry** (`Failure(allow_retries=False)`): the server + refused the statement or the token, or the input is wrong -- the message + names the file and line. +- **A check failed**: the graph disagrees with the files. The check's + description names each condition that failed. + +A failed run can leave part of a load in the graph. Fix the cause and +materialize again: every write is an upsert keyed on the source id, so the +second run completes the first rather than duplicating it. + +## How this page is tested + +`examples/orchestration/tests/test_dagster.py` materializes the assets and +checks against a real server: a full load with both checks passing, a second +run that must leave the graph unchanged, a file naming an unknown account (the +step fails, is not retried, writes nothing), a failing accounts check that must +stop `transfers` from loading, a token from the resource, a missing token, and +the shipped definitions reading `SAMYAMA_URL` and `SAMYAMA_TOKEN` from the +environment. The commands above were run as written against a local server +before this page was published; they are not re-run by CI. diff --git a/docs/ORCHESTRATION.md b/docs/ORCHESTRATION.md new file mode 100644 index 00000000..0263eeed --- /dev/null +++ b/docs/ORCHESTRATION.md @@ -0,0 +1,151 @@ +# Loading Samyama from Airflow and Dagster + +A scheduled pipeline that loads files into the graph and checks the result, in +the two orchestrators most data teams already run. One reference pipeline per +tool, both run against a real server by the test suite: + +| Tool | Reference | How-to | +|---|---|---| +| Apache Airflow 3 | [`examples/orchestration/samyama_airflow_dag.py`](../examples/orchestration/samyama_airflow_dag.py) | [ORCHESTRATION-AIRFLOW.md](./ORCHESTRATION-AIRFLOW.md) | +| Dagster | [`examples/orchestration/samyama_dagster.py`](../examples/orchestration/samyama_dagster.py) | [ORCHESTRATION-DAGSTER.md](./ORCHESTRATION-DAGSTER.md) | + +Both are thin. What a step *does* is in +[`samyama_pipeline.py`](../examples/orchestration/samyama_pipeline.py), once, so +the two cannot drift into loading the same files two different ways. It is +standard library only, so it adds nothing to an orchestrator's worker image. + +--- + +## What the pipeline does + +``` +schema -> accounts -> transfers -> validate +``` + +| Step | What it does | +|---|---| +| schema | A unique constraint on `:Account(id)`. It backs the id lookups with an index, and makes the engine refuse a second account with a taken id. `IF NOT EXISTS`, so repeating it is harmless. | +| accounts | Reads `accounts.csv` (`id, holder, country, opened`) and upserts one `:Account` per row, in batches. | +| transfers | Reads `transfers.csv` (`id, src, dst, amount, at`), checks every account it names exists, then upserts one `:TRANSFER` relationship per row, in batches. | +| validate | Reads the files again and asks the graph whether it holds exactly them. Fails the run if not. | + +The sample in `examples/orchestration/sample/` is synthetic: 24 accounts and 60 +transfers, generated, describing no real party. + +## The three properties every orchestrated load needs + +**A retry is safe.** Orchestrators retry on their own initiative -- a worker +restart, a timeout, a node drained for maintenance. Every write is a `MERGE` +keyed on the source row's id, so running a step twice leaves the graph as +running it once. Both test suites run each pipeline twice and compare the graph +with the files. + +**Values are bound, never pasted into the statement.** Rows travel as the +`params` of `POST /api/query` (#1463): + +```cypher +UNWIND $rows AS r +MATCH (s:Account {id: r.src}), (d:Account {id: r.dst}) +MERGE (s)-[t:TRANSFER {id: r.id}]->(d) +SET t.amount = r.amount, t.at = r.at +RETURN count(t) AS written +``` + +A holder name with a quote in it, or text that looks like Cypher, is stored as +the string it is; the tests load a few such values and read them back. This is +also why the pipelines talk HTTP rather than use the Python SDK: the SDK's +`query()` takes no parameters, so using it would mean building statement text +out of file contents. + +**A failure says whether retrying can help.** + +| Failure | Retried? | Why | +|---|---|---| +| Server unreachable, timeout, any 5xx (including 503 when the server refuses writes after a failed flush) | Yes, with backoff | It can clear on its own. | +| The server refused the statement or the credential (4xx) | No | It will refuse it again. | +| The input is wrong: a missing column, an empty field, a duplicate id, an amount that is not a number, a transfer naming an account that does not exist | No | A retry reads the same file. | +| Validation found the graph disagrees with the files | No | A retry re-checks the same graph. | + +Retrying a permanent failure only delays the page to whoever has to fix it. +Airflow gets `AirflowFailException`; Dagster gets `Failure(allow_retries=False)`. + +## What validation checks + +Nothing it checks is taken from the load step's own report -- the files are read +again and the graph is asked: + +| Check | Catches | +|---|---| +| every account, found by id, holds the file's `holder`, `country`, `opened` | a missed row, a value that did not arrive, a value something else changed | +| no account id is held by two nodes | a writer that used `CREATE` where it should `MERGE` (the constraint stops this once the schema step has run; the check covers graphs where it has not) | +| every transfer runs between the accounts the file names and holds its `id`, `amount`, `at` | a dropped or misdirected edge, a changed value | +| no transfer id from this file is held by two relationships | a transfer whose account changed between two runs: `MERGE` matches on both endpoints and the id, so it adds a second relationship rather than moving the first | + +The transfer step also refuses, before writing anything, a file that names an +account the graph does not hold. Without that, the `MATCH` would find no row for +it, the `MERGE` would never run, and the statement would succeed having silently +skipped the transfer. + +## Adapting it to your data + +Copy `samyama_pipeline.py` and change the statements and the two readers; the +orchestrator files need to change only if you add or rename a step. Keep: + +- **`MERGE` on a key that comes from the source**, with a unique constraint on + it. Without a key a retry duplicates. +- **Both endpoints of a relationship in one comma-separated `MATCH`.** Written + as two `MATCH` clauses -- which means the same thing -- the planner does not + look either up by id: it pairs every node with every node and filters. On a + graph of 1,000 accounts a 200-row batch took 4.5 ms in the comma form and + timed out after 281 s in the two-clause form (2026-10-19, release build of + `dev` at `23a96ec`, one laptop, one run each -- the gap is the point, not the + figures). +- **Named variables on nodes whose properties come from the row.** + `(:Account {id: r.src})` with no variable is refused in a `MATCH`; write + `(s:Account {id: r.src})`. +- **Batches.** The default is 500 rows per statement; it is a parameter of both + pipelines. + +## What this does not do + +- **A run is not one transaction.** Each batch commits on its own, so while a + run is in progress -- or after it fails -- readers see a partial load, which + the next run completes. `POST /api/tx/begin` can make a run atomic, but a + transaction holds the store's write lock: every other reader and writer waits + for the whole load, and the transaction is rolled back if it outlives the + session timeout. That trade is not made for you here. +- **Deletions are not propagated.** A row removed from the file stays in the + graph. Upserts cannot tell "deleted" from "not in this extract". +- **One graph per server.** The open-source server serves the graph named + `default`; use a separate server and data directory per dataset. +- **dbt is not covered.** dbt compiles SQL for a SQL warehouse, and there is no + dbt adapter for Cypher. +- **Airflow 2 is not supported.** The DAG imports from `airflow.sdk`, which is + Airflow 3's authoring interface. + +## How it is tested + +`examples/orchestration/tests/`, run by the `orchestration` job in +`.github/workflows/ci.yml` against the release binary the `test` job builds: + +```bash +cargo build --release --bin samyama +cd examples/orchestration +pip install -r requirements.txt \ + -c https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt +python -m pytest -v tests/ +``` + +Each test starts its own server on spare ports with a fresh data directory, so +no test sees another's graph and nothing touches a server you are running. +`SAMYAMA_BIN` names another binary. + +What the tests run: the Airflow DAG through `dag.test()` and the Dagster assets +through `materialize`, in process, with Airflow's own Connection lookup and +Dagster's own resources. What they do **not** run: a deployed scheduler with +separate workers, or any version other than the pinned Airflow 3.3.2 and +Dagster 1.13.26. + +The CI job is not yet one of the checks `ci-ok` waits for, so a failure there +shows on a pull request without blocking its merge; whether it should block is +a separate decision, recorded in `tests/ci_gate_enumeration.rs`. diff --git a/docs/README.md b/docs/README.md index 048efd2e..7174db3b 100644 --- a/docs/README.md +++ b/docs/README.md @@ -48,6 +48,9 @@ eight of eighteen documents (#1450). - **[FAILURE-MODES.md](./FAILURE-MODES.md)** — how the engine behaves when things go wrong. - **[SOC2-CONTROL-SET.md](./SOC2-CONTROL-SET.md)** — SOC 2 gap analysis (documented, not certified): access control, audit, change management, backup and incident response, each control with its evidence and its gap. - **[BI-CONNECTIVITY.md](./BI-CONNECTIVITY.md)** — connecting BI tools. +- **[ORCHESTRATION.md](./ORCHESTRATION.md)** — loading the graph from a scheduled pipeline: a retry-safe, parameter-bound load with validation, as reference pipelines for Airflow and Dagster that the test suite runs against a real server. +- **[ORCHESTRATION-AIRFLOW.md](./ORCHESTRATION-AIRFLOW.md)** — how to run the reference DAG in Airflow 3: credential, Connection, params, what a failure means. +- **[ORCHESTRATION-DAGSTER.md](./ORCHESTRATION-DAGSTER.md)** — how to run the reference assets and checks in Dagster. ### Data, provenance and privacy diff --git a/examples/README.md b/examples/README.md index f801b6d4..0ed2f351 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,7 +1,7 @@ # `examples/` Every program in this directory, what it does, and what data it needs. -`cargo run --release --example ` runs any of the `.rs` files (the two non-Rust +`cargo run --release --example ` runs any of the `.rs` files (the four non-Rust programs say how below). The names are an interface: `scripts/run_all_examples.sh`, `scripts/verify-sweep.sh`, `scripts/regression-test.sh`, `.github/workflows/ci.yml` and external knowledge-graph repositories invoke them by name, so they do not get renamed. @@ -9,7 +9,7 @@ external knowledge-graph repositories invoke them by name, so they do not get re **Start here** if you are new: the demos on built-in data. They need nothing but a checkout. -**Most of this directory is internal.** Of the 127 programs, 60 are performance +**Most of this directory is internal.** Of the 129 programs, 60 are performance investigations, conformance checks or maintainer tools — written to answer one question about the engine, usually once, often to kill a hypothesis. They print verdicts and timings, not results you can use. Those sections say *Internal* in their opening line; @@ -29,6 +29,7 @@ Purposes below are taken from each file's own header comment. | [Conformance and correctness checks](#conformance-and-correctness-checks) | 32 | No - internal | | [Maintainer tools](#maintainer-tools) | 3 | No - internal | | [Hardware client sketch](#hardware-client-sketch) | 1 | No - internal | +| [Orchestration reference pipelines](#orchestration-reference-pipelines) | 2 | Yes | Helper code shared between programs lives in `examples/common/`, which holds no runnable program. A dataset's loader and its benchmarks share `common/_common.rs`, @@ -235,6 +236,14 @@ not need them. |---|---|---| | `esp32/samyama_esp32.ino` | Writes to a Samyama Edge node from an ESP32 over RESP (API-15). RESP rather than HTTP because a TLS stack, a JSON encoder and a chunked reader are most of an ESP32's flash and all of its RAM headroom. Nothing here heap-allocates. | Arduino/ESP32 toolchain and a reachable Samyama RESP endpoint. | +## Orchestration reference pipelines +Not Cargo examples. Python, run by Airflow or Dagster against a running server; both share the load steps in `orchestration/samyama_pipeline.py`. How to run them: [`docs/ORCHESTRATION.md`](../docs/ORCHESTRATION.md). `orchestration/tests/` runs both against a real server, and the `orchestration` CI job runs those tests. + +| Program | What it does | Data it needs | +|---|---|---| +| `orchestration/samyama_airflow_dag.py` | Airflow 3 DAG (INT-10): constraint, then accounts and transfers upserted in batches with bound parameters, then a validation task that compares the graph with the files. Retries only failures a retry can fix. | `orchestration/sample/` (synthetic, bundled), or your own CSVs as DAG params; a server named by an Airflow Connection. | +| `orchestration/samyama_dagster.py` | The same load as Dagster assets, with blocking asset checks: transfers are not loaded on top of accounts that fail theirs. | `orchestration/sample/`, or your own CSVs in the `source_files` resource; `SAMYAMA_URL` and optionally `SAMYAMA_TOKEN`. | + ## Terms used in the file headers Several terms appear in these headers with no definition in this repository. Each one is diff --git a/examples/orchestration/README.md b/examples/orchestration/README.md new file mode 100644 index 00000000..0aa34c74 --- /dev/null +++ b/examples/orchestration/README.md @@ -0,0 +1,16 @@ +# Orchestration reference pipelines + +Loading Samyama from Airflow and Dagster (INT-10). Start with +[`docs/ORCHESTRATION.md`](../../docs/ORCHESTRATION.md). + +| File | | +|---|---| +| `samyama_pipeline.py` | The load steps both pipelines run. Standard library only. | +| `samyama_airflow_dag.py` | Airflow 3 DAG — [how to run it](../../docs/ORCHESTRATION-AIRFLOW.md). | +| `samyama_dagster.py` | Dagster assets and checks — [how to run them](../../docs/ORCHESTRATION-DAGSTER.md). | +| `sample/` | Synthetic accounts and transfers, the default input. | +| `requirements.txt` | The orchestrator versions the tests were run against. | +| `tests/` | Runs both pipelines against a real server: `python -m pytest tests/`. | + +The sample is not in `data/` because the repository ignores every directory of +that name. diff --git a/examples/orchestration/requirements.txt b/examples/orchestration/requirements.txt new file mode 100644 index 00000000..acc6684d --- /dev/null +++ b/examples/orchestration/requirements.txt @@ -0,0 +1,12 @@ +# The versions tests/ were last run against. The DAG and the assets use only +# stable public API, but a version not listed here has not been tested. +# +# Airflow wants its constraints file, which pins its transitive dependencies: +# pip install -r requirements.txt \ +# -c https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt +# +# Each pipeline needs only its own orchestrator; `samyama_pipeline.py` is +# standard library only. +apache-airflow==3.3.2 +dagster==1.13.26 +pytest diff --git a/examples/orchestration/sample/accounts.csv b/examples/orchestration/sample/accounts.csv new file mode 100644 index 00000000..2612a23c --- /dev/null +++ b/examples/orchestration/sample/accounts.csv @@ -0,0 +1,25 @@ +id,holder,country,opened +ACC-0001,Holder 01,DE,2021-11-11 +ACC-0002,Holder 02,GB,2023-07-24 +ACC-0003,Holder 03,IN,2022-06-30 +ACC-0004,Holder 04,GB,2019-05-19 +ACC-0005,Holder 05,GB,2021-06-02 +ACC-0006,Holder 06,IN,2022-08-24 +ACC-0007,Holder 07,SG,2023-06-13 +ACC-0008,Holder 08,IN,2021-09-28 +ACC-0009,Holder 09,AE,2022-12-15 +ACC-0010,Holder 10,DE,2021-10-08 +ACC-0011,Holder 11,US,2020-04-21 +ACC-0012,Holder 12,US,2023-10-04 +ACC-0013,Holder 13,IN,2019-03-12 +ACC-0014,Holder 14,DE,2023-07-20 +ACC-0015,Holder 15,DE,2021-07-10 +ACC-0016,Holder 16,SG,2022-02-11 +ACC-0017,Holder 17,GB,2019-10-21 +ACC-0018,Holder 18,IN,2021-11-13 +ACC-0019,Holder 19,IN,2022-03-04 +ACC-0020,Holder 20,DE,2021-06-19 +ACC-0021,Holder 21,AE,2020-04-19 +ACC-0022,Holder 22,US,2023-07-20 +ACC-0023,Holder 23,IN,2022-04-07 +ACC-0024,Holder 24,AE,2020-09-10 diff --git a/examples/orchestration/sample/transfers.csv b/examples/orchestration/sample/transfers.csv new file mode 100644 index 00000000..9370975b --- /dev/null +++ b/examples/orchestration/sample/transfers.csv @@ -0,0 +1,61 @@ +id,src,dst,amount,at +TX-00024,ACC-0014,ACC-0004,10626.50,2026-09-01T17:26:00 +TX-00008,ACC-0006,ACC-0022,2040.19,2026-09-01T22:51:00 +TX-00023,ACC-0007,ACC-0004,22974.72,2026-09-02T04:31:00 +TX-00046,ACC-0011,ACC-0021,4836.15,2026-09-03T13:20:00 +TX-00003,ACC-0019,ACC-0021,15383.89,2026-09-05T10:03:00 +TX-00034,ACC-0008,ACC-0001,13521.34,2026-09-05T10:10:00 +TX-00007,ACC-0003,ACC-0009,16470.95,2026-09-05T11:15:00 +TX-00001,ACC-0002,ACC-0008,15511.29,2026-09-07T01:34:00 +TX-00038,ACC-0008,ACC-0014,24447.60,2026-09-07T17:05:00 +TX-00016,ACC-0002,ACC-0001,1757.14,2026-09-08T07:38:00 +TX-00013,ACC-0005,ACC-0010,18480.51,2026-09-08T13:21:00 +TX-00014,ACC-0003,ACC-0024,8392.61,2026-09-11T14:42:00 +TX-00055,ACC-0004,ACC-0020,19788.88,2026-09-12T05:15:00 +TX-00053,ACC-0007,ACC-0009,4614.14,2026-09-12T07:13:00 +TX-00036,ACC-0019,ACC-0015,1342.28,2026-09-12T08:49:00 +TX-00012,ACC-0009,ACC-0014,16782.06,2026-09-12T16:45:00 +TX-00021,ACC-0016,ACC-0006,18163.40,2026-09-12T23:31:00 +TX-00020,ACC-0003,ACC-0007,10314.72,2026-09-13T02:15:00 +TX-00032,ACC-0019,ACC-0021,14762.61,2026-09-13T04:29:00 +TX-00035,ACC-0008,ACC-0009,12375.58,2026-09-14T19:56:00 +TX-00047,ACC-0013,ACC-0010,24356.62,2026-09-15T00:01:00 +TX-00019,ACC-0006,ACC-0011,17176.78,2026-09-15T00:28:00 +TX-00051,ACC-0018,ACC-0008,1226.38,2026-09-15T00:44:00 +TX-00026,ACC-0020,ACC-0006,24295.55,2026-09-15T11:58:00 +TX-00004,ACC-0001,ACC-0008,17736.04,2026-09-15T14:08:00 +TX-00005,ACC-0002,ACC-0001,3809.13,2026-09-16T04:33:00 +TX-00050,ACC-0016,ACC-0020,22660.49,2026-09-16T06:04:00 +TX-00059,ACC-0021,ACC-0006,17425.31,2026-09-16T22:17:00 +TX-00011,ACC-0016,ACC-0005,7413.98,2026-09-17T22:37:00 +TX-00037,ACC-0001,ACC-0012,13238.87,2026-09-18T00:30:00 +TX-00044,ACC-0014,ACC-0021,7584.47,2026-09-18T19:27:00 +TX-00056,ACC-0022,ACC-0006,916.03,2026-09-18T23:14:00 +TX-00031,ACC-0016,ACC-0023,18584.87,2026-09-19T02:11:00 +TX-00018,ACC-0001,ACC-0019,16750.03,2026-09-20T01:56:00 +TX-00058,ACC-0006,ACC-0024,9366.96,2026-09-21T07:53:00 +TX-00045,ACC-0004,ACC-0013,19902.21,2026-09-21T13:27:00 +TX-00010,ACC-0009,ACC-0006,5539.39,2026-09-21T15:06:00 +TX-00048,ACC-0017,ACC-0014,7880.60,2026-09-22T06:25:00 +TX-00049,ACC-0013,ACC-0021,8251.42,2026-09-22T18:49:00 +TX-00006,ACC-0014,ACC-0021,22073.56,2026-09-23T06:19:00 +TX-00060,ACC-0014,ACC-0002,1211.54,2026-09-23T18:29:00 +TX-00033,ACC-0002,ACC-0001,1347.28,2026-09-24T10:34:00 +TX-00029,ACC-0024,ACC-0019,6958.20,2026-09-25T08:07:00 +TX-00027,ACC-0022,ACC-0020,16655.47,2026-09-25T16:40:00 +TX-00028,ACC-0012,ACC-0006,8840.92,2026-09-25T17:08:00 +TX-00022,ACC-0004,ACC-0008,11271.48,2026-09-26T04:17:00 +TX-00043,ACC-0008,ACC-0005,16964.93,2026-09-26T06:47:00 +TX-00054,ACC-0013,ACC-0017,17249.54,2026-09-26T22:23:00 +TX-00057,ACC-0006,ACC-0023,14257.22,2026-09-27T03:01:00 +TX-00030,ACC-0005,ACC-0021,144.26,2026-09-27T05:24:00 +TX-00009,ACC-0014,ACC-0008,18448.21,2026-09-27T12:52:00 +TX-00040,ACC-0007,ACC-0018,16884.95,2026-09-27T16:07:00 +TX-00002,ACC-0019,ACC-0021,10370.35,2026-09-28T17:08:00 +TX-00025,ACC-0019,ACC-0020,17488.19,2026-09-29T00:40:00 +TX-00041,ACC-0008,ACC-0015,12644.28,2026-09-29T04:32:00 +TX-00017,ACC-0009,ACC-0023,15700.25,2026-09-29T13:11:00 +TX-00015,ACC-0011,ACC-0023,10958.45,2026-09-29T15:37:00 +TX-00042,ACC-0010,ACC-0022,20761.68,2026-09-29T23:49:00 +TX-00039,ACC-0010,ACC-0014,4582.28,2026-09-30T01:11:00 +TX-00052,ACC-0005,ACC-0013,2075.53,2026-10-01T05:16:00 diff --git a/examples/orchestration/samyama_airflow_dag.py b/examples/orchestration/samyama_airflow_dag.py new file mode 100644 index 00000000..97d9c86b --- /dev/null +++ b/examples/orchestration/samyama_airflow_dag.py @@ -0,0 +1,131 @@ +"""Reference Airflow DAG: load accounts and transfers into Samyama, then check them (INT-10). + + schema -> accounts -> transfers -> validate + +Copy this file and `samyama_pipeline.py` into your DAGs folder. Airflow puts +that folder on `sys.path`, which is how the import below resolves. + +Where the settings come from -- nothing about a deployment is written here: + +- The server and its credential come from an Airflow Connection, by default + `samyama_default`: host, port, `schema` (`http` or `https`, the convention + Airflow's own HTTP hook uses) and, if the server has an auth file, the token + in `password`. Optional `timeout_seconds` in the extras. +- The source files and batch size are DAG params, so a run can be pointed at + other files without editing the DAG. They default to the bundled sample. + +Retries: a step that fails because the server was unreachable is retried, with +backoff. A step that fails because the server refused the statement or the +credential, or because the input is wrong, fails at once -- retrying would read +the same files and get the same answer. See docs/ORCHESTRATION-AIRFLOW.md. + +Tested against Airflow 3.3 by `tests/test_airflow.py`. +""" + +from __future__ import annotations + +import functools +from datetime import timedelta +from urllib.parse import urlsplit + +from airflow.sdk import BaseHook, Param, dag, get_current_context, task +from airflow.sdk.exceptions import AirflowFailException + +import samyama_pipeline as sp + +DEFAULT_CONN_ID = "samyama_default" + + +def client_from_connection(conn_id: str) -> sp.SamyamaHttp: + conn = BaseHook.get_connection(conn_id) + host = conn.host or "" + if not host: + raise AirflowFailException(f"connection {conn_id!r} has no host") + base = host if "://" in host else f"{conn.schema or 'http'}://{host}" + if conn.port and urlsplit(base).port is None: + base = f"{base}:{conn.port}" + timeout = float(conn.extra_dejson.get("timeout_seconds", 30)) + return sp.SamyamaHttp(base, token=conn.password or None, timeout_seconds=timeout) + + +def fail_fast_on_permanent_errors(fn): + """Turn a failure that a retry cannot fix into one Airflow does not retry.""" + + @functools.wraps(fn) + def wrapper(*args, **kwargs): + try: + return fn(*args, **kwargs) + except (sp.SamyamaError, sp.SourceDataError) as e: + if sp.is_retryable(e): + raise + raise AirflowFailException(str(e)) from e + + return wrapper + + +def _settings() -> tuple[sp.SamyamaHttp, dict]: + params = get_current_context()["params"] + return client_from_connection(params["samyama_conn_id"]), params + + +@dag( + dag_id="samyama_accounts_transfers", + description="Load accounts and transfers from CSV into Samyama, then check the graph matches the files", + schedule=None, + catchup=False, + max_active_runs=1, + tags=["samyama", "reference"], + default_args={ + "retries": 3, + "retry_delay": timedelta(seconds=30), + "retry_exponential_backoff": True, + }, + params={ + "samyama_conn_id": Param(DEFAULT_CONN_ID, type="string", + description="Airflow Connection holding the server URL and token"), + "accounts_csv": Param(str(sp.SAMPLE_DIR / "accounts.csv"), type="string", + description="CSV with columns id, holder, country, opened"), + "transfers_csv": Param(str(sp.SAMPLE_DIR / "transfers.csv"), type="string", + description="CSV with columns id, src, dst, amount, at"), + "batch_size": Param(500, type="integer", minimum=1, + description="Rows per statement"), + }, +) +def samyama_accounts_transfers(): + @task + @fail_fast_on_permanent_errors + def schema() -> int: + client, _ = _settings() + return sp.ensure_schema(client) + + @task + @fail_fast_on_permanent_errors + def accounts() -> int: + client, params = _settings() + rows = sp.read_accounts(params["accounts_csv"]) + return sp.load_accounts(client, rows, params["batch_size"]) + + @task + @fail_fast_on_permanent_errors + def transfers() -> int: + client, params = _settings() + rows = sp.read_transfers(params["transfers_csv"]) + return sp.load_transfers(client, rows, params["batch_size"]) + + @task + @fail_fast_on_permanent_errors + def validate() -> dict: + client, params = _settings() + checks = sp.check_accounts(client, sp.read_accounts(params["accounts_csv"])) + checks += sp.check_transfers(client, sp.read_transfers(params["transfers_csv"])) + failed = sp.failures(checks) + if failed: + # The graph disagrees with the files it was loaded from. Retrying + # re-checks the same graph against the same files. + raise AirflowFailException("; ".join(failed)) + return {c.name: c.found for c in checks} + + schema() >> accounts() >> transfers() >> validate() + + +samyama_accounts_transfers() diff --git a/examples/orchestration/samyama_dagster.py b/examples/orchestration/samyama_dagster.py new file mode 100644 index 00000000..fb729dea --- /dev/null +++ b/examples/orchestration/samyama_dagster.py @@ -0,0 +1,145 @@ +"""Reference Dagster assets: load accounts and transfers into Samyama, with checks (INT-10). + + samyama_schema -> accounts -> transfers + (check) (check) + +Load it as a code location with this directory on the path: + + dagster dev -f samyama_dagster.py + +Where the settings come from -- nothing about a deployment is written here: + +- `SAMYAMA_URL` (required) names the server's HTTP endpoint. `SAMYAMA_TOKEN` + is the bearer token, needed only when the server runs with an auth file. Both + are resolved in the execution environment: the URL through `dg.EnvVar`, + and the optional token when the resource creates its client. No token value + is stored with the code location. +- The source files and batch size are the `source_files` resource. They default + to the bundled sample; set them per run in the launchpad's resource config. + +Each asset's check is blocking: if the accounts in the graph do not match the +file, the transfers are not loaded on top of them. + +Retries: an unreachable server is retried with backoff. A refused statement or +credential, or wrong input, fails at once with `allow_retries=False` -- retrying +would read the same files and get the same answer. See +docs/ORCHESTRATION-DAGSTER.md. + +Tested against Dagster 1.13 by `tests/test_dagster.py`. +""" + +from __future__ import annotations + +import functools +import os +from typing import Optional + +import dagster as dg + +import samyama_pipeline as sp + + +class SamyamaResource(dg.ConfigurableResource): + """The server, its credential, and how long one statement may take.""" + + url: str + token: Optional[str] = None + token_env_var: Optional[str] = None + timeout_seconds: float = 30.0 + + def client(self) -> sp.SamyamaHttp: + token = self.token + if token is None and self.token_env_var is not None: + token = os.environ.get(self.token_env_var) + return sp.SamyamaHttp(self.url, token=token or None, timeout_seconds=self.timeout_seconds) + + +class SourceFiles(dg.ConfigurableResource): + """Where the rows come from, and how many go in one statement.""" + + accounts_csv: str = str(sp.SAMPLE_DIR / "accounts.csv") + transfers_csv: str = str(sp.SAMPLE_DIR / "transfers.csv") + batch_size: int = 500 + + +RETRY = dg.RetryPolicy(max_retries=3, delay=30, backoff=dg.Backoff.EXPONENTIAL) + + +def fail_fast_on_permanent_errors(fn): + """Turn a failure that a retry cannot fix into one Dagster does not retry.""" + + @functools.wraps(fn) + def wrapper(*args, **kwargs): + try: + return fn(*args, **kwargs) + except (sp.SamyamaError, sp.SourceDataError) as e: + if sp.is_retryable(e): + raise + raise dg.Failure(description=str(e), allow_retries=False) from e + + return wrapper + + +@dg.asset(retry_policy=RETRY, group_name="samyama") +@fail_fast_on_permanent_errors +def samyama_schema(samyama: SamyamaResource) -> dg.MaterializeResult: + return dg.MaterializeResult(metadata={"statements": sp.ensure_schema(samyama.client())}) + + +@dg.asset(deps=[samyama_schema], retry_policy=RETRY, group_name="samyama") +@fail_fast_on_permanent_errors +def accounts(samyama: SamyamaResource, source_files: SourceFiles) -> dg.MaterializeResult: + rows = sp.read_accounts(source_files.accounts_csv) + written = sp.load_accounts(samyama.client(), rows, source_files.batch_size) + return dg.MaterializeResult(metadata={"rows": written, "source": source_files.accounts_csv}) + + +@dg.asset(deps=[accounts], retry_policy=RETRY, group_name="samyama") +@fail_fast_on_permanent_errors +def transfers(samyama: SamyamaResource, source_files: SourceFiles) -> dg.MaterializeResult: + rows = sp.read_transfers(source_files.transfers_csv) + written = sp.load_transfers(samyama.client(), rows, source_files.batch_size) + return dg.MaterializeResult(metadata={"rows": written, "source": source_files.transfers_csv}) + + +def _result(checks: list[sp.Check]) -> dg.AssetCheckResult: + failed = sp.failures(checks) + return dg.AssetCheckResult( + passed=not failed, + description="; ".join(failed) or "the graph matches the source file", + metadata={c.name: f"{c.found} of {c.expected}" for c in checks}, + ) + + +@dg.asset_check(asset=accounts, blocking=True, retry_policy=RETRY) +@fail_fast_on_permanent_errors +def accounts_match_source(samyama: SamyamaResource, source_files: SourceFiles) -> dg.AssetCheckResult: + return _result(sp.check_accounts(samyama.client(), sp.read_accounts(source_files.accounts_csv))) + + +@dg.asset_check(asset=transfers, blocking=True, retry_policy=RETRY) +@fail_fast_on_permanent_errors +def transfers_match_source(samyama: SamyamaResource, source_files: SourceFiles) -> dg.AssetCheckResult: + return _result(sp.check_transfers(samyama.client(), sp.read_transfers(source_files.transfers_csv))) + + +ASSETS = [samyama_schema, accounts, transfers] +CHECKS = [accounts_match_source, transfers_match_source] + +def build_definitions() -> dg.Definitions: + return dg.Definitions( + assets=ASSETS, + asset_checks=CHECKS, + resources={ + "samyama": SamyamaResource( + url=dg.EnvVar("SAMYAMA_URL"), + # Resolve optional credentials on workers, where secrets may be + # injected after the code location builds these definitions. + token_env_var="SAMYAMA_TOKEN", + ), + "source_files": SourceFiles(), + }, + ) + + +defs = build_definitions() diff --git a/examples/orchestration/samyama_pipeline.py b/examples/orchestration/samyama_pipeline.py new file mode 100644 index 00000000..8da59809 --- /dev/null +++ b/examples/orchestration/samyama_pipeline.py @@ -0,0 +1,374 @@ +"""The load both reference pipelines run, with no orchestrator in it (INT-10). + +`samyama_airflow_dag.py` and `samyama_dagster.py` are thin: they decide when a +step runs, where its settings come from, and what a failure means to their +scheduler. What a step *does* lives here, once, so the two pipelines cannot +drift into loading the same files two different ways. + +Three properties every orchestrated load needs, and how this module gets each: + +- **A retry is safe.** Every write is a `MERGE` keyed on the source row's id, + so running a step twice leaves the graph as running it once. An orchestrator + retries on its own initiative -- a worker restart, a timeout -- and a load + that duplicates on retry corrupts the graph the first time that happens. +- **Values are bound, never interpolated.** Rows travel in the `params` of + `POST /api/query` (#1463), so a holder name containing a quote, or text that + looks like Cypher, is stored as the string it is. +- **A failure says whether retrying can help.** `SamyamaUnavailable` (no + connection, a timeout, a 5xx, a 408 or 429) might clear on its own; anything + else -- `SamyamaRejected` (the server refused the statement or the + credential), `SourceDataError` (the input is wrong), a load the server + reported inconsistently -- will not, and a scheduler that retries it only + delays the page to a human. `is_retryable` is the one place that decides. + +Standard library only: the module runs inside any orchestrator's worker +without adding a dependency to it. +""" + +from __future__ import annotations + +import csv +import json +import math +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterator, Optional, Sequence + +#: The sample files the pipelines read when they are not told otherwise. +SAMPLE_DIR = Path(__file__).resolve().parent / "sample" + + +class SamyamaError(Exception): + """A failure talking to the server. Retrying will not help unless it is + the `SamyamaUnavailable` subclass.""" + + +class SamyamaUnavailable(SamyamaError): + """The server could not be reached, failed, or asked the client to come + back later; a retry may succeed.""" + + +#: Client errors that say "not now" rather than "not this": a request timeout, +#: and a quota or rate limit. +RETRYABLE_4XX = frozenset({408, 429}) + + +def is_retryable(error: BaseException) -> bool: + """Whether a step that raised `error` is worth running again. + + Only an unavailable server is. A refused statement or credential, wrong + input, or a load the server reported inconsistently gets the same answer + the next time. + """ + return isinstance(error, SamyamaUnavailable) + + +class SamyamaRejected(SamyamaError): + """The server refused the request; a retry will get the same answer.""" + + def __init__(self, status: int, message: str): + super().__init__(f"HTTP {status}: {message}") + self.status = status + self.message = message + + +class SourceDataError(ValueError): + """The input files are wrong; a retry will read the same files.""" + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + """API redirects must not forward write credentials to a different server.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +@dataclass +class SamyamaHttp: + """A minimal client for `POST /api/query`. + + The Python SDK's `query` takes no parameters, so a pipeline built on it + would have to splice values into the statement text. This posts the + statement and its parameters separately, which is the property a loader + that reads untrusted files cannot do without. + """ + + url: str + token: Optional[str] = None + timeout_seconds: float = 30.0 + + def __post_init__(self) -> None: + if not self.url: + raise ValueError("a Samyama URL is required, e.g. http://localhost:8080") + self.url = self.url.rstrip("/") + + def query(self, cypher: str, params: Optional[dict] = None, read_only: bool = False) -> dict: + body: dict[str, Any] = {"query": cypher} + if params: + body["params"] = params + if read_only: + # The server refuses the statement if it writes (#1628). + body["read_only"] = True + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + request = urllib.request.Request( + f"{self.url}/api/query", + data=json.dumps(body).encode("utf-8"), + headers=headers, + method="POST", + ) + try: + with urllib.request.build_opener(_NoRedirect()).open( + request, timeout=self.timeout_seconds) as response: + return json.loads(response.read()) + except urllib.error.HTTPError as e: + message = _error_message(e) + if e.code >= 500 or e.code in RETRYABLE_4XX: + # 503 is the server refusing writes after a failed flush (#1274): + # it can clear, so it is retryable along with every other 5xx. + raise SamyamaUnavailable(f"HTTP {e.code}: {message}") from e + raise SamyamaRejected(e.code, message) from e + except (urllib.error.URLError, TimeoutError, ConnectionError) as e: + raise SamyamaUnavailable(f"{self.url}: {e}") from e + + def scalar(self, cypher: str, params: Optional[dict] = None) -> Any: + """The single value a one-row, one-column read returns.""" + records = self.query(cypher, params, read_only=True)["records"] + if len(records) != 1 or len(records[0]) != 1: + raise SamyamaError(f"expected one value, got {records!r} for: {cypher}") + return records[0][0] + + +def _error_message(e: urllib.error.HTTPError) -> str: + raw = e.read().decode("utf-8", "replace") + try: + return json.loads(raw).get("error", raw) + except (ValueError, AttributeError): + return raw or e.reason + + +# -------------------------------------------------------------------------- +# Reading the source files +# -------------------------------------------------------------------------- + +ACCOUNT_COLUMNS = ("id", "holder", "country", "opened") +TRANSFER_COLUMNS = ("id", "src", "dst", "amount", "at") + + +def read_accounts(path: str | Path) -> list[dict]: + rows = _read_csv(path, ACCOUNT_COLUMNS) + _reject_duplicate_ids(path, rows) + return [{"id": r["id"], "holder": r["holder"], "country": r["country"], "opened": r["opened"]} + for _, r in rows] + + +def read_transfers(path: str | Path) -> list[dict]: + rows = _read_csv(path, TRANSFER_COLUMNS) + _reject_duplicate_ids(path, rows) + out = [] + for line, r in rows: + try: + amount = float(r["amount"]) + except ValueError: + raise SourceDataError(f"{path}:{line}: amount {r['amount']!r} is not a number") from None + if not math.isfinite(amount): + raise SourceDataError(f"{path}:{line}: amount {r['amount']!r} is not finite") + out.append({"id": r["id"], "src": r["src"], "dst": r["dst"], "amount": amount, "at": r["at"]}) + return out + + +def _read_csv(path: str | Path, columns: Sequence[str]) -> list[tuple[int, dict]]: + path = Path(path) + if not path.is_file(): + raise SourceDataError(f"{path}: no such file") + with path.open(newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + missing = [c for c in columns if c not in (reader.fieldnames or [])] + if missing: + raise SourceDataError(f"{path}: missing column(s) {', '.join(missing)}") + rows = [] + # Line 1 is the header, so the first data row is line 2. + for line, row in enumerate(reader, start=2): + for c in columns: + if row.get(c) is None or row[c].strip() == "": + raise SourceDataError(f"{path}:{line}: empty {c!r}") + rows.append((line, row)) + return rows + + +def _reject_duplicate_ids(path: str | Path, rows: list[tuple[int, dict]]) -> None: + # Two rows with one id would MERGE into one node or edge, and the second + # row's values would silently win. Which one is right is a question for + # whoever owns the file, not for the loader. + first_seen: dict[str, int] = {} + for line, r in rows: + if r["id"] in first_seen: + raise SourceDataError(f"{path}:{line}: id {r['id']!r} already on line {first_seen[r['id']]}") + first_seen[r["id"]] = line + + +def batches(rows: Sequence[dict], size: int) -> Iterator[list[dict]]: + if size < 1: + raise ValueError(f"batch size must be at least 1, got {size}") + for start in range(0, len(rows), size): + yield list(rows[start:start + size]) + + +# -------------------------------------------------------------------------- +# The steps +# -------------------------------------------------------------------------- + +#: Run before any load. A unique constraint rather than a plain index: it backs +#: the `{id: ...}` lookups below with an index the same way, and it also makes +#: the engine refuse a second `:Account` with an id that is already taken, so a +#: writer that uses CREATE where it should MERGE fails instead of duplicating. +#: `IF NOT EXISTS` makes the step safe to repeat. Over a graph that already +#: holds a duplicate id the statement is refused, naming the value. +SCHEMA_STATEMENTS = ( + "CREATE CONSTRAINT IF NOT EXISTS FOR (a:Account) REQUIRE a.id IS UNIQUE", +) + +UPSERT_ACCOUNTS = """ +UNWIND $rows AS r +MERGE (a:Account {id: r.id}) +SET a.holder = r.holder, a.country = r.country, a.opened = r.opened +RETURN count(a) AS written +""" + +FOUND_ACCOUNTS = """ +UNWIND $ids AS id +MATCH (a:Account {id: id}) +RETURN collect(a.id) AS found +""" + +# Both endpoints in ONE comma-separated MATCH. Written as two MATCH clauses -- +# which means the same thing -- the planner does not look each one up by id: it +# builds every pair of accounts and then filters, so the cost grows with the +# square of the number of accounts in the graph rather than with the rows. +UPSERT_TRANSFERS = """ +UNWIND $rows AS r +MATCH (s:Account {id: r.src}), (d:Account {id: r.dst}) +MERGE (s)-[t:TRANSFER {id: r.id}]->(d) +SET t.amount = r.amount, t.at = r.at +RETURN count(t) AS written +""" + +#: Each account, found by its id, holding the values the file holds. +CHECK_ACCOUNTS = """ +UNWIND $rows AS r +MATCH (a:Account {id: r.id}) +WHERE a.holder = r.holder AND a.country = r.country AND a.opened = r.opened +RETURN count(a) AS matched +""" + +# Each transfer, between the accounts the file names, holding its values. The +# endpoints are named variables: the engine refuses a property map that is not +# a literal on an anonymous node (`(:Account {id: r.src})`). +CHECK_TRANSFERS = """ +UNWIND $rows AS r +MATCH (s:Account {id: r.src})-[t:TRANSFER]->(d:Account {id: r.dst}) +WHERE t.id = r.id AND t.amount = r.amount AND t.at = r.at +RETURN count(t) AS matched +""" + +# Edge ids have no index to look them up by, so this is one pass over the +# TRANSFER edges, grouped by id -- linear in the edges, where asking id by id +# would be linear in the edges once per id. +DUPLICATED_TRANSFER_IDS = """ +MATCH ()-[t:TRANSFER]->() +WITH t.id AS id, count(t) AS n +WHERE n > 1 +RETURN collect(id) AS duplicated +""" + + +DUPLICATED_ACCOUNT_IDS = """ +MATCH (a:Account) +WITH a.id AS id, count(a) AS n +WHERE n > 1 +RETURN collect(id) AS duplicated +""" + + +def ensure_schema(client: SamyamaHttp) -> int: + for statement in SCHEMA_STATEMENTS: + client.query(statement) + return len(SCHEMA_STATEMENTS) + + +def load_accounts(client: SamyamaHttp, rows: Sequence[dict], batch_size: int) -> int: + written = 0 + for batch in batches(rows, batch_size): + written += client.query(UPSERT_ACCOUNTS, {"rows": batch})["records"][0][0] + if written != len(rows): + raise SamyamaError(f"wrote {written} accounts for {len(rows)} rows") + return written + + +def load_transfers(client: SamyamaHttp, rows: Sequence[dict], batch_size: int) -> int: + # Checked before anything is written. A transfer whose account does not + # exist would otherwise match nothing and be dropped without a word: the + # MATCH finds no row, so the MERGE never runs and the statement succeeds. + endpoints = sorted({r["src"] for r in rows} | {r["dst"] for r in rows}) + found = set(client.scalar(FOUND_ACCOUNTS, {"ids": endpoints})) if endpoints else set() + missing = [i for i in endpoints if i not in found] + if missing: + raise SourceDataError( + f"{len(missing)} account(s) referenced by transfers do not exist: " + + ", ".join(sorted(missing)[:20]) + + (" ..." if len(missing) > 20 else "") + ) + written = 0 + for batch in batches(rows, batch_size): + written += client.query(UPSERT_TRANSFERS, {"rows": batch})["records"][0][0] + if written != len(rows): + raise SamyamaError(f"wrote {written} transfers for {len(rows)} rows") + return written + + +@dataclass +class Check: + name: str + expected: int + found: int + detail: str = field(default="") + + @property + def passed(self) -> bool: + return self.found == self.expected + + +def check_accounts(client: SamyamaHttp, rows: Sequence[dict]) -> list[Check]: + ids = {r["id"] for r in rows} + duplicated = sorted(ids.intersection(client.scalar(DUPLICATED_ACCOUNT_IDS))) + return [ + Check("accounts present with the file's values", len(rows), + client.scalar(CHECK_ACCOUNTS, {"rows": list(rows)})), + # Count duplicates by ID: an extra copy of A must not cancel a missing B. + Check("no account id held by more than one node", 0, len(duplicated), + ", ".join(duplicated[:20])), + ] + + +def check_transfers(client: SamyamaHttp, rows: Sequence[dict]) -> list[Check]: + ids = {r["id"] for r in rows} + # Only this file's ids count against it: a duplicate some other load wrote + # is that load's to answer for. + duplicated = sorted(ids.intersection(client.scalar(DUPLICATED_TRANSFER_IDS))) + return [ + Check("transfers present between the file's accounts with its values", len(rows), + client.scalar(CHECK_TRANSFERS, {"rows": list(rows)})), + # MERGE matches on both endpoints and the id, so a source row whose + # account changed between two runs adds a second edge with the old + # id rather than moving the first. This is what sees it. + Check("no transfer id held by more than one edge", 0, len(duplicated), + ", ".join(duplicated[:20])), + ] + + +def failures(checks: Sequence[Check]) -> list[str]: + return [f"{c.name}: expected {c.expected}, found {c.found}" + (f" ({c.detail})" if c.detail else "") + for c in checks if not c.passed] diff --git a/examples/orchestration/tests/conftest.py b/examples/orchestration/tests/conftest.py new file mode 100644 index 00000000..f1781584 --- /dev/null +++ b/examples/orchestration/tests/conftest.py @@ -0,0 +1,199 @@ +"""A real Samyama server for each test, on spare ports, in a throwaway directory. + +The pipelines are tested against the engine, not a mock of it: what INT-10 +asks for is an integration that has been run, and the defects worth catching +here -- a statement the engine refuses, a MATCH that silently finds nothing, a +parameter that arrives as the wrong type -- are the ones a mock would agree +with. + +The binary is `$SAMYAMA_BIN`, or the repository's release build if that exists. +Neither present is an error, not a skip: a suite that skips when its server is +missing reports green while testing nothing. +""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +import time +import urllib.error +import urllib.request +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + +import pytest + +HERE = Path(__file__).resolve().parent +PIPELINE_DIR = HERE.parent +REPO_ROOT = PIPELINE_DIR.parent.parent + +sys.path.insert(0, str(PIPELINE_DIR)) + +STARTUP_SECONDS = float(os.environ.get("SAMYAMA_STARTUP_SECONDS", "60")) + + +def samyama_bin() -> Path: + configured = os.environ.get("SAMYAMA_BIN") + path = Path(configured) if configured else REPO_ROOT / "target" / "release" / "samyama" + if not path.is_file() or not os.access(path, os.X_OK): + pytest.fail( + f"no Samyama server binary at {path}. Build one with " + "`cargo build --release --bin samyama`, or set SAMYAMA_BIN.", + pytrace=False, + ) + return path + + +def free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@dataclass +class Engine: + url: str + token: Optional[str] + log: Path + + def query(self, cypher: str, params: Optional[dict] = None) -> list: + body = {"query": cypher, "params": params or {}} + headers = {"Content-Type": "application/json"} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + req = urllib.request.Request(f"{self.url}/api/query", json.dumps(body).encode(), headers) + try: + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read())["records"] + except urllib.error.HTTPError as e: + raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8', 'replace')}") from None + + def count(self, cypher: str, params: Optional[dict] = None) -> int: + return self.query(cypher, params)[0][0] + + +def _start(tmp_path: Path, auth_line: Optional[str] = None): + binary = samyama_bin() + data = tmp_path / "samyama-data" + data.mkdir() + http_port, resp_port = free_port(), free_port() + args = [str(binary), "--host", "127.0.0.1", "--port", str(resp_port), + "--http-port", str(http_port), "--data-path", str(data)] + if auth_line: + auth_file = tmp_path / "credentials" + auth_file.write_text(auth_line + "\n") + args += ["--auth-file", str(auth_file)] + log = tmp_path / "samyama.log" + # cwd is the scratch directory: the server resolves some defaults against + # its working directory, and the repository is not the place for them. + proc = subprocess.Popen(args, cwd=tmp_path, stdout=log.open("w"), stderr=subprocess.STDOUT) + url = f"http://127.0.0.1:{http_port}" + deadline = time.monotonic() + STARTUP_SECONDS + while True: + if proc.poll() is not None: + pytest.fail(f"samyama exited with {proc.returncode}:\n{log.read_text()[-4000:]}", pytrace=False) + try: + urllib.request.urlopen(f"{url}/api/status", timeout=2).close() + break + except urllib.error.HTTPError: + break # answering, even if only to say 401 + except (urllib.error.URLError, ConnectionError, TimeoutError): + if time.monotonic() > deadline: + proc.kill() + pytest.fail(f"samyama did not answer on {url} within {STARTUP_SECONDS:.0f}s:\n" + f"{log.read_text()[-4000:]}", pytrace=False) + time.sleep(0.2) + return proc, url, log + + +def _stop(proc: subprocess.Popen) -> None: + proc.terminate() + try: + proc.wait(timeout=20) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + + +@pytest.fixture +def engine(tmp_path): + proc, url, log = _start(tmp_path) + try: + yield Engine(url, None, log) + finally: + _stop(proc) + + +@pytest.fixture +def secured_engine(tmp_path): + """A server that refuses requests without a token, and the token it accepts.""" + # The server's own subcommand writes the credential, so the test does not + # restate how a token is hashed. It prints the file line, then the token. + out = subprocess.run([str(samyama_bin()), "auth-token", "pipeline"], + capture_output=True, text=True, check=True, cwd=tmp_path).stdout + lines = [line for line in out.splitlines() if line and not line.startswith("#")] + credential, token = lines[0], lines[1] + # Write, not admin: a pipeline that loads data has no business managing tenants. + proc, url, log = _start(tmp_path, auth_line=f"{credential}:roles=write") + try: + yield Engine(url, token, log) + finally: + _stop(proc) + + +# --------------------------------------------------------------------------- +# Source files, and what the graph should hold after loading them +# --------------------------------------------------------------------------- + +import samyama_pipeline as sp # noqa: E402 (needs the sys.path entry above) + + +@dataclass +class Source: + accounts_csv: Path + transfers_csv: Path + + @property + def accounts(self) -> list[dict]: + return sp.read_accounts(self.accounts_csv) + + @property + def transfers(self) -> list[dict]: + return sp.read_transfers(self.transfers_csv) + + +@pytest.fixture +def sample() -> Source: + return Source(sp.SAMPLE_DIR / "accounts.csv", sp.SAMPLE_DIR / "transfers.csv") + + +def assert_graph_matches(engine: Engine, source: Source) -> None: + """The graph holds exactly the source files, read back with plain Cypher. + + Deliberately not `samyama_pipeline.check_*`: the pipelines validate with + those, and a test that asked them whether they were right would pass + whenever they were wrong in the same way twice. + """ + accounts, transfers = source.accounts, source.transfers + assert engine.count("MATCH (a:Account) RETURN count(a)") == len(accounts) + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == len(transfers) + + got_accounts = { + r[0]: {"holder": r[1], "country": r[2], "opened": r[3]} + for r in engine.query("MATCH (a:Account) RETURN a.id, a.holder, a.country, a.opened") + } + assert got_accounts == { + a["id"]: {"holder": a["holder"], "country": a["country"], "opened": a["opened"]} + for a in accounts + } + + got_transfers = { + r[0]: (r[1], r[2], r[3], r[4]) + for r in engine.query( + "MATCH (s:Account)-[t:TRANSFER]->(d:Account) RETURN t.id, s.id, d.id, t.amount, t.at") + } + assert got_transfers == {t["id"]: (t["src"], t["dst"], t["amount"], t["at"]) for t in transfers} diff --git a/examples/orchestration/tests/test_airflow.py b/examples/orchestration/tests/test_airflow.py new file mode 100644 index 00000000..b5af348a --- /dev/null +++ b/examples/orchestration/tests/test_airflow.py @@ -0,0 +1,111 @@ +"""The reference Airflow DAG, run by Airflow against a real server. + +`dag.test()` runs every task in this process through Airflow's own task +runner: params are resolved, the Connection is looked up through Airflow's +secrets backends, and a task's failure is classified the way a scheduler would +classify it. It needs a metadata database, so the session gets a throwaway one. + +Airflow reads its configuration when it is first imported, so nothing here +imports it at module level: the environment is set first, in the fixture. +""" + +from __future__ import annotations + +import csv +import importlib +import json +import os +import subprocess +import sys + +import pytest + +from conftest import PIPELINE_DIR, Source, assert_graph_matches + +DAG_ID = "samyama_accounts_transfers" + + +@pytest.fixture(scope="session") +def airflow_dag(tmp_path_factory): + home = tmp_path_factory.mktemp("airflow-home") + env = { + "AIRFLOW_HOME": str(home), + "AIRFLOW__CORE__LOAD_EXAMPLES": "False", + "AIRFLOW__CORE__DAGS_FOLDER": str(PIPELINE_DIR), + } + os.environ.update(env) + migrate = subprocess.run([sys.executable, "-m", "airflow", "db", "migrate"], + capture_output=True, text=True, env={**os.environ, **env}) + assert migrate.returncode == 0, migrate.stdout[-3000:] + migrate.stderr[-3000:] + module = importlib.import_module("samyama_airflow_dag") + return module.samyama_accounts_transfers() + + +def connect(monkeypatch, engine, conn_id="samyama_default"): + """Point an Airflow Connection at the server, the way an operator would.""" + host_port = engine.url.split("://", 1)[1] + host, port = host_port.rsplit(":", 1) + conn = {"conn_type": "http", "host": host, "port": int(port), "schema": "http"} + if engine.token: + conn["password"] = engine.token + monkeypatch.setenv(f"AIRFLOW_CONN_{conn_id.upper()}", json.dumps(conn)) + + +def run(dag, source: Source, **conf): + run_conf = {"accounts_csv": str(source.accounts_csv), + "transfers_csv": str(source.transfers_csv), **conf} + dr = dag.test(run_conf=run_conf) + states = {ti.task_id: (ti.state, ti.try_number) for ti in dr.get_task_instances()} + return dr.state, states + + +def test_the_dag_loads_the_sample_and_validates_it(airflow_dag, engine, sample, monkeypatch): + connect(monkeypatch, engine) + state, tasks = run(airflow_dag, sample, batch_size=7) + assert state == "success", tasks + assert {t: s for t, (s, _) in tasks.items()} == { + "schema": "success", "accounts": "success", "transfers": "success", "validate": "success"} + assert_graph_matches(engine, sample) + + +def test_running_the_dag_twice_leaves_the_graph_as_once(airflow_dag, engine, sample, monkeypatch): + connect(monkeypatch, engine) + assert run(airflow_dag, sample, batch_size=7)[0] == "success" + assert run(airflow_dag, sample, batch_size=13)[0] == "success" + assert_graph_matches(engine, sample) + + +def test_bad_input_fails_the_run_without_retrying(airflow_dag, engine, sample, monkeypatch, tmp_path): + connect(monkeypatch, engine) + bad = tmp_path / "transfers.csv" + with bad.open("w", newline="") as f: + w = csv.writer(f, lineterminator="\n") + w.writerow(["id", "src", "dst", "amount", "at"]) + w.writerow(["TX-1", sample.accounts[0]["id"], "ACC-NOWHERE", "1.00", "2026-01-01T00:00:00"]) + state, tasks = run(airflow_dag, Source(sample.accounts_csv, bad)) + assert state == "failed" + assert tasks["transfers"] == ("failed", 1), tasks # one try: the DAG allows three retries + assert tasks["validate"][0] == "upstream_failed" + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == 0 + + +def test_the_token_comes_from_the_connection(airflow_dag, secured_engine, sample, monkeypatch): + connect(monkeypatch, secured_engine) + state, tasks = run(airflow_dag, sample) + assert state == "success", tasks + assert_graph_matches(secured_engine, sample) + + +def test_a_connection_without_the_token_fails_without_retrying(airflow_dag, secured_engine, sample, + monkeypatch): + connect(monkeypatch, secured_engine.__class__(secured_engine.url, None, secured_engine.log)) + state, tasks = run(airflow_dag, sample) + assert state == "failed" + assert tasks["schema"] == ("failed", 1), tasks + + +def test_the_connection_id_is_a_param(airflow_dag, engine, sample, monkeypatch): + connect(monkeypatch, engine, conn_id="graph_prod_like") + state, tasks = run(airflow_dag, sample, samyama_conn_id="graph_prod_like") + assert state == "success", tasks + assert_graph_matches(engine, sample) diff --git a/examples/orchestration/tests/test_dagster.py b/examples/orchestration/tests/test_dagster.py new file mode 100644 index 00000000..dd1beba9 --- /dev/null +++ b/examples/orchestration/tests/test_dagster.py @@ -0,0 +1,158 @@ +"""The reference Dagster assets and checks, run by Dagster against a real server. + +`materialize` executes the assets and their checks in this process through +Dagster's own executor, with the resources an operator would configure. +""" + +from __future__ import annotations + +import csv + +import dagster as dg +import pytest + +import samyama_dagster as sd +from conftest import Source, assert_graph_matches + + +def resources(engine, source: Source, batch_size: int = 500, token=None) -> dict: + return { + "samyama": sd.SamyamaResource(url=engine.url, token=token), + "source_files": sd.SourceFiles(accounts_csv=str(source.accounts_csv), + transfers_csv=str(source.transfers_csv), + batch_size=batch_size), + } + + +def run(res: dict) -> dg.ExecuteInProcessResult: + return dg.materialize([*sd.ASSETS, *sd.CHECKS], resources=res, raise_on_error=False) + + +def checks(result: dg.ExecuteInProcessResult) -> dict: + return {e.check_name: e.passed for e in result.get_asset_check_evaluations()} + + +def materialized(result: dg.ExecuteInProcessResult) -> set: + return {e.asset_key.to_user_string() for e in result.get_asset_materialization_events()} + + +def test_the_assets_load_the_sample_and_the_checks_pass(engine, sample): + result = run(resources(engine, sample, batch_size=7)) + assert result.success + assert materialized(result) == {"samyama_schema", "accounts", "transfers"} + assert checks(result) == {"accounts_match_source": True, "transfers_match_source": True} + assert_graph_matches(engine, sample) + + +def test_materializing_twice_leaves_the_graph_as_once(engine, sample): + assert run(resources(engine, sample, batch_size=7)).success + assert run(resources(engine, sample, batch_size=13)).success + assert_graph_matches(engine, sample) + + +def test_bad_input_fails_the_asset_without_retrying(engine, sample, tmp_path): + bad = tmp_path / "transfers.csv" + with bad.open("w", newline="") as f: + w = csv.writer(f, lineterminator="\n") + w.writerow(["id", "src", "dst", "amount", "at"]) + w.writerow(["TX-1", sample.accounts[0]["id"], "ACC-NOWHERE", "1.00", "2026-01-01T00:00:00"]) + result = run(resources(engine, Source(sample.accounts_csv, bad))) + assert not result.success + assert "transfers" not in materialized(result) + failures = [e for e in result.all_events if e.event_type_value == "STEP_FAILURE"] + assert [e.step_key for e in failures] == ["transfers"] + assert "ACC-NOWHERE" in failures[0].event_specific_data.error.message + assert not [e for e in result.all_events if e.event_type_value == "STEP_UP_FOR_RETRY"] + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == 0 + + +def test_a_failed_accounts_check_blocks_the_transfers(engine, sample, monkeypatch): + # Make the graph disagree with the file between the load and its check: + # the check must fail, and being blocking, stop `transfers` from loading. + real = sd.sp.load_accounts + + def load_then_corrupt(client, rows, batch_size): + written = real(client, rows, batch_size) + client.query("MATCH (a:Account {id: $id}) SET a.holder = 'tampered'", {"id": rows[0]["id"]}) + return written + + monkeypatch.setattr(sd.sp, "load_accounts", load_then_corrupt) + result = run(resources(engine, sample)) + assert not result.success + assert checks(result) == {"accounts_match_source": False} + assert "transfers" not in materialized(result) + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == 0 + + +def test_the_token_comes_from_the_resource(secured_engine, sample): + result = run(resources(secured_engine, sample, token=secured_engine.token)) + assert result.success + assert_graph_matches(secured_engine, sample) + + +def test_a_resource_without_the_token_fails_without_retrying(secured_engine, sample): + result = run(resources(secured_engine, sample, token=None)) + assert not result.success + failures = [e for e in result.all_events if e.event_type_value == "STEP_FAILURE"] + assert [e.step_key for e in failures] == ["samyama_schema"] + assert "401" in failures[0].event_specific_data.error.message + assert not [e for e in result.all_events if e.event_type_value == "STEP_UP_FOR_RETRY"] + + +def test_the_definitions_read_the_server_from_the_environment(engine, sample, monkeypatch): + # The shipped definitions, not a test-built copy: the URL arrives through EnvVar. + monkeypatch.delenv("SAMYAMA_TOKEN", raising=False) + monkeypatch.setenv("SAMYAMA_URL", engine.url) + job = sd.build_definitions().resolve_implicit_global_asset_job_def() + result = job.execute_in_process(raise_on_error=False) + assert result.success + assert_graph_matches(engine, sample) + + +def test_the_definitions_read_the_token_from_the_environment(secured_engine, sample, monkeypatch): + monkeypatch.setenv("SAMYAMA_URL", secured_engine.url) + # Code locations may load before workers receive their secret environment. + monkeypatch.delenv("SAMYAMA_TOKEN", raising=False) + job = sd.build_definitions().resolve_implicit_global_asset_job_def() + monkeypatch.setenv("SAMYAMA_TOKEN", secured_engine.token) + result = job.execute_in_process(raise_on_error=False) + assert result.success + assert_graph_matches(secured_engine, sample) + + +@pytest.mark.parametrize("check_name", ["check_accounts", "check_transfers"]) +def test_transient_check_failure_is_retried_then_succeeds(engine, sample, monkeypatch, check_name): + real = getattr(sd.sp, check_name) + attempts = [] + + def transient_once(*args, **kwargs): + attempts.append(1) + if len(attempts) == 1: + raise sd.sp.SamyamaUnavailable("HTTP 503: temporarily unavailable") + return real(*args, **kwargs) + + monkeypatch.setattr(sd.sp, check_name, transient_once) + # Exercise Dagster's real retry scheduler without waiting thirty seconds. + monkeypatch.setattr(dg.RetryPolicy, "calculate_delay", lambda self, attempt_num: 0) + result = run(resources(engine, sample)) + assert result.success + assert len(attempts) == 2 + retries = [e for e in result.all_events if e.event_type_value == "STEP_UP_FOR_RETRY"] + assert len(retries) == 1 + assert checks(result) == {"accounts_match_source": True, "transfers_match_source": True} + assert_graph_matches(engine, sample) + + +@pytest.mark.parametrize("check_name", ["check_accounts", "check_transfers"]) +def test_permanent_check_failure_is_not_retried(engine, sample, monkeypatch, check_name): + attempts = [] + + def rejected(*args, **kwargs): + attempts.append(1) + raise sd.sp.SamyamaRejected(401, "missing credential") + + monkeypatch.setattr(sd.sp, check_name, rejected) + result = run(resources(engine, sample)) + assert not result.success + assert len(attempts) == 1 + assert not [e for e in result.all_events if e.event_type_value == "STEP_UP_FOR_RETRY"] diff --git a/examples/orchestration/tests/test_docs.py b/examples/orchestration/tests/test_docs.py new file mode 100644 index 00000000..66931fb8 --- /dev/null +++ b/examples/orchestration/tests/test_docs.py @@ -0,0 +1,43 @@ +"""The Cypher the orchestration docs show is the Cypher the pipelines run. + +CH-DOC-EXEC checks that a documented statement parses. It cannot see that the +statement on the page is no longer the one in `samyama_pipeline.py` -- and a +reader copies the page, not the module. +""" + +from __future__ import annotations + +import re + +import pytest + +import samyama_pipeline as sp +from conftest import REPO_ROOT + +DOCS = ["ORCHESTRATION.md", "ORCHESTRATION-AIRFLOW.md", "ORCHESTRATION-DAGSTER.md"] + + +def normalise(text: str) -> str: + return " ".join(text.split()) + + +def pipeline_statements() -> set[str]: + return {normalise(v) for k, v in vars(sp).items() + if k.isupper() and isinstance(v, str) and ("MATCH" in v or "MERGE" in v)} | { + normalise(s) for s in sp.SCHEMA_STATEMENTS} + + +@pytest.mark.parametrize("doc", DOCS) +def test_every_cypher_block_is_a_statement_the_pipeline_runs(doc): + text = (REPO_ROOT / "docs" / doc).read_text(encoding="utf-8") + blocks = re.findall(r"```cypher\n(.*?)```", text, flags=re.S) + known = pipeline_statements() + for block in blocks: + assert normalise(block) in known, ( + f"docs/{doc} shows a statement samyama_pipeline.py does not run:\n{block}") + + +def test_the_overview_shows_at_least_one_statement(): + # Without this, deleting every block would pass the test above. + text = (REPO_ROOT / "docs" / "ORCHESTRATION.md").read_text(encoding="utf-8") + assert "```cypher" in text diff --git a/examples/orchestration/tests/test_pipeline.py b/examples/orchestration/tests/test_pipeline.py new file mode 100644 index 00000000..df94e142 --- /dev/null +++ b/examples/orchestration/tests/test_pipeline.py @@ -0,0 +1,332 @@ +"""The shared load steps, against a real server and with no orchestrator.""" + +from __future__ import annotations + +import csv +from pathlib import Path + +import pytest + +import samyama_pipeline as sp +from conftest import Source, assert_graph_matches + + +def write_csv(path: Path, columns, rows) -> Path: + with path.open("w", newline="", encoding="utf-8") as f: + w = csv.writer(f, lineterminator="\n") + w.writerow(columns) + w.writerows(rows) + return path + + +def load_all(client, source: Source, batch_size: int) -> None: + sp.ensure_schema(client) + sp.load_accounts(client, source.accounts, batch_size) + sp.load_transfers(client, source.transfers, batch_size) + + +# --- reading ----------------------------------------------------------------- + +def test_a_duplicate_id_names_both_lines(tmp_path): + f = write_csv(tmp_path / "a.csv", sp.ACCOUNT_COLUMNS, + [["A1", "x", "IN", "2020-01-01"], ["A2", "y", "IN", "2020-01-01"], + ["A1", "z", "GB", "2021-01-01"]]) + with pytest.raises(sp.SourceDataError, match=r"a\.csv:4: id 'A1' already on line 2"): + sp.read_accounts(f) + + +def test_a_missing_column_is_named(tmp_path): + f = write_csv(tmp_path / "t.csv", ["id", "src", "dst", "at"], [["T1", "A1", "A2", "x"]]) + with pytest.raises(sp.SourceDataError, match="missing column.*amount"): + sp.read_transfers(f) + + +@pytest.mark.parametrize("amount", ["twelve", "nan", "inf"]) +def test_an_amount_that_is_not_a_finite_number_is_refused(tmp_path, amount): + f = write_csv(tmp_path / "t.csv", sp.TRANSFER_COLUMNS, [["T1", "A1", "A2", amount, "x"]]) + with pytest.raises(sp.SourceDataError, match=r"t\.csv:2: amount"): + sp.read_transfers(f) + + +def test_an_empty_field_is_refused(tmp_path): + f = write_csv(tmp_path / "a.csv", sp.ACCOUNT_COLUMNS, [["A1", " ", "IN", "2020-01-01"]]) + with pytest.raises(sp.SourceDataError, match="empty 'holder'"): + sp.read_accounts(f) + + +def test_batches_cover_every_row_once(): + rows = [{"id": i} for i in range(23)] + got = list(sp.batches(rows, 5)) + assert [len(b) for b in got] == [5, 5, 5, 5, 3] + assert [r for b in got for r in b] == rows + with pytest.raises(ValueError): + list(sp.batches(rows, 0)) + + +# --- loading ----------------------------------------------------------------- + +def test_the_sample_loads_exactly(engine, sample): + # A batch size that divides neither file, so the last batch of each is short. + load_all(sp.SamyamaHttp(engine.url), sample, batch_size=7) + assert_graph_matches(engine, sample) + + +def test_loading_twice_is_loading_once(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, batch_size=7) + load_all(client, sample, batch_size=11) + assert_graph_matches(engine, sample) + + +def test_values_are_stored_as_given_not_read_as_cypher(engine, tmp_path): + hostile = [ + "O'Brien", + 'say "hi"', + "x'}) DETACH DELETE (n) //", + "back\\slash", + "$rows", + "名前 — ünïcödé", + ] + accounts = write_csv(tmp_path / "a.csv", sp.ACCOUNT_COLUMNS, + [[f"A{i}", h, "IN", "2020-01-01"] for i, h in enumerate(hostile)]) + transfers = write_csv(tmp_path / "t.csv", sp.TRANSFER_COLUMNS, + [["T0", "A0", "A1", "1.5", "2026-01-01T00:00:00"]]) + source = Source(accounts, transfers) + load_all(sp.SamyamaHttp(engine.url), source, batch_size=2) + assert_graph_matches(engine, source) + + +def test_a_transfer_to_an_unknown_account_writes_nothing(engine, sample, tmp_path): + client = sp.SamyamaHttp(engine.url) + sp.ensure_schema(client) + sp.load_accounts(client, sample.accounts, 500) + + rows = [[t["id"], t["src"], t["dst"], t["amount"], t["at"]] for t in sample.transfers] + rows.append(["TX-ORPHAN", sample.accounts[0]["id"], "ACC-DOES-NOT-EXIST", "1.00", + "2026-01-01T00:00:00"]) + bad = Source(sample.accounts_csv, write_csv(tmp_path / "t.csv", sp.TRANSFER_COLUMNS, rows)) + + with pytest.raises(sp.SourceDataError, match="ACC-DOES-NOT-EXIST"): + sp.load_transfers(client, bad.transfers, 7) + # Checked before the first batch, so not even the valid rows went in. + assert engine.count("MATCH ()-[t:TRANSFER]->() RETURN count(t)") == 0 + + +# --- validating -------------------------------------------------------------- + +def test_the_checks_pass_on_a_correct_load(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + checks = sp.check_accounts(client, sample.accounts) + sp.check_transfers(client, sample.transfers) + assert sp.failures(checks) == [] + # Not vacuous: the checks that count what is present expected the files' + # rows, not zero of them. (The duplicate check expects zero by design.) + present = {c.name: c.expected for c in checks if "present" in c.name} + assert present == { + "accounts present with the file's values": len(sample.accounts), + "transfers present between the file's accounts with its values": len(sample.transfers), + } + + +def test_the_checks_see_a_changed_value(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + victim = sample.transfers[0] + engine.query("MATCH ()-[t:TRANSFER {id: $id}]->() SET t.amount = $a", + {"id": victim["id"], "a": victim["amount"] + 0.01}) + failed = sp.failures(sp.check_transfers(client, sample.transfers)) + assert len(failed) == 1 and "values" in failed[0], failed + + +def _create_account(engine, a): + engine.query("CREATE (:Account {id: $id, holder: $h, country: $c, opened: $o})", + {"id": a["id"], "h": a["holder"], "c": a["country"], "o": a["opened"]}) + + +def test_after_the_schema_step_the_engine_refuses_a_duplicate_account(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + with pytest.raises(Exception, match="Constraint violation"): + _create_account(engine, sample.accounts[0]) + assert engine.count("MATCH (a:Account) RETURN count(a)") == len(sample.accounts) + + +def test_the_schema_step_refuses_a_graph_that_already_holds_a_duplicate(engine, sample): + _create_account(engine, sample.accounts[0]) + _create_account(engine, sample.accounts[0]) + with pytest.raises(sp.SamyamaRejected, match="duplicate value"): + sp.ensure_schema(sp.SamyamaHttp(engine.url)) + + +def test_the_checks_see_a_duplicate_node_where_no_constraint_stops_it(engine, sample): + # A graph some other writer populated, without the schema step. + client = sp.SamyamaHttp(engine.url) + sp.load_accounts(client, sample.accounts, 500) + _create_account(engine, sample.accounts[0]) + failed = sp.failures(sp.check_accounts(client, sample.accounts)) + assert any("more than one node" in f for f in failed), failed + + +def test_the_checks_see_a_duplicate_edge(engine, sample): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + t = sample.transfers[0] + engine.query("MATCH (s:Account {id: $s}), (d:Account {id: $d}) " + "CREATE (s)-[:TRANSFER {id: $id, amount: $a, at: $at}]->(d)", + {"s": t["src"], "d": t["dst"], "id": t["id"], "a": t["amount"], "at": t["at"]}) + failed = sp.failures(sp.check_transfers(client, sample.transfers)) + # An exact copy -- same id, endpoints and values -- trips both checks: + # it is one row too many present, and its id is held twice. + n = len(sample.transfers) + assert failed == [ + f"transfers present between the file's accounts with its values: expected {n}, found {n + 1}", + f"no transfer id held by more than one edge: expected 0, found 1 ({t['id']})", + ] + + +def test_a_transfer_whose_account_changed_between_runs_is_reported(engine, sample, tmp_path): + client = sp.SamyamaHttp(engine.url) + load_all(client, sample, 500) + # The same transfer id, re-sent with a different destination. + moved = dict(sample.transfers[0]) + moved["dst"] = next(a["id"] for a in sample.accounts if a["id"] not in (moved["src"], moved["dst"])) + rows = [moved] + sample.transfers[1:] + sp.load_transfers(client, rows, 500) + failed = sp.failures(sp.check_transfers(client, rows)) + assert failed == [f"no transfer id held by more than one edge: expected 0, found 1 ({moved['id']})"] + + +# --- failing the right way --------------------------------------------------- + +def test_a_refused_statement_is_not_retryable(engine): + with pytest.raises(sp.SamyamaRejected) as e: + sp.SamyamaHttp(engine.url).query("THIS IS NOT CYPHER") + assert e.value.status == 400 + + +def test_an_unreachable_server_is_retryable(): + from conftest import free_port + client = sp.SamyamaHttp(f"http://127.0.0.1:{free_port()}", timeout_seconds=2) + with pytest.raises(sp.SamyamaUnavailable) as e: + client.query("RETURN 1") + assert sp.is_retryable(e.value) + + +@pytest.mark.parametrize("error, retryable", [ + (sp.SamyamaUnavailable("HTTP 503: writes refused"), True), + (sp.SamyamaRejected(400, "syntax"), False), + (sp.SamyamaRejected(401, "no credential"), False), + (sp.SamyamaError("wrote 59 transfers for 60 rows"), False), + (sp.SourceDataError("a.csv:3: empty 'holder'"), False), +]) +def test_only_an_unavailable_server_is_worth_retrying(error, retryable): + assert sp.is_retryable(error) == retryable + + +@pytest.mark.parametrize("status, retryable", [ + (400, False), (401, False), (403, False), (404, False), + (408, True), (429, True), (500, True), (503, True), +]) +def test_an_http_status_is_classified_as_the_scheduler_needs(status, retryable): + # The engine does not answer every status on demand, so a stub does: what + # is under test is the client's reading of the status, not the server. + import http.server + import json as _json + import threading + + class Stub(http.server.BaseHTTPRequestHandler): + def do_POST(self): + self.rfile.read(int(self.headers["Content-Length"])) + body = _json.dumps({"error": f"stub {status}"}).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + server = http.server.HTTPServer(("127.0.0.1", 0), Stub) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + with pytest.raises(sp.SamyamaError) as e: + sp.SamyamaHttp(f"http://127.0.0.1:{server.server_port}").query("RETURN 1") + assert sp.is_retryable(e.value) == retryable + assert f"stub {status}" in str(e.value), "the server's own message is kept" + finally: + server.shutdown() + server.server_close() + + +def test_a_missing_or_wrong_token_is_refused_and_the_right_one_works(secured_engine, sample): + for token in (None, "0" * 64): + with pytest.raises(sp.SamyamaRejected) as e: + sp.SamyamaHttp(secured_engine.url, token=token).query("RETURN 1") + assert e.value.status == 401 + load_all(sp.SamyamaHttp(secured_engine.url, token=secured_engine.token), sample, 500) + assert_graph_matches(secured_engine, sample) + + +def test_duplicate_and_missing_account_cannot_cancel_each_other(engine, sample): + a, b = sample.accounts[:2] + _create_account(engine, a) + _create_account(engine, a) + checked = sp.check_accounts(sp.SamyamaHttp(engine.url), [a, b]) + # The old total-count check returned two for both checks and passed. + assert checked[0].passed + failed = sp.failures(checked) + assert len(failed) == 1 and "more than one node" in failed[0] + assert a["id"] in failed[0] + + +@pytest.mark.parametrize("status", [301, 302, 303, 307, 308]) +def test_redirect_cannot_forward_write_credentials(status): + import http.server + import json + import threading + + source_auth, redirected_auth = [], [] + + class Receiver(http.server.BaseHTTPRequestHandler): + def do_GET(self): + redirected_auth.append(self.headers.get("Authorization")) + body = json.dumps({"records": [[1]]}).encode() + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + do_POST = do_GET + + def log_message(self, *args): + pass + + receiver = http.server.HTTPServer(("127.0.0.1", 0), Receiver) + + class Redirect(http.server.BaseHTTPRequestHandler): + def do_POST(self): + source_auth.append(self.headers.get("Authorization")) + self.rfile.read(int(self.headers["Content-Length"])) + self.send_response(status) + self.send_header("Location", f"http://127.0.0.1:{receiver.server_port}/capture") + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *args): + pass + + source = http.server.HTTPServer(("127.0.0.1", 0), Redirect) + for server in (source, receiver): + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + client = sp.SamyamaHttp(f"http://127.0.0.1:{source.server_port}", token="test-only-token") + with pytest.raises(sp.SamyamaRejected) as error: + client.query("RETURN 1") + assert error.value.status == status + assert source_auth == ["Bearer test-only-token"] + assert redirected_auth == [] + finally: + for server in (source, receiver): + server.shutdown() + server.server_close() diff --git a/src/query/executor/operator.rs b/src/query/executor/operator.rs index 2355fdb2..e03d8921 100644 --- a/src/query/executor/operator.rs +++ b/src/query/executor/operator.rs @@ -16958,6 +16958,14 @@ pub struct MatchMergeEdgeOperator { /// `source -> target` and MERGE created a duplicate beside an existing /// `target -> source` (#938). edges_to_merge: Vec<(String, String, EdgeType, HashMap, Option, bool)>, + /// The non-literal half of each relationship's property map, by position + /// in `edges_to_merge`: `{id: r.id}` from an UNWIND, `{k: a.name}` from a + /// bound node. `edges_to_merge` carries only the literals, and these were + /// never handed over, so `UNWIND $rows AS r MATCH (a), (b) MERGE + /// (a)-[:T {id: r.id}]->(b)` matched any `:T` between the pair and created + /// edges without the property -- two rows for one pair wrote one edge, + /// with no `id`, and reported success. #642 fixed the node half. + property_exprs: Vec>>, /// `ON CREATE SET` property and whole-entity writes, in the order written; /// see `MergeOperator` (#874, #1919). on_create_writes: Vec, @@ -16980,6 +16988,7 @@ impl MatchMergeEdgeOperator { Self { input, edges_to_merge, + property_exprs: Vec::new(), on_create_writes: writes(on_create_set), on_match_writes: writes(on_match_set), done: false, @@ -17003,6 +17012,12 @@ impl MatchMergeEdgeOperator { self } + /// Attach each relationship's non-literal properties, in the order of + /// `edges_to_merge`. A missing entry means the pattern has none. + pub fn with_property_exprs(mut self, exprs: Vec>>) -> Self { + self.property_exprs = exprs; + self + } /// The `ON CREATE` / `ON MATCH` writes, applied in exactly this order /// (#1919). Replaces whatever `new` and `with_entity_sets` were given. pub fn with_ordered_sets(mut self, on_create: Vec, on_match: Vec) -> Self { @@ -17065,9 +17080,22 @@ impl PhysicalOperator for MatchMergeEdgeOperator { fn next_mut(&mut self, store: &mut GraphStore, tenant_id: &str) -> ExecutionResult> { if !self.done { while let Some(record) = self.input.next_mut(store, tenant_id)? { - for (source_var, target_var, edge_type, properties, edge_var, undirected) in - &self.edges_to_merge + for (i, (source_var, target_var, edge_type, literal_properties, edge_var, undirected)) in + self.edges_to_merge.iter().enumerate() { + // Resolved against this row, and the one map used both to + // match and to create -- the property #642 holds to for + // nodes: a MERGE that searched on one set of values and + // wrote another would create an edge its own pattern + // could not find, and running it twice would make two. + let resolved = MergeOperator::resolved_props( + Some(literal_properties), + self.property_exprs.get(i).and_then(|e| e.as_ref()), + &record, + store, + )? + .unwrap_or_default(); + let properties = &resolved; let source_id = match record.get(source_var).and_then(|v| v.node_id()) { Some(id) => id, None => continue, @@ -17200,12 +17228,16 @@ impl PhysicalOperator for MatchMergeEdgeOperator { let mut parts: Vec = self .edges_to_merge .iter() - .map(|(source, target, edge_type, properties, edge_var, undirected)| { + .enumerate() + .map(|(i, (source, target, edge_type, properties, edge_var, undirected))| { let edge = format!( "[{}:{}{}]", edge_var.as_deref().unwrap_or(""), edge_type.as_str(), - format_properties(Some(properties), None) + format_properties( + Some(properties), + self.property_exprs.get(i).and_then(|e| e.as_ref()) + ) ); if *undirected { format!("({})-{}-({})", source, edge, target) diff --git a/src/query/executor/planner.rs b/src/query/executor/planner.rs index cc2d7e29..31eb4e00 100644 --- a/src/query/executor/planner.rs +++ b/src/query/executor/planner.rs @@ -9443,6 +9443,8 @@ impl QueryPlanner { // Extract edge patterns from MERGE clause let mut edges_to_merge = Vec::new(); + // Each edge's non-literal properties, by position in `edges_to_merge`. + let mut edge_property_exprs = Vec::new(); // `MERGE p = (a)-[:R]->(b)` binds `p` (#876). An anonymous // relationship inside a named path is given a synthetic handle, for // the same reason `CREATE` gives one to an anonymous node: the path @@ -9491,6 +9493,7 @@ impl QueryPlanner { edge_var, undirected, )); + edge_property_exprs.push(edge.property_exprs.clone()); } current_var = target_var; } @@ -9528,7 +9531,8 @@ impl QueryPlanner { use crate::query::executor::operator::MatchMergeEdgeOperator; operator = Box::new( MatchMergeEdgeOperator::new(operator, edges_to_merge, Vec::new(), Vec::new()) - .with_ordered_sets(on_create, on_match), + .with_ordered_sets(on_create, on_match) + .with_property_exprs(edge_property_exprs), ); if !merge_named_paths.is_empty() { operator = Box::new(crate::query::executor::operator::BindPathOperator::new( diff --git a/tests/ci_gate_enumeration.rs b/tests/ci_gate_enumeration.rs index 0aad12d3..6acb256e 100644 --- a/tests/ci_gate_enumeration.rs +++ b/tests/ci_gate_enumeration.rs @@ -36,11 +36,18 @@ const AGGREGATOR: &str = "ci-ok"; /// that is in neither fails. So an entry cannot outlive the decision it records, /// and a new suite cannot quietly escape the gate -- escaping it means adding a /// line here, with a reason, in a diff someone reads. -const UNGATED: &[(&str, &str)] = &[( - "gpml", - "conformance has never gated a merge; requiring it is a policy decision, \ - separate from folding it into this workflow (#1865)", -)]; +const UNGATED: &[(&str, &str)] = &[ + ( + "gpml", + "conformance has never gated a merge; requiring it is a policy decision, \ + separate from folding it into this workflow (#1865)", + ), + ( + "orchestration", + "a new suite that installs Airflow and Dagster from PyPI; making it block \ + every merge is a policy decision, separate from adding it (#1715)", + ), +]; fn jobs(doc: &serde_yaml::Value) -> Vec { doc["jobs"] diff --git a/tests/merge_row_properties.rs b/tests/merge_row_properties.rs index 4115fa8b..2e007055 100644 --- a/tests/merge_row_properties.rs +++ b/tests/merge_row_properties.rs @@ -101,3 +101,98 @@ fn merge_between_two_bound_nodes_still_reuses_them() { assert_eq!(count(&store, "MATCH (n:B) RETURN n"), 1); assert_eq!(count(&store, "MATCH ()-[r:R]->() RETURN r"), 1); } + +// --------------------------------------------------------------------------- +// The relationship half: a MERGE that wires an edge between endpoints a MATCH +// already bound, keyed on a value from the row. +// +// ```cypher +// UNWIND $rows AS r MATCH (s {id: r.src}), (d {id: r.dst}) MERGE (s)-[:T {id: r.id}]->(d) +// ``` +// +// is the bulk edge-upsert idiom, and it goes to `MatchMergeEdgeOperator`, +// which was only ever handed the literal half of the property map. The +// expressions were dropped: MERGE matched any `:T` between the pair, and when +// it created one, created it without the property. Two rows for one pair wrote +// ONE edge with no `id`, and the statement reported success. Found loading the +// INT-10 reference pipeline, where 60 transfers became 55 edges (#1715). +// --------------------------------------------------------------------------- + +#[test] +fn an_edge_keyed_on_the_row_is_one_edge_per_distinct_key_between_bound_nodes() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (:A {id: 1}), (:B {id: 1})"); + let upsert = "UNWIND [{id: 't1'}, {id: 't2'}, {id: 't1'}] AS r \ + MATCH (a:A {id: 1}), (b:B {id: 1}) MERGE (a)-[:T {id: r.id}]->(b)"; + run(&mut store, upsert); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2, "'t1' twice is still one edge"); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 't1'}]->() RETURN t"), 1, "the key was written"); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 't2'}]->() RETURN t"), 1); + + run(&mut store, upsert); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2, "still idempotent"); +} + +#[test] +fn an_edge_with_a_different_key_is_not_a_match() { + // The silent half of the defect: the existing edge was "matched" whatever + // its key, so a later SET overwrote it instead of creating a second. + let mut store = GraphStore::new(); + run(&mut store, "CREATE (a:A {id: 1})-[:T {id: 'old', amount: 1}]->(b:B {id: 1})"); + run( + &mut store, + "UNWIND [{id: 'new', amount: 2}] AS r MATCH (a:A {id: 1}), (b:B {id: 1}) \ + MERGE (a)-[t:T {id: r.id}]->(b) SET t.amount = r.amount", + ); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 'old', amount: 1}]->() RETURN t"), 1, "untouched"); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 'new', amount: 2}]->() RETURN t"), 1); +} + +#[test] +fn an_edge_can_key_on_a_property_of_another_bound_node() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (:A {id: 1}), (:B {id: 1}), (:Src {k: 'x'}), (:Src {k: 'y'})"); + let merge = "MATCH (s:Src) MATCH (a:A {id: 1}), (b:B {id: 1}) MERGE (a)-[:T {k: s.k}]->(b)"; + run(&mut store, merge); + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2); + assert_eq!(count(&store, "MATCH ()-[t:T {k: 'x'}]->() RETURN t"), 1); + assert_eq!(count(&store, "MATCH ()-[t:T {k: 'y'}]->() RETURN t"), 1); +} + +#[test] +fn literal_and_row_properties_on_one_edge_must_both_match() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (a:A {id: 1})-[:T {kind: 'other', id: 't1'}]->(b:B {id: 1})"); + let merge = "UNWIND ['t1'] AS x MATCH (a:A {id: 1}), (b:B {id: 1}) \ + MERGE (a)-[:T {kind: 'wire', id: x}]->(b)"; + run(&mut store, merge); + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2, "kind differs, so no match"); + assert_eq!(count(&store, "MATCH ()-[t:T {kind: 'wire', id: 't1'}]->() RETURN t"), 1); +} + +#[test] +fn on_create_and_on_match_see_the_row_keyed_edge() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (:A {id: 1}), (:B {id: 1})"); + let merge = "UNWIND ['t1'] AS x MATCH (a:A {id: 1}), (b:B {id: 1}) \ + MERGE (a)-[t:T {id: x}]->(b) ON CREATE SET t.created = true ON MATCH SET t.matched = true"; + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 't1', created: true}]->() RETURN t"), 1); + assert_eq!(count(&store, "MATCH ()-[t:T]->() WHERE t.matched IS NULL RETURN t"), 1); + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T {id: 't1', matched: true}]->() RETURN t"), 1); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 1); +} + +#[test] +fn an_undirected_edge_keyed_on_the_row_matches_either_way_round() { + let mut store = GraphStore::new(); + run(&mut store, "CREATE (a:A {id: 1})<-[:T {id: 't1'}]-(b:B {id: 1})"); + let merge = "UNWIND ['t1', 't2'] AS x MATCH (a:A {id: 1}), (b:B {id: 1}) MERGE (a)-[:T {id: x}]-(b)"; + run(&mut store, merge); + run(&mut store, merge); + assert_eq!(count(&store, "MATCH ()-[t:T]->() RETURN t"), 2, "t1 existed reversed; only t2 is new"); +}