[DRAFT] Build Enriched Traces & Transactions + Aggregation Tables

.vscode/settings.json

@@ -22,5 +22,8 @@
     "sqlfluff.executablePath": ".venv/bin/sqlfluff",
     "files.associations": {
         "*.sql.j2": "sql"
-    }
+    },
+    "sqlfluff.excludeRules": [
+        "LT04"
+    ]
 }

src/op_analytics/datapipeline/models/code/daily_address_summary_old.py

@@ -0,0 +1,21 @@
+# import duckdb
+
+# from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+# from op_analytics.datapipeline.models.compute.registry import register_model
+# from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+# @register_model(
+#     input_datasets=["intermediate/refined_transactions_fees_v1"],
+#     expected_outputs=["summary_v1"],
+#     auxiliary_views=[
+#         TemplatedSQLQuery(
+#             template_name="daily_address_summary",
+#             context={},
+#         ),
+#     ],
+# )
+# def daily_address_summary_old(duckdb_client: duckdb.DuckDBPyConnection) -> NamedRelations:
+#     return {
+#         "summary_v1": duckdb_client.view("daily_address_summary"),
+#     }

src/op_analytics/datapipeline/models/code/daily_transactions_fees_by_to.py

@@ -0,0 +1,26 @@
+import duckdb
+
+from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+from op_analytics.datapipeline.models.compute.registry import register_model
+from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+@register_model(
+    input_datasets=["intermediate/enriched_transactions_v1"],
+    expected_outputs=["daily_transactions_fees_by_to_v1"],
+    # TODO: Uncomment if we do this as a view (or some element as a view)
+    # auxiliary_views=[
+    #     TemplatedSQLQuery(
+    #         template_name="daily_transactions_fees_by_to",
+    #         context={},
+    #     ),
+    # ],
+)
+def daily_transactions_fees_by_to(duckdb_client: duckdb.DuckDBPyConnection) -> NamedRelations:
+    return {
+        "daily_transactions_fees_by_to_v1": duckdb_client.view(
+            """
+            TODO: AGGREGATION CODE
+            """
+        ),
+    }

src/op_analytics/datapipeline/models/code/event_emitting_transactions.py

@@ -0,0 +1,31 @@
+# # TO DEPRECATE?
+
+# import duckdb
+
+# from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+# from op_analytics.datapipeline.models.compute.registry import register_model
+# from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+# @register_model(
+#     input_datasets=["ingestion/logs_v1", "ingestion/transactions_v1", "ingestion/blocks_v1"],
+#     expected_outputs=["event_emitting_transactions_v1"],
+#     auxiliary_views=[
+#         TemplatedSQLQuery(
+#             template_name="refined_transactions_fees",
+#             context={},
+#         ),
+#         TemplatedSQLQuery(
+#             template_name="logs_topic0_filters",
+#             context={},
+#         ),
+#         TemplatedSQLQuery(
+#             template_name="event_emitting_transactions",
+#             context={},
+#         ),
+#     ],
+# )
+# def event_emitting_transactions(duckdb_client: duckdb.DuckDBPyConnection) -> NamedRelations:
+#     return {
+#         "event_emitting_transactions_v1": duckdb_client.view("event_emitting_transactions"),
+#     }

src/op_analytics/datapipeline/models/code/refined_trace_calls.py

@@ -0,0 +1,21 @@
+# import duckdb
+
+# from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+# from op_analytics.datapipeline.models.compute.registry import register_model
+# from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+# @register_model(
+#     input_datasets=["ingestion/traces_v1", "refined_transactions_fees_v1"],
+#     expected_outputs=["refined_trace_calls_v1"],
+#     auxiliary_views=[
+#         TemplatedSQLQuery(
+#             template_name="refined_trace_calls",
+#             context={},
+#         ),
+#     ],
+# )
+# def refined_trace_calls(duckdb_client: duckdb.DuckDBPyConnection) -> NamedRelations:
+#     return {
+#         "refined_trace_calls_v1": duckdb_client.view("refined_trace_calls"),
+#     }

src/op_analytics/datapipeline/models/code/refined_transactions_traces_address_models.py

@@ -0,0 +1,69 @@
+import duckdb
+
+from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+from op_analytics.datapipeline.models.compute.registry import register_model
+from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+@register_model(
+    input_datasets=[
+        "ingestion/transactions_v1",
+        "ingestion/blocks_v1",
+        "ingestion/logs_v1",
+        "ingestion/traces_v1",
+    ],
+    expected_outputs=[
+        "refined_transactions_fees_v1",
+        "refined_trace_calls_v1",
+        "event_emitting_transactions_v1",
+        "summary_v1",
+    ],
+    auxiliary_views=[
+        TemplatedSQLQuery(
+            template_name="refined_transactions_fees",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="refined_trace_calls",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="logs_topic0_filters",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="event_emitting_transactions",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="daily_address_summary",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="refined_trace_calls_agg_from_to_hash",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="refined_trace_calls_agg_to_hash",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="daily_trace_calls_agg_to",
+            context={},
+        ),
+    ],
+)
+def refined_transactions_traces_address_models(
+    duckdb_client: duckdb.DuckDBPyConnection,
+) -> NamedRelations:
+    return {
+        "refined_transactions_fees_v1": duckdb_client.view("refined_transactions_fees"),
+        "refined_trace_calls_v1": duckdb_client.view("refined_trace_calls"),
+        "event_emitting_transactions_v1": duckdb_client.view("event_emitting_transactions"),
+        "summary_v1": duckdb_client.view("daily_address_summary"),
+        "refined_trace_calls_agg_from_to_hash_v1": duckdb_client.view(
+            "refined_trace_calls_agg_from_to_hash"
+        ),
+        "refined_trace_calls_agg_to_hash_v1": duckdb_client.view("refined_trace_calls_agg_to_hash"),
+        "daily_trace_calls_agg_to_v1": duckdb_client.view("daily_trace_calls_agg_to"),
+    }

