Implement BigQuery Atomic View Swapping & High-Performance GCS Streaming

.gcp/terraforms/bigquery.tf

@@ -0,0 +1,90 @@
+# ------------------------------------------------------------
+# OPS EXTERNALIZED TABLES (For metadata caching)
+# ------------------------------------------------------------
+
+resource "google_bigquery_connection" "biglake_connection" {
+  connection_id = "ops_biglake_connection"
+  location      = var.region
+  friendly_name = "BigLake Connection for GCS Parquet Scanning"
+  cloud_resource {}
+}
+
+# Enable connection service to access pipeline bucket
+resource "google_storage_bucket_iam_member" "biglake_storage_viewer" {
+  bucket = google_storage_bucket.ops_pipeline_bucket.name
+  role   = "roles/storage.objectViewer"
+  member = "serviceAccount:${google_bigquery_connection.biglake_connection.cloud_resource[0].service_account_id}"
+}
+
+resource "google_bigquery_dataset" "silver_dataset" {
+  dataset_id = var.bq_dataset_id
+  location   = var.region
+
+  delete_contents_on_destroy = false
+}
+
+locals {
+  external_tables = [
+    "df_orders",
+    "df_customers",
+    "df_order_items",
+    "df_products",
+    "df_payments"
+  ]
+}
+
+resource "google_bigquery_table" "external_tables" {
+  for_each   = toset(local.external_tables)
+  dataset_id = google_bigquery_dataset.silver_dataset.dataset_id
+  table_id   = each.key
+
+  # Might throw error if contracted/ is empty
+  external_data_configuration {
+    autodetect    = true
+    source_format = "PARQUET"
+    connection_id = google_bigquery_connection.biglake_connection.name
+    source_uris   = ["gs://${google_storage_bucket.ops_pipeline_bucket.name}/contracted/${each.key}_*.parquet"]
+
+    # Triggered manually by pipeline
+    metadata_cache_mode = "MANUAL"
+  }
+  lifecycle {
+    prevent_destroy = true
+  }
+}
+
+
+# ------------------------------------------------------------
+# BIGQUERY SEMANTTIC DATASETS (For table versionining)
+# ------------------------------------------------------------
+
+locals {
+  # Expiration for versioned tables
+  one_month_ms = 2678400000
+
+  semantic_datasets = [
+    "seller_semantic",
+    "customer_semantic",
+    "product_semantic"
+  ]
+}
+
+resource "google_bigquery_dataset" "semantic_datasets" {
+  for_each   = toset(local.semantic_datasets)
+  dataset_id = each.key
+  location   = var.region
+
+  delete_contents_on_destroy  = false
+  default_table_expiration_ms = local.one_month_ms
+
+  description = "Semantic layer for ${each.key}. Tables expire after 1 month."
+
+  labels = {
+    env   = var.environment
+    layer = "semantic"
+  }
+
+  lifecycle {
+    prevent_destroy = true
+  }
+}

.gcp/terraforms/iam_bindings.tf

@@ -24,7 +24,9 @@ locals {
     "roles/monitoring.admin",                # Manage Monitoring in monitoring.tf
     "roles/logging.configWriter",            # Required for log-based alert policies
     "roles/iam.serviceAccountAdmin",         # Manage Alert policies in monitoring.tf
-    "roles/iam.admin"                        # Manage Iam roles
+    "roles/iam.admin",                       # Manage Iam roles
+    "roles/bigquery.admin",                  # Manage BigQuery datasets and views
+    "roles/serviceusage.serviceUsageAdmin",  # Manage APIs
   ]
 }
 
@@ -74,6 +76,21 @@ resource "google_storage_bucket_iam_member" "pipeline_runner_pipeline_access" {
   member = "serviceAccount:${google_service_account.platform_accounts["ops-pipeline-sa"].email}"
 }
 
