store_api/storage/
requests.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashMap;
16use std::fmt::{Display, Formatter};
17
18use common_recordbatch::OrderOption;
19use datafusion_expr::expr::Expr;
20use datatypes::types::json_type::JsonNativeType;
21use itertools::Itertools;
22use strum::Display;
23
24use crate::storage::SequenceNumber;
25
26/// A hint on how to select rows from a time-series.
27#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Display)]
28pub enum TimeSeriesRowSelector {
29    /// Only keep the last row of each time-series.
30    #[strum(to_string = "LastRow {{ after_merge: {after_merge} }}")]
31    LastRow {
32        /// Whether selection runs after cross-source merge and deduplication.
33        after_merge: bool,
34    },
35}
36
37/// A hint on how to distribute time-series data on the scan output.
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Display)]
39pub enum TimeSeriesDistribution {
40    /// Data are distributed by time window first. The scanner will
41    /// return all data within one time window before moving to the next one.
42    TimeWindowed,
43    /// Data are organized by time-series first. The scanner will return
44    /// all data for one time-series before moving to the next one.
45    PerSeries,
46}
47
48#[derive(Default, Clone, Debug, PartialEq)]
49pub struct ScanRequest {
50    /// Optional projection information for the scan. `None` reads all root
51    /// columns.
52    pub projection: Option<Vec<usize>>,
53    /// Filters pushed down
54    pub filters: Vec<Expr>,
55    /// Expected output ordering. This is only a hint and isn't guaranteed.
56    pub output_ordering: Option<Vec<OrderOption>>,
57    /// limit can be used to reduce the amount scanned
58    /// from the datasource as a performance optimization.
59    /// If set, it contains the amount of rows needed by the caller,
60    /// The data source should return *at least* this number of rows if available.
61    pub limit: Option<usize>,
62    /// Optional hint to select rows from time-series.
63    pub series_row_selector: Option<TimeSeriesRowSelector>,
64    /// Optional constraint on the sequence number of the rows to read.
65    /// If set, only rows with a sequence number **lesser or equal** to this value
66    /// will be returned.
67    /// This is the effective memtable upper bound used by the scan, whether provided
68    /// explicitly or bound on scan open.
69    pub memtable_max_sequence: Option<SequenceNumber>,
70    /// Optional constraint on the minimal sequence number in the memtable.
71    /// If set, only the memtables that contain sequences **greater than** this value will be scanned
72    pub memtable_min_sequence: Option<SequenceNumber>,
73    /// Optional constraint on the minimal sequence number in the SST files.
74    /// If set, only the SST files that contain sequences greater than this value will be scanned.
75    pub sst_min_sequence: Option<SequenceNumber>,
76    /// Whether to skip all SST files.
77    /// This is stronger than `sst_min_sequence` and also skips SST files without sequence metadata.
78    pub skip_sst_files: bool,
79    /// Whether to bind the effective snapshot upper bound when opening the scan.
80    pub snapshot_on_scan: bool,
81    /// Explicit intent to read an exact row-level sequence delta `(min, max]`
82    /// across memtables and all SST files (Flow's `sequence_range` incremental
83    /// mode). The engine performs exact row-level filtering only when the region
84    /// preserves per-row sequences and every participating SST file is trusted;
85    /// otherwise it returns a structured stale/unsupported error so the caller
86    /// falls back instead of silently approximating.
87    ///
88    /// Historical `memtable_only` reads must never set this flag.
89    pub exact_sequence_range: bool,
90    /// Optional hint for the distribution of time-series data.
91    pub distribution: Option<TimeSeriesDistribution>,
92    /// Optional hint from query-driven JSON type concretization.
93    pub json_type_hint: HashMap<String, JsonNativeType>,
94    /// Whether Mito should keep string primary-key columns dictionary encoded in its output.
95    pub preserve_pk_dictionary_encoding: bool,
96}
97
98impl Display for ScanRequest {
99    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
100        enum Delimiter {
101            None,
102            Init,
103        }
104
105        impl Delimiter {
106            fn as_str(&mut self) -> &str {
107                match self {
108                    Delimiter::None => {
109                        *self = Delimiter::Init;
110                        ""
111                    }
112                    Delimiter::Init => ", ",
113                }
114            }
115        }
116
117        let mut delimiter = Delimiter::None;
118
119        write!(f, "ScanRequest {{ ")?;
120        if let Some(projection) = &self.projection {
121            write!(f, "{}projection: {:?}", delimiter.as_str(), projection)?;
122        }
123        if !self.filters.is_empty() {
124            write!(
125                f,
126                "{}filters: [{}]",
127                delimiter.as_str(),
128                self.filters
129                    .iter()
130                    .map(|f| f.to_string())
131                    .collect::<Vec<_>>()
132                    .join(", ")
133            )?;
134        }
135        if let Some(output_ordering) = &self.output_ordering {
136            write!(
137                f,
138                "{}output_ordering: {:?}",
139                delimiter.as_str(),
140                output_ordering
141            )?;
142        }
143        if let Some(limit) = &self.limit {
144            write!(f, "{}limit: {}", delimiter.as_str(), limit)?;
145        }
146        if let Some(series_row_selector) = &self.series_row_selector {
147            write!(
148                f,
149                "{}series_row_selector: {}",
150                delimiter.as_str(),
151                series_row_selector
152            )?;
153        }
154        if let Some(sequence) = &self.memtable_max_sequence {
155            write!(f, "{}sequence: {}", delimiter.as_str(), sequence)?;
156        }
157        if let Some(sst_min_sequence) = &self.sst_min_sequence {
158            write!(
159                f,
160                "{}sst_min_sequence: {}",
161                delimiter.as_str(),
162                sst_min_sequence
163            )?;
164        }
165        if self.skip_sst_files {
166            write!(
167                f,
168                "{}skip_sst_files: {}",
169                delimiter.as_str(),
170                self.skip_sst_files
171            )?;
172        }
173        if self.snapshot_on_scan {
174            write!(
175                f,
176                "{}snapshot_on_scan: {}",
177                delimiter.as_str(),
178                self.snapshot_on_scan
179            )?;
180        }
181        if self.exact_sequence_range {
182            write!(f, "{}exact_sequence_range: true", delimiter.as_str())?;
183        }
184        if self.preserve_pk_dictionary_encoding {
185            write!(
186                f,
187                "{}preserve_pk_dictionary_encoding: true",
188                delimiter.as_str()
189            )?;
190        }
191        if let Some(distribution) = &self.distribution {
192            write!(f, "{}distribution: {}", delimiter.as_str(), distribution)?;
193        }
194        if !self.json_type_hint.is_empty() {
195            write!(
196                f,
197                "{}json_type_hint: {}",
198                delimiter.as_str(),
199                self.json_type_hint
200                    .iter()
201                    .map(|(column, json_type)| format!("({column}: {json_type})"))
202                    .join(", ")
203            )?;
204        }
205        write!(f, " }}")
206    }
207}
208
209#[cfg(test)]
210mod tests {
211    use datafusion_expr::{Operator, binary_expr, col, lit};
212
213    use super::*;
214
215    #[test]
216    fn test_display_scan_request() {
217        let request = ScanRequest {
218            ..Default::default()
219        };
220        assert_eq!(request.to_string(), "ScanRequest {  }");
221
222        let projection = Some(vec![1, 2]);
223        let request = ScanRequest {
224            projection,
225            filters: vec![
226                binary_expr(col("i"), Operator::Gt, lit(1)),
227                binary_expr(col("s"), Operator::Eq, lit("x")),
228            ],
229            limit: Some(10),
230            ..Default::default()
231        };
232        assert_eq!(
233            request.to_string(),
234            r#"ScanRequest { projection: [1, 2], filters: [i > Int32(1), s = Utf8("x")], limit: 10 }"#
235        );
236
237        let request = ScanRequest {
238            filters: vec![
239                binary_expr(col("i"), Operator::Gt, lit(1)),
240                binary_expr(col("s"), Operator::Eq, lit("x")),
241            ],
242            limit: Some(10),
243            ..Default::default()
244        };
245        assert_eq!(
246            request.to_string(),
247            r#"ScanRequest { filters: [i > Int32(1), s = Utf8("x")], limit: 10 }"#
248        );
249
250        let projection = Some(vec![1, 2]);
251        let request = ScanRequest {
252            projection,
253            limit: Some(10),
254            ..Default::default()
255        };
256        assert_eq!(
257            request.to_string(),
258            "ScanRequest { projection: [1, 2], limit: 10 }"
259        );
260
261        let request = ScanRequest {
262            series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }),
263            snapshot_on_scan: true,
264            exact_sequence_range: true,
265            ..Default::default()
266        };
267        assert_eq!(
268            request.to_string(),
269            "ScanRequest { series_row_selector: LastRow { after_merge: true }, snapshot_on_scan: true, exact_sequence_range: true }"
270        );
271
272        assert_eq!(
273            TimeSeriesRowSelector::LastRow { after_merge: false }.to_string(),
274            "LastRow { after_merge: false }"
275        );
276
277        let request = ScanRequest {
278            skip_sst_files: true,
279            ..Default::default()
280        };
281        assert_eq!(request.to_string(), "ScanRequest { skip_sst_files: true }");
282    }
283}
store_api/storage/requests.rs

store_api/storage/
requests.rs
