DENG-9849: Update copy dedup code to null out city and subdivision fields for browser apps (#8362)

* Null geo in copy_dedup

* fixed ci failures

* Fixed param type

* Used channel to app name mapping

* Update bigquery_etl/copy_deduplicate.py

Co-authored-by: Ben Wu <[email protected]>

* Use app id

* Added client id check

* Update bigquery_etl/copy_deduplicate.py

Co-authored-by: Ben Wu <[email protected]>

* Added regex table id

---------

Co-authored-by: Ben Wu <[email protected]>

Author: wwyc

bigquery_etl/copy_deduplicate.py

@@ -9,16 +9,19 @@
 
 import json
 import logging
+import re
 from datetime import datetime, timedelta
 from functools import partial
 from itertools import groupby
 from multiprocessing.pool import ThreadPool
+from typing import List
 
 import click
 from google.api_core.exceptions import BadRequest
 from google.cloud import bigquery
 
 from bigquery_etl.cli.utils import table_matches_patterns
+from bigquery_etl.config import ConfigLoader
 from bigquery_etl.util.bigquery_id import sql_table_id
 from bigquery_etl.util.client_queue import ClientQueue
 from bigquery_etl.util.common import TempDatasetReference
@@ -73,14 +76,72 @@
   --
   -- Retain only one document for each ID.
 SELECT
-  * EXCEPT(_n)
+  * EXCEPT(_n){replace_geo}
 FROM
   numbered_duplicates
 WHERE
   _n = 1
 """
 
 
+def _has_field_path(schema: List[bigquery.SchemaField], path: List[str]) -> bool:
+    """Return True if nested field path (e.g., ['metadata','geo','city']) exists."""
+    for name in path:
+        f = next((field for field in schema if field.name == name), None)
+        if not f:
+            return False
+        schema = getattr(f, "fields", []) or []
+    return True
+
+
+def _select_geo(live_table: str, client: bigquery.Client) -> str:
+    """Build a SELECT REPLACE clause that NULLs metadata.geo.* if applicable."""
+    _, dataset_id, table_id = live_table.split(".")
+
+    excluded_tables = set(
+        ConfigLoader.get("geo_deprecation", "skip_tables", fallback=[])
+    )
+    if re.sub(r"_v\d+$", "", table_id) in excluded_tables:
+        return ""
+
+    app_id = dataset_id.removesuffix("_live")
+    included_apps = set(
+        ConfigLoader.get("geo_deprecation", "include_app_ids", fallback=[])
+    )
+    if app_id not in included_apps:
+        return ""
+
+    table = client.get_table(live_table)
+
+    include_client_id = table.labels.get("include_client_id") == "true"
+    if not include_client_id:
+        return ""
+
+    # Check schema to ensure geo fields exists
+    schema = table.schema
+    required_fields = ("city", "subdivision1", "subdivision2")
+    has_required_fields = all(
+        _has_field_path(schema, ["metadata", "geo", field]) for field in required_fields
+    )
+    if not has_required_fields:
+        return ""
+
+    return """
+        REPLACE (
+        (SELECT AS STRUCT
+           metadata.* REPLACE (
+             (SELECT AS STRUCT
+                metadata.geo.* REPLACE (
+                  CAST(NULL AS STRING) AS city,
+                  CAST(NULL AS STRING) AS subdivision1,
+                  CAST(NULL AS STRING) AS subdivision2
+                )
+             ) AS geo
+           )
+        ) AS metadata)
+    """
+
+
 def _get_query_job_configs(
     client,
     live_table,
@@ -92,7 +153,8 @@ def _get_query_job_configs(
     num_retries,
     temp_dataset,
 ):
-    sql = QUERY_TEMPLATE.format(live_table=live_table)
+    replace_geo = _select_geo(live_table, client)
+    sql = QUERY_TEMPLATE.format(live_table=live_table, replace_geo=replace_geo)
     stable_table = f"{live_table.replace('_live.', '_stable.', 1)}${date:%Y%m%d}"
     kwargs = dict(use_legacy_sql=False, dry_run=dry_run, priority=priority)
     start_time = datetime(*date.timetuple()[:6])

bqetl_project.yaml

@@ -558,6 +558,12 @@ generate:
     - firefox_desktop_background_defaultagent
     - firefox_desktop_background_tasks
 
+geo_deprecation:
+  include_app_ids:
+  - org_mozilla_ios_klar
+  skip_tables:
+  - newtab
+
 retention_exclusion_list:
 - sql/moz-fx-data-shared-prod/search_derived/acer_cohort_v1
 - sql/moz-fx-data-shared-prod/telemetry_derived/clients_first_seen_v3