From 44d036e7cfde0f9f539a9b7a8042268c1034ad41 Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Mon, 10 Aug 2026 09:07:00 +0000
Subject: [PATCH 01/13] Update CDC500 state query to use new TimeSeries and
 Observation tables

---
 scripts/us_cdc/cdc500_state/process.py | 99 +++++++++++++++-----------
 1 file changed, 58 insertions(+), 41 deletions(-)

diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index 0115b96e43..ea9b2ecbe7 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -22,49 +22,66 @@
     os.mkdir(_OUTPUT_FILE_PATH)
 
 query = """
-SELECT distinct * from(
-SELECT 
-  statvar, 
-  SUBSTR(observation_about,0,8) as observation_about,
-  observation_date, 
-  CONCAT('dcAggregate/',measurement_method) as measurement_method,
-  population_statvar,
-  SUM(CAST(pop_count AS FLOAT64))*100/SUM(CAST(population AS FLOAT64)) as percent
-FROM 
-(
+WITH cdc_sv AS (
   SELECT
-    SVO1.variable_measured as statvar,
-    SVO1.observation_about as observation_about,
-    SVO1.observation_date as observation_date,
-    SVO1.value as percent,
-    SVO1.measurement_method as measurement_method,
-    SVO2.variable_measured as population_statvar,
-    SVO2.value as population,
-    CAST(SVO2.value AS FLOAT64) * CAST(SVO1.value AS FLOAT64) / 100 as pop_count
-  FROM `datcom-store.dc_kg_latest.StatVarObservation` as SVO1
-  JOIN `datcom-store.dc_kg_latest.StatVarObservation` as SVO2 ON TRUE
-  JOIN (
-    # Get the statvars and corresponding population statvar
-    # with ‘Percent_’ replaced with ‘Count_’ and
-    # dropping the non-age, non-gender constraints.
+    variable_measured AS cdc500,
+    CONCAT('Count_', REGEXP_SUBSTR(variable_measured, '(Person_.*ale|Person_.*Years|Person)')) AS pop_statvar
+  FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries`
+  WHERE provenance = 'dc/base/CDC500'
+    AND variable_measured LIKE 'Percent_%'
+  GROUP BY cdc500, pop_statvar
+),
+svo_percent AS (
+  SELECT
+    O.variable_measured AS statvar,
+    O.entity1 AS observation_about,
+    O.date AS observation_date,
+    O.value AS percent,
+    T.measurement_method AS measurement_method,
+    cdc_sv.pop_statvar
+  FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation` AS O
+  JOIN `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` AS T
+    ON O.variable_measured = T.variable_measured
+    AND O.entity1 = T.entity1
+    AND O.facet_id = T.facet_id
+  JOIN cdc_sv ON O.variable_measured = cdc_sv.cdc500
+  WHERE O.entity1 LIKE 'geoId/%'
+),
+svo_count AS (
+  SELECT
+    variable_measured AS population_statvar,
+    entity1 AS observation_about,
+    date AS observation_date,
+    value AS population
+  FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation`
+  WHERE entity1 LIKE 'geoId/%'
+    AND variable_measured IN (SELECT DISTINCT pop_statvar FROM cdc_sv)
+)
+SELECT DISTINCT * FROM (
+  SELECT 
+    statvar, 
+    SUBSTR(observation_about, 0, 8) AS observation_about,
+    observation_date, 
+    CONCAT('dcAggregate/', measurement_method) AS measurement_method,
+    population_statvar,
+    SUM(CAST(pop_count AS FLOAT64)) * 100 / SUM(CAST(population AS FLOAT64)) AS percent
+  FROM (
     SELECT
-      SVO.variable_measured as CDC500,
-      CONCAT('Count_', REGEXP_SUBSTR(SVO.variable_measured, '(Person_.*ale|Person_.*Years|Person)')) as pop_statvar
-      FROM `datcom-store.dc_kg_latest.StatVarObservation` as SVO
-    WHERE
-      SVO.prov_id = 'dc/base/CDC500'
-      AND SVO.variable_measured like 'Percent_%'
-      GROUP BY CDC500, pop_statvar
-   ) AS CDC_SV ON TRUE
-  WHERE
-    SVO1.prov_id = 'dc/base/CDC500'
-    AND SVO1.variable_measured LIKE 'Percent%'
-    AND SVO1.observation_about = SVO2.observation_about
-    AND SVO1.observation_date = SVO2.observation_date
-    AND SVO1.variable_measured = CDC_SV.CDC500
-    AND SVO2.variable_measured = CDC_SV.pop_statvar
-    AND SVO1.observation_about like "geoId/%"
-) group by 1,2,3,4,5
+      p.statvar,
+      p.observation_about,
+      p.observation_date,
+      p.percent,
+      p.measurement_method,
+      c.population_statvar,
+      c.population,
+      CAST(c.population AS FLOAT64) * CAST(p.percent AS FLOAT64) / 100 AS pop_count
+    FROM svo_percent AS p
+    JOIN svo_count AS c
+      ON p.observation_about = c.observation_about
+      AND p.observation_date = c.observation_date
+      AND p.pop_statvar = c.population_statvar
+  )
+  GROUP BY 1, 2, 3, 4, 5
 )
 """
 

From ebc79b75c1a48b9163aa7f0d88c16a43275e0609 Mon Sep 17 00:00:00 2001
From: shvngisingh <shvngisingh@google.com>
Date: Mon, 10 Aug 2026 17:16:40 +0530
Subject: [PATCH 02/13] Update scripts/us_cdc/cdc500_state/process.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 scripts/us_cdc/cdc500_state/process.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index ea9b2ecbe7..63c8e25bda 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -64,7 +64,7 @@
     observation_date, 
     CONCAT('dcAggregate/', measurement_method) AS measurement_method,
     population_statvar,
-    SUM(CAST(pop_count AS FLOAT64)) * 100 / SUM(CAST(population AS FLOAT64)) AS percent
+    SAFE_DIVIDE(SUM(CAST(pop_count AS FLOAT64)) * 100, SUM(CAST(population AS FLOAT64))) AS percent
   FROM (
     SELECT
       p.statvar,

From fdcdbb5378a334274919b364df6182b339de8903 Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Thu, 27 Aug 2026 11:43:10 +0000
Subject: [PATCH 03/13] Address review comments: optimize joins, add provenance
 filters, and streamline aggregation in CDC500 state query

---
 scripts/us_cdc/cdc500_state/process.py | 71 +++++++++++++-------------
 1 file changed, 36 insertions(+), 35 deletions(-)

diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index 63c8e25bda..36e1f11af6 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -31,6 +31,7 @@
     AND variable_measured LIKE 'Percent_%'
   GROUP BY cdc500, pop_statvar
 ),
