Feat concurrent replay wal

src/analytic_engine/src/instance/wal_replayer.rs

@@ -186,9 +186,24 @@ impl Replay for TableBasedReplay {
             batch_size: context.wal_replay_batch_size,
             ..Default::default()
         };
-        for table_data in table_datas {
-            let table_id = table_data.id;
-            if let Err(e) = Self::recover_table_logs(context, table_data, &read_ctx).await {
+
+        let mut tasks = futures::stream::iter(
+            table_datas
+                .iter()
+                .map(|table_data| {
+                    let table_id = table_data.id;
+                    let read_ctx = &read_ctx;
+                    async move {
+                        let ret = Self::recover_table_logs(context, table_data, read_ctx).await;
+                        (table_id, ret)
+                    }
+                })
+                .collect::<Vec<_>>(),
+        )
+        .buffer_unordered(20);
+        while let Some((table_id, ret)) = tasks.next().await {
+            if let Err(e) = ret {
+                // If occur error, mark this table as failed and store the cause.
                 failed_tables.insert(table_id, e);
             }
         }