Merge branch 'main' into relud-experiment-monitoring

Author: mikewilli

.gitignore

@@ -16,6 +16,7 @@ sql/**/docs
 sql/**/mkdocs.yml
 generated_docs/
 generated-sql/
+CLAUDE.md
 
 # ignore Bigeye artifacts
 moz-fx-data-shared-prod_APPLY.yml

CODEOWNERS

@@ -16,6 +16,7 @@ dags.yaml
 /sql/**/active_users_aggregates_v3 @mozilla/kpi_table_reviewers
 /sql_generators/active_users_aggregates_v4/templates/ @mozilla/kpi_table_reviewers
 /sql/**/active_users_aggregates_v4 @mozilla/kpi_table_reviewers
+/sql/moz-fx-data-shared-prod/telemetry/active_users_aggregates @mozilla/kpi_table_reviewers
 # Search
 /sql/moz-fx-data-shared-prod/search_terms @whd @jasonthomas
 /sql/moz-fx-data-shared-prod/search_terms_derived @whd @jasonthomas

bigquery_etl/cli/metadata.py

@@ -234,6 +234,7 @@ def publish(
     table_metadata_files = paths_matching_name_pattern(
         name, sql_dir, project_id=project_id, files=["metadata.yaml"]
     )
+    skip_deploy = ConfigLoader.get("metadata", "deploy", "skip", fallback=[])
 
     if parallelism > 0:
         credentials = get_credentials()
@@ -245,7 +246,10 @@ def publish(
             )
     else:
         for metadata_file in table_metadata_files:
-            _publish_metadata(project_id, credentials=None, metadata_file=metadata_file)
+            if str(metadata_file) not in skip_deploy:
+                _publish_metadata(
+                    project_id, credentials=None, metadata_file=metadata_file
+                )
 
 
 def _publish_metadata(project_id, credentials, metadata_file):

bigquery_etl/cli/stage.py

@@ -295,7 +295,7 @@ def _view_dependencies(artifact_files, sql_dir):
                     )
                 project, dataset, name = dependency_components
 
-                file_path = Path(view.path).parent.parent.parent / dataset / name
+                file_path = Path(sql_dir) / project / dataset / name
 
                 file_exists_for_dependency = False
                 for file in [VIEW_FILE, QUERY_FILE, QUERY_SCRIPT, MATERIALIZED_VIEW]:

bigquery_etl/cli/utils.py

@@ -4,11 +4,13 @@
 import os
 import re
 from fnmatch import fnmatchcase
+from functools import cache
 from glob import glob
 from pathlib import Path
-from typing import Iterator, List, Optional, Tuple
+from typing import Dict, Iterator, List, Optional, Tuple
 
 import click
+import requests
 from google.auth.exceptions import DefaultCredentialsError
 from google.cloud import bigquery
 
@@ -23,6 +25,7 @@
     r"^.*/([a-zA-Z0-9-]+)/([a-zA-Z0-9_]+)/([a-zA-Z0-9_]+(_v[0-9]+)?)/"
     r"(?:checks\.sql)$"
 )
+GLEAN_APP_LISTINGS_URL = "https://probeinfo.telemetry.mozilla.org/v2/glean/app-listings"
 
 
 def is_valid_dir(ctx, param, value):
@@ -250,3 +253,26 @@ def temp_dataset_option(
         help="Dataset where intermediate query results will be temporarily stored, "
         "formatted as PROJECT_ID.DATASET_ID",
     )
+
+
+@cache
+def get_glean_app_id_to_app_name_mapping() -> Dict[str, str]:
+    """Return a dict where key is the channel app id and the value is the shared app name.
+
+    e.g. {
+        "org_mozilla_firefox": "fenix",
+        "org_mozilla_firefox_beta": "fenix",
+        "org_mozilla_ios_firefox": "firefox_ios",
+        "org_mozilla_ios_firefoxbeta": "firefox_ios",
+    }
+    """
+    response = requests.get(GLEAN_APP_LISTINGS_URL)
+    response.raise_for_status()
+
+    app_listings = response.json()
+
+    return {
+        app["bq_dataset_family"]: app["app_name"]
+        for app in app_listings
+        if "bq_dataset_family" in app and "app_name" in app
+    }

bigquery_etl/copy_deduplicate.py

