From ccd666aa9b71aba752c7d21fba447848b1604335 Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com>
Date: Fri, 2 Jun 2023 13:16:59 +0800
Subject: [PATCH] fix: avoid writing manifest and wal if no files are actually
 flushed (#1698)

* fix: avoid writing manifest and wal if no files are actually flushed

* fix: simplify log
---
 src/storage/src/compaction/task.rs | 6 +++++-
 src/storage/src/flush.rs           | 9 +++++++--
 2 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/src/storage/src/compaction/task.rs b/src/storage/src/compaction/task.rs
index b37b3ff9bf..233cf1a2b7 100644
--- a/src/storage/src/compaction/task.rs
+++ b/src/storage/src/compaction/task.rs
@@ -16,7 +16,7 @@ use std::collections::HashSet;
 use std::fmt::{Debug, Formatter};
 
 use common_base::readable_size::ReadableSize;
-use common_telemetry::{debug, error, timer};
+use common_telemetry::{debug, error, info, timer};
 use store_api::logstore::LogStore;
 use store_api::storage::RegionId;
 
@@ -148,6 +148,10 @@ impl<S: LogStore> CompactionTask for CompactionTaskImpl<S> {
             e
         })?;
         compacted.extend(self.expired_ssts.iter().map(FileHandle::meta));
+
+        let input_ids = compacted.iter().map(|f| f.file_id).collect::<Vec<_>>();
+        let output_ids = output.iter().map(|f| f.file_id).collect::<Vec<_>>();
+        info!("Compacting SST files, input: {input_ids:?}, output: {output_ids:?}");
         self.write_manifest_and_apply(output, compacted)
             .await
             .map_err(|e| {
diff --git a/src/storage/src/flush.rs b/src/storage/src/flush.rs
index 6501537c62..a5bd8f9149 100644
--- a/src/storage/src/flush.rs
+++ b/src/storage/src/flush.rs
@@ -239,6 +239,10 @@ impl<S: LogStore> FlushJob<S> {
         let _timer = timer!(FLUSH_ELAPSED);
 
         let file_metas = self.write_memtables_to_layer().await?;
+        if file_metas.is_empty() {
+            // skip writing manifest and wal if no files are flushed.
+            return Ok(());
+        }
         self.write_manifest_and_apply(&file_metas).await?;
 
         Ok(())
@@ -287,13 +291,14 @@ impl<S: LogStore> FlushJob<S> {
             });
         }
 
-        let metas = futures_util::future::try_join_all(futures)
+        let metas: Vec<_> = futures_util::future::try_join_all(futures)
             .await?
             .into_iter()
             .flatten()
             .collect();
 
-        logging::info!("Successfully flush memtables to files: {:?}", metas);
+        let file_ids = metas.iter().map(|f| f.file_id).collect::<Vec<_>>();
+        logging::info!("Successfully flush memtables, region:{region_id}, files: {file_ids:?}");
         Ok(metas)
     }