feat: implement direct GCS streaming by URIs via BigQuery metadata; update docs and unit tests accordingly

Author: BLMgithub

assets/benchmarks/polars/40mrows_dataset_stats_log.csv

@@ -0,0 +1,111 @@
+view,timestamp,logger,memory,unit
+DEFAULT,2026-04-20T14:59:18.784422Z,METRIC_MEM:,429,MB
+DEFAULT,2026-04-20T14:59:19.784315Z,METRIC_MEM:,467.35,MB
+DEFAULT,2026-04-20T14:59:20.784602Z,METRIC_MEM:,525.09,MB
+DEFAULT,2026-04-20T14:59:21.784888Z,METRIC_MEM:,530.13,MB
+DEFAULT,2026-04-20T14:59:22.785227Z,METRIC_MEM:,533.62,MB
+DEFAULT,2026-04-20T14:59:23.785545Z,METRIC_MEM:,534.02,MB
+DEFAULT,2026-04-20T14:59:24.785893Z,METRIC_MEM:,534.46,MB
+DEFAULT,2026-04-20T14:59:25.786286Z,METRIC_MEM:,533.57,MB
+DEFAULT,2026-04-20T14:59:26.786608Z,METRIC_MEM:,533.07,MB
+DEFAULT,2026-04-20T14:59:27.786989Z,METRIC_MEM:,533.24,MB
+DEFAULT,2026-04-20T14:59:28.787246Z,METRIC_MEM:,534.23,MB
+DEFAULT,2026-04-20T14:59:29.787577Z,METRIC_MEM:,534.72,MB
+DEFAULT,2026-04-20T14:59:30.787938Z,METRIC_MEM:,534.72,MB
+DEFAULT,2026-04-20T14:59:31.788315Z,METRIC_MEM:,534.96,MB
+DEFAULT,2026-04-20T14:59:32.788644Z,METRIC_MEM:,536.2,MB
+DEFAULT,2026-04-20T14:59:33.788993Z,METRIC_MEM:,538.16,MB
+DEFAULT,2026-04-20T14:59:34.789341Z,METRIC_MEM:,538.66,MB
+DEFAULT,2026-04-20T14:59:35.789670Z,METRIC_MEM:,538.66,MB
+DEFAULT,2026-04-20T14:59:36.789550Z,METRIC_MEM:,539.64,MB
+DEFAULT,2026-04-20T14:59:37.789398Z,METRIC_MEM:,539.64,MB
+DEFAULT,2026-04-20T14:59:38.789317Z,METRIC_MEM:,541.61,MB
+DEFAULT,2026-04-20T14:59:39.789179Z,METRIC_MEM:,541.61,MB
+DEFAULT,2026-04-20T14:59:40.789326Z,METRIC_MEM:,546.59,MB
+DEFAULT,2026-04-20T14:59:41.789650Z,METRIC_MEM:,546.59,MB
+DEFAULT,2026-04-20T14:59:42.790023Z,METRIC_MEM:,546.59,MB
+DEFAULT,2026-04-20T14:59:43.790569Z,METRIC_MEM:,720.57,MB
+DEFAULT,2026-04-20T14:59:44.790956Z,METRIC_MEM:,950.8,MB
+DEFAULT,2026-04-20T14:59:45.791405Z,METRIC_MEM:,1116.99,MB
+DEFAULT,2026-04-20T14:59:46.791919Z,METRIC_MEM:,1334.91,MB
+DEFAULT,2026-04-20T14:59:47.792322Z,METRIC_MEM:,1531.51,MB
+DEFAULT,2026-04-20T14:59:48.792605Z,METRIC_MEM:,1672.36,MB
+DEFAULT,2026-04-20T14:59:49.793038Z,METRIC_MEM:,1945.44,MB
+DEFAULT,2026-04-20T14:59:50.796118Z,METRIC_MEM:,2140.77,MB
+DEFAULT,2026-04-20T14:59:51.793792Z,METRIC_MEM:,2344.64,MB
+DEFAULT,2026-04-20T14:59:52.794297Z,METRIC_MEM:,2603.11,MB
+DEFAULT,2026-04-20T14:59:53.795767Z,METRIC_MEM:,2983.83,MB
+DEFAULT,2026-04-20T14:59:54.796083Z,METRIC_MEM:,3151.56,MB
+DEFAULT,2026-04-20T14:59:55.822665Z,METRIC_MEM:,3624.16,MB
+DEFAULT,2026-04-20T14:59:56.818814Z,METRIC_MEM:,3774.27,MB
+DEFAULT,2026-04-20T14:59:57.818682Z,METRIC_MEM:,4308.78,MB
+DEFAULT,2026-04-20T14:59:58.818511Z,METRIC_MEM:,4809.82,MB
+DEFAULT,2026-04-20T14:59:59.821162Z,METRIC_MEM:,5383.96,MB
+DEFAULT,2026-04-20T15:00:00.818704Z,METRIC_MEM:,5452.28,MB
+DEFAULT,2026-04-20T15:00:01.819104Z,METRIC_MEM:,5468.47,MB
+DEFAULT,2026-04-20T15:00:02.819475Z,METRIC_MEM:,5457.18,MB
+DEFAULT,2026-04-20T15:00:03.819830Z,METRIC_MEM:,5447.39,MB
+DEFAULT,2026-04-20T15:00:04.820259Z,METRIC_MEM:,5465.26,MB
+DEFAULT,2026-04-20T15:00:05.820598Z,METRIC_MEM:,5463.11,MB
+DEFAULT,2026-04-20T15:00:06.820974Z,METRIC_MEM:,5449.82,MB
+DEFAULT,2026-04-20T15:00:07.821433Z,METRIC_MEM:,5462.01,MB
+DEFAULT,2026-04-20T15:00:08.821801Z,METRIC_MEM:,5461.89,MB
+DEFAULT,2026-04-20T15:00:09.822611Z,METRIC_MEM:,5466.36,MB
+DEFAULT,2026-04-20T15:00:10.839283Z,METRIC_MEM:,5462.5,MB
+DEFAULT,2026-04-20T15:00:11.826381Z,METRIC_MEM:,5461.91,MB
+DEFAULT,2026-04-20T15:00:12.826815Z,METRIC_MEM:,5458.66,MB
+DEFAULT,2026-04-20T15:00:13.827094Z,METRIC_MEM:,5472.28,MB
+DEFAULT,2026-04-20T15:00:14.830802Z,METRIC_MEM:,5777.54,MB
+DEFAULT,2026-04-20T15:00:15.834612Z,METRIC_MEM:,6242.66,MB
+DEFAULT,2026-04-20T15:00:16.831243Z,METRIC_MEM:,6952.15,MB
+DEFAULT,2026-04-20T15:00:17.832603Z,METRIC_MEM:,7086.52,MB
+DEFAULT,2026-04-20T15:00:18.837562Z,METRIC_MEM:,7565.95,MB
+DEFAULT,2026-04-20T15:00:19.832365Z,METRIC_MEM:,7538.39,MB
+DEFAULT,2026-04-20T15:00:20.837168Z,METRIC_MEM:,7572.7,MB
+DEFAULT,2026-04-20T15:00:21.833118Z,METRIC_MEM:,7471.39,MB
+DEFAULT,2026-04-20T15:00:22.833394Z,METRIC_MEM:,7300.56,MB
+DEFAULT,2026-04-20T15:00:23.834075Z,METRIC_MEM:,7132.16,MB
+DEFAULT,2026-04-20T15:00:24.834825Z,METRIC_MEM:,6974.58,MB
+DEFAULT,2026-04-20T15:00:25.835584Z,METRIC_MEM:,6812.74,MB
+DEFAULT,2026-04-20T15:00:26.836323Z,METRIC_MEM:,6683.75,MB
+DEFAULT,2026-04-20T15:00:27.837011Z,METRIC_MEM:,6570.68,MB
+DEFAULT,2026-04-20T15:00:28.840331Z,METRIC_MEM:,6553.34,MB
+DEFAULT,2026-04-20T15:00:29.842342Z,METRIC_MEM:,6449.74,MB
+DEFAULT,2026-04-20T15:00:30.842779Z,METRIC_MEM:,6420.25,MB
+DEFAULT,2026-04-20T15:00:31.843102Z,METRIC_MEM:,6447.04,MB
+DEFAULT,2026-04-20T15:00:32.846716Z,METRIC_MEM:,6428.24,MB
+DEFAULT,2026-04-20T15:00:33.850433Z,METRIC_MEM:,7085.44,MB
+DEFAULT,2026-04-20T15:00:34.847560Z,METRIC_MEM:,7094.02,MB
+DEFAULT,2026-04-20T15:00:35.847891Z,METRIC_MEM:,7108.24,MB
+DEFAULT,2026-04-20T15:00:36.847936Z,METRIC_MEM:,7095.18,MB
+DEFAULT,2026-04-20T15:00:37.847913Z,METRIC_MEM:,7092.72,MB
+DEFAULT,2026-04-20T15:00:38.847807Z,METRIC_MEM:,7062.34,MB
+DEFAULT,2026-04-20T15:00:39.853210Z,METRIC_MEM:,6921.07,MB
+DEFAULT,2026-04-20T15:00:40.853684Z,METRIC_MEM:,6909.97,MB
+DEFAULT,2026-04-20T15:00:41.854214Z,METRIC_MEM:,6891.98,MB
+DEFAULT,2026-04-20T15:00:42.854490Z,METRIC_MEM:,6879.31,MB
+DEFAULT,2026-04-20T15:00:43.855020Z,METRIC_MEM:,6877.95,MB
+DEFAULT,2026-04-20T15:00:44.855450Z,METRIC_MEM:,6894.8,MB
+DEFAULT,2026-04-20T15:00:45.855873Z,METRIC_MEM:,6756.96,MB
+DEFAULT,2026-04-20T15:00:46.857784Z,METRIC_MEM:,6485.72,MB
+DEFAULT,2026-04-20T15:00:47.856652Z,METRIC_MEM:,6450.91,MB
+DEFAULT,2026-04-20T15:00:48.857031Z,METRIC_MEM:,6517.63,MB
+DEFAULT,2026-04-20T15:00:49.862250Z,METRIC_MEM:,6513.41,MB
+DEFAULT,2026-04-20T15:00:50.857773Z,METRIC_MEM:,6521.28,MB
+DEFAULT,2026-04-20T15:00:51.858133Z,METRIC_MEM:,6505.07,MB
+DEFAULT,2026-04-20T15:00:52.858414Z,METRIC_MEM:,6429.95,MB
+DEFAULT,2026-04-20T15:00:53.858797Z,METRIC_MEM:,6476.4,MB
+DEFAULT,2026-04-20T15:00:54.859127Z,METRIC_MEM:,6480.01,MB
+DEFAULT,2026-04-20T15:00:55.859485Z,METRIC_MEM:,6481.43,MB
+DEFAULT,2026-04-20T15:00:56.859441Z,METRIC_MEM:,6478.38,MB
+DEFAULT,2026-04-20T15:00:57.859322Z,METRIC_MEM:,6479.16,MB
+DEFAULT,2026-04-20T15:00:58.859179Z,METRIC_MEM:,6474.57,MB
+DEFAULT,2026-04-20T15:00:59.859043Z,METRIC_MEM:,6344.71,MB
+DEFAULT,2026-04-20T15:01:00.859298Z,METRIC_MEM:,6363.68,MB
+DEFAULT,2026-04-20T15:01:01.859673Z,METRIC_MEM:,6307.33,MB
+DEFAULT,2026-04-20T15:01:02.859970Z,METRIC_MEM:,6302.04,MB
+DEFAULT,2026-04-20T15:01:03.860228Z,METRIC_MEM:,6297.34,MB
+DEFAULT,2026-04-20T15:01:04.860579Z,METRIC_MEM:,6293.2,MB
+DEFAULT,2026-04-20T15:01:05.860960Z,METRIC_MEM:,6287.33,MB
+DEFAULT,2026-04-20T15:01:06.861270Z,METRIC_MEM:,6284.43,MB
+DEFAULT,2026-04-20T15:01:07.861650Z,METRIC_MEM:,6284.43,MB

