chore: rebase main

Author: v0y4g3r

src/mito2/src/access_layer.rs

@@ -278,7 +278,7 @@ impl FilePathProvider for WriteCachePathProvider {
 /// Path provider that builds paths in region storage path.
 #[derive(Clone, Debug)]
 pub(crate) struct RegionFilePathFactory {
-    region_dir: String,
+    pub(crate) region_dir: String,
 }
 
 impl RegionFilePathFactory {

src/mito2/src/compaction/twcs.rs

@@ -323,8 +323,7 @@ mod tests {
 
     use super::*;
     use crate::compaction::test_util::{new_file_handle, new_file_handles};
-    use crate::sst::file::{FileId, FileMeta, Level};
-    use crate::test_util::NoopFilePurger;
+    use crate::sst::file::{FileId, Level};
 
     #[test]
     fn test_get_latest_window_in_seconds() {

src/mito2/src/sst/parquet.rs

@@ -102,7 +102,7 @@ mod tests {
     use tokio_util::compat::FuturesAsyncWriteCompatExt;
 
     use super::*;
-    use crate::access_layer::FilePathProvider;
+    use crate::access_layer::{FilePathProvider, RegionFilePathFactory};
     use crate::cache::{CacheManager, CacheStrategy, PageKey};
     use crate::read::BatchReader;
     use crate::sst::index::{Indexer, IndexerBuilder};
@@ -567,7 +567,7 @@ mod tests {
             ..Default::default()
         };
 
-        let path_provider = RegionFilePathProvider {
+        let path_provider = RegionFilePathFactory {
             region_dir: "test".to_string(),
         };
         let mut writer = ParquetWriter::new_with_object_store(

src/mito2/src/sst/parquet/writer.rs

@@ -185,19 +185,6 @@ where
         Ok(())
     }
 
-    async fn get_or_create_indexer(&mut self) -> &mut Indexer {
-        match self.current_indexer {
-            None => {
-                self.current_file = FileId::random();
-                let indexer = self.indexer_builder.build(self.current_file).await;
-                self.current_indexer = Some(indexer);
-                // safety: self.current_indexer already set above.
-                self.current_indexer.as_mut().unwrap()
-            }
-            Some(ref mut indexer) => indexer,
-        }
-    }
-
     /// Iterates source and writes all rows to Parquet file.
     ///
     /// Returns the [SstInfo] if the SST is written.
@@ -221,7 +208,11 @@ where
                 Ok(mut batch) => {
                     stats.update(&batch);
                     // safety: self.current_indexer must be set when first batch has been written.
-                    self.current_indexer.as_mut().unwrap().update(&batch).await;
+                    self.current_indexer
+                        .as_mut()
+                        .unwrap()
+                        .update(&mut batch)
+                        .await;
                     if let Some(max_file_size) = opts.max_file_size
                         && self.bytes_written.load(Ordering::Relaxed) > max_file_size
                     {
@@ -239,8 +230,6 @@ where
 
         self.finish_current_file(&mut results, &mut stats).await?;
 
-        let file_id = self.current_file;
-
         // object_store.write will make sure all bytes are written or an error is raised.
         Ok(results)
     }
@@ -314,11 +303,7 @@ where
                     .context(WriteParquetSnafu)?;
             self.writer = Some(arrow_writer);
 
-            let index_file_path = self.path_provider.build_index_file_path(self.current_file);
-            let indexer = self
-                .indexer_builder
-                .build(self.current_file, index_file_path)
-                .await;
+            let indexer = self.indexer_builder.build(self.current_file).await;
             self.current_indexer = Some(indexer);
 
             // safety: self.writer is assigned above