fix(us_dol_oflc): resolve the crosswalk by column layout, not by file name

pipelines/datasets/us_dol_oflc/constants.py

@@ -32,13 +32,23 @@ class constants(Enum):
     IMPERSONATE = "chrome"
     BASE_URL = "https://www.dol.gov"
 
-    # Link text on the performance page is inconsistent, so files are found by
-    # matching the href against these patterns, one per program.
+    # Files are found by matching the published file name against these
+    # patterns, one per program. They are anchored and deliberately narrow: the
+    # performance page publishes companion workbooks beside each disclosure file
+    # — LCA Appendix A and Worksites, the H-2A Addendums, the H-2B Appendixes —
+    # which carry different layouts and, under a looser pattern, collapse onto
+    # the same local name and overwrite the file we actually want.
+    #
+    # Only the case-level disclosure file is matched, and only in the modern
+    # naming the pipeline ever sees: the historical layouts (H-1B_Case_Data_FY2008,
+    # Icert_ LCA_ FY2009, LCA_FY2012_Q4 …) were onboarded once and are never
+    # re-fetched, because a run only ever touches the open fiscal year and the
+    # one before it.
     FILE_PATTERNS = {
-        "lca": r"(LCA|H-1B|H1B|Icert)[_ ].*(Disclosure|Case_Data|iCert|FY)",
-        "perm": r"PERM.*(Disclosure|FY)",
-        "h2a": r"H-?2A.*(Disclosure|FY)",
-        "h2b": r"H-?2B.*(Disclosure|FY)",
+        "lca": r"^LCA_Disclosure_Data_FY_?\d{4}(_Q[1-4])?\.xlsx?$",
+        "perm": r"^PERM_Disclosure_Data_(New_Form_)?FY_?\d{4}(_Q[1-4])?\.xlsx?$",
+        "h2a": r"^H-?2A_Disclosure_Data_FY_?\d{4}(_Q[1-4])?(_(new|old)_form)?\.xlsx?$",
+        "h2b": r"^H-?2B_Disclosure_(Data_)?FY_?\d{4}(_Q[1-4])?\.xlsx?$",
     }
 
     # A fiscal year is frozen once its final file has landed and the year is

pipelines/datasets/us_dol_oflc/flows.py

@@ -198,6 +198,15 @@ def us_dol_oflc_flow(
     {"cron": "23 14 5,12,19,26 2,5,8,11 *", "timezone": "America/Sao_Paulo"}
 ]
 # The clean step holds one fiscal year of LCA (~700k rows x 57 columns) in
-# pandas while it is written.
+# pandas while it is written, and calamine builds a Python object per cell of
+# the workbook it is reading.
+#
+# The keys are memory_limit / memory_request: this work pool's job template
+# defines no "memory" variable, so the {"memory": "8Gi"} spelling used by most
+# datasets in this repo is silently discarded and the pod runs at the pool
+# default of 4Gi. That is what OOM-killed the first full run here.
 # pyrefly: ignore [missing-attribute]
-us_dol_oflc_flow.job_variables = {"memory": "8Gi"}
+us_dol_oflc_flow.job_variables = {
+    "memory_limit": "12Gi",
+    "memory_request": "4Gi",
+}

pipelines/datasets/us_dol_oflc/utils.py

@@ -40,6 +40,16 @@
     "prevailing_wage_annual",
 }
 
+
+class UnknownLayoutError(RuntimeError):
+    """A source workbook whose column layout the crosswalk does not describe.
+
+    Raised rather than ``SystemExit`` so a Prefect run reports it as Failed —
+    an ordinary data problem — instead of Crashed, which reads like the
+    infrastructure died.
+    """
+
+
 # Values the source uses for "blank".
 NULLISH = {"", "NA", "N/A", "NULL", "NONE", "UNKNOWN", "-", "--", "."}
 
@@ -125,7 +135,7 @@ def _to_date(v: object) -> str | None:
 
 
 def load_crosswalk(program: str) -> dict[tuple[int, str], dict[str, str]]:
-    """(fiscal_year, source_file) -> {source_column: canonical_column}."""
+    """(fiscal_year, local_file) -> {source_column: canonical_column}."""
     out: dict[tuple[int, str], dict[str, str]] = defaultdict(dict)
     with open(CROSSWALK_DIR / f"{program}.csv") as fh:
         for row in csv.DictReader(fh):
@@ -137,6 +147,50 @@ def load_crosswalk(program: str) -> dict[tuple[int, str], dict[str, str]]:
     return out
 
 
