feat: invoke flush_table and compact_table in fuzz tests (#4045)

* feat: invoke `flush_table` and `compact_table` in fuzz tests

* feat: support to flush and compact physical metric table

* fix: avoid to create tables with the same name

* feat: validate values after flushing or compacting table

src/metric-engine/src/engine.rs

@@ -34,6 +34,7 @@ use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_recordbatch::SendableRecordBatchStream;
 use mito2::engine::MitoEngine;
+use snafu::ResultExt;
 use store_api::metadata::RegionMetadataRef;
 use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
 use store_api::region_engine::{
@@ -44,7 +45,7 @@ use store_api::storage::{RegionId, ScanRequest};
 
 use self::state::MetricEngineState;
 use crate::data_region::DataRegion;
-use crate::error::{Result, UnsupportedRegionRequestSnafu};
+use crate::error::{self, Result, UnsupportedRegionRequestSnafu};
 use crate::metadata_region::MetadataRegion;
 use crate::utils;
 
@@ -144,10 +145,33 @@ impl RegionEngine for MetricEngine {
                     .alter_region(region_id, alter, &mut extension_return_value)
                     .await
             }
-            RegionRequest::Delete(_)
-            | RegionRequest::Flush(_)
-            | RegionRequest::Compact(_)
-            | RegionRequest::Truncate(_) => UnsupportedRegionRequestSnafu { request }.fail(),
+            RegionRequest::Flush(_) => {
+                if self.inner.is_physical_region(region_id) {
+                    self.inner
+                        .mito
+                        .handle_request(region_id, request)
+                        .await
+                        .context(error::MitoFlushOperationSnafu)
+                        .map(|response| response.affected_rows)
+                } else {
+                    UnsupportedRegionRequestSnafu { request }.fail()
+                }
+            }
+            RegionRequest::Compact(_) => {
+                if self.inner.is_physical_region(region_id) {
+                    self.inner
+                        .mito
+                        .handle_request(region_id, request)
+                        .await
+                        .context(error::MitoFlushOperationSnafu)
+                        .map(|response| response.affected_rows)
+                } else {
+                    UnsupportedRegionRequestSnafu { request }.fail()
+                }
+            }
+            RegionRequest::Delete(_) | RegionRequest::Truncate(_) => {
+                UnsupportedRegionRequestSnafu { request }.fail()
+            }
             RegionRequest::Catchup(ref req) => self.inner.catchup_region(region_id, *req).await,
         };
 

src/metric-engine/src/error.rs

@@ -121,6 +121,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Mito flush operation fails"))]
+    MitoFlushOperation {
+        source: BoxedError,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Mito catchup operation fails"))]
     MitoCatchupOperation {
         source: BoxedError,
@@ -128,6 +135,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Mito compact operation fails"))]
+    MitoCompactOperation {
+        source: BoxedError,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Failed to collect record batch stream"))]
     CollectRecordBatchStream {
         source: common_recordbatch::error::Error,
@@ -275,7 +289,9 @@ impl ErrorExt for Error {
             | CloseMitoRegion { source, .. }
             | MitoReadOperation { source, .. }
             | MitoWriteOperation { source, .. }
-            | MitoCatchupOperation { source, .. } => source.status_code(),
+            | MitoCatchupOperation { source, .. }
+            | MitoFlushOperation { source, .. }
+            | MitoCompactOperation { source, .. } => source.status_code(),
 
             CollectRecordBatchStream { source, .. } => source.status_code(),