+
 svo_percent AS (
   SELECT
     O.variable_measured AS statvar,
@@ -40,49 +41,49 @@
     T.measurement_method AS measurement_method,
     cdc_sv.pop_statvar
   FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation` AS O
-  JOIN `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` AS T
+  INNER JOIN `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` AS T
     ON O.variable_measured = T.variable_measured
     AND O.entity1 = T.entity1
     AND O.facet_id = T.facet_id
-  JOIN cdc_sv ON O.variable_measured = cdc_sv.cdc500
+    AND T.provenance = 'dc/base/CDC500'
+    AND T.variable_measured LIKE 'Percent_%'
+  INNER JOIN cdc_sv
+    ON O.variable_measured = cdc_sv.cdc500
   WHERE O.entity1 LIKE 'geoId/%'
+    AND O.variable_measured LIKE 'Percent_%'
 ),
+
 svo_count AS (
   SELECT
-    variable_measured AS population_statvar,
-    entity1 AS observation_about,
-    date AS observation_date,
-    value AS population
-  FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation`
-  WHERE entity1 LIKE 'geoId/%'
-    AND variable_measured IN (SELECT DISTINCT pop_statvar FROM cdc_sv)
-)
-SELECT DISTINCT * FROM (
-  SELECT 
-    statvar, 
-    SUBSTR(observation_about, 0, 8) AS observation_about,
-    observation_date, 
-    CONCAT('dcAggregate/', measurement_method) AS measurement_method,
-    population_statvar,
-    SAFE_DIVIDE(SUM(CAST(pop_count AS FLOAT64)) * 100, SUM(CAST(population AS FLOAT64))) AS percent
-  FROM (
-    SELECT
-      p.statvar,
-      p.observation_about,
-      p.observation_date,
-      p.percent,
-      p.measurement_method,
-      c.population_statvar,
-      c.population,
-      CAST(c.population AS FLOAT64) * CAST(p.percent AS FLOAT64) / 100 AS pop_count
-    FROM svo_percent AS p
-    JOIN svo_count AS c
-      ON p.observation_about = c.observation_about
-      AND p.observation_date = c.observation_date
-      AND p.pop_statvar = c.population_statvar
-  )
-  GROUP BY 1, 2, 3, 4, 5
+    O.variable_measured AS population_statvar,
+    O.entity1 AS observation_about,
+    O.date AS observation_date,
+    O.value AS population
+  FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation` AS O
+  INNER JOIN (
+    SELECT DISTINCT pop_statvar
+    FROM cdc_sv
+  ) AS pop
+    ON O.variable_measured = pop.pop_statvar
+  WHERE O.entity1 LIKE 'geoId/%'
 )
+
+SELECT 
+  p.statvar, 
+  SUBSTR(p.observation_about, 0, 8) AS observation_about,
+  p.observation_date, 
+  CONCAT('dcAggregate/', p.measurement_method) AS measurement_method,
+  p.pop_statvar AS population_statvar,
+  SAFE_DIVIDE(
+    SUM(CAST(c.population AS FLOAT64) * CAST(p.percent AS FLOAT64) / 100) * 100,
+    SUM(CAST(c.population AS FLOAT64))
+  ) AS percent
+FROM svo_percent AS p
+INNER JOIN svo_count AS c
+  ON p.observation_about = c.observation_about
+  AND p.observation_date = c.observation_date
+  AND p.pop_statvar = c.population_statvar
+GROUP BY 1, 2, 3, 4, 5
 """
 
 client = bigquery.Client()

From d2d76104f7bccc60788edffee505d554a91ed1f4 Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Mon, 31 Aug 2026 11:51:38 +0000
Subject: [PATCH 04/13] Address code review findings for CDC500 state
 aggregation: fix Census ACS 5Yr facet join, explicit cohort CASE mapping,
 modular entrypoint, and add unit test suite

---
 scripts/us_cdc/cdc500_state/README.md       |  4 +-
 scripts/us_cdc/cdc500_state/manifest.json   |  3 -
 scripts/us_cdc/cdc500_state/process.py      | 83 +++++++++++++++------
 scripts/us_cdc/cdc500_state/process_test.py | 66 ++++++++++++++++
 4 files changed, 129 insertions(+), 27 deletions(-)
 create mode 100644 scripts/us_cdc/cdc500_state/process_test.py

diff --git a/scripts/us_cdc/cdc500_state/README.md b/scripts/us_cdc/cdc500_state/README.md
index 79f5abb00c..07ce31a391 100644
--- a/scripts/us_cdc/cdc500_state/README.md
+++ b/scripts/us_cdc/cdc500_state/README.md
@@ -12,7 +12,7 @@ Author: Padma Gundapaneni @padma-g
 ## About the Dataset
 
 ### Overview
-The state level data is aggragated from city level data coming from CDC500 import. 
+The state level data is aggregated from city level data coming from CDC500 import. 
 
 To get the data for this import run:
 ```bash
@@ -24,7 +24,7 @@ $ python3 process.py
 ### Artifacts
 
 #### Scripts
-[`process.py`](https://github.com/datacommonsorg/data/blob/master//scripts/us_cdc/cdc500_state/process.py)
+[`process.py`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/process.py)
 
 
 #### tMCFs
diff --git a/scripts/us_cdc/cdc500_state/manifest.json b/scripts/us_cdc/cdc500_state/manifest.json
index eae459b1c6..1d3909a12f 100644
--- a/scripts/us_cdc/cdc500_state/manifest.json
+++ b/scripts/us_cdc/cdc500_state/manifest.json
@@ -21,9 +21,6 @@
                 "memory": 64,
                 "disk": 100
             },
-            "source_files": [
-                "CDC500State_Output/CDC500State_Output.csv"
-            ],
             "cron_schedule": "0 1 * * 1"
         }
     ]
diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index 36e1f11af6..6e5d17877d 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -1,4 +1,4 @@
-# Copyright 2021 Google LLC
+# Copyright 2026 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -11,21 +11,36 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+"""Processes CDC 500 cities data into aggregated state-level health indicators."""
 
 import os
+from absl import app
+from absl import flags
 from absl import logging
 from google.cloud import bigquery
+import pandas as pd
 
+_FLAGS = flags.FLAGS
 _MODULE_DIR = os.path.dirname(os.path.abspath(__file__))
-_OUTPUT_FILE_PATH = os.path.join(_MODULE_DIR + '/CDC500State_Output')
-if not os.path.exists(_OUTPUT_FILE_PATH):
-    os.mkdir(_OUTPUT_FILE_PATH)
+_DEFAULT_OUTPUT_DIR = os.path.join(_MODULE_DIR, 'CDC500State_Output')
 
