delete 'shallow blocks' in block cache when not using firehose (#4691)

* add mechanism to delete 'shallow blocks' in block cache when a chain is not firehose-only

* add proper indexes for chainXX.blocks in migration

---------

Co-authored-by: Stéphane Duchesneau <stephane.duchesneau@streamingfast.io>

Author: sduchesneau

node/src/main.rs

@@ -378,6 +378,16 @@ async fn main() {
             Some(&eth_firehose_only_networks)
         };
 
+        let eth_firehose_only_network_names = match eth_firehose_only_networks {
+            Some(firehose_only) => Some(Vec::from_iter(firehose_only.networks.keys())),
+            None => None,
+        };
+
+        network_store
+            .block_store()
+            .cleanup_shallow_blocks(eth_firehose_only_network_names)
+            .unwrap();
+
         let ethereum_chains = ethereum_networks_as_chains(
             &mut blockchain_map,
             &logger,

store/postgres/migrations/2023-06-22_050000_index_ethereum_nulldata/down.sql

@@ -0,0 +1,17 @@
+drop index ethereum_blocks_nulldata_idx ;
+
+do $$
+declare
+    tables cursor for select namespace
+                        from ethereum_networks
+                       where namespace != 'public';
+begin
+	for table_record in tables loop
+		execute
+			'drop index '
+			|| table_record.namespace
+			|| '.'
+			|| 'blocks_nulldata_idx';
+	end loop;
+end;
+$$;

store/postgres/migrations/2023-06-22_050000_index_ethereum_nulldata/up.sql

@@ -0,0 +1,26 @@
+create index if not exists
+    ethereum_blocks_nulldata_idx
+on
+	ethereum_blocks
+using
+    BTREE ((data->'block'->'data'))
+where
+    data->'block'->'data' = 'null'::jsonb;
+
+do $$
+declare
+    tables cursor for select namespace
+                        from ethereum_networks
+                       where namespace != 'public';
+begin
+	for table_record in tables loop
+		execute
+			'create index if not exists blocks_nulldata_idx on '
+			|| table_record.namespace
+			|| '.'
+			|| 'blocks '
+            || 'using BTREE ((data->''block''->''data'')) '
+            || 'where data->''block''->''data'' = ''null''::jsonb';
+	end loop;
+end;
+$$;

store/postgres/src/block_store.rs

@@ -7,7 +7,7 @@ use std::{
 use graph::{
     blockchain::ChainIdentifier,
     components::store::BlockStore as BlockStoreTrait,
-    prelude::{error, warn, BlockNumber, BlockPtr, Logger, ENV_VARS},
+    prelude::{error, info, warn, BlockNumber, BlockPtr, Logger, ENV_VARS},
 };
 use graph::{constraint_violation, prelude::CheapClone};
 use graph::{
@@ -442,6 +442,23 @@ impl BlockStore {
         Ok(())
     }
 
+    pub fn cleanup_shallow_blocks(
+        &self,
+        firehose_networks: Option<Vec<&String>>,
+    ) -> Result<(), StoreError> {
+        for store in self.stores.read().unwrap().values() {
+            if let Some(fh_nets) = firehose_networks.clone() {
+                if fh_nets.contains(&&store.chain) {
+                    continue;
+                };
+            }
+
+            info!(&self.logger, "Cleaning shallow blocks on non-firehose chain"; "network" => &store.chain);
+            store.cleanup_shallow_blocks()?
+        }
+        Ok(())
+    }
+
     fn truncate_block_caches(&self) -> Result<(), StoreError> {
         for store in self.stores.read().unwrap().values() {
             store.truncate_block_cache()?

store/postgres/src/chain_store.rs

@@ -346,6 +346,7 @@ mod data {
                   data         jsonb not null
                 );
                 create index blocks_number ON {nsp}.blocks using btree(number);
+                create index blocks_expr_idx ON {nsp}.blocks USING BTREE ((data->'block'->'data')) where data->'block'->'data' = 'null'::jsonb;
 
                 create table {nsp}.call_cache (
 	              id               bytea not null primary key,
@@ -418,6 +419,18 @@ mod data {
             Ok(())
         }
 
+        pub(super) fn cleanup_shallow_blocks(&self, conn: &PgConnection) -> Result<(), StoreError> {
+            let table_name = match &self {
+                Storage::Shared => ETHEREUM_BLOCKS_TABLE_NAME,
+                Storage::Private(Schema { blocks, .. }) => &blocks.qname,
+            };
+            conn.batch_execute(&format!(
+                "delete from {} WHERE data->'block'->'data' = 'null'::jsonb;",
+                table_name
+            ))?;
+            Ok(())
+        }
+
         /// Insert a block. If the table already contains a block with the
         /// same hash, then overwrite that block since it may be adding
         /// transaction receipts. If `overwrite` is `true`, overwrite a
@@ -1553,6 +1566,12 @@ impl ChainStore {
             .delete_blocks_by_hash(&conn, &self.chain, block_hashes)
     }
 
+    pub fn cleanup_shallow_blocks(&self) -> Result<(), StoreError> {
+        let conn = self.get_conn()?;
+        self.storage.cleanup_shallow_blocks(&conn)?;
+        Ok(())
+    }
+
     pub fn truncate_block_cache(&self) -> Result<(), StoreError> {
         let conn = self.get_conn()?;
         self.storage.truncate_block_cache(&conn)?;