Replace dry run check in glean_usage generator to check referenced tables exist (#8083)

* Remove dry run check in glean_usage generator to check referenced tables exist

* Cache baseline tables

* Cache stable schemas in local directory

* Cache app info in events monitoring generator

* run loop instead of threading pool

* use threadpool

* check outdir for existing schemas

* Derive metrics_clients_last_seen from metrics_clients_daily

* Skip generating metrics_clients_last_seen and _daily for certain apps

* Bug 1988207 - Look for schema in output_dir in glean_app_ping_views

* debug

* add default sql_dir

* remove debug

* Rename baseline_tables to base_tables in glean_usage generation

* Update path-filtering orb

---------

Co-authored-by: bwu <[email protected]>

Author: scholtzan

.circleci/config.yml

@@ -3,7 +3,7 @@ version: 2.1
 setup: true
 
 orbs:
-  path-filtering: circleci/path-filtering@0.1.3
+  path-filtering: circleci/path-filtering@2.0.4
   continuation: circleci/[email protected]
 
 parameters:

bigquery_etl/schema/stable_table_schema.py

@@ -1,7 +1,10 @@
 """Methods for working with stable table schemas."""
 
 import json
+import os
+import pickle
 import tarfile
+import tempfile
 import urllib.request
 from dataclasses import dataclass
 from functools import cache
@@ -68,6 +71,23 @@ def prod_schemas_uri():
 def get_stable_table_schemas() -> List[SchemaFile]:
     """Fetch last schema metadata per doctype by version."""
     schemas_uri = prod_schemas_uri()
+
+    # create cache file path based on the schemas URI
+    commit_hash = schemas_uri.split("/")[-1].replace(".tar.gz", "")
+    cache_dir = os.path.join(tempfile.gettempdir(), "bigquery_etl_schemas")
+    os.makedirs(cache_dir, exist_ok=True)
+    cache_file = os.path.join(cache_dir, f"schemas_{commit_hash}.pkl")
+
+    # check if cached file exists and load it
+    if os.path.exists(cache_file):
+        print(f"Loading cached schemas from {cache_file}")
+        try:
+            with open(cache_file, "rb") as f:
+                return pickle.load(f)
+        except (pickle.PickleError, EOFError, OSError) as e:
+            print(f"Failed to load cached schemas: {e}, re-downloading...")
+
+    print(f"Downloading schemas from {schemas_uri}")
     with urllib.request.urlopen(schemas_uri) as f:
         tarbytes = BytesIO(f.read())
 
@@ -132,4 +152,12 @@ def get_stable_table_schemas() -> List[SchemaFile]:
         )
     ]
 
+    # Cache the processed schemas
+    try:
+        with open(cache_file, "wb") as f:
+            pickle.dump(schemas, f)
+        print(f"Cached schemas to {cache_file}")
+    except (pickle.PickleError, OSError) as e:
+        print(f"Failed to cache schemas: {e}")
+
     return schemas

bqetl_project.yaml

@@ -410,6 +410,32 @@ generate:
       - focus_android
       - fenix
       - firefox_desktop
+    metrics_clients_last_seen:
+      skip_apps:
+      - accounts_cirrus
+      - experimenter_cirrus
+      - firefox_desktop_background_defaultagent
+      - firefox_desktop_background_tasks
+      - firefox_crashreporter
+      - monitor_cirrus
+      - mozilla_vpn
+      - mozillavpn_backend_cirrus
+      - pine
+      - thunderbird_android
+      - thunderbird_desktop
+    clients_last_seen_joined:
+      skip_apps:
+      - accounts_cirrus
+      - experimenter_cirrus
+      - firefox_desktop_background_defaultagent
+      - firefox_desktop_background_tasks
+      - firefox_crashreporter
+      - monitor_cirrus
+      - mozilla_vpn
+      - mozillavpn_backend_cirrus
+      - pine
+      - thunderbird_android
+      - thunderbird_desktop
     bigconfig:
       skip_apps:
       - firefox_echo_show

sql_generators/glean_usage/__init__.py