-query = """
+flags.DEFINE_string('output_dir', _DEFAULT_OUTPUT_DIR,
+                    'Directory to write output CSV.')
+
+QUERY = """
 WITH cdc_sv AS (
   SELECT
     variable_measured AS cdc500,
-    CONCAT('Count_', REGEXP_SUBSTR(variable_measured, '(Person_.*ale|Person_.*Years|Person)')) AS pop_statvar
+    CASE
+      WHEN variable_measured LIKE '%Female_50To74Years%' OR variable_measured LIKE '%50To74Years_Female%' THEN 'Count_Person_Female_50To74Years'
+      WHEN variable_measured LIKE '%Female_21To65Years%' OR variable_measured LIKE '%21To65Years_Female%' THEN 'Count_Person_Female_21To65Years'
+      WHEN variable_measured LIKE '%Female_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Female%' THEN 'Count_Person_Female_65OrMoreYears'
+      WHEN variable_measured LIKE '%Male_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Male%' THEN 'Count_Person_Male_65OrMoreYears'
+      WHEN variable_measured LIKE '%65OrMoreYears%' THEN 'Count_Person_65OrMoreYears'
+      WHEN variable_measured LIKE '%18To64Years%' THEN 'Count_Person_18To64Years'
+      WHEN variable_measured LIKE '%18OrMoreYears%' THEN 'Count_Person_18OrMoreYears'
+      ELSE 'Count_Person'
+    END AS pop_statvar
   FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries`
   WHERE provenance = 'dc/base/CDC500'
     AND variable_measured LIKE 'Percent_%'
@@ -60,6 +75,11 @@
     O.date AS observation_date,
     O.value AS population
   FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation` AS O
+  INNER JOIN `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` AS T
+    ON O.variable_measured = T.variable_measured
+    AND O.entity1 = T.entity1
+    AND O.facet_id = T.facet_id
+    AND T.provenance = 'dc/base/CensusACS5YearSurvey'
   INNER JOIN (
     SELECT DISTINCT pop_statvar
     FROM cdc_sv
@@ -70,12 +90,12 @@
 
 SELECT 
   p.statvar, 
-  SUBSTR(p.observation_about, 0, 8) AS observation_about,
+  SUBSTR(p.observation_about, 1, 8) AS observation_about,
   p.observation_date, 
   CONCAT('dcAggregate/', p.measurement_method) AS measurement_method,
   p.pop_statvar AS population_statvar,
   SAFE_DIVIDE(
-    SUM(CAST(c.population AS FLOAT64) * CAST(p.percent AS FLOAT64) / 100) * 100,
+    SUM(CAST(c.population AS FLOAT64) * CAST(p.percent AS FLOAT64)),
     SUM(CAST(c.population AS FLOAT64))
   ) AS percent
 FROM svo_percent AS p
@@ -86,18 +106,37 @@
 GROUP BY 1, 2, 3, 4, 5
 """
 
-client = bigquery.Client()
-try:
-    logging.info("Running the query")
-    query_job = client.query(query)
-except Exception as e:
-    logging.fatal(f"Error faced while running the query {e}")
-try:
-    logging.info("Converting to dataframe")
-    results = query_job.to_dataframe()
-except Exception as e:
-    logging.info(f"Error faced while fetching results: {e}")
+def get_query() -> str:
+    """Returns the SQL query string for CDC 500 state aggregation."""
+    return QUERY
+
+def run_process(client: bigquery.Client, output_file: str) -> pd.DataFrame:
+    """Executes the BigQuery query and writes the resulting DataFrame to output_file."""
+    logging.info("Running BigQuery aggregation query...")
+    try:
+        query_job = client.query(get_query())
+    except Exception as e:
+        logging.fatal("Failed to submit BigQuery query: %s", e)
+        raise
+
+    logging.info("Fetching query results into dataframe...")
+    try:
+        df = query_job.to_dataframe()
+    except Exception as e:
+        logging.fatal("Failed to fetch query results into dataframe: %s", e)
+        raise
+
+    output_dir = os.path.dirname(output_file)
+    os.makedirs(output_dir, exist_ok=True)
+    logging.info("Writing %d rows to %s", len(df), output_file)
+    df.to_csv(output_file, index=False)
+    return df
+
+def main(argv):
+    del argv  # Unused.
+    client = bigquery.Client()
+    output_file = os.path.join(_FLAGS.output_dir, 'CDC500State_Output.csv')
+    run_process(client, output_file)
 
-logging.info("Writing output to CSV")
-output_file = os.path.join(_OUTPUT_FILE_PATH + "/CDC500State_Output.csv")
-results.to_csv(output_file, index=False)
+if __name__ == '__main__':
+    app.run(main)
diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py
new file mode 100644
index 0000000000..d00cc29dec
--- /dev/null
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -0,0 +1,66 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Unit tests for CDC 500 State aggregation script."""
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+import pandas as pd
+
+from scripts.us_cdc.cdc500_state import process
+
+class CDC500StateProcessTest(unittest.TestCase):
+
+    def test_get_query(self):
+        query = process.get_query()
+        self.assertIn("spanner_dc_graph_prod_DEFAULT.TimeSeries", query)
+        self.assertIn("spanner_dc_graph_prod_DEFAULT.Observation", query)
+        self.assertIn("dc/base/CDC500", query)
+        self.assertIn("dc/base/CensusACS5YearSurvey", query)
+        self.assertIn("SAFE_DIVIDE", query)
+        self.assertIn("SUBSTR(p.observation_about, 1, 8)", query)
+
+    def test_run_process_success(self):
+        mock_client = mock.MagicMock()
+        sample_data = pd.DataFrame({
+            'statvar': ['Percent_Person_18OrMoreYears_WithAnyDisability'],
+            'observation_about': ['geoId/06'],
+            'observation_date': ['2022'],
+            'measurement_method': ['dcAggregate/CrudePrevalence'],
+            'population_statvar': ['Count_Person_18OrMoreYears'],
+            'percent': [29.6479]
+        })
+        mock_client.query.return_value.to_dataframe.return_value = sample_data
+        
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
+            result_df = process.run_process(mock_client, output_file)
+            
+            mock_client.query.assert_called_once()
+            self.assertTrue(os.path.exists(output_file))
+            saved_df = pd.read_csv(output_file)
+            self.assertEqual(len(saved_df), 1)
+            self.assertEqual(saved_df['observation_about'].iloc[0], 'geoId/06')
+
+    def test_run_process_query_error(self):
+        mock_client = mock.MagicMock()
+        mock_client.query.side_effect = Exception("BigQuery Access Denied")
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
+            with self.assertRaises(Exception):
+                process.run_process(mock_client, output_file)
+
+if __name__ == '__main__':
+    unittest.main()

From 3415550706c01dd2366e579e2e325d2736a57dca Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Mon, 31 Aug 2026 12:52:34 +0000
Subject: [PATCH 05/13] Add validation_config.json, update manifest, add tMCF
 units and guard output path

---
 scripts/us_cdc/cdc500_state/cdc500_state.tmcf |  2 +
 scripts/us_cdc/cdc500_state/manifest.json     |  3 +-
 scripts/us_cdc/cdc500_state/process.py        |  3 +-
 .../cdc500_state/validation_config.json       | 54 +++++++++++++++++++
 4 files changed, 60 insertions(+), 2 deletions(-)
 create mode 100644 scripts/us_cdc/cdc500_state/validation_config.json

diff --git a/scripts/us_cdc/cdc500_state/cdc500_state.tmcf b/scripts/us_cdc/cdc500_state/cdc500_state.tmcf
index 73a294f9cf..b7d6106b85 100644
--- a/scripts/us_cdc/cdc500_state/cdc500_state.tmcf
+++ b/scripts/us_cdc/cdc500_state/cdc500_state.tmcf
@@ -4,5 +4,7 @@ variableMeasured: C:CDC->statvar
 observationAbout: C:CDC->observation_about
 observationDate: C:CDC->observation_date
 value: C:CDC->percent
+unit: Percent
+scalingFactor: 100
 measurementMethod: C:CDC->measurement_method
 observationPeriod: "P1Y"
\ No newline at end of file
diff --git a/scripts/us_cdc/cdc500_state/manifest.json b/scripts/us_cdc/cdc500_state/manifest.json
index 1d3909a12f..4a2ac714b8 100644
--- a/scripts/us_cdc/cdc500_state/manifest.json
+++ b/scripts/us_cdc/cdc500_state/manifest.json
@@ -21,7 +21,8 @@
                 "memory": 64,
                 "disk": 100
             },
-            "cron_schedule": "0 1 * * 1"
+            "cron_schedule": "0 1 * * 1",
+            "validation_config_file": "validation_config.json"
         }
     ]
 }
\ No newline at end of file
diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index 6e5d17877d..c1b2608b4c 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -127,7 +127,8 @@ def run_process(client: bigquery.Client, output_file: str) -> pd.DataFrame:
         raise
 
     output_dir = os.path.dirname(output_file)
-    os.makedirs(output_dir, exist_ok=True)
+    if output_dir:
+        os.makedirs(output_dir, exist_ok=True)
     logging.info("Writing %d rows to %s", len(df), output_file)
     df.to_csv(output_file, index=False)
     return df
diff --git a/scripts/us_cdc/cdc500_state/validation_config.json b/scripts/us_cdc/cdc500_state/validation_config.json
new file mode 100644
index 0000000000..ec7154e2fd
--- /dev/null
+++ b/scripts/us_cdc/cdc500_state/validation_config.json
@@ -0,0 +1,54 @@
+{
+    "schema_version": "1.0",
+    "rules": [
+        {
+            "rule_id": "check_deleted_records_percent",
+            "description": "Checks that the percentage of deleted points is within the threshold.",
+            "validator": "DELETED_RECORDS_PERCENT",
+            "params": {
+                "threshold": 0
+            }
+        },
+        {
+            "rule_id": "check_missing_refs_count",
+            "description": "Checks that there are no missing entity references in lint report.",
+            "validator": "MISSING_REFS_COUNT",
+            "params": {
+                "threshold": 0
+            }
+        },
+        {
+            "rule_id": "check_lint_error_count",
+            "description": "Checks that there are no lint errors during MCF generation.",
+            "validator": "LINT_ERROR_COUNT",
+            "params": {
+                "threshold": 0
+            }
+        },
+        {
+            "rule_id": "check_max_value_percentage",
+            "description": "Checks that all percentage StatVars do not exceed 100%.",
+            "validator": "MAX_VALUE_CHECK",
+            "params": {
+                "maximum": 100
+            }
+        },
+        {
+            "rule_id": "check_min_value_percentage",
+            "description": "Checks that all percentage StatVars are not below 0%.",
+            "validator": "MIN_VALUE_CHECK",
+            "params": {
+                "minimum": 0
+            }
+        },
+        {
+            "rule_id": "check_num_places_state_count",
+            "description": "Checks that state-level observations cover all 50-52 US state entities.",
+            "validator": "NUM_PLACES_COUNT",
+            "params": {
+                "minimum": 50,
+                "maximum": 52
+            }
+        }
+    ]
+}

From dbc69620be7711e6437a2472c42e4bc3cbe0f1f1 Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Tue, 1 Sep 2026 13:04:43 +0000
Subject: [PATCH 06/13] Address code review feedback on CDC500 state
 aggregation: update README docs, remove default validation rules, use
 logging.error, and clean up query helpers

---
 scripts/us_cdc/cdc500_state/README.md         | 39 +++++++++++++++----
 scripts/us_cdc/cdc500_state/process.py        | 13 ++-----
 scripts/us_cdc/cdc500_state/process_test.py   |  6 +--
 .../cdc500_state/validation_config.json       | 24 ------------
 4 files changed, 39 insertions(+), 43 deletions(-)

