Add --transform hook to copy verb for filter/reshape/redact (#93)

[monikagadage]

Summary

Adds an optional --transform <name> argument to the copy verb. It dynamically loads a module from server/src/python_modules/copy/transform/, the same convention load-export and revert-export use for their --transform argument.

Each transform module exposes transform_item(item) and returns a list, matching the export verbs' contract:

• [item] — the copy proceeds as a PUT of that item, modified or not. A bare item is coerced to a one-element list for convenience; None is not a skip signal.
• [] — the item is skipped (not written to the target). The count is reported at the end as "Items excluded by transform".
• [item_a, item_b, ...] — fan-out: each item is written to the target.

Ships three example modules under copy/transform/: default (passthrough), pii_redact (replaces named attributes with a placeholder), attribute_filter (copies only items matching a configurable attribute/value).

Closes #93.

Notes when --transform is used

• A bad --transform name fails immediately on the driver, in run() before the RDD is submitted — one BulkExecutorError sentence, not a Spark task that dies 400x with the cause buried in a Py4J wrapper.
• Transform exceptions go through record_worker_failure(..., understood=False), so the worker traceback needed to debug a user transform_item is preserved.
• Per-worker error accumulation is bounded: the first transform failure and the first post-transform key failure are recorded, then the worker stays quiet.
• Post-transform key validation: an item that loses its pk/sk in the transform is dropped and reported, not sent into a batch_writer flush that would fail mid-segment and leave the target partially written.
• The up-front cost estimate assumes a 1:1 source-to-target copy, so run() prints a caveat that it no longer holds.

Structural changes (matching the export verbs)

• copy.py becomes a copy/ package with a transform/ subpackage and a default.py passthrough, like fill/, load/, update/, scancount/.
• transform_loader moves from shared/export/pipeline/ to shared/, so the copy path no longer imports the export pipeline.

Test plan

make install && make test: 1751 passed, 48 skipped. copy/__init__.py at 100% coverage. New unit coverage for the transform logic (modify / [] skip / fan-out / bounded exception / key validation), the three example modules, and a module_zipper test asserting copy/transform/ and shared/transform_loader.py land in the bootstrap archive.

Verified end to end against a real Glue job:

• --transform pii_redact — 200 rows copied
• --transform attribute_filter — 0 rows copied, Items excluded by transform: 200
• a bad --transform name — job fails in ~48s on the driver (Cannot import transform module ...), before the scan; a real copy of the same table takes ~2:05
• a transform that drops the sort key — job fails with missing key attribute(s) ['sk'], zero rows written to the target

tests/e2e/commands/test_copy_smoke.py is included but not wired into make test (the e2e suite is opt-in and needs a bootstrapped account).

(Supersedes #288, closed due to a branch rename that orphaned its head ref.)

[hunterhacker]

Review

The core idea and wiring are solid, the tests are real, and the suite reproduces the claim in the description (1570 passed, 49 skipped; server/src/python_modules/copy.py and both example transforms at 100% coverage). But it does not fully follow the --transform convention the export verbs established, and two of the gaps have real operational consequences.

Should fix

1. The transform module is loaded inside the worker, not on the driver. _copy_data calls load_transform_module at the top, outside its try. A typo'd --transform therefore raises inside a Spark task — precisely what shared/worker_errors.py's module docstring says to avoid: "an exception escaping a worker costs four Spark task retries, aborts the job, and reaches the driver as a Py4J wrapper with the cause buried." So instead of load_transform_module's carefully-worded "Cannot import transform module 'typo'", the user pays for a Glue job that grinds through 400 tasks × 4 retries and dies with a buried traceback. The export path loads on the driver (_apply_transform_and_resolve in shared/export/pipeline/__init__.py) before touching the RDD, exactly so a typo is "a sentence rather than a traceback." Add a driver-side load_transform_module(...) in run() to fail fast; keep the per-worker load.

2. Transform exceptions record a bare string, so the user loses the traceback they need. This PR does error_accumulator.add([f"Transform '...' raised an exception..."]). The export path uses record_worker_failure(error_accumulator, e, "Transform function raised an exception", understood=False) — deliberately, because worker_errors.py calls out "a user-supplied generator or transform doing something we cannot anticipate. This one stays an exception... the frames that name the bug are the worker's." A bare string is treated as understood and surfaces as a one-line BulkExecutorError with no frames — someone debugging their own transform_item gets nothing to work with. Note that copy.py predates these helpers (its own worker-error path is also a bare string, and run() does raise Exception(first_error) rather than raise_first_worker_error), so switching to the helper means migrating copy's error path too.

3. Unbounded per-item error accumulation. One accumulator entry per failing item. A transform with a KeyError on a missing attribute raises on every item, so a 10M-item table ships ~10M strings to the driver (≈1GB against a 10g G.1X driver) — and run() only ever reads [0]. The export path has the same shape, so this isn't a new convention break, but given #343 / 8e6ad96 this repo is actively bounding exactly this kind of growth. Record the first transform error per worker and stop, or count-and-keep-one.

Convention gaps vs. the export verbs

4. The return contract is looser than the export one. Export transforms always return a list ([record] / []); None is not part of the contract. This PR accepts a bare item, None, or a list, and documents None as the skip signal in the README. More forgiving is defensible, but now the two --transform contracts disagree and someone moving between verbs gets bitten. Either align on list-only ([] to skip) or state explicitly in the README that copy is deliberately more lenient.

5. No post-transform key validation. Export validates that the transformed record still has its key attributes and reports "Item missing key attributes after resolve: {...}". Copy has no equivalent: a transform that drops or renames the PK produces a raw boto3 ValidationException from a batch_writer flush, which kills the rest of that segment while earlier batches stay written — a partially-populated target with an opaque message.

6. Nothing is reported about what the transform did. Export tracks excluded vs. modified/included counts and reports them. Here, --transform attribute_filter prints Total records copied: 0 with no hint the transform caused it. The up-front cost estimate also assumes 1:1 source→target writes, which --transform silently invalidates. A line like "N items excluded by transform" fits the ethos of b51299c (count the noise and say so).

7. Layout: python_modules/copy_transform/ vs. the established python_modules/<verb>/transform/. Both export verbs use the nested form. copy.py is a module rather than a package, so this is the cheap way out — but fill/, load/, update/ and scancount/ are all packages, so converting copy.py → copy/ with a transform/ subpackage is precedented and would make the convention uniform. Also missing the default.py passthrough that both export transform packages ship.

8. Importing the loader from the export subtree drags in the whole export pipeline. from python_modules.shared.export.pipeline.transform_loader import ... executes shared/export/pipeline/__init__.py, which pulls in the readers, parsers, writers, validators and cost estimator — in every one of the 400 tasks, for a verb that touches no exports. transform_loader is generic; move it to python_modules/shared/transform_loader.py and have the export pipeline import it from there.

Smaller things

• pii_redact.py redacts key attributes. The existing pii_mask_attribute.py deliberately excludes pk/sk via the key schema. This one doesn't — and on a table whose PK is named Name, every item collapses onto a single redacted key, i.e. silent near-total data loss. The docstring's claim that it "keeps... any key attributes that happen to share a name intact" reads as if values are preserved; they aren't. Fix the wording at minimum, and ideally skip key attributes.
• The two examples duplicate existing ones (pii_redact ≈ pii_mask_attribute, attribute_filter ≈ load_only_active) under different naming. Matching the existing names would help discoverability.
• Numbers arrive as Decimal, not int — both example docstrings say {"Id": 42} and the tests use 42. Since these are the templates users copy, worth stating (a transform doing isinstance(v, int) will silently not match).
• Fan-out + batch_writer without overwrite_by_pkeys: two fanned-out items sharing a key in the same 25-item flush fails the whole BatchWriteItem with "Provided list of item keys contains duplicates." Worth a README caveat.
• The client-side change is untested — client/src/python_modules/copy.py has no test file at all and doesn't appear in coverage. Pre-existing gap; load_export.py is the same.
• The two new transform test files lack the module docstring AGENTS.md/CLAUDE.md asks for, but the existing tests/server/load_export/transform/ files don't have them either, so this matches actual practice.
• README badges correctly untouched.

Not verified

The e2e smoke test hasn't been run. It's well designed — asserting 0 items landed means a silently-ignored --transform can't pass — and it's the only thing that would prove copy_transform/ actually reaches the Glue job through the bootstrap zip. module_zipper.py walks the tree generically so it should be picked up, but that's unverified.

[hunterhacker]

Findings are in the review above: eight items plus a handful of smaller ones. #1 (transform loaded in the worker, so a typo'd --transform dies after 400 tasks × 4 retries with a buried Py4J traceback instead of one sentence from the driver) and #2 (transform exceptions recorded as a bare string, so a broken transform_item loses the traceback worker_errors.py exists to preserve) are the ones with operational consequences.

[monikagadage]

Thanks for the review, plan:

• Rebase onto main first (this branch predates worker_errors.py).
• Items 1–2: driver-side transform load in run(), and record_worker_failure(..., understood=False) for the transform exception path.
• Items 3–6: bounded per-worker error accumulation; list-only return contract to match the export verbs; post-transform key validation; excluded/modified counts.
• Items 7–8: copy.py → copy/ package with transform/; move transform_loader to shared/.
• Plus the smaller items, and a module_zipper test for the bootstrap-zip packaging.

Will re-request review once items 1–6 are in.

[monikagadage]

Rebased onto current main and pushed. All eight items plus the smaller ones are addressed.

1 — driver-side fail-fast. load_transform_module now runs in run() before the RDD is submitted; the per-worker load stays. A bad --transform name raises one BulkExecutorError.

2 — traceback preservation. Transform exceptions go through record_worker_failure(..., understood=False). copy.py's own worker-error path and raise_first_worker_error were already on main after the rebase, so nothing extra there.

3 — bounded accumulation. The first transform failure and the first post-transform key failure per worker are recorded, then the worker stays quiet — a broken transform on a large table can't flood the driver.

4 — return contract. List-only, matching the export verbs: [item] writes, [] skips, a list fans out. A bare item is coerced to [item]; None is no longer a skip signal. README and the examples updated.

5 — post-transform key validation. An item that loses its pk/sk in the transform is dropped and reported ("missing key attribute(s) ..."), instead of a raw ValidationException from a batch_writer flush that leaves the segment half-written.

6 — reporting. run() prints Items excluded by transform: N and a caveat that the up-front cost estimate assumes a 1:1 copy and no longer holds.

7 — layout. copy.py is now a copy/ package with a transform/ subpackage and a default.py passthrough, matching fill/, load/, update/, scancount/.

8 — loader location. transform_loader moved to python_modules/shared/; the export pipeline imports it from there, so the copy path no longer pulls in the export pipeline.

Smaller items: Decimal-not-int note in both examples; fan-out + batch_writer duplicate-key caveat in the README; module_zipper test asserting copy/transform/ and shared/transform_loader.py land in the bootstrap archive (covers the packaging question from the review).

Two things I did differently — happy to change if you'd rather:

• Kept the example names pii_redact / attribute_filter rather than renaming to pii_mask_attribute / load_only_active. They aren't the same modules: pii_redact replaces the value outright (it can't use the key schema the way pii_mask_attribute does), and attribute_filter is a configurable filter rather than a hardcoded active check. Each module's docstring cross-references the export analogue, and all three are listed in the README.
• pii_redact still can't auto-skip key attributes, because transform_item(item) has no access to the key schema (unlike export's transform_full_record(record) with record.table_key_schema). Fixed the docstring and added an explicit warning instead.

Verification. make test: 1751 passed, 48 skipped; copy/__init__.py at 100%. Also run end to end against a real Glue job:

• --transform pii_redact -> 200 copied
• --transform attribute_filter -> 0 copied, Items excluded by transform: 200
• bad --transform name -> job fails in ~48s on the driver ("Cannot import transform module ..."), before the scan (a real copy takes ~2:05)
• a transform that drops the sort key -> job fails with missing key attribute(s) ['sk'], zero rows written to the target

tests/e2e/commands/test_copy_smoke.py is included but still not wired into make test.