Merge branch 'main' into reuse-influxql-logical-planner

Author: Rachelint

Cargo.lock

@@ -114,6 +114,7 @@ system_catalog = { path = "system_catalog" }
 table_engine = { path = "table_engine" }
 table_kv = { path = "components/table_kv" }
 tempfile = "3.1.0"
+toml = "0.7"
 tracing_util = { path = "components/tracing_util" }
 trace_metric = { path = "components/trace_metric" }
 trace_metric_derive = { path = "components/trace_metric_derive" }
@@ -143,6 +144,7 @@ server = { workspace = true }
 signal-hook = "0.3"
 sort = "0.8.5"
 table_engine = { workspace = true }
+toml = { workspace = true }
 tracing_util = { workspace = true }
 
 [build-dependencies]

Cargo.toml

@@ -5,7 +5,7 @@
 use std::{collections::HashMap, str::FromStr, sync::Arc};
 
 use common_util::config::{ReadableSize, TimeUnit};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use snafu::{ensure, Backtrace, GenerateBacktrace, ResultExt, Snafu};
 use tokio::sync::oneshot;
 
@@ -57,15 +57,15 @@ pub enum Error {
     InvalidOption { error: String, backtrace: Backtrace },
 }
 
-#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Default)]
+#[derive(Debug, Clone, Copy, Deserialize, Default, PartialEq, Serialize)]
 pub enum CompactionStrategy {
     #[default]
     Default,
     TimeWindow(TimeWindowCompactionOptions),
     SizeTiered(SizeTieredCompactionOptions),
 }
 
