rc_core/
select.rs

1//! S3 Select domain types (no AWS SDK types).
2
3use thiserror::Error;
4
5/// Object payload format for S3 Select input.
6#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
7pub enum SelectInputFormat {
8    #[default]
9    Csv,
10    Json,
11    Parquet,
12}
13
14/// Result row format for S3 Select output.
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
16pub enum SelectOutputFormat {
17    #[default]
18    Csv,
19    Json,
20}
21
22/// Compression applied to the **stored object** (input decompression).
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
24pub enum SelectCompression {
25    #[default]
26    None,
27    Gzip,
28    Bzip2,
29}
30
31/// CSV header handling for S3 Select input.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
33pub enum SelectCsvFileHeaderInfo {
34    #[default]
35    None,
36    Ignore,
37    Use,
38}
39
40/// JSON input shape for S3 Select.
41#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
42pub enum SelectJsonInputType {
43    #[default]
44    Lines,
45    Document,
46}
47
48/// CSV output quote behavior.
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
50pub enum SelectQuoteFields {
51    Always,
52    #[default]
53    AsNeeded,
54}
55
56/// Supported CSV input serialization options.
57#[derive(Debug, Clone, PartialEq, Eq, Default)]
58pub struct SelectCsvInputOptions {
59    pub file_header_info: SelectCsvFileHeaderInfo,
60    pub field_delimiter: Option<String>,
61    pub record_delimiter: Option<String>,
62    pub quote_character: Option<String>,
63    pub quote_escape_character: Option<String>,
64    pub comments: Option<String>,
65}
66
67/// Supported CSV output serialization options.
68#[derive(Debug, Clone, PartialEq, Eq, Default)]
69pub struct SelectCsvOutputOptions {
70    pub field_delimiter: Option<String>,
71    pub record_delimiter: Option<String>,
72    pub quote_character: Option<String>,
73    pub quote_escape_character: Option<String>,
74    pub quote_fields: SelectQuoteFields,
75}
76
77/// Supported JSON input serialization options.
78#[derive(Debug, Clone, PartialEq, Eq, Default)]
79pub struct SelectJsonInputOptions {
80    pub input_type: SelectJsonInputType,
81}
82
83/// Supported JSON output serialization options.
84#[derive(Debug, Clone, PartialEq, Eq, Default)]
85pub struct SelectJsonOutputOptions {
86    pub record_delimiter: Option<String>,
87}
88
89/// ScanRange request body parameters.
90#[derive(Debug, Clone, PartialEq, Eq, Default)]
91pub struct SelectScanRangeOptions {
92    pub start: Option<i64>,
93    pub end: Option<i64>,
94}
95
96/// SSE-C parameters for encrypted objects.
97#[derive(Debug, Clone, PartialEq, Eq, Default)]
98pub struct SelectSseCustomerOptions {
99    pub algorithm: Option<String>,
100    pub key: Option<String>,
101    pub key_md5: Option<String>,
102}
103
104/// Options for running an S3 Select query on one object.
105#[derive(Debug, Clone, PartialEq, Eq)]
106pub struct SelectOptions {
107    /// SQL expression (S3 Select / `s3object`).
108    pub expression: String,
109    pub input_format: SelectInputFormat,
110    pub output_format: SelectOutputFormat,
111    pub compression: SelectCompression,
112    pub csv_input: SelectCsvInputOptions,
113    pub csv_output: SelectCsvOutputOptions,
114    pub json_input: SelectJsonInputOptions,
115    pub json_output: SelectJsonOutputOptions,
116    pub scan_range: SelectScanRangeOptions,
117    pub sse_customer: SelectSseCustomerOptions,
118}
119
120/// Invalid combinations or values in an S3 Select request.
121#[derive(Debug, Clone, PartialEq, Eq, Error)]
122pub enum SelectOptionsError {
123    #[error("{field} must be exactly one byte")]
124    InvalidSingleByte { field: &'static str },
125    #[error("CSV input record delimiter must be one or two bytes")]
126    InvalidCsvInputRecordDelimiter,
127    #[error("CSV output record delimiter must be exactly one byte or CRLF")]
128    InvalidCsvOutputRecordDelimiter,
129    #[error("Parquet input does not support whole-object GZIP or BZIP2 compression")]
130    CompressedParquetInput,
131    #[error("ScanRange is not supported for JSON document input")]
132    JsonDocumentScanRange,
133    #[error("ScanRange is not supported for compressed input")]
134    CompressedInputScanRange,
135    #[error("ScanRange start and end must be non-negative")]
136    NegativeScanRange,
137    #[error("ScanRange start must not be greater than end")]
138    ReversedScanRange,
139}
140
141impl SelectOptions {
142    /// Validate format-specific options before a request reaches an object store.
143    pub fn validate(&self) -> std::result::Result<(), SelectOptionsError> {
144        if matches!(self.input_format, SelectInputFormat::Csv) {
145            validate_single_byte(
146                "CSV input field delimiter",
147                self.csv_input.field_delimiter.as_deref(),
148            )?;
149            validate_input_record_delimiter(self.csv_input.record_delimiter.as_deref())?;
150            validate_single_byte(
151                "CSV input quote character",
152                self.csv_input.quote_character.as_deref(),
153            )?;
154            validate_single_byte(
155                "CSV input quote escape character",
156                self.csv_input.quote_escape_character.as_deref(),
157            )?;
158            validate_single_byte(
159                "CSV input comment character",
160                self.csv_input.comments.as_deref(),
161            )?;
162        }
163
164        if matches!(self.output_format, SelectOutputFormat::Csv) {
165            validate_single_byte(
166                "CSV output field delimiter",
167                self.csv_output.field_delimiter.as_deref(),
168            )?;
169            validate_output_record_delimiter(self.csv_output.record_delimiter.as_deref())?;
170            validate_single_byte(
171                "CSV output quote character",
172                self.csv_output.quote_character.as_deref(),
173            )?;
174            validate_single_byte(
175                "CSV output quote escape character",
176                self.csv_output.quote_escape_character.as_deref(),
177            )?;
178        }
179
180        if matches!(self.input_format, SelectInputFormat::Parquet)
181            && !matches!(self.compression, SelectCompression::None)
182        {
183            return Err(SelectOptionsError::CompressedParquetInput);
184        }
185
186        self.validate_scan_range()
187    }
188
189    fn validate_scan_range(&self) -> std::result::Result<(), SelectOptionsError> {
190        let scan_range = &self.scan_range;
191        if scan_range.start.is_none() && scan_range.end.is_none() {
192            return Ok(());
193        }
194        if matches!(self.input_format, SelectInputFormat::Json)
195            && matches!(self.json_input.input_type, SelectJsonInputType::Document)
196        {
197            return Err(SelectOptionsError::JsonDocumentScanRange);
198        }
199        let is_noop = scan_range.start == Some(0) && scan_range.end.is_none();
200        if !matches!(self.compression, SelectCompression::None) && !is_noop {
201            return Err(SelectOptionsError::CompressedInputScanRange);
202        }
203        if scan_range.start.is_some_and(|start| start < 0)
204            || scan_range.end.is_some_and(|end| end < 0)
205        {
206            return Err(SelectOptionsError::NegativeScanRange);
207        }
208        if let (Some(start), Some(end)) = (scan_range.start, scan_range.end)
209            && start > end
210        {
211            return Err(SelectOptionsError::ReversedScanRange);
212        }
213        Ok(())
214    }
215}
216
217fn validate_single_byte(
218    field: &'static str,
219    value: Option<&str>,
220) -> std::result::Result<(), SelectOptionsError> {
221    if value.is_some_and(|value| value.len() != 1) {
222        return Err(SelectOptionsError::InvalidSingleByte { field });
223    }
224    Ok(())
225}
226
227fn validate_input_record_delimiter(
228    value: Option<&str>,
229) -> std::result::Result<(), SelectOptionsError> {
230    if value.is_some_and(|value| !(1..=2).contains(&value.len())) {
231        return Err(SelectOptionsError::InvalidCsvInputRecordDelimiter);
232    }
233    Ok(())
234}
235
236fn validate_output_record_delimiter(
237    value: Option<&str>,
238) -> std::result::Result<(), SelectOptionsError> {
239    if value.is_some_and(|value| value.len() != 1 && value != "\r\n") {
240        return Err(SelectOptionsError::InvalidCsvOutputRecordDelimiter);
241    }
242    Ok(())
243}
244
245impl Default for SelectOptions {
246    fn default() -> Self {
247        Self {
248            expression: String::new(),
249            input_format: SelectInputFormat::Csv,
250            output_format: SelectOutputFormat::Csv,
251            compression: SelectCompression::None,
252            csv_input: SelectCsvInputOptions::default(),
253            csv_output: SelectCsvOutputOptions::default(),
254            json_input: SelectJsonInputOptions::default(),
255            json_output: SelectJsonOutputOptions::default(),
256            scan_range: SelectScanRangeOptions::default(),
257            sse_customer: SelectSseCustomerOptions::default(),
258        }
259    }
260}
261
262#[cfg(test)]
263mod tests {
264    use super::*;
265
266    #[test]
267    fn validation_allows_parquet_scan_range() {
268        let options = SelectOptions {
269            input_format: SelectInputFormat::Parquet,
270            scan_range: SelectScanRangeOptions {
271                start: Some(1024),
272                end: Some(2047),
273            },
274            ..SelectOptions::default()
275        };
276
277        options
278            .validate()
279            .expect("Parquet scan range should be supported");
280    }
281
282    #[test]
283    fn validation_rejects_non_noop_scan_range_for_compressed_input() {
284        let options = SelectOptions {
285            compression: SelectCompression::Gzip,
286            scan_range: SelectScanRangeOptions {
287                start: Some(1),
288                end: None,
289            },
290            ..SelectOptions::default()
291        };
292
293        assert_eq!(
294            options.validate(),
295            Err(SelectOptionsError::CompressedInputScanRange)
296        );
297    }
298
299    #[test]
300    fn validation_allows_two_byte_csv_input_record_delimiter() {
301        let options = SelectOptions {
302            csv_input: SelectCsvInputOptions {
303                record_delimiter: Some("\r\n".to_string()),
304                ..SelectCsvInputOptions::default()
305            },
306            ..SelectOptions::default()
307        };
308
309        options
310            .validate()
311            .expect("two-byte CSV input record delimiter should be supported");
312    }
313
314    #[test]
315    fn validation_rejects_empty_csv_input_record_delimiter() {
316        let options = SelectOptions {
317            csv_input: SelectCsvInputOptions {
318                record_delimiter: Some(String::new()),
319                ..SelectCsvInputOptions::default()
320            },
321            ..SelectOptions::default()
322        };
323
324        assert_eq!(
325            options.validate(),
326            Err(SelectOptionsError::InvalidCsvInputRecordDelimiter)
327        );
328    }
329
330    #[test]
331    fn validation_ignores_csv_options_for_non_csv_formats() {
332        let options = SelectOptions {
333            input_format: SelectInputFormat::Json,
334            output_format: SelectOutputFormat::Json,
335            csv_input: SelectCsvInputOptions {
336                record_delimiter: Some(String::new()),
337                ..SelectCsvInputOptions::default()
338            },
339            csv_output: SelectCsvOutputOptions {
340                field_delimiter: Some("||".to_string()),
341                ..SelectCsvOutputOptions::default()
342            },
343            ..SelectOptions::default()
344        };
345
346        options
347            .validate()
348            .expect("inactive CSV options should not affect JSON requests");
349    }
350}
rc_core/select.rs

rc_core/
select.rs