@@ -20,14 +20,17 @@
 from google.api_core.exceptions import BadRequest
 from google.cloud import bigquery
 
-from bigquery_etl.cli.utils import table_matches_patterns
+from bigquery_etl.cli.utils import (
+    get_glean_app_id_to_app_name_mapping,
+    parallelism_option,
+    project_id_option,
+    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
 
-from .cli.utils import parallelism_option, project_id_option
-
 QUERY_TEMPLATE = """
 WITH
   -- Distinct document_ids and their minimum submission_timestamp today
@@ -97,33 +100,43 @@ def _has_field_path(schema: List[bigquery.SchemaField], path: List[str]) -> bool
 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(".")
+    channel_to_app_name = get_glean_app_id_to_app_name_mapping()
+    app_id = re.sub("_live$", "", dataset_id)
+
+    excluded_apps = set(ConfigLoader.get("geo_deprecation", "skip_apps", fallback=[]))
+    app_name = channel_to_app_name.get(app_id)
+    if app_name in excluded_apps:
+        return ""
 
     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)
 
+    # Only deprecating the geo fields for glean apps.  Legacy tables would be deprecated after glean migration
+    if app_id not in channel_to_app_name.keys():
+        return ""
+
+    # only glean tables have this label
     include_client_id = table.labels.get("include_client_id") == "true"
     if not include_client_id:
         return ""
 
-    # Check schema to ensure geo fields exists
+    # Check schema to ensure required 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
+    has_client_id_field = _has_field_path(schema, ["client_info", "client_id"])
+    if not has_client_id_field:
+        return ""
+
+    required_geo_fields = ("city", "subdivision1", "subdivision2")
+    has_required_geo_fields = all(
+        _has_field_path(schema, ["metadata", "geo", field])
+        for field in required_geo_fields
     )
-    if not has_required_fields:
+    if not has_required_geo_fields:
         return ""
 
     return """

bigquery_etl/metadata/parse_metadata.py

@@ -37,6 +37,14 @@ def literal_presenter(dumper, data):
 yaml.add_representer(Literal, literal_presenter)
 
 
+class AssetLevel(enum.Enum):
+    """Represents BigQuery table level based on requirements for quality and maturity."""
+
+    GOLD = "gold"
+    SILVER = "silver"
+    BRONZE = "bronze"
+
+
 class PartitionType(enum.Enum):
     """Represents BigQuery table partition types."""
 
@@ -198,13 +206,31 @@ class Metadata:
     deletion_date: Optional[date] = attr.ib(None)
     monitoring: Optional[MonitoringMetadata] = attr.ib(None)
     require_column_descriptions: bool = attr.ib(False)
+    level: Optional[str] = attr.ib(None)
 
     @owners.validator
     def validate_owners(self, attribute, value):
         """Check that provided email addresses or github identities for owners are valid."""
         if not all(map(lambda e: is_email_or_github_identity(e), value)):
             raise ValueError(f"Invalid email or Github identity for owners: {value}.")
 
+    @level.validator
+    def validate_level(self, attribute, value):
+        """Check that the level label is a string and one of the expected values."""
+        allowed = [e.value for e in AssetLevel]
+
+        if value is None:
+            return
+
+        if not isinstance(value, str):
+            raise ValueError(
+                f"ERROR. Invalid level in metadata with type '{type(value).__name__}'. Must be a string."
+            )
+        if value not in allowed:
+            raise ValueError(
+                f"ERROR. Invalid level in metadata: {value}. Must be only one of {sorted(allowed)}."
+            )
+
     @labels.validator
     def validate_labels(self, attribute, value):
         """Check that labels are valid."""
@@ -276,6 +302,7 @@ def from_file(cls, metadata_file):
         deletion_date = None
         monitoring = None
         require_column_descriptions = False
+        level = None
 
         with open(metadata_file, "r") as yaml_stream:
             try:
@@ -363,6 +390,9 @@ def from_file(cls, metadata_file):
                         "require_column_descriptions"
                     ]
 
+                if "level" in metadata:
+                    level = metadata["level"]
+
                 return cls(
                     friendly_name,
                     description,
@@ -378,6 +408,7 @@ def from_file(cls, metadata_file):
                     deletion_date,
                     monitoring,
                     require_column_descriptions,
+                    level,
                 )
             except yaml.YAMLError as e:
                 raise e