+def load_crosswalk_headers(
+    program: str,
+) -> dict[frozenset[str], dict[str, str]]:
+    """Header signature -> mapping, for files the crosswalk knows by layout.
+
+    The crosswalk is keyed on the file name the onboarding run happened to give
+    each workbook, but the recurring pipeline derives its own names from the
+    published file names, and the two do not always agree — the FY2025 LCA Q4
+    file is ``lca_2025.xlsx`` in the crosswalk and ``lca_2025q4.xlsx`` when the
+    pipeline downloads it.
+
+    Matching on the set of source columns instead removes that coupling
+    entirely, and it is the more meaningful key: what determines how a workbook
+    is read is its layout, not its name. A new quarterly file with an unchanged
+    layout therefore resolves on its own, while a genuine form revision still
+    finds no match and fails loudly, which is what the crosswalk is for.
+    """
+    by_file: dict[tuple[int, str], set[str]] = defaultdict(set)
+    with open(CROSSWALK_DIR / f"{program}.csv") as fh:
+        for row in csv.DictReader(fh):
+            if row["source_column"]:
+                by_file[(int(row["fiscal_year"]), row["source_file"])].add(
+                    row["source_column"]
+                )
+    mapped = load_crosswalk(program)
+    return {
+        frozenset(columns): mapped[key]
+        for key, columns in by_file.items()
+        if key in mapped
+    }
+
+
+def read_header(path: Path) -> list[str]:
+    """The header row of a workbook, without materialising the rest of it.
+
+    ``to_python()`` builds a Python object per cell, so reading a 437k x 98
+    workbook to look at one row costs gigabytes — enough to OOM the worker. The
+    layout pre-flight only needs the header, so it reads only the header.
+    """
+    ws = pc.CalamineWorkbook.from_path(str(path)).get_sheet_by_index(0)
+    rows = ws.to_python(nrows=1)
+    return [str(c).strip() for c in rows[0]] if rows else []
+
+
 def read_sheet(path: Path) -> tuple[list[str], list[list]]:
     ws = pc.CalamineWorkbook.from_path(str(path)).get_sheet_by_index(0)
     rows = ws.to_python()
@@ -174,13 +228,20 @@ def read_file(
     order: list[str],
     types: dict[str, str],
     xw,
+    by_header,
     unknown_units,
 ) -> pd.DataFrame:
     """One source workbook as a canonical-schema DataFrame."""
+    header, rows = read_sheet(path)
     mapping = xw.get((fy, path.name))
     if not mapping:
-        raise SystemExit(f"No crosswalk entry for {path.name} (FY{fy})")
-    header, rows = read_sheet(path)
+        mapping = by_header.get(frozenset(header))
+    if not mapping:
+        raise UnknownLayoutError(
+            f"No crosswalk entry for {path.name} (FY{fy}) and its column layout "
+            f"matches no known layout for {program}. Rebuild the crosswalk with "
+            f"build_crosswalk.py and review what changed."
+        )
     idx = {col: i for i, col in enumerate(header)}
     data: dict[str, list] = {}
     for src, canon in mapping.items():
@@ -260,6 +321,7 @@ def build(
     order = [c for c, _ in spec]
     types = dict(spec)
     xw = load_crosswalk(program)
+    by_header = load_crosswalk_headers(program)
     unknown_units: dict[str, int] = defaultdict(int)
 
     typed = pa.schema(
@@ -290,7 +352,9 @@ def build(
             print(f"  FY{fy}: already written, skipping", flush=True)
             continue
         frames = [
-            read_file(p, fy, program, order, types, xw, unknown_units)
+            read_file(
+                p, fy, program, order, types, xw, by_header, unknown_units
+            )
             for p in paths
         ]
         df = (
@@ -429,11 +493,25 @@ def fiscal_year_of(name: str) -> int | None:
 
 
 def local_name(program: str, name: str) -> str:
-    """Local file name for a source workbook, matching the crosswalk key."""
+    """Local file name for a source workbook.
+
+    Two source files must never collapse onto one name — that silently replaces
+    one with the other. A fiscal year can legitimately be published as several
+    files: one per quarter, and in a form-transition year one per form version
+    (PERM FY2024, H-2A FY2025), so both are carried into the name.
+
+    The crosswalk is resolved by column layout rather than by this name, so the
+    name only has to be unique, not to match anything.
+    """
     fy = fiscal_year_of(name)
     quarter = re.search(r"_Q([1-4])", name, re.I)
-    suffix = f"q{quarter.group(1)}" if quarter and fy and fy >= 2020 else ""
-    return f"{program}_{fy}{suffix}{Path(name).suffix}"
+    parts = [program, str(fy)]
+    if quarter and fy and fy >= 2020:
+        parts.append(f"q{quarter.group(1)}")
+    form = re.search(r"(new|old)[_ ]form", name, re.I)
+    if form:
+        parts.append(form.group(1).lower())
+    return "".join([parts[0], "_", "".join(parts[1:])]) + Path(name).suffix
 
 
 # --------------------------------------------------------------------------
@@ -476,10 +554,23 @@ def download_fiscal_years(
     session = _session()
     input_dir.mkdir(parents=True, exist_ok=True)
     got: list[Path] = []
-    for name, url in sorted(list_source_files(program).items()):
-        fy = fiscal_year_of(name)
-        if fy not in years:
-            continue
+    wanted = {
+        name: url
+        for name, url in sorted(list_source_files(program).items())
+        if fiscal_year_of(name) in years
+    }
+    # A collision would silently replace one source file with another, so it is
+    # an error rather than something to resolve by ordering.
+    names: dict[str, str] = {}
+    for name in wanted:
+        local = local_name(program, name)
+        if local in names:
+            raise UnknownLayoutError(
+                f"{program}: {name} and {names[local]} both map to {local}. "
+                f"Two source files cannot share one local name."
+            )
+        names[local] = name
+    for name, url in wanted.items():
         dest = input_dir / local_name(program, name)
         if dest.exists() and dest.stat().st_size > 10_000:
             got.append(dest)