Address ballooning costs (#1784)

* Significantly reduces costs of operating the pull request checking

* Ensure that external-prs only run if there are changes to sql files

ops/external-prs/package.json

@@ -51,6 +51,7 @@
     "mustache": "^4.2.0",
     "octokit": "^3.1.0",
     "oss-directory": "^0.0.15",
+    "simple-git": "^3.25.0",
     "tmp-promise": "^3.0.3",
     "ts-dedent": "^2.2.0",
     "winston": "^3.11.0",

ops/external-prs/src/oso/deploy.ts

@@ -5,6 +5,8 @@ import dayjs from "dayjs";
 import { exec } from "child_process";
 import * as util from "util";
 import * as path from "path";
+import { simpleGit } from "simple-git";
+import _ from "lodash";
 
 import { Repo } from "../github.js";
 import { logger } from "../utils/logger.js";
@@ -61,6 +63,25 @@ export class PRTestDeployCoordinator {
       repo: this.repo,
     });
 
+    const prInfo = await this.octo.rest.pulls.get({
+      owner: this.repo.owner,
+      repo: this.repo.name,
+      pull_number: pr,
+    });
+    const baseSha = prInfo.data.base.sha;
+
+    // Compare the baseSha and the current sha to see if anything has changed in the dbt directory.
+    // If not then report a success and that this check is not needed for this PR.
+    if (!(await this.hasDbtChanges(baseSha, sha, checkoutPath))) {
+      await this.noRelevantChanges(sha);
+      await this.appUtils.setStatusComment(
+        pr,
+        dedent`
+        Test deployment unnecessary, no dbt files have been changed.
+        `,
+      );
+    }
+
     const datasetName = this.datasetNameFromPR(pr);
 
     await this.getOrCreateDataset(datasetName);
@@ -121,6 +142,19 @@ export class PRTestDeployCoordinator {
     );
   }
 
+  async hasDbtChanges(
+    baseSha: string,
+    headSha: string,
+    checkoutPath: string,
+  ): Promise<boolean> {
+    const git = simpleGit({ baseDir: checkoutPath });
+    const diffResult = await git.diffSummary([baseSha, headSha]);
+    const dbtFiles = diffResult.files.filter((f) => {
+      return _.startsWith(f.file, "warehouse/dbt");
+    });
+    return dbtFiles.length > 0;
+  }
+
   async teardown(pr: number) {
     // Generally this should be called from a pull_request_target "closed" event
     // This will delete a pull request's test deployment
@@ -306,6 +340,19 @@ export class PRTestDeployCoordinator {
     });
   }
 
