perf(pruning): Use more efficient query to delete past storage logs (#…

…2179)

## What ❔

Uses an SQL query for deleting past storage logs which promises to be
more efficient than the one currently used.

## Why ❔

The current DB query used for this purpose has fluctuating, but overall
slow, performance.

## Checklist

- [x] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [x] Tests for the changes have been added / updated.
- [x] Code has been formatted via `zk fmt` and `zk lint`.
- [x] Spellcheck has been run via `zk spellcheck`.

core/lib/dal/src/pruning_dal/mod.rs

@@ -318,29 +318,51 @@ impl PruningDal<'_, '_> {
         &mut self,
         l2_blocks_to_prune: ops::RangeInclusive<L2BlockNumber>,
     ) -> DalResult<u64> {
-        let execution_result = sqlx::query!(
-            r#"
-            DELETE FROM storage_logs
-            WHERE
-                storage_logs.miniblock_number < $1
-                AND hashed_key IN (
-                    SELECT
-                        hashed_key
-                    FROM
-                        storage_logs
-                    WHERE
-                        miniblock_number BETWEEN $1 AND $2
-                )
-            "#,
-            i64::from(l2_blocks_to_prune.start().0),
-            i64::from(l2_blocks_to_prune.end().0)
-        )
-        .instrument("hard_prune_batches_range#prune_storage_logs_from_past_l2_blocks")
-        .with_arg("l2_blocks_to_prune", &l2_blocks_to_prune)
-        .report_latency()
-        .execute(self.storage)
-        .await?;
-        Ok(execution_result.rows_affected())
+        /// Number of past logs to delete in a single query run.
+        const BATCHING_LIMIT: i64 = 10_000;
+
+        let mut total_rows_affected = 0;
+        loop {
+            let execution_result = sqlx::query!(
+                r#"
+                DELETE FROM storage_logs
+                WHERE
+                    ctid IN (
+                        SELECT
+                            prev_logs.ctid
+                        FROM
+                            storage_logs AS prev_logs
+                            INNER JOIN LATERAL (
+                                SELECT
+                                    1
+                                FROM
+                                    storage_logs AS current_logs
+                                WHERE
+                                    current_logs.miniblock_number BETWEEN $1 AND $2
+                                    AND current_logs.hashed_key = prev_logs.hashed_key
+                            ) AS current_logs ON TRUE
+                        WHERE
+                            prev_logs.miniblock_number < $1
+                        LIMIT
+                            $3
+                    )
+                "#,
+                i64::from(l2_blocks_to_prune.start().0),
+                i64::from(l2_blocks_to_prune.end().0),
+                BATCHING_LIMIT
+            )
+            .instrument("hard_prune_batches_range#prune_storage_logs_from_past_l2_blocks")
+            .with_arg("l2_blocks_to_prune", &l2_blocks_to_prune)
+            .report_latency()
+            .execute(self.storage)
+            .await?;
+
+            if execution_result.rows_affected() > 0 {
+                total_rows_affected += execution_result.rows_affected();
+            } else {
+                return Ok(total_rows_affected);
+            }
+        }
     }
 
     async fn prune_storage_logs_in_range(

core/lib/dal/src/pruning_dal/tests.rs

@@ -361,7 +361,7 @@ async fn storage_logs_pruning_works_correctly() {
     )
     .await;
 
-    transaction
+    let stats = transaction
         .pruning_dal()
         .hard_prune_batches_range(L1BatchNumber(4), L2BlockNumber(9))
         .await
@@ -377,8 +377,10 @@ async fn storage_logs_pruning_works_correctly() {
         &[random_storage_log(2, 3), random_storage_log(3, 4)],
     );
     assert_l2_block_storage_logs_equal(L2BlockNumber(1), &actual_logs, &[random_storage_log(1, 1)]);
+    assert_eq!(stats.deleted_storage_logs_from_past_batches, 0);
+    assert_eq!(stats.deleted_storage_logs_from_pruned_batches, 1);
 
-    transaction
+    let stats = transaction
         .pruning_dal()
         .hard_prune_batches_range(L1BatchNumber(10), L2BlockNumber(21))
         .await
@@ -400,6 +402,8 @@ async fn storage_logs_pruning_works_correctly() {
         &actual_logs,
         &[random_storage_log(5, 7)],
     );
+    assert_eq!(stats.deleted_storage_logs_from_past_batches, 1);
+    assert_eq!(stats.deleted_storage_logs_from_pruned_batches, 1);
 }
 
 #[tokio::test]