Fix epacriteriaairpollutants

[shourya116]

Summary

Resolves execution failures and structural automation issues for the EPACriteriaAirPollutants import pipeline across 1980–2025 historical data (64.8M rows, 305.2M nodes, 123.27 GB MCF):

1. Incident 1 (Validation Gate Failure): Blocked at ImportStatus.VALIDATION due to 18.1M missing measurement method warnings (NO2_1hour_2010), ~875K fatal observationAbout reference errors from unregistered cross-border monitors (epa/800060001), and differ summary DATA_ERROR.
2. Incident 2 (Differ In-Memory OOM Crash / Exit Code 50002): Cloud Batch VM kernel freeze caused by the in-memory Python differ (mcf_file_util.load_mcf_nodes) attempting to load 121+ GB of MCF text into Python dictionaries on a 128 GiB RAM machine.
3. Review Findings Resolution (Commit 573d49a8):
   • Canonical Schema De-duplication: Removed 18 canonical StatVars and 2 core units from EPA_AirQuality.mcf to eliminate duplicate instance node ingestion.
   • Dynamic Station Generation: Replaced 8,404 hardcoded static station nodes in Git with dynamic runtime emission (EPA_AirQuality_sites.mcf) and wildcard "node_mcf": "EPA_AirQuality*.mcf", ensuring future automated refreshes never break against threshold: 0 when upstream EPA commissions new monitors.
   • Dynamic Date Freshness: Replaced hardcoded '2024' date cutoff with dynamic DuckDB date expression.

---

Key Changes

1. Data Extraction & Dynamic Site Generation (scripts/us_epa/airdata/air_quality.py)

• Geographic Scoping: Filter out non-US cross-border monitors (State Code '80' for Mexico and 'CC' for Canada) that lack Data Commons place entities, eliminating ~875K fatal lint errors.
• Dynamic Station MCF Generation: Updated write_csv to dynamically emit unique AirQualitySite nodes to EPA_AirQuality_sites.mcf at runtime. Added create_sites_mcf helper and preserved cross-call uniqueness across all 270+ annual archive extractions.
• Unit Normalization: Map observation units via UNIT_MAP (MicrogramsPerCubicMeter_lc, MicrogramsPerCubicMeter_25C, PartsPerMillion, PartsPerBillion) with camel-case fallback.
• Resilient HTTP Connection Pooling: Wrapped downloads in requests.Session() with urllib3.util.Retry (10 retries, backoff factor 2, status codes [429, 500, 502, 503, 504], and 120s timeout) to prevent dropped connections during archive downloads.

2. Schema Declarations & De-duplication (scripts/us_epa/airdata/EPA_AirQuality.mcf)

• Retain Only Custom Methods: Retained the 21 MeasurementMethodEnum nodes (Ozone_8hour_2015, NO2_1hour_2010, SO2_1hour_2010, PM25_24hour_2012, PM10_24hour_2006, etc.) that do not exist in Google3 schema.
• Eliminated Duplicate Schemas: Removed lines 1–158 (18 canonical StatVars and 2 units already defined in manual_epa_stat_vars.mcf and unit.mcf), avoiding double-ingestion into instance_mcf_nodes.mcf.
• Eliminated Static Git Bloat: Removed all 8,404 hardcoded station nodes (lines 246–50671), reducing file size from 1.37 MB (50,671 lines) to 1.8 KB (86 lines).

3. Automation Configuration & Differ Guardrail (scripts/us_epa/airdata/manifest.json)

• Set "config_override": {"invoke_differ_tool": false} to safely bypass the in-memory differ on the 123+ GB MCF output and prevent VM memory exhaustion.
• Configured wildcard pattern "node_mcf": "EPA_AirQuality*.mcf" under import_inputs so genmcf loads both EPA_AirQuality.mcf (methods) and dynamically generated EPA_AirQuality_sites.mcf (sites) into local_nodes.
• Allocated appropriate resources: cpu: 32, memory: 128, disk: 512.

