Skip to content

feat(deflation): add pd_deflate_pipeline() batch orchestrator #32

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
randrescastaneda wants to merge 5 commits into
base: PROD
Choose a base branch
from
+1,879 −60
Open

feat(deflation): add pd_deflate_pipeline() batch orchestrator #32

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1761d64
feat(deflation): add pd_deflate_pipeline() batch orchestrator
randrescastaneda Aug 18, 2026
cacb853
docs(deflation): document pipeline stage and deflation summary
randrescastaneda Aug 18, 2026
57a1543
test(deflation): cover pipeline, schema, and report section
randrescastaneda Aug 18, 2026
fa4450b
docs(plans): add deflation pipeline wrapper plan and records
randrescastaneda Aug 18, 2026
ec7fec6
chore(state): mark deflate-pipeline-wrapper done in roadmap
randrescastaneda Aug 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 19 additions & 31 deletions .cg-docs/active-state/current.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,40 +1,28 @@
{
"plan": ".cg-docs/plans/2026-08-06-aux-version-gate-valid-dlw-load-revised.md",
"executionReport": ".cg-docs/work-reports/2026-08-06-aux-version-gate-valid-dlw-load-revised.md",
"plan": ".cg-docs/plans/2026-08-17-deflate-pipeline-wrapper.md",
"executionReport": ".cg-docs/work-reports/2026-08-17-deflate-pipeline-wrapper.md",
"currentPhase": null,
"completedPhases": [1, 2],
"lastUpdated": "2026-08-07",
"completedPhases": [],
"lastUpdated": "2026-08-17",
"refs": {
"roadmapFeature": "aux-version-gate-valid-dlw-load",
"roadmapFeature": "deflate-pipeline-wrapper",
"roadmapMilestone": "Pipeline Alignment Audit"
},
"decisions": [
"Roadmap feature aux-version-gate-valid-dlw-load updated to status active and plan path repointed to the revised 2026-08-06 plan.",
"Phase 1 complete (Steps 1-2): get_aux_hashes() resolver added in R/utils.R; wired into pd_process_data() before aux loading; build_pip_inventory() gains aux_hashes param and writes aux_<measure>_hash columns on current-run rows, initialised to NA on assembled master for legacy/retained rows.",
"Phase 2 complete (Step 3): valid_dlw_load() rewritten with two-stage aux gating. Master loaded once (shared dt_master) and passed to inv_to_process() (now accepts dt_master) and aux_hash_candidates(). Stage 1 builds candidates from per-survey aux hash comparison over full filtered/latest inventory; missing historical hash is a candidate; conflicting same-survey/content hashes abort. Stage 2 runs valid_aux_load() only for changed measures and intersects affected surveys with candidates. force=TRUE skips master/aux comparisons and processes all rows.",
"Tests: test-get-aux-hashes.R (7), test-build_pip_inventory.R (3 aux-hash), test-valid_dlw_load.R (9 new gating + updated existing), test-valid_aux_load.R (updated verbose propagation). Full suite PASS (2 pre-existing skips). roxygenise regenerated .Rd files.",
"Plan marked status: completed, completed-date: 2026-08-07. Roadmap feature aux-version-gate-valid-dlw-load set to done.",
"Verification review 2026-08-07: full R suite PASS (0 failures, 2 pre-existing skips). Both light verification agents returned no usable output due to model-service 'Response contained no choices' errors; not retried per protocol."
"Plan validation preflight failed on Step 4 unknown requirement ID 'R1-R11'; corrected to enumerate R1..R11 and re-validated (EXIT=0).",
"Roadmap feature deflate-pipeline-wrapper updated to active via @cg-roadmap.",
"Steps 1-6 complete: pip_deflated alias registered (R/pipdata_dlw_process.R + Pipdata_script.R), deflation columns initialized in build_pip_inventory(), R/pd_deflate_pipeline.R created, 60 pipeline tests + 2 inventory + 3 log_report tests pass, full suite green, devtools::check 0 errors/0 warnings.",
"Plan marked status completed, completed-date 2026-08-17.",
"Evidence V1-V7 executed and passed (V4 passed-by-review; live round-trip requires configured environment)."
],
"evidenceStatus": {
"V1": "passed (resolver tests: all six measures incl. PFW)",
"V2": "passed (missing/ambiguous/empty/failure/missing-hash abort tests)",
"V3": "passed (pd_process_data resolves hashes once before aux loading)",
"V4": "passed (build_pip_inventory persists hashes on current-run rows)",
"V5": "passed (legacy/retained rows keep NA; schema initialised)",
"V6": "passed (master inventory loaded exactly once)",
"V7": "passed (changed measures only invoke valid_aux_load)",
"V8": "passed (COL/ARG vs USA/GER intersection test)",
"V9": "passed (conflicting same-survey hashes abort)",
"V10": "passed (new/DLW-changed surveys remain selected)",
"V11": "passed (force mode skips master/aux comparisons)",
"V12": "passed (no .joyn column, no duplicate survey IDs)",
"V13": "passed (targeted valid_dlw_load/valid_aux_load tests)",
"V14": "passed (full suite, 2 pre-existing skips)",
"V15": "passed (roxygen regenerated; context.md unchanged - no canonical logging semantics changed)",
"verification": "full suite passed; agent verification incomplete due to model-service errors"
"V1": "passed (5 deflate_one tests)",
"V2": "passed (8 pipeline tests)",
"V3": "passed (check 0 errors/0 warnings)",
"V4": "passed by review (st_init alias registered)",
"V5": "passed (2 inventory tests)",
"V6": "passed (3 log_report tests)",
"V7": "passed (pk asserted in test)"
},
"nextCommand": "/cg-review mode:verify (rerun light agents if independent reviewer output is required)"
}


