Appearance
Data Lineage / Field-Source Mapping — Mirakl / PIM / Offers
Companion to docs/fullstack/README.md and index.md, which document pipeline architecture (what system talks to what). This document tracks data lineage: per data domain, which system is authoritative, which systems merge/gap-fill it, and what breaks if a source is missing.
See also: Open Questions (things we don't know, need the team to answer), Risks (things we do know, that are structurally fragile), and Opportunities (dormant data sources found in the codebase — NeuCo, Rheem, Hydros — that are real and working, but don't currently feed supplysync.com and are intentionally kept out of this document).
Grounded entirely in code read in this session from:
airflow-pim-dags/airflow/dags/bloomreach/normalization/{availabilityNormalizer,erpJoin,xrefNormalizer,attributeReduce,stockNormalizer,regionNormalizer,branchNormalizer,filterSales}.pyairflow-pim-dags/airflow/dags/bloomreach/xmlbuilder/{mirakl,all_products}.pyairflow-pim-dags/airflow/dags/jobs-inf-w1-full.py,jobs-alg-all.py,jobs-alg-cene.py,jobs-alg-gm.py,jobs-alg-bk.py,jobs-alg-hm.py,jobs-alg-pp.py,jobs-alg-ec.pywatsco1-etl-offers/dags/pfx_offer_sync.py,dags/etl_lib/transform.py
Sources
Both source repos are on Bitbucket. Citations below are pinned to the commit read at analysis time — re-verify against main if it's been a while, since neither pin is enforced/updated automatically.
| Repo | Remote | Commit pin (at analysis time) |
|---|---|---|
airflow-pim-dags | bitbucket.org/watscoenterprise/airflow-pim-dags | 58359a03 (2026-08-11, re-verified after 26b73c4 — see §2 for what changed) |
watsco1-etl-offers | bitbucket.org/watscomagento/watsco1-etl-offers | 72fa008 (2026-06-02, unchanged since analysis) |
airflow-pim-dags has ~150 generated DAGs (per-BU/locale variants); only the file list above was read for this analysis. Domains/claims not backed by one of these files are marked unconfirmed, needs verification with the team rather than inferred from repo structure alone.
Anything not backed by a specific function/file citation below is marked unconfirmed, needs verification with the team.
Coarse table
| Data domain | PriceFX | BI/Snowflake | BU PIM (per-BU) | BU ERP (per-BU) | ALL/W1 PIM export | Conduit | Used by |
|---|---|---|---|---|---|---|---|
| Long description | ○ | ○ | ● | ○ | ○ | ○ | A, S |
| Other attributes (structure, category, flex) | ○ | ○ | ◐ | ○ | ● | ○ | A, S |
| PID ↔ ERP ID matching | ○ | ○ | ● | ○ | ○ | ○ | S, M, A |
| Branch inventory / stock (batch) | ○ | ◇ | ○ | ● | ○ | ○ | A, S |
| Live inventory (PLP/PDP/Cart/checkout) | ○ | ○ | ○ | ● | ○ | ◇ | S |
| Pricing (offers/pricing pushed to Mirakl) | ● | ◆ | ○ | ○ | ○ | ○ | S |
| Product-to-sell selection | ● | ◆ | ○ | ○ | ○ | ○ | M, S, A |
| Historical sales | ○ | ● | ○ | ？ | ○ | ○ | A |
Legend: ● authoritative (source of record) · ◐ gap-fill (fills holes left by the authoritative source, doesn't override it) · ◇ proxy (stands in for a different real-time source) · ◆ confirmed but out-of-band (human/business process, not implemented in either repo's code) · ？ unconfirmed, needs verification with the team · ○ not sourced here (no code evidence this system supplies this domain) · small dot after an icon = click it for detail, full citation in the numbered section below. Used by: A Algolia, S Shopware, M Mirakl.
BU PIM vs. BU ERP — these are different systems, not two names for the same thing. BU PIM is each BU's own PIM/normalization output ({bu}_product.json, generated by that BU's own Algolia/Shopware DAG) — this is where the BU's own PIM team performs PID↔ERP ID matching, and it's what merge_bu_erp() reads for Section 2. BU ERP is the raw transactional/availability feed in that BU's native format (SAP CSV, Mincron JSON, TE-specific CSV, etc.) — this is what feeds Section 3's stock/availability converters (convert_ce_data, convert_mincron_json, etc.) directly, and it is not what Section 2's ID-matching join reads. A BU's raw ERP export and that same BU's PIM output can, in principle, disagree — this doc doesn't currently trace whether/how that's reconciled.
1. Long description / attributes
Corrected 2026-08-11: an earlier version of this section cited
xrefNormalizer.join()/attributeReduce.process()(both inbloomreach/normalization/) as the base merge and reduction mechanism. Neither is imported anywhere in any of the four W1 DAG files (jobs-inf-w1-{full,full-pre,delta,delta-pre}.py), nor anywhere in thebloomreach.xmlbuilder/bloomreach.util/bloomreach.environmentsmodules those DAGs actually import — confirmed by grep across the wholebloomreachtree forxrefNormalizer(zero hits). This was the same class of mistake as the NeuCo/Rheem citations (seeopportunities.md): citing a real function without confirming it's reachable from W1. Rewritten below against the functions the W1 DAG actually calls.
Long description — authoritative per-BU, same mechanism as §2's ID matching, not a separate gap-fill layer. Merge-Descriptions (all_products.merge_bu_erp(input_dir_path, colname='description', input_file_path, output_file_path, env), all_products.py:242) is the same merge_bu_erp fan-in documented in §2, called a second time with colname='description' instead of erp_id. No separate base/default description field appears elsewhere in the W1 DAG's item CSV — description for a product comes entirely from whichever BU's {bu}_product.json feed matches it (missing='' if none do). Treat this the same way §2 treats erp_id: per-BU authoritative, no fallback, silent-empty on no match.
Other attributes (not description) — base from raw CSV exports, gap-filled per-BU. Handle-Attributes (all_products.merge_attrs(merged_attr_path, env), all_products.py:288) reads the raw attribute export (env.attr_file) and renames columns per env.attr_colmap. Handle-Structures/process_category/process_flex (all_products.py:415 and neighboring functions) handle structure/category/flex attribute data similarly — each reads its own CSV export directly, no normalization-layer indirection. Merge-BU-Attrs (all_products.merge_bu_attrs(input_dir_path, item_file_path, output_file_path, env, attr_map, drop=[]), all_products.py:297) left-joins each BU's own {bu}_product.json feed (bu_data.cut('objectID', 'attributes')) onto the base item data by product_id↔objectID — missing={} on the join, so a BU with no matching product simply contributes an empty attributes dict for that field, not a dropped row.
Attribute value normalization (Algolia-facing): supply_sync_algolia_process (bloomreach.xmlbuilder.supply_sync, called from the W1 DAG) imports bloomreach.normalization.attributeReduce (aliased attribute_reduce) and uses attribute_reduce._attrs_normalized/_normalize_alt to normalize attribute values for search — this is the one place attributeReduce.py genuinely is reachable from W1, via supply_sync.py's import, not via a direct DAG-level import. Scoped to the Algolia output specifically, not the base attribute merge.
Consequence if a source is missing: for description, no match in merge_bu_erp → empty string, same as §2's erp_id case. For other attributes, merge_bu_attrs's missing={} leftjoin means a BU with no matching {bu}_product.json row for a product simply contributes an empty attributes object — no hard failure, no row dropped, just an absent per-BU attribute contribution for that product. This is the same missing-data pattern documented elsewhere in this doc (§2, §3): silent, not a raised error.
2. PID ↔ ERP ID matching
Re-verified 2026-08-11 against
airflow-pim-dags@58359a03(was pinned to26b73c4). TheWWIES-5783refactor changed this domain's mechanism, not just line numbers — see corrections below.
Authoritative per-BU source, one-hop join (was two-hop before the refactor). all_products.merge_bu_erp(input_dir_path, colname, input_file_path, output_file_path, env, drop=[], drop_val=[], key=None) (bloomreach/xmlbuilder/all_products.py:242) left-joins the ALL/W1 item file directly against each BU's own {bu}_product.json feed (lkey='product_id', rkey='objectID', missing=''), producing a per-BU {bu}_{colname} field. Called via python_callable=all_etl.merge_bu_erp for 7 fields (erp_id, catalog_code, description, status, title, uom, url_key) at jobs-inf-w1-full.py lines 333, 347, 361, 375, 389, 403, 420.
This replaces the older all_products.merge_erp() for the W1 ID-matching fan-in — that function still exists (all_products.py:219) but is now scoped only to jobs-alg-all.py (5 calls) and jobs-alg-hydros.py (2 calls), a different domain. Its former two-hop join via each BU's erp_pim_data_{bu}.csv xref file (written by convert_te_data/convert_hy_data/convert_mincron_json/xrefNormalizer.norm_ext_attr/xrefNormalizer.join) is no longer in the ID-matching join's path — merge_bu_erp reads each BU's {bu}_product.json directly. Those convert_* functions are otherwise unchanged (0 commits) and their xref CSVs may still feed other domains (e.g. Section 1's attribute gap-fill, still on xrefNormalizer.join() — untouched by this refactor) — don't assume they're dead code.
Mirakl seller-record generation — signature and data source changed. mirakl.mirakl_seller_process(input_path, w1_feed_path, output_dir_path, output_file_pattern, bu, env, drop_ahri=False) (mirakl.py:310) now reads erp_id directly off the BU's own per-BU feed (input_path → cut to objectID, active, attributes, catalog_code, description, erp_id, image, mpn, title), then left-joins w1_feed_path only to filter to products present in the W1 feed (_w1.cut('objectID', 'cat_id').addfield('w1', True), select('w1', lambda p: p is True)) and attach category. Call sites moved from jobs-alg-{cene,gm,bk,hm,pp,ec}.py to jobs-shw-{cene,gm,bk,hm,pp,ec}.py (plus new -inr variants), per the "move Mirakl API operators from W1 into BU DAGs" / "move Mirakl files export from Algolia to Shopware BU DAGs" commits.
mirakl_seller_wso_process() — the function this section previously cited for the "no ERP match → no seller" consequence — no longer exists. It has been removed from mirakl.py along with the diff_list/pim=True filtering it implemented. There is no direct replacement doing the same "diff against current PIM run" filter; mirakl_seller_process now does its own W1-presence check inline via the w1_feed_path join described above.
Corrected consequence claim. The previous version of this doc stated: "a product that doesn't come through a given BU's PIM/ERP feed is never associated with that BU as a Mirakl seller — the BU→PID match only happens via this feed, there's no fallback path." Based on the current code, the mechanism changed but the outcome the team confirms is the same in substance: mirakl_seller_process reads erp_id straight from the per-BU feed with no row-drop on a missing value — a merge_bu_erp join with no match produces erp_id='' (per missing=''), and that empty string still flows through to the generated seller CSV rather than excluding the product from the file.
Per the team: this doesn't require Mirakl to reject anything at the API layer — the failure is structural, not a validation rejection. A missing/empty erp_id means that BU's seller-channel entity doesn't get correctly matched to the product in Mirakl, so there's no seller record for that BU to attach an offer to. PriceFX can have a valid offer for that PID, but if the seller-record side has nothing to attach it to for that BU, the offer has no destination. If none of a product's BUs match, the product has no seller record anywhere in Mirakl, and PriceFX's offer for that PID — regardless of which BU it's for — has nothing to attach to at all. This is a cross-cutting failure mode, not specific to one BU: a PID-level ID-matching gap (across all BUs) breaks the offers pipeline even though the offers pipeline itself (PriceFX → watsco1-etl-offers) has no code-level dependency on merge_bu_erp.
3. Branch inventory / stock
Authoritative, per-BU, no single canonical format. Confirmed BU→converter/format mapping (call sites in jobs-alg-all.py unless noted):
| BU code(s) | Converter function | Source format | Call site |
|---|---|---|---|
| GM (Gemaire) | availabilityNormalizer.convert_mincron_json | Mincron JSON export (product_headers.json) | jobs-alg-all.py:291 (Convert-GM-Availability) |
| AC, HC (split out of GM) | availabilityNormalizer.split_gm_availability | (post-Mincron-conversion split by branch list) | jobs-alg-all.py:303 (Split-GM-Availability) |
| CE (Carrier Enterprise, and by extension all CE* regional variants sharing the feed) | availabilityNormalizer.convert_ce_data | SAP-style CSV (BRANCH_NUMBER, SUB_BUSINESS_UNIT_NUMBER, INTERNAL_PRODUCT_NUMBER, NET_AVAILABLE) | jobs-alg-all.py:353 (Convert-CE-Availability) |
| TE | availabilityNormalizer.convert_te_data | TE-specific CSV (BUName, BranchNumber, InternalProductNumber, AvailableQty, ProductId) | jobs-alg-all.py:397 (Convert-TE-Availability) |
| BK (Baker), FC | availabilityNormalizer.merge_baker | Baker + FC dual-source CSV, deduplicated by (branch, erp_id), qty conflict resolved via max() | referenced via Fix-Ext-Case/Remove-Extra-Delimiters steps in jobs-alg-all.py (lines ~ near "AVAILABILITY_BK.CSV|AVAILABILITY_FC.CSV") |
| HM (Homans) | availabilityNormalizer.split_homans | split out of a combined feed by sub == '6' | bloomreach/normalization/availabilityNormalizer.py:split_homans (call site not directly observed this session — unconfirmed exact DAG line) |
| any BU using region/division splits (e.g. CE sub-regions) | availabilityNormalizer.split_ce_reg | splits one input file into per-BU files by branch→region mapping (env.bu_availability_files) | bloomreach/normalization/availabilityNormalizer.py:split_ce_reg |
| generic SAP-fed BUs | stockNormalizer.process_sap | generic SAP stock-status CSV filtered by Centre regex | bloomreach/normalization/stockNormalizer.py:process_sap |
Fan-in / merge: availabilityNormalizer.merge()/merge_pid() cat all per-BU availability files together, select only rows with qty > 0, and aggregate branches per erp_id. merge_pid() resolves to product_id (PIM PID) via each BU's own erp_pim_data_{bu} file rather than staying at erp_id — i.e. this is where the PID↔ERP-ID matching from Section 2 becomes load-bearing for stock: a stock row with no matching product_id in that BU's xref is dropped (.select('product_id', lambda p: p not in [None, ''])).
Region roll-up: stockNormalizer.create_stocks() / regionNormalizer.create_regions() further reduce branch-level stock into region-level stock+status flags for search-facing consumption (region_bu_active, stock_status Y/N, idx_hi/idx_md search-index hints) — these are derived views, not new sources.
Consequence if missing: a BU's stock file that's empty, absent, or fails its converter step causes that BU's branches to simply not appear in the merged/region-rolled-up files (no row to join) — no explicit code path forces a hard failure at this stage; it silently reads as "no stock at that branch."
Batch usage (this lineage). The merged/region-rolled-up stock has two confirmed batch consumers: Algolia search ranking (via the BI inventory-by-branch proxy signal — see the coarse table's BI/Snowflake column and the "Key relationships" note in README.md), and — per the team — the batch BU ERP feed's stock is separately matched to BU offers to drive pickup availability and delivery decisions at checkout in Shopware.
Live path — separate row in the coarse table ("Live inventory (PLP/PDP/Cart/checkout)"), not part of this batch lineage. Real-time per-branch inventory across the full ecom experience — product listing, product detail, cart, and checkout, not checkout alone — is served by a separate, live query path through Conduit, bypassing this entire batch pipeline: both Shopware and a GraphQL proxy layer in front of it call Conduit directly for live stock, and Conduit fronts the per-BU ERPs with per-BU auth (see README.md's "Shopware also queries live inventory via Conduit" note and the context diagram). For this live path, the BU ERP is the source of truth (●) and Conduit is the proxy (◇) that fronts it — the storefront never queries the ERP directly, it always goes through Conduit. This means a product's batch stock/availability value (documented above) and what a customer sees across the site can come from two different code paths reading the same underlying ERPs — treat the batch value as directional (search ranking, general availability signal), not as the live source of truth.
4. Pricing (offers/pricing pushed to Mirakl)
Authoritative: PriceFX, via watsco1-etl-offers — confirmed in dags/pfx_offer_sync.py: the DAG factory builds pricefx_mirakl_load_{stg,prd} with task chain prepare_offers >> process_offers >> prepare_pricing >> process_pricing >> report_stats. process_phase() (imported from etl_lib.pipeline) drives PriceFX batch pull → S3 raw JSON → validate → transform.records_to_csv() (renames PriceFX JSON fields to Mirakl CSV columns per cfg.offer_field_mapping/price_field_mapping) → mkl.import_offers/import_prices (Mirakl OF01/PRI01) → poll → error report. No code path in this DAG reads from the PIM pipeline for price values — this repo's pricing values originate solely from PriceFX's batch API responses.
"PriceFX pushes a priced-items list into PIM via S3": this claim from the existing docs (docs/fullstack/README.md) is not evidenced in code in either repo searched this session — no pricefx_pim_priced_items, priced_items, or equivalent string/path appears anywhere in airflow-pim-dags or watsco1-etl-offers. Flagging as unconfirmed, needs verification with the team — it may be true but implemented outside both repos (e.g. directly in Informatica Product360, or a script not in these two codebases), or it may be stale/aspirational documentation.
5. Product-to-sell selection
Confirmed out-of-band, not implemented in either repo's code: the PriceFX team selects which products to sell, informed by BI reports. This is a human/business process, not an automated data pipeline — there's no code in airflow-pim-dags or watsco1-etl-offers implementing a "BI tells PriceFX" mechanism because none exists; the decision is made by people reading BI reports, and the output of that decision enters the code lineage only downstream (e.g. wherever PriceFX's resulting item/price selection is fed into these pipelines, per the batch pull in pfx_offer_sync.py — see §4).
(Previously flagged as unconfirmed after a code-only search — searched both repos for snowflake/Snowflake; all airflow-pim-dags hits were false positives on the string "sales", and the only genuine references were in this doc's own architecture-narrative siblings, README.md/index.md/diagrams/context.py. Confirmed directly by the team as an out-of-band process rather than a missing/undiscovered code path.)
Downstream consumers. Per the team: this selection flows to Mirakl, Shopware, and Algolia. In Mirakl specifically, it's broken out into seller channels, one per BU — i.e. the product-to-sell decision doesn't just gate whether a product exists in Mirakl at all, it also determines which BU-specific seller channel(s) carry it. This is consistent with, and adds business context to, the per-BU seller-record generation traced in §2 (mirakl_seller_process), though the exact code path connecting "PriceFX's selection" to "which BU seller channels get a record" was not traced in this session — flagged as unconfirmed, needs verification with the team.
6. Historical sales
Confirmed, but narrower than the general "BI supplies historical sales" claim: bloomreach/normalization/filterSales.py has three functions:
process_ce(salesdata_file_path, env)— CE-specific, filters sales rows byregion_bu_code(env.salesdata_filter) and roundssales_quantity. Called fromjobs-alg-cene.py:408(filterSales.process_ce) — i.e. this runs inside the CE Algolia DAG, not the shared W1/PIM/Mirakl pipeline. (Line shifted from:394at26b73c4to:408at58359a03due to unrelated insertions earlier in the file — same call, same behavior.)process(salesdata_file_path, branchregion_file_path, env)— generic branch→region sales aggregation (rolls branch-level sales up to region level using a branch-region map).process_sap(...)— explicitly commented as a placeholder, not yet active:"This code is a place-holder for the future use, once and if SALES are available as a SAP export for Latam CEMX".
No call site for filterSales.process (the generic one) was found in the DAG files read this session — only process_ce's call site was confirmed. Per the team: historical sales come from BI, delivered via a nightly feed to the far end of the ETL pipeline through S3 — i.e. BI drops a file to S3 on a nightly cadence, and that's the salesdata_file_path process_ce reads. BI is the source of truth (●) this pipeline actually reads for historical sales — unlike the branch inventory/stock domain (§3), where the pipeline's BI signal is explicitly a proxy for a separately-identified authoritative BU ERP record, this domain has no such confirmed BU ERP counterpart to point to as "the real source" — see the open question below.
This confirms the ingestion mechanism (S3-delivered, BI-originated, nightly) that the code alone couldn't show. It does not settle a separate, narrower question the team also flagged: whether BI's own feed is itself derived from CE's ERP upstream of BI — i.e. whether BI is the true point of origin for the values, or just the last system to touch them before this pipeline reads them. Marked unconfirmed (？) in the BU ERP column, not resolved either way — this pipeline doesn't read the BU ERP directly for this domain regardless of how that question is eventually answered, so it doesn't change what's actually implemented, only where the values ultimately trace back to.
The existing docs' claim that "BI also supplies historical item sales data, feeding into the same normalization step as the inventory-by-branch signal" is confirmed by the above.
Corrections to the CONTEXT section's claims
The "PriceFX pushes a priced-items list into PIM via S3" and "BI tells PriceFX which products to sell" claims are unconfirmed by code in either repo searched this session (see Sections 4 and 5 above). These may be accurate — they may simply live outside the ~150+ DAG files and both repos' code that were searched — but they should not be treated as code-verified facts until traced to an actual source file.
The CONTEXT section's assumption that
merge_erp(xrefNormalizer.py,colname='erp_id') is the PID↔ERP mechanism undersells the picture — there are twomerge_erpfunctions (one per-BU inxrefNormalizer.py, one ALL/W1 fan-in inall_products.py), and the fan-in version is called 7 times for 7 different fields (erp_id,catalog_code,description,status,title,uom,url_key), not just for ID matching — the ID-matching consequence for Mirakl sellers specifically rides on theerp_id-flavored call, but the same mechanism doubles as the primary "gap-fill" path for Section 1 (description) and other domains simultaneously. These are the same join, not two separate systems.NeuCo, Rheem, Hydros, and the
ecom_products_xrefDAG are not part of this pipeline — moved toopportunities.md. An earlier version of this document spent significant space documenting NeuCo's active-but-unused status, its BU code, and its relationship to Rheem/Hydros/xref; per the team, since none of that data is used by supplysync.com, it doesn't belong in a document about what actually feeds the site. Seeopportunities.mdfor the full detail on all four.