4. Import Validation Configuration (scripts/us_epa/airdata/validation_config.json)

• Disabled check_deleted_records_percent to prevent DATA_ERROR when the differ tool is bypassed.
• Strictly enforced threshold: 0 for check_missing_refs_count and check_lint_error_count.
• check_max_date_consistent (MAX_DATE_CONSISTENT): Verifies uniform MaxDate across all 18 air quality StatVars.
• check_date_freshness (SQL_VALIDATOR): Enforces dynamic 1-year publication lag freshness using DuckDB:

  "condition": "CAST(SUBSTR(latest_date, 1, 4) AS INT) >= YEAR(CURRENT_DATE) - 1"

5. Unit Tests (scripts/us_epa/airdata/air_quality_test.py, README.md)

• Added tests asserting dynamic site MCF emission, uniqueness, and cross-border monitor filtering.
• Maintained unit tests for unit mapping and CSV output generation (7/7 OK).
• Documented EPA_AirQuality_sites.mcf in README.md.

---

Verification and Test Run Proof

Cloud Batch Staging Execution

• Latest Commit: 573d49a867981d146f2371b5d0e2d0f992a4bb38 (shourya116:fix-epacriteriaairpollutants)
• Cloud Batch Job: epacriteriaairpollutants-shouryasingh-20260916-091708 (UID: epacriteriaairpoll-9beed9ad-4c41-487c0, Project: datcom-infosys-dev, Region: us-west4)
• GCS Staging Version: gs://datcom-import-test/scripts/us_epa/airdata/EPACriteriaAirPollutants/2026_09_16T02_19_58_109207_07_00
• Execution Latencies:
  • Script (CSV Extraction & Dynamic MCF Generation): ~33.9 mins (64,802,212 rows / 16.45 GB CSV / 1.27 MB sites MCF)
  • GenMCF (Transformation): ~3.23 hours (305,269,418 nodes / 123.27 GB)
  • Differ: Safely bypassed (0 min, 0 RAM consumed)
  • Validation: ~0.8 sec
  • Total Wall Clock Time: ~3.88 hours (Exit Code: 0)
• Pipeline Status: Successfully promoted to ImportStatus.STAGING.

Validation Report (input0/validation/validation_output.csv)

Validation Check	Status	Details
check_empty_import	PASSED	305,269,418 nodes, 64,802,212 rows processed
check_missing_refs_count	PASSED	0 missing references (threshold: 0)
check_lint_error_count	PASSED	0 fatal lint errors (threshold: 0)
check_max_date_consistent	PASSED	Uniform 2025-12-31 across all 18 StatVars
check_date_freshness	PASSED	Dynamic 1-year lag satisfied (latest_date = 2025-12-31 >= 2025)

Artifact & De-duplication Audit

• Double-Ingestion Check: Audited input0/genmcf/instance_mcf_nodes.mcf: confirmed 0 canonical StatisticalVariable nodes and 0 canonical UnitOfMeasure nodes (contains only 8,404 AirQualitySite, 8,337 GeoCoordinates, and 21 MeasurementMethodEnum instances).
• Local Unit Tests: python3 -m unittest discover -v -s scripts/us_epa/airdata -p "*_test.py": 7/7 tests passed.

---

References

• Diagnostic / RCA Report: https://paste.googleplex.com/5339962954874880
• Code Review Analysis (CRA): https://paste.googleplex.com/5391115948261376
• Postmortem Report: https://paste.googleplex.com/4823148197969920

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[gemini-code-assist]

Code Review

This pull request introduces a diagnostic runbook, configures validation rules, and updates the air_quality.py script to handle unit mapping, filter out cross-border monitors, and use a robust HTTP session with retries for downloading data. Feedback on the changes includes addressing a critical runtime AttributeError when configuring retries, relaxing the lint error threshold in validation_config.json to prevent pipeline failures from new stations, and resolving a potential KeyError and eager evaluation issue when parsing observation units.