@@ -102,7 +102,7 @@ def generate(
 
     @cache
     def get_tables(table_name="baseline_v1"):
-        baseline_tables = list_tables(
+        tables = list_tables(
             project_id=target_project,
             only_tables=[only] if only else None,
             table_filter=table_filter,
@@ -111,9 +111,9 @@ def get_tables(table_name="baseline_v1"):
 
         # filter out skipped apps
         return [
-            baseline_table
-            for baseline_table in baseline_tables
-            if baseline_table.split(".")[1]
+            table
+            for table in tables
+            if table.split(".")[1]
             not in [
                 f"{skipped_app}_stable"
                 for skipped_app in ConfigLoader.get(
@@ -153,12 +153,31 @@ def get_tables(table_name="baseline_v1"):
                 parallelism=parallelism,
                 id_token=id_token,
             ),
-            baseline_table,
+            base_table,
         )
         for table in GLEAN_TABLES
-        for baseline_table in get_tables(table_name=table.base_table_name)
+        for base_table in get_tables(table_name=table.base_table_name)
     ]
 
+    base_tables = {}
+    unique_base_table_names = {table.base_table_name for table in GLEAN_TABLES}
+    for table_name in unique_base_table_names:
+        base_tables[table_name] = get_tables(table_name=table_name)
+
+    def all_base_tables_exist(app_info, table_name="baseline_v1"):
+        """Check if baseline tables exist for all app datasets."""     
+        # Extract dataset names from table names (format: project.dataset.table)
+        existing_datasets = {table.split(".")[1] for table in base_tables[table_name]}
+        
+        # Check if all app datasets have corresponding tables
+        if isinstance(app_info, dict):
+            required_datasets = {f"{app_info['bq_dataset_family']}_stable"}
+        else:
+            required_datasets = {f"{app['bq_dataset_family']}_stable" for app in app_info}
+
+        return all(dataset in existing_datasets for dataset in required_datasets)
+    
+
     # Parameters to generate per-app datasets consist of the function to be called
     # and app_info
     generate_per_app = [
@@ -170,6 +189,9 @@ def get_tables(table_name="baseline_v1"):
                 use_cloud_function=use_cloud_function,
                 parallelism=parallelism,
                 id_token=id_token,
+                all_base_tables_exist=all_base_tables_exist(info, table_name=table.base_table_name) 
+                if hasattr(table, 'per_app_requires_all_base_tables') and table.per_app_requires_all_base_tables 
+                else None
             ),
             info,
         )

sql_generators/glean_usage/baseline_clients_daily.py

@@ -15,3 +15,4 @@ def __init__(self):
         self.target_table_id = BASELINE_DAILY_TABLE_ID
         self.prefix = PREFIX
         self.custom_render_kwargs = {}
+        self.per_app_requires_all_base_tables = True

sql_generators/glean_usage/baseline_clients_first_seen.py

@@ -15,6 +15,7 @@ def __init__(self):
         self.target_table_id = TARGET_TABLE_ID
         self.prefix = PREFIX
         self.custom_render_kwargs = {}
+        self.per_app_requires_all_base_tables = True
 
     def generate_per_app_id(
         self,

sql_generators/glean_usage/baseline_clients_last_seen.py

@@ -25,3 +25,4 @@ def __init__(self):
         self.custom_render_kwargs = dict(
             usage_types=USAGE_TYPES,
         )
+        self.per_app_requires_all_base_tables = True

sql_generators/glean_usage/clients_last_seen_joined.py

@@ -2,6 +2,8 @@
 
 from sql_generators.glean_usage.common import GleanTable
 
+from bigquery_etl.config import ConfigLoader
+
 TARGET_TABLE_ID = "clients_last_seen_joined_v1"
 PREFIX = "clients_last_seen_joined"
 
@@ -15,3 +17,33 @@ def __init__(self):
         self.target_table_id = TARGET_TABLE_ID
         self.per_app_id_enabled = False
         self.cross_channel_template = None
+        self.per_app_requires_all_base_tables = True
+
+    def generate_per_app(
+        self,
+        project_id,
+        app_info,
+        output_dir=None,
+        use_cloud_function=True,
+        parallelism=8,
+        id_token=None,
+        all_base_tables_exist=None,
+    ):
+        """Generate per-app datasets."""
+        skip_apps = ConfigLoader.get(
+            "generate", "glean_usage", "clients_last_seen_joined", "skip_apps", fallback=[]
+        )
+        if app_info[0]["app_name"] in skip_apps:
+            print(
+                f"Skipping clients_last_seen_joined generation for {app_info[0]['app_name']}"
+            )
+            return
+        return super().generate_per_app(
+            project_id,
+            app_info,
+            output_dir,
+            use_cloud_function,
+            parallelism,
+            id_token,
+            all_base_tables_exist,
+        )

sql_generators/glean_usage/common.py

@@ -149,24 +149,6 @@ def table_names_from_baseline(baseline_table, include_project_id=True):
     )
 
 
-def referenced_table_exists(view_sql, id_token=None):
-    """Dry run the given view SQL to see if its referent exists."""
-    dryrun = DryRun("foo/bar/view.sql", content=view_sql, id_token=id_token)
-    # 403 is returned if referenced dataset doesn't exist; we need to check that the 403 is due to dataset not existing
-    # since dryruns on views will also return 403 due to the table CREATE
-    # 404 is returned if referenced table or view doesn't exist
-    return not any(
-        [
-            404 == e.get("code")
-            or (
-                403 == e.get("code")
-                and "bigquery.tables.create denied" not in e.get("message")
-            )
-            for e in dryrun.errors()
-        ]
-    )
-
-
 def _contains_glob(patterns):
     return any({"*", "?", "["}.intersection(pattern) for pattern in patterns)
 
@@ -204,6 +186,7 @@ def __init__(self):
         self.custom_render_kwargs = {}
         self.per_app_id_enabled = True
         self.per_app_enabled = True
+        self.per_app_requires_all_base_tables = False
         self.across_apps_enabled = True
         self.cross_channel_template = "cross_channel.view.sql"
         self.base_table_name = "baseline_v1"
@@ -361,10 +344,7 @@ def generate_per_app_id(
         if query_python:
             artifacts.append(Artifact(table, "query.py", query_python))
 
-        if not (referenced_table_exists(view_sql, id_token)):
-            logging.info("Skipping view for table which doesn't exist:" f" {table}")
-        else:
-            artifacts.append(Artifact(view, "view.sql", view_sql))
+        artifacts.append(Artifact(view, "view.sql", view_sql))
 
         skip_existing_artifact = self.skip_existing(output_dir, project_id)
 
@@ -409,15 +389,22 @@ def generate_per_app(
         use_cloud_function=True,
         parallelism=8,
         id_token=None,
+        all_base_tables_exist=None,
     ):
         """Generate the baseline table query per app_name."""
         if not self.per_app_enabled:
             return
-
+        
         app_name = app_info[0]["app_name"]
 
         target_view_name = "_".join(self.target_table_id.split("_")[:-1])
         target_dataset = app_name
+        
+        if self.per_app_requires_all_base_tables and not all_base_tables_exist:
+            logging.info(
+                f"Skipping per-app generation for {target_dataset}.{target_view_name} as not all baseline tables exist"
+            )
+            return
 
         datasets = [
             (a["bq_dataset_family"], a.get("app_channel", "release")) for a in app_info
@@ -465,10 +452,6 @@ def generate_per_app(
             )
             view = f"{project_id}.{target_dataset}.{target_view_name}"
 
-            if not (referenced_table_exists(sql, id_token=id_token)):
-                logging.info("Skipping view for table which doesn't exist:" f" {view}")
-                return
-
             if output_dir:
                 write_dataset_metadata(output_dir, view)
 
@@ -500,13 +483,6 @@ def generate_per_app(
             table = f"{project_id}.{target_dataset}_derived.{self.target_table_id}"
             view = f"{project_id}.{target_dataset}.{target_view_name}"
 
-            if not (referenced_table_exists(query_sql, id_token=id_token)):
-                logging.info(
-                    "Skipping query for table which doesn't exist:"
-                    f" {self.target_table_id}"
-                )
-                return
-
             if output_dir:
                 artifacts = [
                     Artifact(table, "query.sql", query_sql),

sql_generators/glean_usage/event_monitoring_live.py

@@ -94,6 +94,9 @@ def generate_per_app_id(
         parallelism=8,
         id_token=None,
     ):
+        # Cache app_info to avoid repeated calls
+        cached_app_info = get_app_info()
+        
         # Get the app ID from the baseline_table name.
         # This is what `common.py` also does.
         app_id = re.sub(r"_stable\..+", "", baseline_table)
@@ -120,7 +123,7 @@ def generate_per_app_id(
 
         app_name = [
             app_dataset["app_name"]
-            for _, app in get_app_info().items()
+            for _, app in cached_app_info.items()
             for app_dataset in app
             if dataset == app_dataset["bq_dataset_family"]
         ][0]
@@ -130,7 +133,7 @@ def generate_per_app_id(
         else:
             v1_name = [
                 app_dataset["v1_name"]
-                for _, app in get_app_info().items()
+                for _, app in cached_app_info.items()
                 for app_dataset in app
                 if dataset == app_dataset["bq_dataset_family"]
             ][0]
@@ -166,7 +169,7 @@ def generate_per_app_id(
             current_date=datetime.today().strftime("%Y-%m-%d"),
             app_name=[
                 app_dataset["canonical_app_name"]
-                for _, app in get_app_info().items()
+                for _, app in cached_app_info.items()
                 for app_dataset in app
                 if dataset == app_dataset["bq_dataset_family"]
             ][0],
@@ -227,6 +230,9 @@ def generate_across_apps(
         if not self.across_apps_enabled:
             return
 
+        # Cache app_info to avoid repeated calls
+        cached_app_info = get_app_info()
+
         aggregate_table = "event_monitoring_aggregates_v1"
         target_view_name = "_".join(self.target_table_id.split("_")[:-1])
 
@@ -249,7 +255,7 @@ def generate_across_apps(
                 dataset = app_dataset["bq_dataset_family"]
                 app_name = [
                     app_dataset["app_name"]
-                    for _, app in get_app_info().items()
+                    for _, app in cached_app_info.items()
                     for app_dataset in app
                     if dataset == app_dataset["bq_dataset_family"]
                 ][0]
@@ -261,7 +267,7 @@ def generate_across_apps(
                 else:
                     v1_name = [
                         app_dataset["v1_name"]
-                        for _, app in get_app_info().items()
+                        for _, app in cached_app_info.items()
                         for app_dataset in app
                         if dataset == app_dataset["bq_dataset_family"]
                     ][0]