diff --git a/scripts/us_cdc/cdc500_state/README.md b/scripts/us_cdc/cdc500_state/README.md
index 07ce31a391..e651276c93 100644
--- a/scripts/us_cdc/cdc500_state/README.md
+++ b/scripts/us_cdc/cdc500_state/README.md
@@ -5,6 +5,8 @@ Author: Padma Gundapaneni @padma-g
 ## Table of Contents
 1. [About the Dataset](#about-the-dataset)
     1. [Overview](#overview)
+    2. [Data Sources and Tables](#data-sources-and-tables)
+    3. [Aggregation Methodology](#aggregation-methodology)
 2. [About the Import](#about-the-import)
     1. [Artifacts](#artifacts)
     2. [Import Procedure](#import-procedure)
@@ -12,12 +14,30 @@ Author: Padma Gundapaneni @padma-g
 ## About the Dataset
 
 ### Overview
-The state level data is aggregated from city level data coming from CDC500 import. 
+The state-level dataset calculates aggregated health indicator prevalence estimates for US states from the city-level CDC 500 Cities (`CDC500`) project data, weighted by corresponding Census ACS 5-Year population counts.
 
-To get the data for this import run:
-```bash
-$ python3 process.py
-```
+### Data Sources and Tables
+
+The aggregation script queries Google Cloud BigQuery graph tables in dataset `datcom-store.spanner_dc_graph_prod_DEFAULT`:
+
+1. **`TimeSeries`**:
+   - **CDC 500 Series**: Identifies CDC 500 Statistical Variables (`provenance = 'dc/base/CDC500'` and `variable_measured LIKE 'Percent_%'`) and extracts their measurement methods (`measurement_method`). It maps each percentage health metric to its appropriate denominator demographic cohort StatVar (e.g., `Count_Person_18OrMoreYears`, `Count_Person_Female_50To74Years`, `Count_Person_Female_21To65Years`, `Count_Person_65OrMoreYears`, etc.).
+   - **Census ACS 5-Year Series**: Filters and joins population counts from Census ACS 5-Year Survey (`provenance = 'dc/base/CensusACS5YearSurvey'`).
+
+2. **`Observation`**:
+   - **Health Indicator Percentages**: Fetches city-level percentage values (`value AS percent`), observation dates (`date`), and city geoIds (`entity1 LIKE 'geoId/%'`) for CDC 500 StatVars.
+   - **City Cohort Populations**: Fetches city-level population counts (`value AS population`) for the corresponding demographic cohort StatVars.
+
+### Aggregation Methodology
+
+For each state, indicator StatVar, and observation date:
+- City observations are joined with their corresponding demographic population counts.
+- City geoIds (`geoId/XXXXXXX`) are mapped to state geoIds (`geoId/XX`) using the first 8 characters (including the prefix).
+- State-level prevalence percentages are computed as a population-weighted average:
+
+$$\text{State Percent} = \frac{\sum (\text{City Population} \times \text{City Percent})}{\sum \text{City Population}}$$
+
+The output measurement method is prefixed with `dcAggregate/` (e.g., `dcAggregate/CrudePrevalence`).
 
 ## About the Import
 
@@ -26,15 +46,20 @@ $ python3 process.py
 #### Scripts
 [`process.py`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/process.py)
 
+#### Unit Tests
+[`process_test.py`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/process_test.py)
 
-#### tMCFs
+#### tMCF Template
 [`cdc500_state.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/cdc500_state.tmcf)
 
+#### Validation Config
+[`validation_config.json`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/validation_config.json)
+
 ### Import Procedure
 
 #### Data Download and Processing Steps
 
-To get the data for this import run:
+To run the BigQuery aggregation and generate the output CSV:
 
 ```bash
 $ python3 process.py
diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index c1b2608b4c..e20cdf8a11 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -106,24 +106,20 @@
 GROUP BY 1, 2, 3, 4, 5
 """
 
-def get_query() -> str:
-    """Returns the SQL query string for CDC 500 state aggregation."""
-    return QUERY
-
-def run_process(client: bigquery.Client, output_file: str) -> pd.DataFrame:
+def run_process(client: bigquery.Client, output_file: str) -> None:
     """Executes the BigQuery query and writes the resulting DataFrame to output_file."""
     logging.info("Running BigQuery aggregation query...")
     try:
-        query_job = client.query(get_query())
+        query_job = client.query(QUERY)
     except Exception as e:
-        logging.fatal("Failed to submit BigQuery query: %s", e)
+        logging.error("Failed to submit BigQuery query: %s", e)
         raise
 
     logging.info("Fetching query results into dataframe...")
     try:
         df = query_job.to_dataframe()
     except Exception as e:
-        logging.fatal("Failed to fetch query results into dataframe: %s", e)
+        logging.error("Failed to fetch query results into dataframe: %s", e)
         raise
 
     output_dir = os.path.dirname(output_file)
@@ -131,7 +127,6 @@ def run_process(client: bigquery.Client, output_file: str) -> pd.DataFrame:
         os.makedirs(output_dir, exist_ok=True)
     logging.info("Writing %d rows to %s", len(df), output_file)
     df.to_csv(output_file, index=False)
-    return df
 
 def main(argv):
     del argv  # Unused.
diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py
index d00cc29dec..eaec37e2f0 100644
--- a/scripts/us_cdc/cdc500_state/process_test.py
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -23,8 +23,8 @@
 
 class CDC500StateProcessTest(unittest.TestCase):
 
-    def test_get_query(self):
-        query = process.get_query()
+    def test_query_constants(self):
+        query = process.QUERY
         self.assertIn("spanner_dc_graph_prod_DEFAULT.TimeSeries", query)
         self.assertIn("spanner_dc_graph_prod_DEFAULT.Observation", query)
         self.assertIn("dc/base/CDC500", query)
@@ -46,7 +46,7 @@ def test_run_process_success(self):
         
         with tempfile.TemporaryDirectory() as tmp_dir:
             output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
-            result_df = process.run_process(mock_client, output_file)
+            process.run_process(mock_client, output_file)
             
             mock_client.query.assert_called_once()
             self.assertTrue(os.path.exists(output_file))
diff --git a/scripts/us_cdc/cdc500_state/validation_config.json b/scripts/us_cdc/cdc500_state/validation_config.json
index ec7154e2fd..5ac0fba24f 100644
--- a/scripts/us_cdc/cdc500_state/validation_config.json
+++ b/scripts/us_cdc/cdc500_state/validation_config.json
@@ -1,30 +1,6 @@
 {
     "schema_version": "1.0",
     "rules": [
-        {
-            "rule_id": "check_deleted_records_percent",
-            "description": "Checks that the percentage of deleted points is within the threshold.",
-            "validator": "DELETED_RECORDS_PERCENT",
-            "params": {
-                "threshold": 0
-            }
-        },
-        {
-            "rule_id": "check_missing_refs_count",
-            "description": "Checks that there are no missing entity references in lint report.",
-            "validator": "MISSING_REFS_COUNT",
-            "params": {
-                "threshold": 0
-            }
-        },
-        {
-            "rule_id": "check_lint_error_count",
-            "description": "Checks that there are no lint errors during MCF generation.",
-            "validator": "LINT_ERROR_COUNT",
-            "params": {
-                "threshold": 0
-            }
-        },
         {
             "rule_id": "check_max_value_percentage",
             "description": "Checks that all percentage StatVars do not exceed 100%.",

From 74dd551c3a9e44afa0529a8c68ea0009f213bfbe Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Wed, 2 Sep 2026 06:26:08 +0000
Subject: [PATCH 07/13] Address code review findings: remove unused pandas
 import, add dataframe error test, and fix senior cohort statvars

---
 scripts/us_cdc/cdc500_state/process.py      |  5 ++---
 scripts/us_cdc/cdc500_state/process_test.py | 10 ++++++++++
 2 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index e20cdf8a11..f4476810f3 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -18,7 +18,6 @@
 from absl import flags
 from absl import logging
 from google.cloud import bigquery
-import pandas as pd
 
 _FLAGS = flags.FLAGS
 _MODULE_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -34,8 +33,8 @@
     CASE
       WHEN variable_measured LIKE '%Female_50To74Years%' OR variable_measured LIKE '%50To74Years_Female%' THEN 'Count_Person_Female_50To74Years'
       WHEN variable_measured LIKE '%Female_21To65Years%' OR variable_measured LIKE '%21To65Years_Female%' THEN 'Count_Person_Female_21To65Years'
-      WHEN variable_measured LIKE '%Female_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Female%' THEN 'Count_Person_Female_65OrMoreYears'
-      WHEN variable_measured LIKE '%Male_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Male%' THEN 'Count_Person_Male_65OrMoreYears'
+      WHEN variable_measured LIKE '%Female_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Female%' THEN 'Count_Person_65OrMoreYears_Female'
+      WHEN variable_measured LIKE '%Male_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Male%' THEN 'Count_Person_65OrMoreYears_Male'
       WHEN variable_measured LIKE '%65OrMoreYears%' THEN 'Count_Person_65OrMoreYears'
       WHEN variable_measured LIKE '%18To64Years%' THEN 'Count_Person_18To64Years'
       WHEN variable_measured LIKE '%18OrMoreYears%' THEN 'Count_Person_18OrMoreYears'
diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py
index eaec37e2f0..c013d94153 100644
--- a/scripts/us_cdc/cdc500_state/process_test.py
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -62,5 +62,15 @@ def test_run_process_query_error(self):
             with self.assertRaises(Exception):
                 process.run_process(mock_client, output_file)
 
+    def test_run_process_dataframe_error(self):
+        mock_client = mock.MagicMock()
+        mock_query_job = mock.MagicMock()
+        mock_query_job.to_dataframe.side_effect = Exception("Failed to fetch dataframe")
+        mock_client.query.return_value = mock_query_job
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
+            with self.assertRaises(Exception):
+                process.run_process(mock_client, output_file)
+
 if __name__ == '__main__':
     unittest.main()

From f7d2dfc868e6db08865fff523b16953fb79e164e Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Wed, 2 Sep 2026 07:34:48 +0000
Subject: [PATCH 08/13] Return True on successful process completion and update
 unit tests

---
 scripts/us_cdc/cdc500_state/process.py      |  6 +++++-
 scripts/us_cdc/cdc500_state/process_test.py | 11 +++++++----
 2 files changed, 12 insertions(+), 5 deletions(-)

diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index f4476810f3..1b39961d6e 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -105,7 +105,8 @@
 GROUP BY 1, 2, 3, 4, 5
 """
 
-def run_process(client: bigquery.Client, output_file: str) -> None:
+
+def run_process(client: bigquery.Client, output_file: str) -> bool:
     """Executes the BigQuery query and writes the resulting DataFrame to output_file."""
     logging.info("Running BigQuery aggregation query...")
     try:
@@ -126,6 +127,8 @@ def run_process(client: bigquery.Client, output_file: str) -> None:
         os.makedirs(output_dir, exist_ok=True)
     logging.info("Writing %d rows to %s", len(df), output_file)
     df.to_csv(output_file, index=False)
+    return True
+
 
 def main(argv):
     del argv  # Unused.
@@ -133,5 +136,6 @@ def main(argv):
     output_file = os.path.join(_FLAGS.output_dir, 'CDC500State_Output.csv')
     run_process(client, output_file)
 
+
 if __name__ == '__main__':
     app.run(main)
diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py
index c013d94153..93d18f8274 100644
--- a/scripts/us_cdc/cdc500_state/process_test.py
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -21,6 +21,7 @@
 
 from scripts.us_cdc.cdc500_state import process
 
+
 class CDC500StateProcessTest(unittest.TestCase):
 
     def test_query_constants(self):
@@ -43,11 +44,11 @@ def test_run_process_success(self):
             'percent': [29.6479]
         })
         mock_client.query.return_value.to_dataframe.return_value = sample_data
-        
+
         with tempfile.TemporaryDirectory() as tmp_dir:
             output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
-            process.run_process(mock_client, output_file)
-            
+            result = process.run_process(mock_client, output_file)
+            self.assertTrue(result)
             mock_client.query.assert_called_once()
             self.assertTrue(os.path.exists(output_file))
             saved_df = pd.read_csv(output_file)
@@ -65,12 +66,14 @@ def test_run_process_query_error(self):
     def test_run_process_dataframe_error(self):
         mock_client = mock.MagicMock()
         mock_query_job = mock.MagicMock()
-        mock_query_job.to_dataframe.side_effect = Exception("Failed to fetch dataframe")
+        mock_query_job.to_dataframe.side_effect = Exception(
+            "Failed to fetch dataframe")
         mock_client.query.return_value = mock_query_job
         with tempfile.TemporaryDirectory() as tmp_dir:
             output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
             with self.assertRaises(Exception):
                 process.run_process(mock_client, output_file)
 
+
 if __name__ == '__main__':
     unittest.main()

From 475be721966bf21071576ab00ea6ecd607cca95e Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Wed, 2 Sep 2026 11:20:56 +0000
Subject: [PATCH 09/13] Restrict entity resolution to city-level places and
 restore source_files in manifest

---
 scripts/us_cdc/cdc500_state/README.md       | 2 +-
 scripts/us_cdc/cdc500_state/manifest.json   | 3 +++
 scripts/us_cdc/cdc500_state/process.py      | 2 ++
 scripts/us_cdc/cdc500_state/process_test.py | 1 +
 4 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/scripts/us_cdc/cdc500_state/README.md b/scripts/us_cdc/cdc500_state/README.md
index e651276c93..45d26db85c 100644
--- a/scripts/us_cdc/cdc500_state/README.md
+++ b/scripts/us_cdc/cdc500_state/README.md
@@ -25,7 +25,7 @@ The aggregation script queries Google Cloud BigQuery graph tables in dataset `da
    - **Census ACS 5-Year Series**: Filters and joins population counts from Census ACS 5-Year Survey (`provenance = 'dc/base/CensusACS5YearSurvey'`).
 
 2. **`Observation`**:
-   - **Health Indicator Percentages**: Fetches city-level percentage values (`value AS percent`), observation dates (`date`), and city geoIds (`entity1 LIKE 'geoId/%'`) for CDC 500 StatVars.
+   - **Health Indicator Percentages**: Fetches city-level percentage values (`value AS percent`), observation dates (`date`), and city geoIds (`entity1 LIKE 'geoId/%' AND LENGTH(entity1) = 13`) for CDC 500 StatVars.
    - **City Cohort Populations**: Fetches city-level population counts (`value AS population`) for the corresponding demographic cohort StatVars.
 
 ### Aggregation Methodology
diff --git a/scripts/us_cdc/cdc500_state/manifest.json b/scripts/us_cdc/cdc500_state/manifest.json
index 4a2ac714b8..3ea02369d0 100644
--- a/scripts/us_cdc/cdc500_state/manifest.json
+++ b/scripts/us_cdc/cdc500_state/manifest.json
@@ -22,6 +22,9 @@
                 "disk": 100
             },
             "cron_schedule": "0 1 * * 1",
+            "source_files": [
+                "CDC500State_Output/CDC500State_Output.csv"
+            ],
             "validation_config_file": "validation_config.json"
         }
     ]
diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index 1b39961d6e..e9e7f31836 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -64,6 +64,7 @@
   INNER JOIN cdc_sv
     ON O.variable_measured = cdc_sv.cdc500
   WHERE O.entity1 LIKE 'geoId/%'
+    AND LENGTH(O.entity1) = 13
     AND O.variable_measured LIKE 'Percent_%'
 ),
 
@@ -85,6 +86,7 @@
   ) AS pop
     ON O.variable_measured = pop.pop_statvar
   WHERE O.entity1 LIKE 'geoId/%'
+    AND LENGTH(O.entity1) = 13
 )
 
 SELECT 
diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py
index 93d18f8274..6660acf1e7 100644
--- a/scripts/us_cdc/cdc500_state/process_test.py
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -32,6 +32,7 @@ def test_query_constants(self):
         self.assertIn("dc/base/CensusACS5YearSurvey", query)
         self.assertIn("SAFE_DIVIDE", query)
         self.assertIn("SUBSTR(p.observation_about, 1, 8)", query)
+        self.assertIn("LENGTH(O.entity1) = 13", query)
 
     def test_run_process_success(self):
         mock_client = mock.MagicMock()

From da5e51cf2ae28614bf78bdf117a64ccb258d2b9c Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Tue, 8 Sep 2026 13:01:10 +0000
Subject: [PATCH 10/13] Address code review feedback on CDC500 state
 aggregation

- Exclude age-bracketed cancer screening indicators lacking composite Census ACS cohorts (Mammography, Cervical screening, Pap smear, Colorectal screening).
- Add QUALIFY deduplication in svo_count for 1:1 population weighting.
- Raise RuntimeError on empty query results and write CSV atomically.
- Add scoped SQL_VALIDATOR check for cohort vintage freshness in validation_config.json.
- Remove duplicate source_files in manifest.json and ensure trailing newlines.
- Expand unit tests to verify query constants, error handling, and CLI flags.
- Update README.md with rationale, prerequisites, testing, and automation details.
---
 scripts/us_cdc/cdc500_state/README.md         | 43 +++++++++++++++++--
 scripts/us_cdc/cdc500_state/cdc500_state.tmcf |  2 +-
 scripts/us_cdc/cdc500_state/manifest.json     |  5 +--
 scripts/us_cdc/cdc500_state/process.py        | 34 ++++++++++++---
 scripts/us_cdc/cdc500_state/process_test.py   | 39 +++++++++++++++--
 .../cdc500_state/validation_config.json       |  9 ++++
 6 files changed, 113 insertions(+), 19 deletions(-)

diff --git a/scripts/us_cdc/cdc500_state/README.md b/scripts/us_cdc/cdc500_state/README.md
index 45d26db85c..033a271b91 100644
--- a/scripts/us_cdc/cdc500_state/README.md
+++ b/scripts/us_cdc/cdc500_state/README.md
@@ -39,6 +39,16 @@ $$\text{State Percent} = \frac{\sum (\text{City Population} \times \text{City Pe
 
 The output measurement method is prefixed with `dcAggregate/` (e.g., `dcAggregate/CrudePrevalence`).
 
+#### Excluded Indicators
+
+The following age-bracketed cancer screening indicators are omitted from state-level aggregation:
+- `Percent_Person_50To74Years_Female_ReceivedMammography`
+- `Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening`
+- `Percent_Person_21To65Years_Female_ReceivedPapSmearTest`
+- `Percent_Person_50To75Years_ReceivedColorectalCancerScreening`
+
+**Rationale**: The Census ACS 5-Year Survey does not publish single composite population StatVars for these non-standard multi-year age brackets (`50To74Years`, `21To65Years`, `50To75Years`). Rather than applying arbitrary proxy weights or risking silent row omission, these indicators are explicitly excluded from state aggregation.
+
 ## About the Import
 
 ### Artifacts
@@ -57,10 +67,37 @@ The output measurement method is prefixed with `dcAggregate/` (e.g., `dcAggregat
 
 ### Import Procedure
 
-#### Data Download and Processing Steps
+#### Prerequisites
+
+Ensure Google Cloud authentication is configured with access to BigQuery dataset `datcom-store.spanner_dc_graph_prod_DEFAULT`:
+
+```bash
+$ gcloud auth application-default login
+```
+
+#### Running the Script
+
+To run the BigQuery aggregation and write the output CSV to the default output directory (`CDC500State_Output/CDC500State_Output.csv`):
+
+```bash
+$ python3 scripts/us_cdc/cdc500_state/process.py
+```
+
+To specify a custom output directory:
+
+```bash
+$ python3 scripts/us_cdc/cdc500_state/process.py --output_dir=/path/to/output
+```
+
+#### Running Unit Tests
 
-To run the BigQuery aggregation and generate the output CSV:
+Run the test suite using Python's `unittest` runner from the repository root:
 
 ```bash
-$ python3 process.py
+$ python3 -m unittest scripts.us_cdc.cdc500_state.process_test
 ```
+
+#### Automation
+
+This import is automated via Data Commons Import Automation and scheduled to run weekly via Cloud Batch every Monday at 01:00 UTC (`cron_schedule: "0 1 * * 1"` in `manifest.json`).
+
diff --git a/scripts/us_cdc/cdc500_state/cdc500_state.tmcf b/scripts/us_cdc/cdc500_state/cdc500_state.tmcf
index b7d6106b85..e5246d284d 100644
--- a/scripts/us_cdc/cdc500_state/cdc500_state.tmcf
+++ b/scripts/us_cdc/cdc500_state/cdc500_state.tmcf
@@ -7,4 +7,4 @@ value: C:CDC->percent
 unit: Percent
 scalingFactor: 100
 measurementMethod: C:CDC->measurement_method
-observationPeriod: "P1Y"
\ No newline at end of file
+observationPeriod: "P1Y"
diff --git a/scripts/us_cdc/cdc500_state/manifest.json b/scripts/us_cdc/cdc500_state/manifest.json
index 3ea02369d0..3a314ec2a5 100644
--- a/scripts/us_cdc/cdc500_state/manifest.json
+++ b/scripts/us_cdc/cdc500_state/manifest.json
@@ -22,10 +22,7 @@
                 "disk": 100
             },
             "cron_schedule": "0 1 * * 1",
-            "source_files": [
-                "CDC500State_Output/CDC500State_Output.csv"
-            ],
             "validation_config_file": "validation_config.json"
         }
     ]
-}
\ No newline at end of file
+}
diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index e9e7f31836..c5489eeaa4 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -31,10 +31,8 @@
   SELECT
     variable_measured AS cdc500,
     CASE
-      WHEN variable_measured LIKE '%Female_50To74Years%' OR variable_measured LIKE '%50To74Years_Female%' THEN 'Count_Person_Female_50To74Years'
-      WHEN variable_measured LIKE '%Female_21To65Years%' OR variable_measured LIKE '%21To65Years_Female%' THEN 'Count_Person_Female_21To65Years'
-      WHEN variable_measured LIKE '%Female_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Female%' THEN 'Count_Person_65OrMoreYears_Female'
-      WHEN variable_measured LIKE '%Male_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Male%' THEN 'Count_Person_65OrMoreYears_Male'
+      WHEN REGEXP_CONTAINS(variable_measured, r'65OrMoreYears.*Female|Female.*65OrMoreYears') THEN 'Count_Person_65OrMoreYears_Female'
+      WHEN REGEXP_CONTAINS(variable_measured, r'65OrMoreYears.*Male|Male.*65OrMoreYears') THEN 'Count_Person_65OrMoreYears_Male'
       WHEN variable_measured LIKE '%65OrMoreYears%' THEN 'Count_Person_65OrMoreYears'
       WHEN variable_measured LIKE '%18To64Years%' THEN 'Count_Person_18To64Years'
       WHEN variable_measured LIKE '%18OrMoreYears%' THEN 'Count_Person_18OrMoreYears'
@@ -43,6 +41,12 @@
   FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries`
   WHERE provenance = 'dc/base/CDC500'
     AND variable_measured LIKE 'Percent_%'
+    AND variable_measured NOT IN (
+      'Percent_Person_50To74Years_Female_ReceivedMammography',
+      'Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening',
+      'Percent_Person_21To65Years_Female_ReceivedPapSmearTest',
+      'Percent_Person_50To75Years_ReceivedColorectalCancerScreening'
+    )
   GROUP BY cdc500, pop_statvar
 ),
 
@@ -66,6 +70,12 @@
   WHERE O.entity1 LIKE 'geoId/%'
     AND LENGTH(O.entity1) = 13
     AND O.variable_measured LIKE 'Percent_%'
+    AND O.variable_measured NOT IN (
+      'Percent_Person_50To74Years_Female_ReceivedMammography',
+      'Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening',
+      'Percent_Person_21To65Years_Female_ReceivedPapSmearTest',
+      'Percent_Person_50To75Years_ReceivedColorectalCancerScreening'
+    )
 ),
 
 svo_count AS (
@@ -87,6 +97,10 @@
     ON O.variable_measured = pop.pop_statvar
   WHERE O.entity1 LIKE 'geoId/%'
     AND LENGTH(O.entity1) = 13
+  QUALIFY ROW_NUMBER() OVER (
+    PARTITION BY O.variable_measured, O.entity1, O.date
+    ORDER BY O.facet_id DESC
+  ) = 1
 )
 
 SELECT 
@@ -114,21 +128,26 @@ def run_process(client: bigquery.Client, output_file: str) -> bool:
     try:
         query_job = client.query(QUERY)
     except Exception as e:
-        logging.error("Failed to submit BigQuery query: %s", e)
+        logging.error("Failed to submit BigQuery query: %s", e, exc_info=True)
         raise
 
     logging.info("Fetching query results into dataframe...")
     try:
         df = query_job.to_dataframe()
     except Exception as e:
-        logging.error("Failed to fetch query results into dataframe: %s", e)
+        logging.error("Failed to fetch query results into dataframe: %s", e, exc_info=True)
         raise
 
+    if df.empty:
+        raise RuntimeError("BigQuery query returned 0 rows.")
+
     output_dir = os.path.dirname(output_file)
     if output_dir:
         os.makedirs(output_dir, exist_ok=True)
     logging.info("Writing %d rows to %s", len(df), output_file)
-    df.to_csv(output_file, index=False)
+    temp_file = output_file + ".tmp"
+    df.to_csv(temp_file, index=False)
+    os.replace(temp_file, output_file)
     return True
 
 
@@ -141,3 +160,4 @@ def main(argv):
 
 if __name__ == '__main__':
     app.run(main)
+
diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py
index 6660acf1e7..1ec0cde445 100644
--- a/scripts/us_cdc/cdc500_state/process_test.py
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -17,10 +17,13 @@
 import tempfile
 import unittest
 from unittest import mock
+from absl import flags
 import pandas as pd
 
 from scripts.us_cdc.cdc500_state import process
 
+FLAGS = flags.FLAGS
+
 
 class CDC500StateProcessTest(unittest.TestCase):
 
@@ -33,6 +36,12 @@ def test_query_constants(self):
         self.assertIn("SAFE_DIVIDE", query)
         self.assertIn("SUBSTR(p.observation_about, 1, 8)", query)
         self.assertIn("LENGTH(O.entity1) = 13", query)
+        self.assertIn("REGEXP_CONTAINS", query)
+        self.assertIn("QUALIFY ROW_NUMBER() OVER", query)
+        self.assertIn("Percent_Person_50To74Years_Female_ReceivedMammography", query)
+        self.assertIn("Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening", query)
+        self.assertIn("Percent_Person_21To65Years_Female_ReceivedPapSmearTest", query)
+        self.assertIn("Percent_Person_50To75Years_ReceivedColorectalCancerScreening", query)
 
     def test_run_process_success(self):
         mock_client = mock.MagicMock()
@@ -52,29 +61,51 @@ def test_run_process_success(self):
             self.assertTrue(result)
             mock_client.query.assert_called_once()
             self.assertTrue(os.path.exists(output_file))
+            self.assertFalse(os.path.exists(output_file + '.tmp'))
             saved_df = pd.read_csv(output_file)
             self.assertEqual(len(saved_df), 1)
             self.assertEqual(saved_df['observation_about'].iloc[0], 'geoId/06')
 
+    def test_run_process_empty_dataframe_raises_runtime_error(self):
+        mock_client = mock.MagicMock()
+        mock_client.query.return_value.to_dataframe.return_value = pd.DataFrame()
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
+            with self.assertRaises(RuntimeError):
+                process.run_process(mock_client, output_file)
+
     def test_run_process_query_error(self):
         mock_client = mock.MagicMock()
-        mock_client.query.side_effect = Exception("BigQuery Access Denied")
+        mock_client.query.side_effect = RuntimeError("BigQuery Access Denied")
         with tempfile.TemporaryDirectory() as tmp_dir:
             output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
-            with self.assertRaises(Exception):
+            with self.assertRaises(RuntimeError):
                 process.run_process(mock_client, output_file)
 
     def test_run_process_dataframe_error(self):
         mock_client = mock.MagicMock()
         mock_query_job = mock.MagicMock()
-        mock_query_job.to_dataframe.side_effect = Exception(
+        mock_query_job.to_dataframe.side_effect = RuntimeError(
             "Failed to fetch dataframe")
         mock_client.query.return_value = mock_query_job
         with tempfile.TemporaryDirectory() as tmp_dir:
             output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
-            with self.assertRaises(Exception):
+            with self.assertRaises(RuntimeError):
                 process.run_process(mock_client, output_file)
 
+    @mock.patch('scripts.us_cdc.cdc500_state.process.run_process')
+    @mock.patch('google.cloud.bigquery.Client')
+    def test_main(self, mock_bq_client_cls, mock_run_process):
+        mock_client_instance = mock.MagicMock()
+        mock_bq_client_cls.return_value = mock_client_instance
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            FLAGS(['test_process', f'--output_dir={tmp_dir}'])
+            process.main([])
+            expected_output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
+            mock_run_process.assert_called_once_with(mock_client_instance,
+                                                     expected_output_file)
+
 
 if __name__ == '__main__':
     unittest.main()
+
diff --git a/scripts/us_cdc/cdc500_state/validation_config.json b/scripts/us_cdc/cdc500_state/validation_config.json
index 5ac0fba24f..3d7b0cc4a2 100644
--- a/scripts/us_cdc/cdc500_state/validation_config.json
+++ b/scripts/us_cdc/cdc500_state/validation_config.json
@@ -25,6 +25,15 @@
                 "minimum": 50,
                 "maximum": 52
             }
+        },
+        {
+            "rule_id": "check_statvar_max_dates",
+            "description": "Verifies that MaxDate meets expected vintage freshness per StatVar cohort.",
+            "validator": "SQL_VALIDATOR",
+            "params": {
+                "query": "SELECT StatVar, CAST(MaxDate AS INTEGER) AS max_year FROM stats",
+                "condition": "CASE WHEN StatVar = 'Percent_Person_WithAllTeethLoss' THEN max_year >= 2016 WHEN StatVar LIKE '%CorePreventiveServices%' THEN max_year >= 2020 WHEN StatVar IN ('Percent_Person_WithHighBloodPressure', 'Percent_Person_18OrMoreYears_WithHighBloodPressure_ReceivedTakingBloodPressureMedication', 'Percent_Person_WithHighCholesterol', 'Percent_Person_ReceivedCholesterolScreening', 'Percent_Person_WithChronicKidneyDisease') THEN max_year >= 2021 ELSE max_year >= 2022 END"
+            }
         }
     ]
 }

From b69963eaf34874e4671e7e6c29e6055b36f4bd19 Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Tue, 8 Sep 2026 17:37:07 +0000
Subject: [PATCH 11/13] Address code review findings for CDC500 state
 aggregation

- Add check_deleted_records_percent override (threshold 100) in validation_config.json for TMCF unit/scalingFactor migration.
- Remove redundant try/except re-raise logging in run_process to eliminate double tracebacks.
- Add unit tests for demographic cohort regex mappings and population-weighted aggregation.
- Fix demographic cohort examples in README.md to reflect excluded cancer screening indicators.
---
 scripts/us_cdc/cdc500_state/README.md         |  2 +-
 scripts/us_cdc/cdc500_state/process.py        | 12 +---
 scripts/us_cdc/cdc500_state/process_test.py   | 61 +++++++++++++++++++
 .../cdc500_state/validation_config.json       |  8 +++
 4 files changed, 72 insertions(+), 11 deletions(-)

diff --git a/scripts/us_cdc/cdc500_state/README.md b/scripts/us_cdc/cdc500_state/README.md
index 033a271b91..e2b6da4d34 100644
--- a/scripts/us_cdc/cdc500_state/README.md
+++ b/scripts/us_cdc/cdc500_state/README.md
@@ -21,7 +21,7 @@ The state-level dataset calculates aggregated health indicator prevalence estima
 The aggregation script queries Google Cloud BigQuery graph tables in dataset `datcom-store.spanner_dc_graph_prod_DEFAULT`:
 
 1. **`TimeSeries`**:
-   - **CDC 500 Series**: Identifies CDC 500 Statistical Variables (`provenance = 'dc/base/CDC500'` and `variable_measured LIKE 'Percent_%'`) and extracts their measurement methods (`measurement_method`). It maps each percentage health metric to its appropriate denominator demographic cohort StatVar (e.g., `Count_Person_18OrMoreYears`, `Count_Person_Female_50To74Years`, `Count_Person_Female_21To65Years`, `Count_Person_65OrMoreYears`, etc.).
+   - **CDC 500 Series**: Identifies CDC 500 Statistical Variables (`provenance = 'dc/base/CDC500'` and `variable_measured LIKE 'Percent_%'`) and extracts their measurement methods (`measurement_method`). It maps each percentage health metric to its appropriate denominator demographic cohort StatVar (e.g., `Count_Person_18OrMoreYears`, `Count_Person_18To64Years`, `Count_Person_65OrMoreYears`, `Count_Person`, etc.).
    - **Census ACS 5-Year Series**: Filters and joins population counts from Census ACS 5-Year Survey (`provenance = 'dc/base/CensusACS5YearSurvey'`).
 
 2. **`Observation`**:
diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index c5489eeaa4..0550882dd2 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -125,18 +125,10 @@
 def run_process(client: bigquery.Client, output_file: str) -> bool:
     """Executes the BigQuery query and writes the resulting DataFrame to output_file."""
     logging.info("Running BigQuery aggregation query...")
-    try:
-        query_job = client.query(QUERY)
-    except Exception as e:
-        logging.error("Failed to submit BigQuery query: %s", e, exc_info=True)
-        raise
+    query_job = client.query(QUERY)
 
     logging.info("Fetching query results into dataframe...")
-    try:
-        df = query_job.to_dataframe()
-    except Exception as e:
-        logging.error("Failed to fetch query results into dataframe: %s", e, exc_info=True)
-        raise
+    df = query_job.to_dataframe()
 
     if df.empty:
         raise RuntimeError("BigQuery query returned 0 rows.")
diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py
index 1ec0cde445..6f68339def 100644
--- a/scripts/us_cdc/cdc500_state/process_test.py
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -43,6 +43,67 @@ def test_query_constants(self):
         self.assertIn("Percent_Person_21To65Years_Female_ReceivedPapSmearTest", query)
         self.assertIn("Percent_Person_50To75Years_ReceivedColorectalCancerScreening", query)
 
+    def test_demographic_cohort_regex_mapping(self):
+        """Verifies that representative StatVars match the intended demographic regex rules."""
+        import re
+        female_pattern = r'65OrMoreYears.*Female|Female.*65OrMoreYears'
+        male_pattern = r'65OrMoreYears.*Male|Male.*65OrMoreYears'
+
+        self.assertIn(female_pattern, process.QUERY)
+        self.assertIn(male_pattern, process.QUERY)
+
+        # Helper mapping that mirrors the SQL CASE WHEN logic
+        def map_statvar(sv: str) -> str:
+            if re.search(female_pattern, sv):
+                return 'Count_Person_65OrMoreYears_Female'
+            elif re.search(male_pattern, sv):
+                return 'Count_Person_65OrMoreYears_Male'
+            elif '65OrMoreYears' in sv:
+                return 'Count_Person_65OrMoreYears'
+            elif '18To64Years' in sv:
+                return 'Count_Person_18To64Years'
+            elif '18OrMoreYears' in sv:
+                return 'Count_Person_18OrMoreYears'
+            else:
+                return 'Count_Person'
+
+        test_cases = [
+            ('Percent_Person_65OrMoreYears_Female_CorePreventiveServices', 'Count_Person_65OrMoreYears_Female'),
+            ('Percent_Person_Female_65OrMoreYears_CorePreventiveServices', 'Count_Person_65OrMoreYears_Female'),
+            ('Percent_Person_65OrMoreYears_Male_CorePreventiveServices', 'Count_Person_65OrMoreYears_Male'),
+            ('Percent_Person_Male_65OrMoreYears_CorePreventiveServices', 'Count_Person_65OrMoreYears_Male'),
+            ('Percent_Person_65OrMoreYears_CorePreventiveServices', 'Count_Person_65OrMoreYears'),
+            ('Percent_Person_18To64Years_HealthInsurance', 'Count_Person_18To64Years'),
+            ('Percent_Person_18OrMoreYears_WithAnyDisability', 'Count_Person_18OrMoreYears'),
+            ('Percent_Person_18OrMoreYears_WithHighBloodPressure', 'Count_Person_18OrMoreYears'),
+            ('Percent_Person_WithArthritis', 'Count_Person'),
+            ('Percent_Person_WithHighCholesterol', 'Count_Person'),
+        ]
+
+        for sv, expected in test_cases:
+            with self.subTest(statvar=sv):
+                self.assertEqual(map_statvar(sv), expected)
+
+    def test_population_weighted_average_calculation(self):
+        """Verifies the population-weighted average calculation and city-to-state FIPS aggregation."""
+        # Simulated city-level records for California (geoId/06)
+        city_records = pd.DataFrame({
+            'city_geoid': ['geoId/0644000', 'geoId/0666000', 'geoId/0667000'],
+            'city_percent': [20.0, 30.0, 40.0],
+            'city_pop': [10000, 20000, 70000]
+        })
+        city_records['state_geoid'] = city_records['city_geoid'].str.slice(0, 8)
+        self.assertTrue((city_records['state_geoid'] == 'geoId/06').all())
+
+        # Formula: SUM(pop * percent) / SUM(pop)
+        total_weighted = (city_records['city_pop'] * city_records['city_percent']).sum()
+        total_pop = city_records['city_pop'].sum()
+        weighted_avg = total_weighted / total_pop
+
+        # Expected: (10000*20 + 20000*30 + 70000*40) / 100000 = (200000 + 600000 + 2800000) / 100000 = 36.0
+        self.assertEqual(total_pop, 100000)
+        self.assertAlmostEqual(weighted_avg, 36.0, places=4)
+
     def test_run_process_success(self):
         mock_client = mock.MagicMock()
         sample_data = pd.DataFrame({
diff --git a/scripts/us_cdc/cdc500_state/validation_config.json b/scripts/us_cdc/cdc500_state/validation_config.json
index 3d7b0cc4a2..ad5285bf8a 100644
--- a/scripts/us_cdc/cdc500_state/validation_config.json
+++ b/scripts/us_cdc/cdc500_state/validation_config.json
@@ -1,6 +1,14 @@
 {
     "schema_version": "1.0",
     "rules": [
+        {
+            "rule_id": "check_deleted_records_percent",
+            "description": "Override default threshold to 100% due to one-time series identity update (adding unit: Percent and scalingFactor: 100).",
+            "validator": "DELETED_RECORDS_PERCENT",
+            "params": {
+                "threshold": 100
+            }
+        },
         {
             "rule_id": "check_max_value_percentage",
             "description": "Checks that all percentage StatVars do not exceed 100%.",

From 81294e4b4f0324a12098eba4bb20ad5f23ea9819 Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Wed, 9 Sep 2026 07:33:05 +0000
Subject: [PATCH 12/13] Remove check_statvar_max_dates and
 check_deleted_records_percent from validation_config.json

---
 .../us_cdc/cdc500_state/validation_config.json  | 17 -----------------
 1 file changed, 17 deletions(-)

diff --git a/scripts/us_cdc/cdc500_state/validation_config.json b/scripts/us_cdc/cdc500_state/validation_config.json
index ad5285bf8a..5ac0fba24f 100644
--- a/scripts/us_cdc/cdc500_state/validation_config.json
+++ b/scripts/us_cdc/cdc500_state/validation_config.json
@@ -1,14 +1,6 @@
 {
     "schema_version": "1.0",
     "rules": [
-        {
-            "rule_id": "check_deleted_records_percent",
-            "description": "Override default threshold to 100% due to one-time series identity update (adding unit: Percent and scalingFactor: 100).",
-            "validator": "DELETED_RECORDS_PERCENT",
-            "params": {
-                "threshold": 100
-            }
-        },
         {
             "rule_id": "check_max_value_percentage",
             "description": "Checks that all percentage StatVars do not exceed 100%.",
@@ -33,15 +25,6 @@
                 "minimum": 50,
                 "maximum": 52
             }
-        },
-        {
-            "rule_id": "check_statvar_max_dates",
-            "description": "Verifies that MaxDate meets expected vintage freshness per StatVar cohort.",
-            "validator": "SQL_VALIDATOR",
-            "params": {
-                "query": "SELECT StatVar, CAST(MaxDate AS INTEGER) AS max_year FROM stats",
-                "condition": "CASE WHEN StatVar = 'Percent_Person_WithAllTeethLoss' THEN max_year >= 2016 WHEN StatVar LIKE '%CorePreventiveServices%' THEN max_year >= 2020 WHEN StatVar IN ('Percent_Person_WithHighBloodPressure', 'Percent_Person_18OrMoreYears_WithHighBloodPressure_ReceivedTakingBloodPressureMedication', 'Percent_Person_WithHighCholesterol', 'Percent_Person_ReceivedCholesterolScreening', 'Percent_Person_WithChronicKidneyDisease') THEN max_year >= 2021 ELSE max_year >= 2022 END"
-            }
         }
     ]
 }

From c4067f960ebd0926f7fc86a4b7385c770b4f680d Mon Sep 17 00:00:00 2001
From: shvngisingh <Shvngisingh@google.com>
Date: Wed, 9 Sep 2026 08:50:34 +0000
Subject: [PATCH 13/13] Address review findings: add deterministic
 last_update_timestamp deduplication in process.py and use TRY_CAST in
 validation_config.json

---
 scripts/us_cdc/cdc500_state/process.py             | 6 +++++-
 scripts/us_cdc/cdc500_state/process_test.py        | 1 +
 scripts/us_cdc/cdc500_state/validation_config.json | 9 +++++++++
 3 files changed, 15 insertions(+), 1 deletion(-)

diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py
index 0550882dd2..1c1f67b567 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -76,6 +76,10 @@
       'Percent_Person_21To65Years_Female_ReceivedPapSmearTest',
       'Percent_Person_50To75Years_ReceivedColorectalCancerScreening'
     )
+  QUALIFY ROW_NUMBER() OVER (
+    PARTITION BY O.variable_measured, O.entity1, O.date, T.measurement_method
+    ORDER BY O.last_update_timestamp DESC
+  ) = 1
 ),
 
 svo_count AS (
@@ -99,7 +103,7 @@
     AND LENGTH(O.entity1) = 13
   QUALIFY ROW_NUMBER() OVER (
     PARTITION BY O.variable_measured, O.entity1, O.date
-    ORDER BY O.facet_id DESC
+    ORDER BY O.last_update_timestamp DESC
   ) = 1
 )
 
diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py
index 6f68339def..e5eecaa8cf 100644
--- a/scripts/us_cdc/cdc500_state/process_test.py
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -38,6 +38,7 @@ def test_query_constants(self):
         self.assertIn("LENGTH(O.entity1) = 13", query)
         self.assertIn("REGEXP_CONTAINS", query)
         self.assertIn("QUALIFY ROW_NUMBER() OVER", query)
+        self.assertIn("O.last_update_timestamp DESC", query)
         self.assertIn("Percent_Person_50To74Years_Female_ReceivedMammography", query)
         self.assertIn("Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening", query)
         self.assertIn("Percent_Person_21To65Years_Female_ReceivedPapSmearTest", query)
diff --git a/scripts/us_cdc/cdc500_state/validation_config.json b/scripts/us_cdc/cdc500_state/validation_config.json
index 5ac0fba24f..1a18f40b33 100644
--- a/scripts/us_cdc/cdc500_state/validation_config.json
+++ b/scripts/us_cdc/cdc500_state/validation_config.json
@@ -25,6 +25,15 @@
                 "minimum": 50,
                 "maximum": 52
             }
+        },
+        {
+            "rule_id": "check_statvar_max_dates",
+            "description": "Verifies that MaxDate meets expected vintage freshness per StatVar cohort.",
+            "validator": "SQL_VALIDATOR",
+            "params": {
+                "query": "SELECT StatVar, TRY_CAST(MaxDate AS INTEGER) AS max_year FROM stats",
+                "condition": "CASE WHEN StatVar = 'Percent_Person_WithAllTeethLoss' THEN max_year >= 2016 WHEN StatVar LIKE '%CorePreventiveServices%' THEN max_year >= 2020 WHEN StatVar IN ('Percent_Person_WithHighBloodPressure', 'Percent_Person_18OrMoreYears_WithHighBloodPressure_ReceivedTakingBloodPressureMedication', 'Percent_Person_WithHighCholesterol', 'Percent_Person_ReceivedCholesterolScreening', 'Percent_Person_WithChronicKidneyDisease') THEN max_year >= 2021 ELSE max_year >= 2022 END"
+            }
         }
     ]
 }