"nextCommand": "pending review handoff (review:auto)"
}
175 changes: 175 additions & 0 deletions .cg-docs/brainstorms/2026-08-17-deflate-pipeline-wrapper.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,175 @@
---
date: 2026-08-17
title: "pd_deflate_pipeline() — Batch Deflation Orchestrator Design"
status: decided
chosen-approach: "Mode B batch with separate stamp alias and hybrid aux loading"
participants:
- wb384996
tags: [deflation, pipeline, orchestration, batch-processing]
---

# pd_deflate_pipeline() — Batch Deflation Orchestrator Design

## Context

`pd_deflation()` works for single surveys (Mode A: pass `dt`, Mode B: pass `pip_id`). The `integrate-deflation` roadmap item is done. What's missing is the batch wrapper (`deflate-pipeline-wrapper` in roadmap.json, status: `idea`) that mirrors `pd_process_data()`'s iteration pattern: split inventory → lapply with tryCatch → log summary → build result.

`pd_deflation()` has a `@note` at line 157 referencing this future wrapper. Several plan and brainstorm docs also reference it as a deferred item.

## Requirements

| ID | Requirement | Source |
|----|-------------|--------|
| R1 | Iterate over an inventory of cleaned surveys and apply `pd_deflation()` to each | user |
| R2 | Save deflated outputs to a dedicated `"pip_deflated"` stamp alias | user |
| R3 | Log a structured summary entry to `pipdata_log` (n_total, n_success, n_failed, failed_ids) | user |
| R4 | Per-survey `tryCatch` with skip-and-log on failure (no fail-fast) | existing pattern |
| R5 | Support both internal inventory loading (`inv = NULL`) and caller-supplied inventory | existing pattern |
| R6 | Use Mode B (`pip_id`) to delegate loading to `pd_deflation()` | design decision |
| R7 | First version: deflate everything not yet deflated (no incremental aux-hash gating) | user |
| R8 | Load master inventory once, pass `version` hints to avoid redundant inventory reads | design decision |
| R9 | Standalone pipeline stage — not called by `pd_process_data()`, invoked independently | user |
| R10 | Add deflation-specific columns to the master inventory for future incremental processing | design decision |