+  private async noRelevantChanges(sha: string) {
+    return setCheckStatus(this.octo, this.repo.owner, this.repo.name, {
+      name: "test-deploy",
+      head_sha: sha,
+      status: CheckStatus.Completed,
+      conclusion: CheckConclusion.Success,
+      output: {
+        title: "Test Deployment Skipped",
+        summary: `No dbt changes to deploy`,
+      },
+    });
+  }
+
   private async failCheckStatus(sha: string) {
     return setCheckStatus(this.octo, this.repo.owner, this.repo.name, {
       name: "test-deploy",

warehouse/dbt/macros/models/factory_deployments.sql

@@ -8,7 +8,7 @@
 %}
 
 {% if not transactions_source %}
-{% set transactions_source = source(network_name, "transactions") %}
+{% set transactions_source = oso_source(network_name, "transactions") %}
 {% endif %}
 {% set transactions_table_hash_value = "`txs`.`%s`" % (
     transactions_table_transaction_hash_column,
@@ -46,7 +46,7 @@ select
   txs.to_address as originating_contract,
   traces.from_address as factory_address, 
   traces.to_address as contract_address
-from {{ source(network_name, traces) }} as traces
+from {{ oso_source(network_name, traces) }} as traces
 inner join transactions as txs
   on {{ transactions_table_hash_value }} = {{ traces_table_hash_value }}
 where

warehouse/dbt/macros/models/first_time_addresses.sql

@@ -21,7 +21,7 @@ from (
         ,min_by(to_address, block_number) as first_tx_to
         ,min_by(`hash`, block_number) as first_tx_hash
         ,min_by(substring(input, 1, 10), block_number) as first_method_id
-    from {{ source(network_name, "transactions") }}
+    from {{ oso_source(network_name, "transactions") }}
     {% if is_incremental() %}
     where {{ block_timestamp_column }} > TIMESTAMP_SUB(_dbt_max_partition, INTERVAL 1 DAY)
     {% else %}

warehouse/dbt/macros/models/goog_blockchain_deployers.sql

@@ -20,7 +20,7 @@ WHERE
 {% else %}
 select
   *
-from {{ source("base_playground", "%s_deployers" % network_name) }}
+from {{ oso_source(network_name, "deployers") }}
 {% if is_incremental() %}
 where block_timestamp > TIMESTAMP_SUB(_dbt_max_partition, INTERVAL 1 DAY)
   {{ playground_filter("block_timestamp", is_start=False) }}

warehouse/dbt/macros/models/known_proxies.sql

@@ -1,11 +1,13 @@
 {% macro known_proxies(network_name, start, traces="traces") %}
 
 {# 
-
   This model is used to help identify smart contract accounts by looking for transactions that interact with the most widespread proxy contracts.
-
 #}
 
+{% if target.name == 'playground' %}
+{% set start = "TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL %s DAY)" % env_var('PLAYGROUND_DAYS', '90') %}
+{% endif %}
+
 with proxy_contracts as (
   select * 
     from UNNEST([ STRUCT
@@ -57,7 +59,7 @@ proxy_txns as (
       then traces.to_address
       else null
     end as proxy_address
-  from {{ source(network_name, traces) }} as traces
+  from {{ oso_source(network_name, traces) }} as traces
   inner join proxy_contracts as proxies
     on lower(traces.from_address) = lower(proxies.factory_address)
     or lower(traces.to_address) = lower(proxies.factory_address)

warehouse/dbt/macros/models/potential_bots.sql

@@ -8,7 +8,7 @@ with sender_transfer_rates as (
         ,min(block_timestamp) as min_block_time
         ,max(block_timestamp) as max_block_time
         ,count(*) as hr_txs
-    from {{ source(network_name, "transactions") }}
+    from {{ oso_source(network_name, "transactions") }}
     where 
         gas_price > 0 
         and receipt_gas_used > 0

warehouse/dbt/macros/models/transactions_with_receipts_deployers.sql

@@ -23,7 +23,7 @@ WHERE
 {% else %}
 select 
   *
-from {{ source("base_playground", "%s_deployers" % network_name) }}
+from {{ oso_source(network_name, "deployers") }}
 {% if is_incremental() %}
 where block_timestamp > TIMESTAMP_SUB(_dbt_max_partition, INTERVAL 1 DAY)
   {{ playground_filter("block_timestamp", is_start=False) }}

warehouse/dbt/models/base_playground_sources.yml

@@ -19,6 +19,9 @@ sources:
 
       - name: ethereum_deployers
         identifier: stg_ethereum__deployers
+
+      - name: ethereum_transactions
+        identifier: int_ethereum_transactions
 
       - name: frax_traces
         identifier: int_frax_traces

warehouse/dbt/models/intermediate/directory/int_potential_bots.sql

@@ -1,12 +1,15 @@
 {% set network_names = [
     'base',
-    'ethereum',
     'frax',
     'metal',
     'mode',
     'pgn',
     'zora'
 ] %}  --
+{% if target.name == 'production' %}
+{# This is a temporary measure for now to cut costs on the playground #}
+{% set network_names = network_names + ['ethereum'] %}
+{% endif %}
 
 {% for network_name in network_names %}
 

warehouse/dbt/models/marts/superchain/verification/rf4_transactions_window.sql

@@ -23,7 +23,7 @@
     lower(from_address) as from_address,
     lower(to_address) as to_address,
     '{{ network_upper }}' as network,
-  from {{ source(network, "transactions") }}
+  from {{ oso_source(network, "transactions") }}
   where
     block_timestamp > '{{ start_date }}'
     and block_timestamp < '{{ end_date }}'

warehouse/dbt/models/staging/ethereum/stg_ethereum__deployers.sql

@@ -31,7 +31,7 @@ where
   {% endif %}
 {% else %}
   select *
-  from {{ source("base_playground", "ethereum_deployers") }}
+  from {{ source(current_playground(), "ethereum_deployers") }}
   {% if is_incremental() %}
     where
       block_timestamp > TIMESTAMP_SUB(_dbt_max_partition, interval 1 day)

warehouse/dbt/models/staging/ethereum/stg_ethereum__factories.sql

@@ -1,5 +1,6 @@
 {{
   config(
+    enabled=target.name == 'production',
     materialized='incremental',
     partition_by={
       "field": "block_timestamp",

warehouse/dbt/models/staging/ethereum/stg_ethereum__potential_bots.sql

@@ -1,4 +1,5 @@
 {{ config(
+    enabled=target.name == 'production',
     materialized='table',
     partition_by={
       "field": "min_block_time",