chore: rebase develop

Author: v0y4g3r

config/datanode.example.toml

@@ -27,4 +27,4 @@ tcp_nodelay = false
 
 [compaction]
 compaction_after_flush = true
-max_inflight_task = 32
+max_inflight_task = 32

src/storage/src/compaction/picker.rs

@@ -57,7 +57,7 @@ impl<S: LogStore> Picker<CompactionRequestImpl<S>, CompactionTaskImpl<S>> for Si
         ctx: &PickerContext,
         req: &CompactionRequestImpl<S>,
     ) -> crate::error::Result<Option<CompactionTaskImpl<S>>> {
-        let levels = &req.levels;
+        let levels = &req.levels();
 
         for level_num in 0..levels.level_num() {
             let level = levels.level(level_num as u8);

src/storage/src/sst.rs

@@ -31,7 +31,7 @@ use crate::error::Result;
 use crate::memtable::BoxedBatchIterator;
 use crate::read::{Batch, BoxedBatchReader};
 use crate::schema::ProjectedSchemaRef;
-use crate::sst::parquet::{ParquetReader, ParquetWriter, Source};
+use crate::sst::parquet::{ParquetReader, ParquetWriter};
 
 /// Maximum level of SSTs.
 pub const MAX_LEVEL: u8 = 2;
@@ -444,102 +444,3 @@ mod tests {
         );
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::collections::HashSet;
-
-    use super::*;
-
-    fn create_handle(name: &str, level: Level) -> FileHandle {
-        FileHandle::new(FileMeta {
-            file_name: name.to_string(),
-            time_range: None,
-            level,
-        })
-    }
-
-    #[test]
-    fn test_level_metas_add_and_remove() {
-        let metas = LevelMetas::new();
-        let merged = metas.merge(
-            vec![create_handle("a", 0), create_handle("b", 0)].into_iter(),
-            vec![].into_iter(),
-        );
-
-        assert_eq!(
-            HashSet::from(["a".to_string(), "b".to_string()]),
-            merged
-                .level(0)
-                .files()
-                .map(|f| f.file_name().to_string())
-                .collect()
-        );
-
-        let merged1 = merged.merge(
-            vec![create_handle("c", 1), create_handle("d", 1)].into_iter(),
-            vec![].into_iter(),
-        );
-        assert_eq!(
-            HashSet::from(["a".to_string(), "b".to_string()]),
-            merged1
-                .level(0)
-                .files()
-                .map(|f| f.file_name().to_string())
-                .collect()
-        );
-
-        assert_eq!(
-            HashSet::from(["c".to_string(), "d".to_string()]),
-            merged1
-                .level(1)
-                .files()
-                .map(|f| f.file_name().to_string())
-                .collect()
-        );
-
-        let removed1 = merged1.merge(
-            vec![].into_iter(),
-            vec![create_handle("a", 0), create_handle("c", 0)].into_iter(),
-        );
-        assert_eq!(
-            HashSet::from(["b".to_string()]),
-            removed1
-                .level(0)
-                .files()
-                .map(|f| f.file_name().to_string())
-                .collect()
-        );
-
-        assert_eq!(
-            HashSet::from(["c".to_string(), "d".to_string()]),
-            removed1
-                .level(1)
-                .files()
-                .map(|f| f.file_name().to_string())
-                .collect()
-        );
-
-        let removed2 = removed1.merge(
-            vec![].into_iter(),
-            vec![create_handle("c", 1), create_handle("d", 1)].into_iter(),
-        );
-        assert_eq!(
-            HashSet::from(["b".to_string()]),
-            removed2
-                .level(0)
-                .files()
-                .map(|f| f.file_name().to_string())
-                .collect()
-        );
-
-        assert_eq!(
-            HashSet::new(),
-            removed2
-                .level(1)
-                .files()
-                .map(|f| f.file_name().to_string())
-                .collect()
-        );
-    }
-}

src/storage/src/sst/parquet.rs

@@ -45,11 +45,9 @@ use parquet::file::properties::WriterProperties;
 use parquet::format::FileMetaData;
 use parquet::schema::types::SchemaDescriptor;
 use snafu::{OptionExt, ResultExt};
-use store_api::storage::ChunkReader;
 use table::predicate::Predicate;
 use tokio::io::BufReader;
 
-use crate::chunk::ChunkReaderImpl;
 use crate::error::{
     self, DecodeParquetTimeRangeSnafu, NewRecordBatchSnafu, ReadObjectSnafu, ReadParquetSnafu,
     Result, WriteObjectSnafu, WriteParquetSnafu,
@@ -320,7 +318,6 @@ impl<'a> ParquetReader<'a> {
         // checks if converting time range unit into ts col unit will result into rounding error.
         if time_unit_lossy(&self.time_range, ts_col_unit) {
             let filter = RowFilter::new(vec![Box::new(PlainTimestampRowFilter::new(
-                ts_col_idx,
                 self.time_range,
                 projection,
             ))]);
@@ -342,15 +339,9 @@ impl<'a> ParquetReader<'a> {
                 .and_then(|s| s.convert_to(ts_col_unit))
                 .map(|t| t.value()),
         ) {
-            Box::new(FastTimestampRowFilter::new(
-                ts_col_idx, projection, lower, upper,
-            )) as _
+            Box::new(FastTimestampRowFilter::new(projection, lower, upper)) as _
         } else {
-            Box::new(PlainTimestampRowFilter::new(
-                ts_col_idx,
-                self.time_range,
-                projection,
-            )) as _
+            Box::new(PlainTimestampRowFilter::new(self.time_range, projection)) as _
         };
         let filter = RowFilter::new(vec![row_filter]);
         Some(filter)
@@ -371,21 +362,14 @@ fn time_unit_lossy(range: &TimestampRange, ts_col_unit: TimeUnit) -> bool {
 /// `FastTimestampRowFilter` is used to filter rows within given timestamp range when reading
 /// row groups from parquet files, while avoids fetching all columns from SSTs file.
 struct FastTimestampRowFilter {
-    timestamp_index: usize,
     lower_bound: i64,
     upper_bound: i64,
     projection: ProjectionMask,
 }
 
 impl FastTimestampRowFilter {
-    fn new(
-        ts_col_idx: usize,
-        projection: ProjectionMask,
-        lower_bound: i64,
-        upper_bound: i64,
-    ) -> Self {
+    fn new(projection: ProjectionMask, lower_bound: i64, upper_bound: i64) -> Self {
         Self {
-            timestamp_index: ts_col_idx,
             lower_bound,
             upper_bound,
             projection,
@@ -400,7 +384,7 @@ impl ArrowPredicate for FastTimestampRowFilter {
 
     /// Selects the rows matching given time range.
     fn evaluate(&mut self, batch: RecordBatch) -> std::result::Result<BooleanArray, ArrowError> {
-        let ts_col = batch.column(self.timestamp_index);
+        let ts_col = batch.column(0);
 
         macro_rules! downcast_and_compute {
             ($typ: ty) => {
@@ -442,15 +426,13 @@ impl ArrowPredicate for FastTimestampRowFilter {
 /// [PlainTimestampRowFilter] iterates each element in timestamp column, build a [Timestamp] struct
 /// and checks if given time range contains the timestamp.
 struct PlainTimestampRowFilter {
-    timestamp_index: usize,
     time_range: TimestampRange,
     projection: ProjectionMask,
 }
 
 impl PlainTimestampRowFilter {
-    fn new(timestamp_index: usize, time_range: TimestampRange, projection: ProjectionMask) -> Self {
+    fn new(time_range: TimestampRange, projection: ProjectionMask) -> Self {
         Self {
-            timestamp_index,
             time_range,
             projection,
         }
@@ -463,7 +445,7 @@ impl ArrowPredicate for PlainTimestampRowFilter {
     }
 
     fn evaluate(&mut self, batch: RecordBatch) -> std::result::Result<BooleanArray, ArrowError> {
-        let ts_col = batch.column(self.timestamp_index);
+        let ts_col = batch.column(0);
 
         macro_rules! downcast_and_compute {
             ($array_ty: ty, $unit: ident) => {{
@@ -531,33 +513,6 @@ impl BatchReader for ChunkStream {
     }
 }
 
-/// Parquet writer data source.
-pub enum Source {
-    /// Writes rows from memtable to parquet
-    Iter(BoxedBatchIterator),
-    /// Writes row from ChunkReaderImpl (maybe a set of SSTs) to parquet.
-    Reader(ChunkReaderImpl),
-}
-
-impl Source {
-    async fn next_batch(&mut self) -> Result<Option<Batch>> {
-        match self {
-            Source::Iter(iter) => iter.next().transpose(),
-            Source::Reader(reader) => reader
-                .next_chunk()
-                .await
-                .map(|p| p.map(|chunk| Batch::new(chunk.columns))),
-        }
-    }
-
-    fn projected_schema(&self) -> ProjectedSchemaRef {
-        match self {
-            Source::Iter(iter) => iter.schema(),
-            Source::Reader(reader) => reader.projected_schema().clone(),
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;