feat(hl7-transformer): split derivative table creation into new activity (#457) by johnflavin · Pull Request #495 · washu-tag/scout

johnflavin · 2026-07-08T19:13:00Z

Description

Fixes #457

Product

Basically no user-facing changes. HL7 ingestion is more resilient.

Technical

Splits the monolithic ingest-and-derive Delta step into two independently-retryable Temporal activities — ingest_hl7_files_to_delta_lake (parse + atomic MERGE into the base table + per-file status writes) and derive_delta_tables (derives curated/latest/dx/mapping from the base table's change data feed) — orchestrated base → signalRefresh → derive (#457). A base failure marks files failed and skips derive; a derive failure fails the workflow without touching the base table or marking files failed. Adds a per-invocation deriveDeltaTablesTimeout workflow arg (defaults to 480 min), extracts a shared spark_activity_session context manager that owns Spark session lifecycle and error/cancellation cleanup, and fixes the base activity's spark.stop() error handler.

Also changed how exceptions are handled so if the activity has received a cancel signal from temporal, we don't rethrow / mark the pod unhealthy / restart the pod. This fixes #458.

Type of change

Impact

Security

Authorization

N/A — no changes to user roles, data visibility, or API surface.

Appsec

N/A — internal Temporal activity/worker changes only; no new external input surface.

Performance

The base table becomes queryable before the longer derivative derivation runs, and derivative failures retry independently and resume incrementally via the CDF rather than re-ingesting. Per-activity timeouts are independently configurable (base and derive each default to 480 min). Verified on dev against a ~10k-report synthetic dataset.

Data

The base MERGE is a single atomic Delta transaction, so a derive failure leaves the base table untouched and no partial base rows are written. The incremental derive dedupes CDF windows spanning multiple commits to avoid the DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW wedge; unparseable/empty/junk messages are handled per existing per-file error behavior.

Backward compatibility

No Delta table/data-model changes. IngestHl7LogWorkflowInput gains optional deriveDeltaTablesTimeout and createMapping (both default sensibly). The internal activity contract between the Java workflow (hl7log-extractor) and Python worker (hl7-transformer) changed — the two images must be deployed together.

Testing

Added Python unit tests (test_derive.py): derivative happy path, derivative-failure-does-not-touch-base, cancel-during-derivative fault isolation, incrementality, the CDF dedup/wedge regression, and spark_activity_session error/cancel handling; plus Java IngestHl7ToDeltaLakeWorkflowImplTest asserting base→signal→derive ordering, base-failure marks files failed + skips derive, and derive-failure does not mark files failed. CI now runs the hl7-transformer pytest suite as its own separate test activity in parallel with unit-test to remove these new tests from the single python environment with all the dependencies running all the other unit tests.

Manually verified on dev03: ran the log-extraction workflow on synthetic 10k data, then exercised the child transformer workflow — cancelling mid-derive left the base table's version and row count unchanged and a rerun resumed to completion; a tight derive timeout self-healed via incremental resume.

Additional testing on dev03 for #458 specifically. Verified red/green: before my fix, a cancel signal caused the pod to restart (at least sometimes, it seems like a race condition) and after the fix it did not.

Note for reviewers

~~This does not fix 458 (cancelation -> spark failure -> pod restart); the error handling logic is moved into a context manager but otherwise left alone.~~ Does fix #458 as well.

This ~~also~~ does not fix 482 (unconditional MERGE of unchanged rows).

Checklist

My code adheres to the coding and style guidelines of the project (and I've run pre-commit run --all-files)
I have commented my code, particularly in hard-to-understand areas
I have added or updated user documentation, if appropriate
I have added or updated technical documentation, including an architectural decision record, if appropriate
I have added unit tests, if appropriate
I have added end-to-end tests, if appropriate

github-actions · 2026-07-08T19:13:13Z

Dependency Review

✅ No vulnerabilities or OpenSSF Scorecard issues found.

Snapshot Warnings

⚠️: No snapshots were found for the head SHA 00480bc.

Ensure that dependencies are being submitted on PR branches and consider enabling retry-on-snapshot-warnings. See the documentation for more information and troubleshooting advice.

Scanned Files

None

Extract derive_delta_tables() out of import_hl7_files_to_deltalake so the derivative cascade (curated/latest/dx/mapping + epic views) runs as its own unit, reading the base table's committed change data feed. A derivative failure no longer re-runs the base merge — the amplifier behind the June wedge. - deltalake.py: base function is base-only (parse+merge+write_successes, reverted pre-#307 tail); new derive_delta_tables(spark=None) with optional session injection for tests. - activities/ingesthl7.py: base activity drops createMapping; new derive_delta_tables activity + DeriveDeltaTablesActivityInput; both registered on the worker. - tests/: new local Spark+Delta pytest harness (conftest fixtures) and red/green tests — happy path, decoupling invariant (derivative failure leaves base untouched), wedge dedup regression, incrementality. Python side only; the Java workflow (IngestHl7ToDeltaLakeWorkflowImpl) still calls the single activity and must be updated before this is pushed/deployed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…ct (#457) Address Spark-version coupling: the unit tests now install the package and inherit pyspark/delta-spark from pyproject.toml instead of re-pinning them in the run command. A Spark 4 upgrade touches only pyproject + the Dockerfile, not the test harness. - pyproject: requires-python >= 3.10 (matches temporalio; drops the --no-project workaround); pin delta-spark == 3.3.0 to match the Dockerfile jars (was unpinned); add [project.optional-dependencies] test. - run recipe is now version-free: `uv run --python 3.11 --extra test pytest tests/`. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

@timeout

…457) IngestHl7ToDeltaLakeWorkflowImpl now runs base ingest -> signalRefresh -> derive as two independently-retryable activities. A derivative failure fails the workflow (existing alert path) but does NOT mark files failed, since they are already committed to the base table. signalRefresh reverts to firing right after the base ingest rather than behind the long derivative step. - New DeriveDeltaTablesActivityInput; createMapping removed from the base activity input. Constants.DERIVE_ACTIVITY matches the Python @activity.defn. - deriveDeltaTablesTimeout (default 480) in application.yaml, DefaultArgs, and the ansible template (server default only). - IngestHl7ToDeltaLakeWorkflowImplTest (TestWorkflowEnvironment + DynamicActivity) covers happy-path ordering and the base/derivative failure routing; @timeout guards against hangs. Seeded DefaultArgs in the test (no Spring context) and added deriveDeltaTablesTimeout to the test application.yaml. Full module suite green (29 tests) + checkstyle clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…ant Jinja defaults (#457) Follows up PR #474 (harmonized hl7_transformer_timeout to 480) and Kate's note to revisit the timeout defaults under #457. Every var in the templated workflowArgDefaults block is defined in the role defaults, so the `| default(...)` Jinja fallbacks were redundant. - Add hl7_transformer_derive_timeout: 480 to defaults/main.yaml (it was the only var relying on the Jinja fallback). - Remove the redundant `| default(...)` from all five lines in the block so role defaults are the single ansible source. All timeout defaults now consistent at 480 (app-side application.yaml, ansible role defaults) / referenced directly in the template. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…457) Drives a real Temporal cancellation into a running derivative activity and asserts the split's invariant: the base table is untouched and a later clean run completes. Cancellation is delivered reliably (blocking in a bounded heartbeat/is_cancelled loop on the activity thread) rather than raced against Spark, so it can't hang; marked flaky(reruns=2) as insurance. Hermetic — no cluster needed. Note: env.cancel() surfaces as temporalio.exceptions.CancelledError, not the concurrent.futures.TimeoutError the activity's except-clause comment assumes; that classification is the separate cancellation-handling issue and is left untouched here. Full Python suite: 7 passed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

New hl7-transformer-unit job: Java 17 (Spark 3.5.x needs <=17; also valid for Spark 4) + Python 3.12 + `pip install ./extractor/hl7-transformer[test]` + pytest. Uses pip (repo convention) and inherits Spark/Delta versions from pyproject.toml, so CI carries no version pins. Verified the suite passes on 3.12 and the workflow YAML parses. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

The hl7-transformer-unit CI job (added in this branch) pinned Java 17 and carried comments justifying it via the Spark 3.5.x "<= Java 17" constraint. Now that main is on Spark 4.1.x (which supports Java 17 or 21), switch the job to the repo default JAVA_VERSION (21) like every other Java job, and refresh the stale conftest/CI comments. Test suite passes on Java 21. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

#457) The hl7-transformer suite ran twice: the dedicated hl7-transformer-unit job (installs the package, works) and an older step in ansible-and-python-tests that pointed PYTHONPATH=src at a hand-curated pip list missing temporalio/delta-spark/s3fs/psycopg — failing collection on tests/test_derive.py. Remove the stale step (plus its now-unused setup-java and pyspark pin; nothing else in that job uses them) and switch hl7-transformer-unit to setup-uv (SHA-pinned, uv.lock-keyed cache) running 'uv run --locked --extra test', so CI resolves deps from pyproject.toml + uv.lock exactly like local runs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

…nit job (#457) 'uv run --locked' failed in CI because uv.lock was globally gitignored (added in b518917 for verify-node's incidental lockfile) and so never committed. Scope that ignore to verify-node/ and track the transformer lockfile, which CI now consumes. Also add hl7-transformer-unit to deploy-and-test's needs so it gates the pipeline like the other unit-test jobs; publish inherits the gate transitively through deploy-and-test. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

…ader test_hl7reader defined its own session-scoped local[1] spark fixture. Because Spark uses a single JVM session via getOrCreate(), it collided with conftest's new Delta-configured session fixture: whichever ran first won, so the Delta tests passed only because collection order put them first. Drop the local fixture so all tests share conftest's Delta session; add a comment so it isn't reintroduced. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

The base ingest activity's finally block logged with exc_info=e in a bare except around spark.stop(), but e is unbound there (Python clears the except-name at block exit), so a stop() failure raised NameError, masking the real error and skipping temp-dir cleanup. Match the derive copy's handler (except Exception, no exc_info). Also add a NOTE(#458) in derive_delta_tables marking the cancellation/error classification as tracked work. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…rkflow arg The derivative-table activity's timeout was hardcoded to the config default (getDeriveDeltaTablesTimeout(null)), so a large backfill couldn't extend it the way deltaIngestTimeout can. Add a deriveDeltaTablesTimeout field to both IngestHl7LogWorkflowInput and IngestHl7FilesToDeltaLakeInput, thread it through the parent workflow (child input + continue-as-new) into the activity stub's start-to-close timeout, and document it in ingest.md. Falls back to the ansible/ application.yaml default when omitted. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…ext manager The ingest and derive activities each carried a near-verbatim copy of the Spark session lifecycle and error ladder (create/heartbeat, Py4JError/ConnectionError -> health file, TimeoutError -> cancel, Exception -> raise, finally clearCache/stop), which had already drifted. Extract it into a single spark_activity_session context manager so the contract lives in one place — in particular the cancellation/error classification (issue #458) is now a single edit both activities inherit. The CM suppresses the cancellation TimeoutError, so the base activity's write_successes moves inside the session block to keep skipping the status-DB write on a cancel (write_status_to_db opens a Postgres connection even for zero rows). The derive activity drops its test-only spark/own_spark parameter: tests now inject a session by patching the context manager, and it gains direct unit coverage. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…t params Keep CLAUDE.md's Workflow Input Parameters block in sync with the new per-invocation derive timeout arg (also documented in docs/source/ingest.md). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…drift Pin the hl7-transformer unit job to Python 3.10 to match the runtime image (spark:4.1.1-...-python3 ships 3.10), so the tests exercise the interpreter that actually ships instead of 3.12. This immediately surfaced a version-sensitive bug in the new spark_activity_session test: concurrent.futures._base.TimeoutError (what the CM catches) is a DISTINCT class from the builtin TimeoutError before Python 3.11, so the test must raise that exact class to pass on 3.10 — matching the deltalake module's 'catching the builtin doesn't work' caveat. Update the conftest guidance accordingly. Add an explicit 'uv lock --check' step before the tests with an actionable ::error:: so a red build from lock drift reads as 're-lock', not 'tests broke', and document the refresh workflow (uv lock / uv lock --upgrade) in pyproject.toml. uv.lock feeds only these tests; the runtime image installs from pyproject.toml via pip, so it is refreshed by hand and this is the reminder. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

uv.lock governed only the unit tests; the runtime image installs via `pip install .` (Dockerfile), so the locked versions were never what shipped. That split meant two toolchains plus a manual re-lock step (the `uv lock --check` gate) the team has no experience operating -- bought for reproducibility and deterministic transitive pinning we don't currently need. Run the tests the way the image is built: setup-python 3.10 + `pip install .[test]` + pytest, resolving from pyproject.toml. Spark/Delta stay pinned in pyproject.toml so tests and image agree on the versions that matter; peripheral deps float in both. Delete uv.lock and restore the root .gitignore's uv.lock entry, dropping the now-redundant verify-node/.gitignore. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Pin the three GitHub Actions in the new hl7-transformer-unit job to commit SHAs, matching the pins in the security SHA-pinning PR (#493): actions/checkout -> 9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 actions/setup-java -> 0f481fcb613427c0f801b606911222b5b6f3083a # v5.5.0 actions/setup-python -> ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

… the pod (#458) A Temporal cancellation is delivered to these threaded activities by async thread-exception injection, which cannot interrupt a blocking py4j/JVM call. So a cancel that lands mid-Spark (e.g. during the derive step's awaitTermination) is deferred until Spark is torn down and the in-flight call returns as a Py4JError/ConnectionError. spark_activity_session classified by exception type, so it treated that as a Spark connectivity failure: it wrote the health file, /healthz returned 503, the liveness probe failed, and kubelet restarted the pod. Classify on cancellation STATE (activity.is_cancelled()) instead of exception type. is_cancelled() is set before the exception is injected, so it is reliably true whichever exception wins the teardown race: * cancelled -> do not touch the health file; re-raise as a Temporal CancelledError so the activity is recorded Cancelled (not a phantom success, not a retryable failure); * not cancelled + Py4JError/ConnectionError -> mark the pod unhealthy and re-raise (unchanged; k8s restarts on a genuine Spark failure); * not cancelled + anything else (including a genuine TimeoutError) -> re-raise so Temporal's retry policy fires. This also fixes the related bug where a genuine TimeoutError was suppressed and the activity returned normally, so Temporal recorded a phantom success and the 5-attempt retry never fired (derivative tables silently went stale). Drops the fragile concurrent.futures._base.TimeoutError import (no longer caught by name, so the 3.10-vs-3.11 class distinction stops mattering). Tests: replace the obsolete suppress-timeout-as-success test with a cancellation matrix (cancel + Py4JError/either TimeoutError -> CancelledError, no health write; uncancelled timeout -> re-raised for retry, no health write) plus an activity-level guard that a cancelled derive raises instead of returning None. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

kathrynalpert

Should we do a full ingest on preprod before merging this, or wait to get #482 and test the whole shebang?

kathrynalpert · 2026-07-09T14:27:55Z

+    assert spark.table(f"default.{table}_curated").count() == 2
+
+
+@pytest.mark.flaky(reruns=2)


Nit: Don't love a flaky test, is there anything we can do to make it more reliable?

Fixed. The test was flaky because it relied on a sleep for timing events happening on other threads, which wasn't reliable. Now we use some threading primitives to verify the derivative activity is running, cancel it, and check that we don't rerun the base ingest activity.

kathrynalpert · 2026-07-09T14:32:39Z


+        // Derive downstream tables from the base table's change data feed. A failure here
+        // fails this workflow (surfaced by the existing workflow-failure alert) but does
+        // NOT mark files failed — they are already committed to the base table (#457).


I think this makes sense - even if reports don't make it to the derivative table, they're still "green" in the ingest status dashboard because they're in the reports table. They should come into the derivative tables with the next run of the "derive" activity.

This means we'll need to attend workflow failures, not just the ingest table, to determine if something's going awry at this step. We have alerting for this, so seems reasonable.

johnflavin · 2026-07-09T15:20:22Z

Should we do a full ingest on preprod before merging this, or wait to get #482 and test the whole shebang?

My preference for the sequence is

Merge this
Test on preprod / full reingest
Turn to reports table merge is inefficient on re-ingest #482, possibly with another reingest

We could flip 2. and 3. if you want, I'm fine either way, but I don't want to wait to merge this until after fixing #482.

The cancel-during-derivative fault-isolation test was marked flaky (reruns=2). Its flakiness came from timing-coupled thread coordination plus a false model of how ActivityEnvironment delivers a cancel: it does NOT route cancellation through activity.heartbeat() (a no-op in the test env) — it sets the cancelled flag and asynchronously injects a CancelledError into the thread that called env.run(). The old version slept a fixed 0.3s before cancelling and relied on heartbeat to raise, so it raced thread scheduling. Make it deterministic: - Replace the wall-clock sleep with a threading.Event handshake so the cancel lands only once the derivative is provably inside the poll loop. - Run env.run() on a dedicated thread joined before the assertions, so the async-injected CancelledError is confined to that thread and can never surface in the assertions or the clean re-run. - Drop the @pytest.mark.flaky marker and the now-unused pytest-rerunfailures test dependency. - Remove the vacuous base-table-untouched assertions (the cancelled run stubs out process_derivative_data, so it performs no Spark writes); fault isolation is covered by test_derivative_failure_does_not_touch_base. Rename the test to reflect what it verifies: delivery + resumability. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

… test test_derivative_failure_does_not_touch_base used a bare pytest.raises(Exception), which would pass for any error — including an unrelated Spark failure that never exercises the fault-isolation path. Spark re-wraps the RuntimeError raised from foreachBatch as a StreamingQueryException but preserves the message, so match on "injected derivative failure" to confirm the failure under test is the one we injected. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

johnflavin requested a review from a team as a code owner July 8, 2026 19:13

github-actions Bot assigned johnflavin Jul 8, 2026

github-advanced-security AI found potential problems Jul 8, 2026

View reviewed changes

Comment thread .github/workflows/ci.yaml Fixed

Comment thread .github/workflows/ci.yaml Fixed

Comment thread .github/workflows/ci.yaml Fixed

johnflavin force-pushed the hl7-transformer-activities branch 4 times, most recently from a143f09 to 7671efc Compare July 8, 2026 22:30

johnflavin and others added 18 commits July 9, 2026 09:32

johnflavin force-pushed the hl7-transformer-activities branch from 0253f2b to 51cc8ef Compare July 9, 2026 14:34

kathrynalpert approved these changes Jul 9, 2026

View reviewed changes

johnflavin merged commit 73e840d into main Jul 9, 2026
43 checks passed

johnflavin deleted the hl7-transformer-activities branch July 9, 2026 18:45

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat(hl7-transformer): split derivative table creation into new activity (#457)#495

feat(hl7-transformer): split derivative table creation into new activity (#457)#495
johnflavin merged 20 commits into
mainfrom
hl7-transformer-activities

johnflavin commented Jul 8, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Jul 8, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kathrynalpert left a comment

Uh oh!

kathrynalpert Jul 9, 2026

Uh oh!

johnflavin Jul 9, 2026

Uh oh!

kathrynalpert Jul 9, 2026

Uh oh!

johnflavin commented Jul 9, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		assert spark.table(f"default.{table}_curated").count() == 2


		@pytest.mark.flaky(reruns=2)

Uh oh!

Conversation

johnflavin commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Product

Technical

Type of change

Impact

Security

Authorization

Appsec

Performance

Data

Backward compatibility

Testing

Note for reviewers

Checklist

Uh oh!

github-actions Bot commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Dependency Review

Snapshot Warnings

Scanned Files

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kathrynalpert left a comment

Choose a reason for hiding this comment

Uh oh!

kathrynalpert Jul 9, 2026

Choose a reason for hiding this comment

Uh oh!

johnflavin Jul 9, 2026

Choose a reason for hiding this comment

Uh oh!

kathrynalpert Jul 9, 2026

Choose a reason for hiding this comment

Uh oh!

johnflavin commented Jul 9, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

johnflavin commented Jul 8, 2026 •

edited

Loading

github-actions Bot commented Jul 8, 2026 •

edited

Loading