assets/benchmarks/polars/README.md

@@ -2,7 +2,7 @@
 
 This section details the methodology used to capture the memory metrics in the [`GCP Stress-Test Metrics (Scaling Efficiency)`](/README.md#gcp-stress-test-metrics-scaling-efficiency)
 
-The telemetry logger below was added **temporarily** to the orchestrator for a specific benchmarking run. This code was pushed directly to the Cloud Artifact Registry as an experimental image tag (`mem-record`) and is not part of the permanent git repository history.
+The telemetry logger below was added to the orchestrator for a specific benchmarking run. 
 
 ```python
 import psutil
@@ -28,7 +28,7 @@ finally:
     stop_event.set()
     logger_thread.join()
 ```
-Since `psutil` requires C-extensions to compile, the **Dockerfile** was modified to include the necessary build tools and the package itself. This allowed for benchmarking without altering the project's permanent `requirements.txt`.
+Since `psutil` requires C-extensions to compile, the **Dockerfile** was modified to include the necessary build tools and the package itself. This allowed for benchmarking without altering the project's permanent [`requirements.txt`](/data_pipeline/requirements.txt).
 
 ```docker
 FROM python:3.11-slim

data_pipeline/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.11-slim
+FROM python:3.12-slim
 
 ENV PYTHONDONTWRITEBYTECODE=1
 ENV PYTHONUNBUFFERED=1

data_pipeline/assembly/assembly_executor.py

@@ -3,11 +3,15 @@
 # =============================================================================
 
 import gc
+from typing import Dict
 import ctypes
 import platform
-from typing import Dict
 from data_pipeline.shared.run_context import RunContext
-from data_pipeline.shared.loader_exporter import load_historical_data, export_file
+from data_pipeline.shared.loader_exporter import (
+    load_historical_data,
+    scan_gcs_uris_from_bigquery,
+    export_file,
+)
 from data_pipeline.shared.modeling_configs import DIMENSION_REFERENCES
 from data_pipeline.assembly.assembly_logic import (
     init_report,
@@ -181,20 +185,26 @@ def orchestrate_dimension_refs(run_context: RunContext, report: Dict) -> bool:
         lf_raw = None
         df_dim = None
 
-        base_contracted_path = run_context.contracted_path
-
         try:
-            lf_raw = load_historical_data(
-                base_path=base_contracted_path,
-                table_name=table,
-                log_info=lambda msg: loaded_data(msg, report),
-            )
+            # Switch between local and gcp IO
+            if run_context.bq_project_id == "PROJECT_ID_NOT_DETECTED":
+                lf_raw = load_historical_data(
+                    base_path=run_context.storage_contracted_path, table_name=table
+                )
+            else:
+                lf_raw = scan_gcs_uris_from_bigquery(
+                    project_id=run_context.bq_project_id,
+                    dataset_id=run_context.bq_dataset_id,
+                    table_id=table,
+                    log_info=lambda msg: loaded_data(msg, report),
+                )
 
             if lf_raw is None:
                 return False
 
             primary_key = config.get("primary_key", [])
             require_col = config.get("required_column", [])
+            dtypes = config.get("dtypes", {})
 
             ok, df_dim = task_wrapper(
                 report=report,
@@ -204,6 +214,7 @@ def orchestrate_dimension_refs(run_context: RunContext, report: Dict) -> bool:
                 lf=lf_raw,
                 primary_key=primary_key,
                 req_column=require_col,
+                dtypes=dtypes,
             )
 
             if not ok:

data_pipeline/assembly/assembly_logic.py

@@ -6,7 +6,10 @@
 from pathlib import Path
 from typing import Dict, Callable, Any, List
 from data_pipeline.shared.run_context import RunContext
-from data_pipeline.shared.loader_exporter import load_historical_data
+from data_pipeline.shared.loader_exporter import (
+    load_historical_data,
+    scan_gcs_uris_from_bigquery,
+)
 from data_pipeline.shared.modeling_configs import ASSEMBLE_SCHEMA, ASSEMBLE_DTYPES
 
 EVENT_TABLES = ["df_orders", "df_order_items", "df_payments"]
@@ -151,7 +154,7 @@ def derive_fields(lf: pl.LazyFrame) -> pl.LazyFrame:
         )
         .dt.total_days()
         .cast(pl.Int16),
-        order_date=pl.col("order_purchase_timestamp").dt.date(),
+        order_date=pl.col("order_purchase_timestamp").dt.date().cast(pl.Datetime("us")),
         order_year_week=pl.col("order_purchase_timestamp")
         .dt.strftime("%G-W%V")
         .cast(pl.Categorical),
@@ -187,9 +190,17 @@ def freeze_schema(lf: pl.LazyFrame) -> pl.LazyFrame:
     if missing_cols:
         raise RuntimeError(f"missing required columns: {sorted(missing_cols)}")
 
-    lf_contract = lf.select(ASSEMBLE_SCHEMA).cast(pl.Schema(ASSEMBLE_DTYPES))
+    lf_contract = lf.select(ASSEMBLE_SCHEMA)
 
-    return lf_contract
+    datetime_cols = [
+        col for col, dtype in ASSEMBLE_DTYPES.items() if isinstance(dtype, pl.Datetime)
+    ]
+
+    lf_contract = lf_contract.with_columns(
+        [pl.col(col).dt.cast_time_unit("us") for col in datetime_cols]
+    )
+
+    return lf_contract.cast(pl.Schema(ASSEMBLE_DTYPES))
 
 
 # ------------------------------------------------------------
@@ -201,12 +212,14 @@ def dimension_references(
     lf: pl.LazyFrame,
     primary_key: list[str],
     req_column: list[str],
+    dtypes: dict,
 ) -> pl.LazyFrame:
     """
     Extracts a unique reference dataset from a historical source.
 
     Contract:
     - Subtractive Filtering: Selects specified 'req_column' set and enforces uniqueness.
+    - Type Enforcement: Casts columns to the formats defined in the provided 'dtypes' schema.
 
     Invariants:
     - Dataset Grain: Strictly one row per 'primary_key'.
@@ -218,7 +231,7 @@ def dimension_references(
     - [Structural] Crashes if input LazyFrame lacks 'primary_key' or 'req_column'.
     """
 
-    lf_dim = lf.select(req_column).unique(subset=primary_key)
+    lf_dim = lf.select(req_column).unique(subset=primary_key).cast(pl.Schema(dtypes))
 
     return lf_dim
 
@@ -283,10 +296,10 @@ def task_wrapper(
 
 def load_event_table(run_context: RunContext, report: Dict) -> Any:
     """
-    Batch-loads core event tables required for assembly.
+    Batch-loads core event tables required for assembly from BigQuery.
 
     Contract:
-    - Hydrate: Iterates through EVENT_TABLES and loads Parquet files from 'contracted_path'.
+    - Hydrate: Iterates through EVENT_TABLES and streams data via scan_gcs_uris_from_bigquery.
 
     Outputs:
     - Dict keyed by table name containing loaded LazyFrames.
@@ -295,22 +308,31 @@ def load_event_table(run_context: RunContext, report: Dict) -> Any:
     - [Operational] Returns None if any required table is missing or fails to load.
     """
 
-    base_contracted_path = run_context.contracted_path
     tables = {}
 
     for table_name in EVENT_TABLES:
         try:
-            df = load_historical_data(
-                base_path=base_contracted_path,
-                table_name=table_name,
-                log_info=lambda msg: loaded_data(msg, report),
-            )
+            # Switch between local and gcp IO
+            if run_context.bq_project_id == "PROJECT_ID_NOT_DETECTED":
+                df = load_historical_data(
+                    base_path=run_context.storage_contracted_path,
+                    table_name=table_name,
+                    log_info=lambda msg: loaded_data(msg, report),
+                )
+
+            else:
+                df = scan_gcs_uris_from_bigquery(
+                    project_id=run_context.bq_project_id,
+                    dataset_id=run_context.bq_dataset_id,
+                    table_id=table_name,
+                    log_info=lambda msg: loaded_data(msg, report),
+                )
 
             if df is not None:
                 tables[table_name] = df
 
         except Exception as e:
-            log_error(f"Required table {table_name} not found: {e}", report)
+            log_error(f"Required table {table_name} not found : {e}", report)
             return None
 
     if len(tables) < len(EVENT_TABLES):

data_pipeline/requirements.txt

@@ -2,4 +2,5 @@ polars==1.39.0
 pyarrow==19.0.0
 google-cloud-storage
 google-cloud-bigquery>=3.0.0
+google-cloud-bigquery-storage>=2.36.0
 psutil==5.9.8