-#[derive(Debug, Clone, Copy, Deserialize, PartialEq)]
+#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Serialize)]
 pub struct SizeTieredCompactionOptions {
     pub bucket_low: f32,
     pub bucket_high: f32,
@@ -75,7 +75,7 @@ pub struct SizeTieredCompactionOptions {
     pub max_input_sstable_size: ReadableSize,
 }
 
-#[derive(Debug, Clone, Copy, Deserialize, PartialEq)]
+#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Serialize)]
 pub struct TimeWindowCompactionOptions {
     pub size_tiered: SizeTieredCompactionOptions,
     // TODO(boyan) In fact right now we only supports TimeUnit::Milliseconds resolution.

analytic_engine/src/compaction/mod.rs

@@ -185,10 +185,12 @@ impl Bucket {
 
     fn with_files(files: Vec<FileHandle>) -> Self {
         let total: usize = files.iter().map(|f| f.size() as usize).sum();
-        Self {
-            avg_size: total / files.len(),
-            files,
-        }
+        let avg_size = if files.is_empty() {
+            0
+        } else {
+            total / files.len()
+        };
+        Self { avg_size, files }
     }
 
     fn insert_file(&mut self, file: &FileHandle) {
@@ -823,4 +825,11 @@ mod tests {
                 .collect::<Vec<_>>()
         );
     }
+
+    #[test]
+    fn empty_bucket() {
+        let bucket = Bucket::with_files(vec![]);
+        assert_eq!(bucket.avg_size, 0);
+        assert!(bucket.files.is_empty());
+    }
 }

analytic_engine/src/compaction/picker.rs

@@ -21,12 +21,12 @@ use common_util::{
     time::DurationExt,
 };
 use log::{debug, error, info, warn};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use snafu::{ResultExt, Snafu};
 use table_engine::table::TableId;
 use tokio::{
     sync::{
-        mpsc::{self, error::SendError, Receiver, Sender},
+        mpsc::{self, Receiver, Sender},
         Mutex,
     },
     time,
@@ -38,9 +38,7 @@ use crate::{
         PickerManager, TableCompactionRequest, WaitError, WaiterNotifier,
     },
     instance::{
-        flush_compaction::{self, TableFlushOptions},
-        write_worker::CompactionNotifier,
-        Instance, SpaceStore,
+        flush_compaction::TableFlushOptions, write_worker::CompactionNotifier, Instance, SpaceStore,
     },
     table::data::TableDataRef,
     TableOptions,
@@ -54,7 +52,7 @@ pub enum Error {
 
 define_result!(Error);
 
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(default)]
 pub struct SchedulerConfig {
     pub schedule_channel_len: usize,
@@ -574,12 +572,11 @@ impl ScheduleWorker {
             None => {
                 // Memory usage exceeds the threshold, let's put pack the
                 // request.
-                debug!(
-                    "Compaction task is ignored, because of high memory usage:{}, task:{:?}",
+                warn!(
+                    "Compaction task is ignored, because of high memory usage:{}, task:{:?}, table:{}",
                     self.memory_limit.usage.load(Ordering::Relaxed),
-                    compaction_task,
+                    compaction_task, table_data.name
                 );
-                self.put_back_compaction_request(compact_req).await;
                 return;
             }
         };
@@ -596,29 +593,6 @@ impl ScheduleWorker {
         );
     }
 
-    async fn put_back_compaction_request(&self, req: TableCompactionRequest) {
-        if let Err(SendError(ScheduleTask::Request(TableCompactionRequest {
-            compaction_notifier,
-            waiter,
-            ..
-        }))) = self.sender.send(ScheduleTask::Request(req)).await
-        {
-            let e = Arc::new(
-                flush_compaction::Other {
-                    msg: "Failed to put back the compaction request for memory usage exceeds",
-                }
-                .build(),
-            );
-            if let Some(notifier) = compaction_notifier {
-                notifier.notify_err(e.clone());
-            }
-
-            let waiter_notifier = WaiterNotifier::new(waiter);
-            let wait_err = WaitError::Compaction { source: e };
-            waiter_notifier.notify_wait_result(Err(wait_err));
-        }
-    }
-
     async fn schedule(&mut self) {
         self.compact_tables().await;
         self.flush_tables().await;

analytic_engine/src/compaction/scheduler.rs

@@ -36,7 +36,7 @@ use wal::{
 pub use crate::{compaction::scheduler::SchedulerConfig, table_options::TableOptions};
 
 /// Config of analytic engine
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(default)]
 pub struct Config {
     /// Storage options of the engine
@@ -115,7 +115,7 @@ impl Default for Config {
 }
 
 /// Config of wal based on obkv
-#[derive(Debug, Default, Clone, Deserialize)]
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
 #[serde(default)]
 pub struct ObkvWalConfig {
     /// Obkv client config
@@ -222,7 +222,7 @@ impl From<WalNamespaceConfig> for NamespaceConfig {
 }
 
 /// Config of wal based on obkv
-#[derive(Debug, Default, Clone, Deserialize)]
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
 #[serde(default)]
 pub struct KafkaWalConfig {
     /// Kafka client config
@@ -235,7 +235,7 @@ pub struct KafkaWalConfig {
 }
 
 /// Config for wal based on RocksDB
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(default)]
 pub struct RocksDBConfig {
     /// Data directory used by RocksDB.
@@ -250,7 +250,7 @@ impl Default for RocksDBConfig {
     }
 }
 /// Options for wal storage backend
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(tag = "type")]
 pub enum WalStorageConfig {
     RocksDB(Box<RocksDBConfig>),

analytic_engine/src/lib.rs

@@ -22,7 +22,7 @@ use log::{debug, info, warn};
 use object_store::{ObjectStoreRef, Path};
 use parquet::data_type::AsBytes;
 use prost::Message;
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use snafu::{Backtrace, ResultExt, Snafu};
 use table_engine::table::TableId;
 use tokio::sync::Mutex;
@@ -157,7 +157,7 @@ impl MetaUpdateLogEntryIterator for MetaUpdateReaderImpl {
 }
 
 /// Options for manifest
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct Options {
     /// Steps to do snapshot
     pub snapshot_every_n_updates: usize,