src/op_analytics/datapipeline/models/compute/udfs.py

@@ -15,6 +15,9 @@ def create_duckdb_macros(duckdb_client: duckdb.DuckDBPyConnection):
 
     CREATE OR REPLACE MACRO wei_to_gwei(a)
     AS a::DECIMAL(28, 0) * 0.000000001::DECIMAL(10, 10);
+
+    CREATE OR REPLACE MACRO gwei_to_eth(a)
+    AS a::DECIMAL(28, 10) * 0.000000001::DECIMAL(10, 10);
 
     CREATE OR REPLACE MACRO safe_div(a, b) AS
     IF(b = 0, NULL, a / b);
@@ -27,14 +30,34 @@ def create_duckdb_macros(duckdb_client: duckdb.DuckDBPyConnection):
     -- Truncate a timestamp to hour.
     CREATE OR REPLACE MACRO epoch_to_hour(a) AS
     date_trunc('hour', make_timestamp(a * 1000000::BIGINT));
+
+    -- Truncate a timestamp to day.
+    CREATE OR REPLACE MACRO epoch_to_day(a) AS
+    date_trunc('day', make_timestamp(a * 1000000::BIGINT));
 
     -- Division by 16 for DECIMAL types.
     CREATE OR REPLACE MACRO div16(a)
     AS a * 0.0625::DECIMAL(5, 5);
 
     --Get the length in bytes for binary data that is encoded as a hex string
     CREATE OR REPLACE MACRO hexstr_bytelen(x)
-    AS (length(x) - 2) / 2 
+    AS (length(x) - 2) / 2;
+
+    --Count non-zero bytes for binary data that is encoded as a hex string. We don't use hexstr_bytelen because we need to substring the input data.
+    CREATE OR REPLACE MACRO hexstr_nonzero_bytes(x)
+    AS length(replace(hex(unhex(substr(x, 3))), '00', '')) / 2;
+
+    --Count non-zero bytes for binary data that is encoded as a hex string
+    CREATE OR REPLACE MACRO hexstr_zero_bytes(x)
+    AS hexstr_bytelen(x) - hexstr_nonzero_bytes(x);
+
+    --Calculate calldata gas used for binary data that is encoded as a hex string (can be updated by an EIP)
+    CREATE OR REPLACE MACRO hexstr_calldata_gas(x)
+    AS 16*hexstr_nonzero_bytes(x) + 4*hexstr_zero_bytes(x);
+
+    --Get the method id for input data. This is the first 4 bytes, or first 10 string characters for binary data that is encoded as a hex string.
+    CREATE OR REPLACE MACRO hexstr_method_id(x)
+    AS substring(x,1,10)
     """)
 
 

src/op_analytics/datapipeline/models/templates/daily_address_summary.sql.j2

@@ -2,6 +2,7 @@ SELECT
   dt,
   chain,
   chain_id,
+  network,
   from_address AS address,
   -- Aggregates
 
@@ -43,9 +44,9 @@ SELECT
 
   sum(if(success, receipt_gas_used, 0)) AS success_l2_gas_used_sum,
 
-  sum(l1_gas_used) AS l1_gas_used_sum,
+  sum(l1_gas_used_unified) AS l1_gas_used_unified_sum,
 
-  sum(if(success, l1_gas_used, 0)) AS success_l1_gas_used_sum,
+  sum(if(success, l1_gas_used_unified, 0)) AS success_l1_gas_used_unified_sum,
 
   wei_to_eth(sum(tx_fee)) AS tx_fee_sum_eth,
 
@@ -58,7 +59,7 @@ SELECT
 
   wei_to_eth(sum(l2_priority_fee)) AS l2_priority_fee_sum_eth,
 
-  wei_to_eth(sum(l2_base_legacy)) AS l2_base_legacy_fee_sum_eth,
+  wei_to_eth(sum(l2_legacy_extra_fee)) AS l2_base_legacy_fee_sum_eth,
 
   -- L1 Fee and breakdown into BASE + BLOB
   wei_to_eth(sum(l1_fee)) AS l1_fee_sum_eth,
@@ -82,11 +83,29 @@ SELECT
   ) AS l1_base_price_avg_gwei,
 
   wei_to_gwei(safe_div(sum(l1_blob_fee), sum(l1_blob_scaled_size)))
-    AS l1_blob_fee_avg_gwei
+    AS l1_blob_fee_avg_gwei,
+
+  -- Data Processed
+  sum(input_zero_bytes) AS input_zero_bytes_sum,
+  sum(if(success, input_zero_bytes, 0)) AS success_input_zero_bytes_sum,
+
+  sum(input_nonzero_bytes) AS input_nonzero_bytes_sum,
+  sum(if(success, input_nonzero_bytes, 0)) AS success_input_nonzero_bytes_sum,
+
+  sum(input_byte_length) AS input_byte_length_sum,
+  sum(if(success, input_byte_length, 0)) AS success_input_byte_length_sum,
+
+  sum(estimated_size) AS estimated_size_sum,
+  sum(if(success, estimated_size, 0)) AS success_estimated_size_sum
+
 FROM
-  transaction_fees
+  refined_transactions_fees
+WHERE
+  NOT is_system_transaction
+  AND gas_price > 0
 GROUP BY
   1,
   2,
   3,
-  4
+  4,
+  5