## Approaches Considered

### Approach 1: Simple Mode B (no aux optimization)

Call `pd_deflation(pip_id = id)` per survey. Each call internally loads the master inventory + data + metadata.

**Pros**: Zero changes to `pd_deflation()`. Simplest implementation.
**Cons**: Redundant inventory load per survey (N=500 → 500 × same inventory file read).
**Effort**: Small

### Approach 2: Hybrid — load inventory once, pass version hints (CHOSEN)

Load master inventory once in batch wrapper. Build `(pip_id → content_hash_data)` lookup. Pass `version` to `pd_deflation()` so `.load_deflation_aux()` skips its internal inventory load.

**Pros**: No interface changes to `pd_deflation()`. Uses existing `version` parameter. Meaningful I/O reduction (saves N inventory reads).
**Cons**: Slightly more complex batch wrapper (one extra lookup table).
**Effort**: Small-Medium

### Approach 3: Full pre-load — pass aux data directly

Load all CPI/PPP/pop once, pass to `pd_deflation()` via legacy path (`cpi`/`ppp`/`pop` args).

**Pros**: Fastest I/O (1 stamp read per survey instead of 2).
**Cons**: Couples batch wrapper to internal legacy path semantics. Loses per-survey version granularity. The legacy path expects `ppp` as a wide data.table with different processing.
**Effort**: Medium

### Pipeline Call Chain Options

| Option | Description | Pros | Cons |
|--------|-------------|------|------|
| Monolithic | `pd_process_data()` calls `pd_deflate_pipeline()` internally | Single entry point | Tight coupling, can't re-deflate independently |
| **Separate stages (CHOSEN)** | Script calls `pd_process_data()` then `pd_deflate_pipeline()` independently | Independent re-runs, matches existing codebase pattern | Two function calls in script |
| Orchestrator | New `pd_run_pipeline()` chains both | Best of both worlds | Another function to maintain |

## Decision

**Chosen approach**: Approach 2 (hybrid aux loading) with separate-stage pipeline architecture.

### Key decisions summary

| Decision | Choice | Rationale |
|----------|--------|-----------|
| Inventory input | `inv = NULL` default, load internally or accept caller-supplied | Matches `pd_process_data()` pattern |
| Stamp alias | Create `"pip_deflated"` | Separate from cleaned data; requires `pipfun::setup_working_release()` registration |
| Master inventory | Add deflation columns (`deflated`, `deflation_version`, aux hash snapshots) | Enables incremental processing later |
| Error handling | Per-survey `tryCatch` + skip-and-log | Matches `pd_process_data()` — `process_data()` returns NULL on failure |
| Call mode | Mode B (`pip_id`) | Delegates loading to `pd_deflation()`; batch wrapper doesn't duplicate load logic |
| Filtering | "Deflate everything not yet deflated" | Simple first pass; incremental aux-hash gating deferred |
| Aux pre-loading | Hybrid — load inventory once, pass `version` hints | Eliminates redundant inventory reads without changing `pd_deflation()` interface |
| Pipeline chain | Separate stage, called independently by script | Independent re-runs, matches codebase pattern |
| Return value | Updated deflation inventory (data.table) | Consistent with `pd_process_data()` return |

### Proposed signature

```r
pd_deflate_pipeline <- function(
inv = NULL, # master inventory; loaded internally when NULL
force = FALSE, # re-deflate even if already deflated
verbose = getOption("pipdata.verbose", default = TRUE)
)
```

### Proposed internal flow

