feat: compaction scheduler and rate limiter

Cargo.toml

@@ -68,6 +68,7 @@ serde = { version = "1.0", features = ["derive"] }
 snafu = { version = "0.7", features = ["backtraces"] }
 sqlparser = "0.28"
 tokio = { version = "1.24.2", features = ["full"] }
+tokio-util = "0.7"
 tonic = "0.8"
 uuid = { version = "1", features = ["serde", "v4", "fast-rng"] }
 

docs/rfcs/2023-02-01-table-compaction.md

@@ -82,7 +82,7 @@ We can first group SSTs in level n into buckets according to some predefined tim
 SSTs are compacted in a size-tired manner (find SSTs with similar size and compact them to level n+1). 
 SSTs from different time windows are neven compacted together.
 That strategy guarantees SSTs in each level are mainly sorted in timestamp order which boosts queries with 
-explict timestamp condition, while size-tired compaction minimizes the impact to foreground writes. 
+explicit timestamp condition, while size-tired compaction minimizes the impact to foreground writes. 
 
 ### Alternatives
 

src/log-store/Cargo.toml

@@ -30,7 +30,7 @@ snafu = { version = "0.7", features = ["backtraces"] }
 store-api = { path = "../store-api" }
 tempdir = "0.3"
 tokio.workspace = true
-tokio-util = "0.7"
+tokio-util.workspace = true
 
 [dev-dependencies]
 rand = "0.8"

src/storage/Cargo.toml

@@ -34,6 +34,7 @@ snafu = { version = "0.7", features = ["backtraces"] }
 store-api = { path = "../store-api" }
 table = { path = "../table" }
 tokio.workspace = true
+tokio-util.workspace = true
 tonic.workspace = true
 uuid.workspace = true
 

src/storage/src/compaction.rs

@@ -0,0 +1,19 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod dedup_deque;
+mod picker;
+mod rate_limit;
+mod scheduler;
+mod task;

src/storage/src/compaction/dedup_deque.rs

@@ -0,0 +1,101 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::{HashMap, VecDeque};
+use std::fmt::{Debug, Formatter};
+use std::hash::Hash;
+
+/// Deque with key deduplication.
+#[derive(Default)]
+pub struct DedupDeque<K, V> {
+    deque: VecDeque<K>,
+    existing: HashMap<K, V>,
+}
+
+impl<K: Eq + Hash + Clone, V> DedupDeque<K, V> {
+    /// Pushes a key value to the back of deque.
+    /// Returns true if the deque does not already contain value with the same key, otherwise
+    /// returns false.
+    pub fn push_back(&mut self, key: K, value: V) -> bool {
+        debug_assert_eq!(self.deque.len(), self.existing.len());
+        if !self.existing.contains_key(&key) {
+            self.existing.insert(key.clone(), value);
+            self.deque.push_back(key);
+            return true;
+        }
+        false
+    }
+
+    /// Pushes a key value to the front of deque.
+    /// Returns true if the deque does not already contain value with the same key, otherwise
+    /// returns false.
+    pub fn push_front(&mut self, key: K, value: V) -> bool {
+        debug_assert_eq!(self.deque.len(), self.existing.len());
+        if !self.existing.contains_key(&key) {
+            self.existing.insert(key.clone(), value);
+            self.deque.push_front(key);
+            return true;
+        }
+        false
+    }
+
+    /// Pops a pair from the back of deque. Returns [None] if the deque is empty.
+    pub fn pop_front(&mut self) -> Option<(K, V)> {
+        debug_assert_eq!(self.deque.len(), self.existing.len());
+        let key = self.deque.pop_front()?;
+        let value = self.existing.remove(&key)?;
+        Some((key, value))
+    }
+
+    pub fn len(&self) -> usize {
+        debug_assert_eq!(self.deque.len(), self.existing.len());
+        self.deque.len()
+    }
+}
+
+impl<K, V> Debug for DedupDeque<K, V>
+where
+    K: Debug,
+    V: Debug,
+{
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DedupDeque")
+            .field("deque", &self.deque)
+            .field("existing", &self.existing)
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_dedup_deque() {
+        let mut deque = DedupDeque::default();
+        assert!(deque.push_back(1, "hello".to_string()));
+        assert_eq!(1, deque.len());
+        assert!(deque.push_back(2, "world".to_string()));
+        assert_eq!(2, deque.len());
+        assert_eq!((1, "hello".to_string()), deque.pop_front().unwrap());
+        assert_eq!(1, deque.len());
+        assert_eq!((2, "world".to_string()), deque.pop_front().unwrap());
+        assert_eq!(0, deque.len());
+
+        // insert duplicated item
+        assert!(deque.push_back(1, "hello".to_string()));
+        assert!(!deque.push_back(1, "world".to_string()));
+        assert_eq!((1, "hello".to_string()), deque.pop_front().unwrap());
+    }
+}

src/storage/src/compaction/picker.rs

@@ -0,0 +1,67 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::compaction::scheduler::CompactionRequestImpl;
+use crate::compaction::task::{CompactionTask, CompactionTaskImpl};
+
+/// Picker picks input SST files and build the compaction task.
+/// Different compaction strategy may implement different pickers.
+pub trait Picker<R, T: CompactionTask>: Send + 'static {
+    fn pick(&self, req: &R) -> crate::error::Result<T>;
+}
+
+/// L0 -> L1 all-to-all compaction based on time windows.
+pub(crate) struct SimplePicker {}
+
+#[allow(unused)]
+impl SimplePicker {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl Picker<CompactionRequestImpl, CompactionTaskImpl> for SimplePicker {
+    fn pick(&self, _req: &CompactionRequestImpl) -> crate::error::Result<CompactionTaskImpl> {
+        todo!()
+    }
+}
+
+#[cfg(test)]
+pub mod tests {
+    use std::marker::PhantomData;
+
+    use super::*;
+    use crate::compaction::scheduler::CompactionRequest;
+    use crate::compaction::task::tests::{CallbackRef, NoopCompactionTask};
+
+    pub(crate) struct MockPicker<R: CompactionRequest> {
+        pub cbs: Vec<CallbackRef>,
+        _phantom_data: PhantomData<R>,
+    }
+
+    impl<R: CompactionRequest> MockPicker<R> {
+        pub fn new(cbs: Vec<CallbackRef>) -> Self {
+            Self {
+                cbs,
+                _phantom_data: Default::default(),
+            }
+        }
+    }
+
+    impl<R: CompactionRequest> Picker<R, NoopCompactionTask> for MockPicker<R> {
+        fn pick(&self, _req: &R) -> crate::error::Result<NoopCompactionTask> {
+            Ok(NoopCompactionTask::new(self.cbs.clone()))
+        }
+    }
+}