feat: use series scan when distribution is PerSeries

src/mito2/src/read/scan_region.rs

@@ -256,7 +256,9 @@ impl ScanRegion {
 
     /// Returns a [Scanner] to scan the region.
     pub(crate) fn scanner(self) -> Result<Scanner> {
-        if self.use_unordered_scan() {
+        if self.use_series_scan() {
+            self.series_scan().map(Scanner::Series)
+        } else if self.use_unordered_scan() {
             // If table is append only and there is no series row selector, we use unordered scan in query.
             // We still use seq scan in compaction.
             self.unordered_scan().map(Scanner::Unordered)
@@ -268,7 +270,9 @@ impl ScanRegion {
     /// Returns a [RegionScanner] to scan the region.
     #[tracing::instrument(level = tracing::Level::DEBUG, skip_all)]
     pub(crate) fn region_scanner(self) -> Result<RegionScannerRef> {
-        if self.use_unordered_scan() {
+        if self.use_series_scan() {
+            self.series_scan().map(|scanner| Box::new(scanner) as _)
+        } else if self.use_unordered_scan() {
             self.unordered_scan().map(|scanner| Box::new(scanner) as _)
         } else {
             self.seq_scan().map(|scanner| Box::new(scanner) as _)
@@ -287,6 +291,12 @@ impl ScanRegion {
         Ok(UnorderedScan::new(input))
     }
 
+    /// Scans by series.
+    pub(crate) fn series_scan(self) -> Result<SeriesScan> {
+        let input = self.scan_input(true)?;
+        Ok(SeriesScan::new(input))
+    }
+
     #[cfg(test)]
     pub(crate) fn scan_without_filter_deleted(self) -> Result<SeqScan> {
         let input = self.scan_input(false)?;
@@ -307,6 +317,11 @@ impl ScanRegion {
                 || self.request.distribution == Some(TimeSeriesDistribution::TimeWindowed))
     }
 
+    /// Returns true if the region can use series scan for current request.
+    fn use_series_scan(&self) -> bool {
+        self.request.distribution == Some(TimeSeriesDistribution::TimeWindowed)
+    }
+
     /// Creates a scan input.
     fn scan_input(mut self, filter_deleted: bool) -> Result<ScanInput> {
         let time_range = self.build_time_range_predicate();