chore: remove useless code (apache#630)

* chore: remove role table module

* remove wal synchronization

analytic_engine/src/instance/alter.rs

@@ -31,20 +31,6 @@ use crate::{
     table_options,
 };
 
-/// Policy of how to perform flush operation.
-#[derive(Default, Debug, Clone, Copy)]
-pub enum TableAlterSchemaPolicy {
-    /// Unknown flush policy, this is the default value.
-    #[default]
-    Unknown,
-    /// Perform Alter Schema operation.
-    Alter,
-    // TODO: use this policy and remove "allow(dead_code)"
-    /// Ignore this operation.
-    #[allow(dead_code)]
-    Noop,
-}
-
 impl Instance {
     // Alter schema need to be handled by write worker.
     pub async fn alter_schema_of_table(
@@ -88,7 +74,6 @@ impl Instance {
         worker_local: &mut WorkerLocal,
         table_data: &TableDataRef,
         request: AlterSchemaRequest,
-        #[allow(unused_variables)] policy: TableAlterSchemaPolicy,
     ) -> Result<()> {
         // Validate alter schema request.
         self.validate_before_alter(table_data, &request)?;

analytic_engine/src/instance/mod.rs

@@ -41,7 +41,6 @@ use crate::{
         meta_data::cache::MetaCacheRef,
     },
     table::data::{TableDataRef, TableShardInfo},
-    wal_synchronizer::WalSynchronizer,
     TableOptions,
 };
 
@@ -55,11 +54,6 @@ pub enum Error {
     StopScheduler {
         source: crate::compaction::scheduler::Error,
     },
-
-    #[snafu(display("Failed to stop WAL Synchronizer, err:{}", source))]
-    StopWalSynchronizer {
-        source: crate::wal_synchronizer::Error,
-    },
 }
 
 define_result!(Error);
@@ -166,7 +160,6 @@ pub struct Instance {
     // End of write group options.
     compaction_scheduler: CompactionSchedulerRef,
     file_purger: FilePurger,
-    wal_synchronizer: WalSynchronizer,
 
     meta_cache: Option<MetaCacheRef>,
     /// Engine memtable memory usage collector
@@ -187,11 +180,6 @@ impl Instance {
     pub async fn close(&self) -> Result<()> {
         self.file_purger.stop().await.context(StopFilePurger)?;
 
-        self.wal_synchronizer
-            .stop()
-            .await
-            .context(StopWalSynchronizer)?;
-
         self.space_store.close().await?;
 
         self.compaction_scheduler

analytic_engine/src/instance/open.rs

@@ -42,7 +42,6 @@ use crate::{
         file::FilePurger,
     },
     table::data::{TableData, TableDataRef},
-    wal_synchronizer::{WalSynchronizer, WalSynchronizerConfig},
 };
 
 impl Instance {
@@ -74,10 +73,6 @@ impl Instance {
 
         let file_purger = FilePurger::start(&bg_runtime, store_picker.default_store().clone());
 
-        let mut wal_synchronizer =
-            WalSynchronizer::new(WalSynchronizerConfig::default(), wal_manager);
-        wal_synchronizer.start(&bg_runtime).await;
-
         let iter_options = IterOptions {
             batch_size: ctx.config.scan_batch_size,
             sst_background_read_parallelism: ctx.config.sst_background_read_parallelism,
@@ -92,7 +87,6 @@ impl Instance {
             write_group_command_channel_cap: ctx.config.write_group_command_channel_cap,
             compaction_scheduler,
             file_purger,
-            wal_synchronizer,
             meta_cache: ctx.meta_cache.clone(),
             mem_usage_collector: Arc::new(MemUsageCollector::default()),
             db_write_buffer_size: ctx.config.db_write_buffer_size,

analytic_engine/src/instance/write.rs

@@ -161,20 +161,6 @@ impl EncodeContext {
     }
 }
 
-/// Policy of how to perform flush operation.
-#[derive(Default, Debug, Clone, Copy)]
-pub enum TableWritePolicy {
-    /// Unknown policy, this is the default value.
-    #[default]
-    Unknown,
-    /// A full write operation. Write to both memtable and WAL.
-    Full,
-    // TODO: use this policy and remove "allow(dead_code)"
-    /// Only write to memtable.
-    #[allow(dead_code)]
-    MemOnly,
-}
-
 impl Instance {
     /// Write data to the table under give space.
     pub async fn write_to_table(
@@ -215,7 +201,6 @@ impl Instance {
         space: &SpaceRef,
         table_data: &TableDataRef,
         request: WriteRequest,
-        #[allow(unused_variables)] policy: TableWritePolicy,
     ) -> Result<usize> {
         let mut encode_ctx = EncodeContext::new(request.row_group);
 

analytic_engine/src/instance/write_worker.rs

@@ -29,7 +29,6 @@ use table_engine::{
 };
 use tokio::sync::{mpsc, oneshot, watch, watch::Ref, Mutex, Notify};
 
-use super::alter::TableAlterSchemaPolicy;
 use crate::{
     compaction::{TableCompactionRequest, WaitResult},
     instance::{
@@ -805,13 +804,7 @@ impl WriteWorker {
 
         let write_res = self
             .instance
-            .process_write_table_command(
-                &mut self.local,
-                &space,
-                &table_data,
-                request,
-                write::TableWritePolicy::Unknown,
-            )
+            .process_write_table_command(&mut self.local, &space, &table_data, request)
             .await;
         if let Err(res) = tx.send(write_res) {
             error!(
@@ -900,12 +893,7 @@ impl WriteWorker {
 
         let alter_res = self
             .instance
-            .process_alter_schema_command(
-                &mut self.local,
-                &table_data,
-                request,
-                TableAlterSchemaPolicy::Unknown,
-            )
+            .process_alter_schema_command(&mut self.local, &table_data, request)
             .await
             .map_err(|e| Box::new(e) as GenericError)
             .context(Channel);

analytic_engine/src/lib.rs

@@ -9,7 +9,6 @@ mod instance;
 mod manifest;
 pub mod memtable;
 mod payload;
-mod role_table;
 pub mod row_iter;
 mod sampler;
 pub mod setup;
@@ -18,7 +17,6 @@ pub mod sst;
 mod storage_options;
 pub mod table;
 pub mod table_options;
-mod wal_synchronizer;
 
 #[cfg(any(test, feature = "test"))]
 pub mod tests;