Fix Warehouse Pipeline (opensource-observer#1417)

* Some small fixes for int_events

* Add base playground to the pipeline

* fix

.github/scripts/create-dbt-profile.sh

@@ -26,6 +26,16 @@ opensource_observer:
       keyfile: ${service_account_path}
       project: ${GOOGLE_PROJECT_ID}
       threads: 32
+    base_playground:
+      type: bigquery
+      dataset: oso_base_playground
+      job_execution_time_seconds: 300
+      job_retries: 1
+      location: US
+      method: service-account
+      keyfile: ${service_account_path}
+      project: ${GOOGLE_PROJECT_ID}
+      threads: 32
     playground:
       type: bigquery
       dataset: oso_playground

.github/workflows/warehouse-run-data-pipeline.yml

@@ -138,10 +138,16 @@ jobs:
       - name: Run dbt for production
         run: |
           poetry run dbt run --target production
+      
+      - name: Run dbt for the base_playground
+        run: |
+          poetry run dbt run --target base_playground
 
       - name: Run dbt for the playground
         run: |
-          poetry run dbt run --target playground
+          poetry run dbt run --target playground --full-refresh
+        env:
+          PLAYGROUND_DAYS: 30
 
       # - name: Copy the bigquery tables to cloudsql
       #   run: |

warehouse/dbt/models/intermediate/blockchain/int_optimism_transactions.sql

@@ -22,10 +22,20 @@ receipts as (
   from {{ source("optimism", "receipts") }}
   {% if is_incremental() %}
   where block_timestamp > TIMESTAMP_SUB(_dbt_max_partition, INTERVAL 1 DAY)
-    {{ playground_filter("block_timestamp", is_start=True) }}
+    {{ playground_filter("block_timestamp", is_start=False) }}
   {% else %}
   {{ playground_filter("block_timestamp") }}
   {% endif %}
+), blocks as (
+  select *
+  from {{ source("optimism", "blocks") }}
+  {% if is_incremental() %}
+  where `timestamp` > TIMESTAMP_SUB(_dbt_max_partition, INTERVAL 1 DAY)
+    {{ playground_filter("`timestamp`", is_start=False) }}
+  {% else %}
+  {{ playground_filter("`timestamp`") }}
+  {% endif %}
+
 ), transactions_with_receipts as (
 
 select
@@ -40,7 +50,7 @@ select
 from {{ source("optimism", "transactions") }} as transactions
 inner join receipts as receipts
   on transactions.transaction_hash = receipts.transaction_hash
-inner join blocks as blocks
+inner join {{ source("optimism", "blocks") }} as blocks
   on transactions.block_hash = blocks.block_hash
 {% if is_incremental() %}
 where

warehouse/dbt/models/optimism_sources.yml

@@ -12,4 +12,7 @@ sources:
         identifier: logs
 
       - name: receipts
-        identifier: receipts
+        identifier: receipts
+
+      - name: blocks
+        identifier: blocks