+# Pipeline Runner BigQuery Access
+locals {
+  pipeline_bq_roles = [
+    "roles/bigquery.dataEditor",
+    "roles/bigquery.jobUser"
+  ]
+}
+
+resource "google_project_iam_member" "pipeline_runner_bq_access" {
+  for_each = toset(local.pipeline_bq_roles)
+  project  = var.project_id
+  role     = each.key
+  member   = "serviceAccount:${google_service_account.platform_accounts["ops-pipeline-sa"].email}"
+}
+
 
 # ------------------------------------------------------------
 # GOOGLE SERVICE AGENTS (Pub/Sub)

.gcp/terraforms/jobs.tf

@@ -16,22 +16,49 @@ resource "google_cloud_run_v2_job" "pipeline" {
 
         resources {
           limits = {
-            cpu    = "2"
+            cpu    = "4"
             memory = "8Gi"
           }
         }
         env {
           name  = "POLARS_MAX_THREADS"
-          value = "2"
+          value = "4"
+        }
+        env {
+          name  = "GCP_REGION"
+          value = var.region
+        }
+        env {
+          name  = "BQ_DATASET_ID"
+          value = var.bq_dataset_id
+        }
+        env {
+          name  = "GCP_PROJECT"
+          value = var.project_id
+        }
+
+        volume_mounts {
+          name       = "ephemeral-disk-1"
+          mount_path = "/tmp"
+        }
+      }
+
+      volumes {
+        name = "ephemeral-disk-1"
+        empty_dir {
+          size_limit = "10Gi"
         }
       }
     }
   }
   lifecycle {
     ignore_changes = [
+      # Github ci-infra updates image every update
       template[0].template[0].containers[0].image,
       client,
-      client_version
+      client_version,
+      # Block terraform from defaulting medium to MEMORY, DISK isn't supported by provider yet
+      template[0].template[0].volumes[0].empty_dir[0].medium
     ]
   }
 }

.gcp/terraforms/main.tf

@@ -31,6 +31,8 @@ locals {
     "cloudscheduler.googleapis.com",
     "iamcredentials.googleapis.com",
     "drive.googleapis.com",
+    "bigquery.googleapis.com",
+    "bigqueryconnection.googleapis.com",
   ]
 }
 

.gcp/terraforms/storage.tf

@@ -46,3 +46,4 @@ resource "google_storage_bucket" "ops_pipeline_bucket" {
     }
   }
 }
+

.gcp/terraforms/variables.tf

@@ -4,7 +4,7 @@ variable "project_id" {
 }
 
 variable "region" {
-  description = "The Default GCP region"
+  description = "The Project GCP region"
   type        = string
   default     = "us-east1"
 }
@@ -24,3 +24,8 @@ variable "alert_email_map" {
   description = "List of emails to receive pipeline alerts"
   sensitive   = true
 }
+
+variable "bq_dataset_id" {
+  description = "BigQuery dataset containing externalized GCS tables"
+  type        = string
+}

.github/workflows/ci-infra.yml

@@ -55,4 +55,5 @@ jobs:
           TF_VAR_region: ${{ env.REGION }}
           TF_VAR_github_repo: ${{ env.GITHUB_REPO }}
           TF_VAR_alert_email_map: ${{ secrets.ALERT_EMAIL_MAP }}
+          TF_VAR_bq_dataset_id: ${{secrets.BQ_DATASET_ID}}
         run: terraform apply -auto-approve

.gitignore

@@ -7,9 +7,11 @@ __pycache__/
 runtime/
 /data/raw
 data/published/
+data/id_mapping/
 data/run_artifact
 data/contracted/
 assets/benchmarks/benchmark.py
+docker-compose.benchmark.yml
 
 # local editor configs
 pyrightconfig.json

.vscode/settings.json

@@ -5,5 +5,6 @@
     "python.testing.unittestEnabled": false,
     "python.testing.pytestEnabled": true,
     "python-envs.defaultEnvManager": "ms-python.python:conda",
-    "python-envs.defaultPackageManager": "ms-python.python:conda"
+    "python-envs.defaultPackageManager": "ms-python.python:conda",
+    "python.terminal.activateEnvironment": false
 }