feat: implement flush method for mito table

src/mito/src/table.rs

@@ -322,6 +322,21 @@ impl<R: Region> Table for MitoTable<R> {
  }
  Ok(rows_deleted)
  }
+
+ async fn flush(&self) -> TableResult<()> {
+ let futs = self
+ .regions
+ .iter()
+ .map(|(_, region)| region.flush())
+ .collect::<Vec<_>>();
+
+ futures::future::try_join_all(futs)
+ .await
+ .map_err(BoxedError::new)
+ .context(table_error::TableOperationSnafu)?;
+
+ Ok(())
+ }
 }
 
 struct ChunkStream {

src/mito/src/table/test_util/mock_engine.rs

@@ -196,6 +196,10 @@ impl Region for MockRegion {
  async fn close(&self) -> Result<()> {
  Ok(())
  }
+
+ async fn flush(&self) -> Result<()> {
+ unimplemented!()
+ }
 }
 
 impl MockRegionInner {

src/storage/src/region.rs

@@ -125,6 +125,10 @@ impl<S: LogStore> Region for RegionImpl<S> {
  async fn close(&self) -> Result<()> {
  self.inner.close().await
  }
+
+ async fn flush(&self) -> Result<()> {
+ self.inner.flush().await
+ }
 }
 
 /// Storage related config for region.
@@ -550,4 +554,18 @@ impl<S: LogStore> RegionInner<S> {
  async fn close(&self) -> Result<()> {
  self.writer.close().await
  }
+
+ async fn flush(&self) -> Result<()> {
+ let writer_ctx = WriterContext {
+ shared: &self.shared,
+ flush_strategy: &self.flush_strategy,
+ flush_scheduler: &self.flush_scheduler,
+ compaction_scheduler: &self.compaction_scheduler,
+ sst_layer: &self.sst_layer,
+ wal: &self.wal,
+ writer: &self.writer,
+ manifest: &self.manifest,
+ };
+ self.writer.flush(writer_ctx).await
+ }
 }

src/storage/src/region/writer.rs

@@ -260,6 +260,15 @@ impl RegionWriter {
  Ok(())
  }
 
+ /// Flush task manually 
+ pub async fn flush<S: LogStore>(&self, writer_ctx: WriterContext<'_, S>) -> Result<()> {
+ let mut inner = self.inner.lock().await;
+
+ ensure!(!inner.is_closed(), error::ClosedRegionSnafu);
+
+ inner.manual_flush(writer_ctx).await
+ }
+
  /// Cancel flush task if any
  async fn cancel_flush(&self) -> Result<()> {
  let mut inner = self.inner.lock().await;
@@ -680,6 +689,13 @@ impl WriterInner {
  Some(schedule_compaction_cb)
  }
 
+ async fn manual_flush<S: LogStore>(
+ &mut self,
+ writer_ctx: WriterContext<'_, S>,
+ ) -> Result<()> {
+ self.trigger_flush(&writer_ctx).await?;
+ Ok(())
+ }
  #[inline]
  fn is_closed(&self) -> bool {
  self.closed

src/store-api/src/storage/region.rs

@@ -74,6 +74,8 @@ pub trait Region: Send + Sync + Clone + std::fmt::Debug + 'static {
  async fn alter(&self, request: AlterRequest) -> Result<(), Self::Error>;
 
  async fn close(&self) -> Result<(), Self::Error>;
+
+ async fn flush(&self) -> Result<(), Self::Error>;
 }
 
 /// Context for write operations.

src/table/src/table.rs

@@ -93,6 +93,11 @@ pub trait Table: Send + Sync {
  }
  .fail()?
  }
+
+ /// Flush table.
+ async fn flush(&self) -> Result<()> {
+ UnsupportedSnafu { operation: "FLUSH" }.fail()?
+ }
 }
 
 pub type TableRef = Arc<dyn Table>;