fix: memory usage during log ingestion

src/mito2/src/compaction.rs

@@ -570,7 +570,6 @@ pub struct SerializedCompactionOutput {
 struct CompactionSstReaderBuilder<'a> {
     metadata: RegionMetadataRef,
     sst_layer: AccessLayerRef,
-    cache: Option<CacheManagerRef>,
     inputs: &'a [FileHandle],
     append_mode: bool,
     filter_deleted: bool,
@@ -584,7 +583,7 @@ impl<'a> CompactionSstReaderBuilder<'a> {
         let mut scan_input = ScanInput::new(self.sst_layer, ProjectionMapper::all(&self.metadata)?)
             .with_files(self.inputs.to_vec())
             .with_append_mode(self.append_mode)
-            .with_cache(self.cache)
+            .with_cache(None)
             .with_filter_deleted(self.filter_deleted)
             // We ignore file not found error during compaction.
             .with_ignore_file_not_found(true)

src/mito2/src/compaction/compactor.rs

@@ -296,7 +296,6 @@ impl Compactor for DefaultCompactor {
                 let reader = CompactionSstReaderBuilder {
                     metadata: region_metadata.clone(),
                     sst_layer: sst_layer.clone(),
-                    cache: Some(cache_manager.clone()),
                     inputs: &output.inputs,
                     append_mode,
                     filter_deleted: output.filter_deleted,

src/mito2/src/read/last_row.rs

@@ -108,10 +108,10 @@ impl RowGroupLastRowCachedReader {
                 // Schema matches, use cache batches.
                 Self::new_hit(value)
             } else {
-                Self::new_miss(key, row_group_reader, cache_manager)
+                Self::new_miss(key, row_group_reader, Some(cache_manager))
             }
         } else {
-            Self::new_miss(key, row_group_reader, cache_manager)
+            Self::new_miss(key, row_group_reader, Some(cache_manager))
         }
     }
 
@@ -125,7 +125,7 @@ impl RowGroupLastRowCachedReader {
     fn new_miss(
         key: SelectorResultKey,
         row_group_reader: RowGroupReader,
-        cache_manager: CacheManagerRef,
+        cache_manager: Option<CacheManagerRef>,
     ) -> Self {
         selector_result_cache_miss();
         Self::Miss(RowGroupLastRowReader::new(
@@ -170,13 +170,13 @@ pub(crate) struct RowGroupLastRowReader {
     reader: RowGroupReader,
     selector: LastRowSelector,
     yielded_batches: Vec<Batch>,
-    cache_manager: CacheManagerRef,
+    cache_manager: Option<CacheManagerRef>,
     /// Index buffer to take a new batch from the last row.
     take_index: UInt32Vector,
 }
 
 impl RowGroupLastRowReader {
-    fn new(key: SelectorResultKey, reader: RowGroupReader, cache_manager: CacheManagerRef) -> Self {
+    fn new(key: SelectorResultKey, reader: RowGroupReader, cache_manager: Option<CacheManagerRef>) -> Self {
         Self {
             key,
             reader,
@@ -216,6 +216,7 @@ impl RowGroupLastRowReader {
             // we always expect that row groups yields batches.
             return;
         }
+        let Some(cache) = &self.cache_manager else { return; };
         let value = Arc::new(SelectorResultValue {
             result: std::mem::take(&mut self.yielded_batches),
             projection: self
@@ -225,7 +226,7 @@ impl RowGroupLastRowReader {
                 .projection_indices()
                 .to_vec(),
         });
-        self.cache_manager.put_selector_result(self.key, value);
+        cache.put_selector_result(self.key, value);
     }
 }