[abhishekjaisw]

Review scope

• Target: PR #2206 (7cdda42598c5cce772ee173325c655b016f6d083)
• Reviewed: EPA_AirQuality.mcf, air_quality.py, air_quality_test.py, manifest.json, test_data/test_import.csv, validation_config.json
• Skipped: None

Summary of Progression

Thank you for updating the PR description with the detailed incident summaries and staging verification (epacriteriaairpollutants-shouryasingh-20260914-075748 -> gs://datcom-import-test/scripts/us_epa/airdata/EPACriteriaAirPollutants/2026_09_14T01_01_32_581690_07_00, achieving ImportStatus.STAGING with 0 missing refs and 0 lint errors), and for adding unit tests (7/7 passing).

Three remaining schema/automation items need attention before merging (see inline comments):

1. [P1] Remove duplicate canonical StatVars & Units from EPA_AirQuality.mcf (lines 1–158): Inspecting input0/genmcf/instance_mcf_nodes.mcf from the Sep 14 run confirms that genmcf ingests all 18 canonical StatVars and 2 units into instance_mcf_nodes.mcf (lines 97460 and 114284). Because genmcf's built-in schema cache already resolves canonical Google3 StatVars (manual_epa_stat_vars.mcf) and units (unit.mcf) locally without node_mcf, lines 1–158 should be removed so only the 21 MeasurementMethodEnum nodes remain in EPA_AirQuality.mcf.
2. [P1] Dynamically generate AirQualitySite nodes (EPA_AirQuality_sites.mcf) instead of hardcoding 8,404 stations in Git: Pre-populating AirQualitySite nodes via node_mcf is necessary so genmcf populates local_nodes before validating table_mcf rows—preventing remote DC API lookups and achieving 0 missing refs and 0 lint errors. However, hardcoding 8,404 static stations in Git paired with threshold: 0 means future automated cron refreshes (15 22 10 1,7 *) will fail whenever EPA adds a single new monitoring station. Having air_quality.py dynamically write unique AirQualitySite nodes to EPA_AirQuality_sites.mcf at runtime and setting "node_mcf": "EPA_AirQuality*.mcf" in manifest.json preserves threshold: 0 automatically on every future refresh without 50,000+ lines of static Git bloat.
3. [P2] Use dynamic year expression in check_date_freshness: Replace static '2024' with CAST(SUBSTR(latest_date, 1, 4) AS INT) >= YEAR(CURRENT_DATE) - 1.

Positive findings

• scripts/us_epa/airdata/air_quality.py:221-236 - Resilient HTTP Session with Exponential Backoff and Retries ✓
  • Finding: Good - Configuring requests.Session with urllib3.util.Retry (10 retries, backoff factor 2, status forcelist [429, 500, 502, 503, 504]) and 120s timeout provides strong fault tolerance across 270+ annual archive downloads.
• scripts/us_epa/airdata/air_quality.py:138-140 & air_quality_test.py:63-151 - Foreign Monitor Filtering and Unit Mapping with Unit Tests ✓
  • Finding: Good - Filtering out cross-border monitors (State Code in ('80', 'CC')) and normalizing unit strings (UNIT_MAP) cleanly resolves 875K fatal observationAbout errors, verified by unit tests (7/7 passing).
• scripts/us_epa/airdata/manifest.json:25-33 - Appropriate Resource Sizing, Differ Bypass, and PR Documentation ✓
  • Finding: Good - Allocating 32 vCPUs / 128 GiB RAM / 512 GB disk, setting invoke_differ_tool: false, and documenting the RCA and staging proof in the PR description prevents the 121.83 GB MCF in-memory diffing OOM kernel freeze.

Coverage

File	Status	Result
scripts/us_epa/airdata/EPA_AirQuality.mcf	Reviewed	Two P1 findings
scripts/us_epa/airdata/manifest.json	Reviewed	Two P1 findings
scripts/us_epa/airdata/validation_config.json	Reviewed	One P1 finding, One P2 finding
scripts/us_epa/airdata/air_quality.py	Reviewed	Positive findings
scripts/us_epa/airdata/air_quality_test.py	Reviewed	Positive findings
scripts/us_epa/airdata/test_data/test_import.csv	Reviewed	No findings

Verification and limitations

• Checks run: Executed all 7 unit tests in scripts/us_epa/airdata (7/7 OK) and verified yapf formatting (0 diffs) in detached worktree at 7cdda42598c5cce772ee173325c655b016f6d083; audited GCS staging artifacts (instance_mcf_nodes.mcf, report.json, import_summary.json) for 2026_09_10T13_01_42_117377_07_00 and 2026_09_14T01_01_32_581690_07_00; verified Google3 schema declarations via Code Search.
• Checks not run: Live download of 1980–2025 EPA AQS archives (~3.7h runtime).
• Limitations: Note that inline comments for EPA_AirQuality.mcf are attached to manifest.json and validation_config.json because EPA_AirQuality.mcf (+50,525 lines) exceeds GitHub's diff size limit for inline comments.

[shourya116]

All three review findings have been resolved and verified with a full Cloud Batch staging run on datcom-infosys-dev.

Summary of Changes (Commit 573d49a8 on shourya116:fix-epacriteriaairpollutants):

1. [P1] Removed Duplicate Canonical StatVars & Units: Stripped lines 1–158 from EPA_AirQuality.mcf, retaining only the 21 MeasurementMethodEnum nodes. Canonical StatVars and units now resolve purely through genmcf's schema cache without duplicate instance node ingestion.
2. [P1] Dynamically Generated Station MCF (EPA_AirQuality_sites.mcf): Updated air_quality.py (write_csv) to dynamically write unique AirQualitySite nodes at runtime, removed the 8,404 hardcoded station nodes (lines 246–50671) from EPA_AirQuality.mcf (-1.37 MB), and updated manifest.json to "node_mcf": "EPA_AirQuality*.mcf". Future cron runs will automatically capture any new upstream EPA stations with zero manual maintenance.
3. [P2] Dynamic Date Freshness: Updated check_date_freshness in validation_config.json to CAST(SUBSTR(latest_date, 1, 4) AS INT) >= YEAR(CURRENT_DATE) - 1.
4. Unit Tests: Updated air_quality_test.py to verify dynamic site MCF emission, uniqueness, and cross-border monitor filtering (7/7 OK).

---

Staging Verification

• Cloud Batch Job: epacriteriaairpollutants-shouryasingh-20260916-091708 (UID: epacriteriaairpoll-9beed9ad-4c41-487c0, datcom-infosys-dev, us-west4)
• GCS Staging Version: gs://datcom-import-test/scripts/us_epa/airdata/EPACriteriaAirPollutants/2026_09_16T02_19_58_109207_07_00
• Status: ImportStatus.STAGING (Exit Code 0, 64,802,212 rows processed, 305,269,418 MCF nodes generated in 3.88 hours).
• Double-Ingestion Check: Audited input0/genmcf/instance_mcf_nodes.mcf: 0 StatVars, 0 Units (only 8,404 AirQualitySite, 8,337 GeoCoordinates, and 21 MeasurementMethodEnum).
• Validation Output (validation_output.csv):
  • check_empty_import: PASSED (305,269,418 nodes, 64,802,212 rows)
  • check_missing_refs_count: PASSED (0 missing refs, threshold: 0)
  • check_lint_error_count: PASSED (0 lint errors, threshold: 0)
  • check_max_date_consistent: PASSED (18/18 StatVars consistent at MaxDate = 2025-12-31)
  • check_date_freshness: PASSED (CAST(SUBSTR(latest_date, 1, 4) AS INT) >= YEAR(CURRENT_DATE) - 1)