```
1. Load master inventory (if inv is NULL) via pipload::load_pip_master_inventory()
2. Filter to deflation candidates:
- deflated == FALSE | is.na(deflated)
- OR force == TRUE (deflate everything)
3. Early return if nothing to deflate (log info, return inv unchanged)
4. Build version lookup: pip_id → content_hash_data (for passing to pd_deflation)
5. Split inventory → lapply with deflate_one() per pip_id
6. deflate_one() per survey:
a. on.exit cleanup of .pipdataenv key
b. tryCatch:
- pd_deflation(pip_id = id, version = ver, verbose = FALSE)
- save_pip_data(list(dt), alias = "pip_deflated", verbose = verbose)
- return list(pip_id = id, success = TRUE)
c. piperr handler: log via pipfun::log_add(), return NULL
d. error handler: log via pipfun::log_add(), return NULL
7. Collect results: build deflation_map from successful results
8. Update master inventory with deflation columns:
- deflated = TRUE
- deflation_version = content_hash from pip_deflated stamp
- aux_cpi_hash_at_deflation, aux_ppp_hash_at_deflation, aux_pop_hash_at_deflation
9. Save updated master inventory to stamp (pip_master alias)
10. Log summary via pipfun::log_info():
- info = "deflate_summary_inf"
- n_total, n_success, n_failed, surveys_success, surveys_failed
11. Return updated master inventory
```

### Stamp alias registration

`"pip_deflated"` must be registered via `pipfun::setup_working_release()` before `pd_deflate_pipeline()` can write to it. This is an infrastructure prerequisite — either:
- Add it to the existing `setup_working_release()` call in `pipdata_dlw_process.R:53`, or
- Register it once manually via `stamp::st_init(root = ..., alias = "pip_deflated")`

### File locations

| What | Where |
|------|-------|
| Main function | `R/pd_deflate_pipeline.R` (new file) |
| Per-survey worker | `deflate_one()` — internal function in same file |
| Inventory update | Inline in the main function (or extracted to `update_deflation_inventory()` if >50 lines) |
| Tests | `tests/testthat/test-pd_deflate_pipeline.R` |
| Roxygen docs | Inline in `R/pd_deflate_pipeline.R` |

### Integration with master inventory

Add columns to `build_pip_inventory()`:

```r
# In build_pip_inventory.R — Step 7 or new step:
deflation_cols <- c("deflated", "deflation_version",
"aux_cpi_hash_at_deflation", "aux_ppp_hash_at_deflation",
"aux_pop_hash_at_deflation")
```

These columns are initialized as `NA` for new surveys, updated by `pd_deflate_pipeline()` after successful deflation.

## Consequences

1. **New stamp alias**: `"pip_deflated"` must be registered. Downstream consumers that read deflated data from `"pip"` will need to switch to `"pip_deflated"`, OR the final pipeline step copies from `"pip_deflated"` back to `"pip"` for downstream compatibility.

2. **Two-stage pipeline**: Users must call `pd_process_data()` then `pd_deflate_pipeline()` separately. This is a feature (independent re-runs) but requires updating `Pipdata_script.R` and the processing vignette.

3. **Inventory schema change**: Adding deflation columns to the master inventory is a schema migration. Existing inventories will have `NA` for these columns until re-deflated.

4. **Future incremental processing**: The deflation columns enable future aux-hash-gated re-deflation (matching `valid_dlw_load()`'s pattern). This is explicitly deferred to a follow-up.

5. **Downstream data source**: If `"pip_deflated"` becomes the source of deflated data, `pipload::pip_read(id, alias = "pip_deflated")` replaces `pipload::pip_read(id, alias = "pip")` for downstream consumers. This needs coordination with `pipload` and any Shiny/reporting code that reads pip data.

## Next Steps

1. Register `"pip_deflated"` stamp alias in `pipfun::setup_working_release()` / `stamp::st_init()`
2. Create `R/pd_deflate_pipeline.R` with `pd_deflate_pipeline()` and `deflate_one()`
3. Add deflation columns to `build_pip_inventory()` schema
4. Write tests in `tests/testthat/test-pd_deflate_pipeline.R`
5. Update `Pipdata_script.R` to call `pd_deflate_pipeline()` after `pd_process_data()`
6. Update the processing vignette (`vignettes/articles/Processing-Data.Rmd`)
7. Update `pd_deflation()` roxygen `@note` to reflect that the wrapper now exists
Loading
Loading