Airflow changes for EIP-1559

airflow/dags/resources/stages/enrich/schemas/blocks.json

@@ -92,5 +92,10 @@
         "name": "transaction_count",
         "type": "INT64",
         "description": "The number of transactions in the block"
+    },
+    {
+        "name": "base_fee_per_gas",
+        "type": "INT64",
+        "description": "Protocol base fee per gas, which can move up or down"
     }
 ]

airflow/dags/resources/stages/enrich/schemas/transactions.json

@@ -90,5 +90,25 @@
         "type": "STRING",
         "mode": "REQUIRED",
         "description": "Hash of the block where this transaction was in"
+    },
+    {
+        "name": "max_fee_per_gas",
+        "type": "INT64",
+        "description": "Total fee that covers both base and priority fees"
+    },
+    {
+        "name": "max_priority_fee_per_gas",
+        "type": "INT64",
+        "description": "Fee given to miners to incentivize them to include the transaction"
+    },
+    {
+        "name": "transaction_type",
+        "type": "INT64",
+        "description": "Transaction type. One of 0 (Legacy), 1 (Legacy), 2 (EIP-1559)"
+    },
+    {
+        "name": "receipt_effective_gas_price",
+        "type": "INT64",
+        "description": "The actual value per gas deducted from the senders account. Replacement of gas_price after EIP-1559"
     }
 ]

airflow/dags/resources/stages/enrich/sqls/balances.sql

@@ -18,15 +18,19 @@ with double_entry_book as (
     and (call_type not in ('delegatecall', 'callcode', 'staticcall') or call_type is null)
     union all
     -- transaction fees debits
-    select miner as address, sum(cast(receipt_gas_used as numeric) * cast(gas_price as numeric)) as value
+    select
+        miner as address,
+        sum(cast(receipt_gas_used as numeric) * cast((receipt_effective_gas_price - coalesce(base_fee_per_gas, 0)) as numeric)) as value
     from `{{params.destination_dataset_project_id}}.{{params.dataset_name}}.transactions` as transactions
     join `{{params.destination_dataset_project_id}}.{{params.dataset_name}}.blocks` as blocks on blocks.number = transactions.block_number
     where true
     and date(transactions.block_timestamp) <= '{{ds}}'
     group by blocks.miner
     union all
     -- transaction fees credits
-    select from_address as address, -(cast(receipt_gas_used as numeric) * cast(gas_price as numeric)) as value
+    select
+        from_address as address,
+        -(cast(receipt_gas_used as numeric) * cast(receipt_effective_gas_price as numeric)) as value
     from `{{params.destination_dataset_project_id}}.{{params.dataset_name}}.transactions`
     where true
     and date(block_timestamp) <= '{{ds}}'

airflow/dags/resources/stages/enrich/sqls/blocks.sql

@@ -16,7 +16,8 @@ SELECT
     blocks.extra_data,
     blocks.gas_limit,
     blocks.gas_used,
-    blocks.transaction_count
+    blocks.transaction_count,
+    blocks.base_fee_per_gas
 FROM {{params.dataset_name_raw}}.blocks AS blocks
 where true
     {% if not params.load_all_partitions %}

airflow/dags/resources/stages/enrich/sqls/merge/merge_blocks.sql

@@ -20,7 +20,8 @@ insert (
     extra_data,
     gas_limit,
     gas_used,
-    transaction_count
+    transaction_count,
+    base_fee_per_gas
 ) values (
     timestamp,
     number,
@@ -39,7 +40,8 @@ insert (
     extra_data,
     gas_limit,
     gas_used,
-    transaction_count
+    transaction_count,
+    base_fee_per_gas
 )
 when not matched by source and date(timestamp) = '{{ds}}' then
 delete

airflow/dags/resources/stages/enrich/sqls/merge/merge_transactions.sql

@@ -19,7 +19,11 @@ insert (
     receipt_status,
     block_timestamp,
     block_number,
-    block_hash
+    block_hash,
+    max_fee_per_gas,
+    max_priority_fee_per_gas,
+    transaction_type,
+    receipt_effective_gas_price
 ) values (
     `hash`,
     nonce,
@@ -37,7 +41,11 @@ insert (
     receipt_status,
     block_timestamp,
     block_number,
-    block_hash
+    block_hash,
+    max_fee_per_gas,
+    max_priority_fee_per_gas,
+    transaction_type,
+    receipt_effective_gas_price
 )
 when not matched by source and date(block_timestamp) = '{{ds}}' then
 delete

airflow/dags/resources/stages/enrich/sqls/transactions.sql

@@ -15,7 +15,11 @@ SELECT
     receipts.status AS receipt_status,
     TIMESTAMP_SECONDS(blocks.timestamp) AS block_timestamp,
     blocks.number AS block_number,
-    blocks.hash AS block_hash
+    blocks.hash AS block_hash,
+    transactions.max_fee_per_gas,
+    transactions.max_priority_fee_per_gas,
+    transactions.transaction_type,
+    receipts.effective_gas_price as receipt_effective_gas_price
 FROM {{params.dataset_name_raw}}.blocks AS blocks
     JOIN {{params.dataset_name_raw}}.transactions AS transactions ON blocks.number = transactions.block_number
     JOIN {{params.dataset_name_raw}}.receipts AS receipts ON transactions.hash = receipts.transaction_hash

airflow/dags/resources/stages/raw/schemas/blocks.json

@@ -92,5 +92,10 @@
         "name": "transaction_count",
         "type": "INT64",
         "description": "The number of transactions in the block"
+    },
+    {
+        "name": "base_fee_per_gas",
+        "type": "INT64",
+        "description": "Protocol base fee per gas, which can move up or down"
     }
 ]

airflow/dags/resources/stages/raw/schemas/receipts.json

@@ -45,5 +45,11 @@
         "name": "status",
         "type": "INT64",
         "description": "Either 1 (success) or 0 (failure) (post Byzantium)"
+    },
+    {
+
+        "name": "effective_gas_price",
+        "type": "INT64",
+        "description": "The amount of gas used by this specific transaction alone (a replacement for gasUsed field)"
     }
 ]

airflow/dags/resources/stages/raw/schemas/transactions.json

@@ -59,5 +59,20 @@
         "name": "input",
         "type": "STRING",
         "description": "The data sent along with the transaction"
+    },
+    {
+        "name": "max_fee_per_gas",
+        "type": "INT64",
+        "description": "Total fee that covers both base and priority fees"
+    },
+    {
+        "name": "max_priority_fee_per_gas",
+        "type": "INT64",
+        "description": "Fee given to miners to incentivize them to include the transaction"
+    },
+    {
+        "name": "transaction_type",
+        "type": "INT64",
+        "description": "An envelope for future transaction types"
     }
 ]