diff --git a/crates/cli/src/commands/mod.rs b/crates/cli/src/commands/mod.rs index cd7ab3c..c6355f0 100644 --- a/crates/cli/src/commands/mod.rs +++ b/crates/cli/src/commands/mod.rs @@ -879,6 +879,28 @@ mod tests { } } + #[test] + fn cli_accepts_sql_csv_input_record_delimiter() { + let cli = Cli::try_parse_from([ + "rc", + "sql", + "local/reports/data.csv", + "--query", + "SELECT * FROM S3Object", + "--csv-input-record-delimiter", + "\r\n", + ]) + .expect("parse CSV input record delimiter"); + + match cli.command { + Commands::Sql(arg) => { + assert!(matches!(arg.input_format, sql::InputFormatArg::Csv)); + assert_eq!(arg.csv_input_record_delimiter.as_deref(), Some("\r\n")); + } + other => panic!("expected sql command, got {:?}", other), + } + } + #[test] fn cli_accepts_sql_defaults() { let cli = Cli::try_parse_from([ diff --git a/crates/cli/src/commands/sql.rs b/crates/cli/src/commands/sql.rs index 797793b..8b3a6cb 100644 --- a/crates/cli/src/commands/sql.rs +++ b/crates/cli/src/commands/sql.rs @@ -42,6 +42,10 @@ pub struct SqlArgs { #[arg(long)] pub csv_input_field_delimiter: Option, + /// CSV input record delimiter (one or two bytes) + #[arg(long)] + pub csv_input_record_delimiter: Option, + /// CSV input quote character #[arg(long)] pub csv_input_quote: Option, @@ -217,8 +221,9 @@ pub async fn execute(args: SqlArgs, output_config: OutputConfig) -> ExitCode { } }; - if let Err(message) = validate_select_args(&args) { - formatter.error(&message); + let options = select_options_from_args(args); + if let Err(error) = options.validate() { + formatter.error(&error.to_string()); return ExitCode::UsageError; } @@ -246,7 +251,22 @@ pub async fn execute(args: SqlArgs, output_config: OutputConfig) -> ExitCode { } }; - let options = SelectOptions { + let mut stdout = tokio::io::stdout(); + + match client + .select_object_content(&remote, &options, &mut stdout) + .await + { + Ok(()) => ExitCode::Success, + Err(e) => { + formatter.error(&e.to_string()); + exit_code_from_error(&e) + } + } +} + +fn select_options_from_args(args: SqlArgs) -> SelectOptions { + SelectOptions { expression: args.query, input_format: args.input_format.into(), output_format: args.output_format.into(), @@ -254,6 +274,7 @@ pub async fn execute(args: SqlArgs, output_config: OutputConfig) -> ExitCode { csv_input: SelectCsvInputOptions { file_header_info: args.csv_file_header_info.into(), field_delimiter: args.csv_input_field_delimiter, + record_delimiter: args.csv_input_record_delimiter, quote_character: args.csv_input_quote, quote_escape_character: args.csv_input_quote_escape, comments: args.csv_input_comment, @@ -280,89 +301,7 @@ pub async fn execute(args: SqlArgs, output_config: OutputConfig) -> ExitCode { key: args.sse_customer_key, key_md5: args.sse_customer_key_md5, }, - }; - - let mut stdout = tokio::io::stdout(); - - match client - .select_object_content(&remote, &options, &mut stdout) - .await - { - Ok(()) => ExitCode::Success, - Err(e) => { - formatter.error(&e.to_string()); - exit_code_from_error(&e) - } - } -} - -fn validate_select_args(args: &SqlArgs) -> std::result::Result<(), String> { - validate_single_byte( - "--csv-input-field-delimiter", - args.csv_input_field_delimiter.as_deref(), - )?; - validate_single_byte("--csv-input-quote", args.csv_input_quote.as_deref())?; - validate_single_byte( - "--csv-input-quote-escape", - args.csv_input_quote_escape.as_deref(), - )?; - validate_single_byte("--csv-input-comment", args.csv_input_comment.as_deref())?; - validate_single_byte( - "--csv-output-field-delimiter", - args.csv_output_field_delimiter.as_deref(), - )?; - validate_record_delimiter( - "--csv-output-record-delimiter", - args.csv_output_record_delimiter.as_deref(), - )?; - validate_single_byte("--csv-output-quote", args.csv_output_quote.as_deref())?; - validate_single_byte( - "--csv-output-quote-escape", - args.csv_output_quote_escape.as_deref(), - )?; - validate_scan_range_args(args) -} - -fn validate_single_byte(name: &str, value: Option<&str>) -> std::result::Result<(), String> { - if let Some(value) = value - && value.len() != 1 - { - return Err(format!("{name} must be exactly one byte")); - } - Ok(()) -} - -fn validate_record_delimiter(name: &str, value: Option<&str>) -> std::result::Result<(), String> { - if let Some(value) = value - && value.len() != 1 - && value != "\r\n" - { - return Err(format!("{name} must be exactly one byte or CRLF")); - } - Ok(()) -} - -fn validate_scan_range_args(args: &SqlArgs) -> std::result::Result<(), String> { - if args.scan_start.is_none() && args.scan_end.is_none() { - return Ok(()); - } - if matches!(args.input_format, InputFormatArg::Parquet) { - return Err("ScanRange is not supported for Parquet input".to_string()); - } - if matches!(args.input_format, InputFormatArg::Json) - && matches!(args.json_type, JsonTypeArg::Document) - { - return Err("ScanRange is not supported for JSON document input".to_string()); } - if args.scan_start.is_some_and(|start| start < 0) || args.scan_end.is_some_and(|end| end < 0) { - return Err("ScanRange start and end must be non-negative".to_string()); - } - if let (Some(start), Some(end)) = (args.scan_start, args.scan_end) - && start > end - { - return Err("ScanRange start must not be greater than end".to_string()); - } - Ok(()) } fn exit_code_from_error(error: &rc_core::Error) -> ExitCode { @@ -384,6 +323,7 @@ mod tests { compression: CompressionArg::None, csv_file_header_info: CsvFileHeaderInfoArg::None, csv_input_field_delimiter: None, + csv_input_record_delimiter: None, csv_input_quote: None, csv_input_quote_escape: None, csv_input_comment: None, @@ -443,6 +383,37 @@ mod tests { assert_eq!(code, ExitCode::UsageError); } + #[test] + fn sql_allows_scan_range_for_parquet() { + let mut args = base_args("a/b/object.parquet", "SELECT * FROM S3Object"); + args.input_format = InputFormatArg::Parquet; + args.scan_start = Some(1024); + args.scan_end = Some(2047); + + assert!(select_options_from_args(args).validate().is_ok()); + } + + #[test] + fn sql_rejects_compressed_scan_range() { + let mut args = base_args("a/b/object.csv.gz", "SELECT * FROM S3Object"); + args.compression = CompressionArg::Gzip; + args.scan_start = Some(1); + + let error = select_options_from_args(args) + .validate() + .expect_err("compressed input should reject a non-noop scan range"); + assert!(error.to_string().contains("compressed input")); + } + + #[test] + fn sql_allows_noop_compressed_scan_range() { + let mut args = base_args("a/b/object.csv.bz2", "SELECT * FROM S3Object"); + args.compression = CompressionArg::Bzip2; + args.scan_start = Some(0); + + assert!(select_options_from_args(args).validate().is_ok()); + } + #[test] fn sql_exit_code_from_backend_errors() { let cases = [ diff --git a/crates/cli/src/commands/table/mod.rs b/crates/cli/src/commands/table/mod.rs index 5301818..1523c50 100644 --- a/crates/cli/src/commands/table/mod.rs +++ b/crates/cli/src/commands/table/mod.rs @@ -406,10 +406,10 @@ fn properties(values: Vec) -> Result> { Ok(result) } fn require_string(body: &Value, field: &str) -> Result<()> { - if !body + if body .get(field) .and_then(Value::as_str) - .is_some_and(|s| !s.trim().is_empty()) + .is_none_or(|s| s.trim().is_empty()) { return Err(Error::Config(format!("Request requires nonempty {field}"))); } @@ -577,10 +577,10 @@ fn prepare_table(command: TableCommands) -> Result { { return Err(Error::Config("Standard updates use Iceberg requirements; version/location guards require new-metadata-location".into())); } - if !body + if body .get("requirements") .and_then(Value::as_array) - .is_some_and(|v| !v.is_empty()) + .is_none_or(Vec::is_empty) { return Err(Error::Config( "Standard commit requires explicit Iceberg requirements".into(), diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index e70e225..b215a41 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -70,8 +70,8 @@ pub use retry::{RetryBuilder, is_retryable_error, retry_with_backoff}; pub use select::{ SelectCompression, SelectCsvFileHeaderInfo, SelectCsvInputOptions, SelectCsvOutputOptions, SelectInputFormat, SelectJsonInputOptions, SelectJsonInputType, SelectJsonOutputOptions, - SelectOptions, SelectOutputFormat, SelectQuoteFields, SelectScanRangeOptions, - SelectSseCustomerOptions, + SelectOptions, SelectOptionsError, SelectOutputFormat, SelectQuoteFields, + SelectScanRangeOptions, SelectSseCustomerOptions, }; pub use traits::{ AbortMultipartUploadRequest, BucketNotification, Capabilities, CopyObjectOptions, diff --git a/crates/core/src/select.rs b/crates/core/src/select.rs index c384058..c9dab6f 100644 --- a/crates/core/src/select.rs +++ b/crates/core/src/select.rs @@ -1,5 +1,7 @@ //! S3 Select domain types (no AWS SDK types). +use thiserror::Error; + /// Object payload format for S3 Select input. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum SelectInputFormat { @@ -56,6 +58,7 @@ pub enum SelectQuoteFields { pub struct SelectCsvInputOptions { pub file_header_info: SelectCsvFileHeaderInfo, pub field_delimiter: Option, + pub record_delimiter: Option, pub quote_character: Option, pub quote_escape_character: Option, pub comments: Option, @@ -114,6 +117,131 @@ pub struct SelectOptions { pub sse_customer: SelectSseCustomerOptions, } +/// Invalid combinations or values in an S3 Select request. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum SelectOptionsError { + #[error("{field} must be exactly one byte")] + InvalidSingleByte { field: &'static str }, + #[error("CSV input record delimiter must be one or two bytes")] + InvalidCsvInputRecordDelimiter, + #[error("CSV output record delimiter must be exactly one byte or CRLF")] + InvalidCsvOutputRecordDelimiter, + #[error("Parquet input does not support whole-object GZIP or BZIP2 compression")] + CompressedParquetInput, + #[error("ScanRange is not supported for JSON document input")] + JsonDocumentScanRange, + #[error("ScanRange is not supported for compressed input")] + CompressedInputScanRange, + #[error("ScanRange start and end must be non-negative")] + NegativeScanRange, + #[error("ScanRange start must not be greater than end")] + ReversedScanRange, +} + +impl SelectOptions { + /// Validate format-specific options before a request reaches an object store. + pub fn validate(&self) -> std::result::Result<(), SelectOptionsError> { + if matches!(self.input_format, SelectInputFormat::Csv) { + validate_single_byte( + "CSV input field delimiter", + self.csv_input.field_delimiter.as_deref(), + )?; + validate_input_record_delimiter(self.csv_input.record_delimiter.as_deref())?; + validate_single_byte( + "CSV input quote character", + self.csv_input.quote_character.as_deref(), + )?; + validate_single_byte( + "CSV input quote escape character", + self.csv_input.quote_escape_character.as_deref(), + )?; + validate_single_byte( + "CSV input comment character", + self.csv_input.comments.as_deref(), + )?; + } + + if matches!(self.output_format, SelectOutputFormat::Csv) { + validate_single_byte( + "CSV output field delimiter", + self.csv_output.field_delimiter.as_deref(), + )?; + validate_output_record_delimiter(self.csv_output.record_delimiter.as_deref())?; + validate_single_byte( + "CSV output quote character", + self.csv_output.quote_character.as_deref(), + )?; + validate_single_byte( + "CSV output quote escape character", + self.csv_output.quote_escape_character.as_deref(), + )?; + } + + if matches!(self.input_format, SelectInputFormat::Parquet) + && !matches!(self.compression, SelectCompression::None) + { + return Err(SelectOptionsError::CompressedParquetInput); + } + + self.validate_scan_range() + } + + fn validate_scan_range(&self) -> std::result::Result<(), SelectOptionsError> { + let scan_range = &self.scan_range; + if scan_range.start.is_none() && scan_range.end.is_none() { + return Ok(()); + } + if matches!(self.input_format, SelectInputFormat::Json) + && matches!(self.json_input.input_type, SelectJsonInputType::Document) + { + return Err(SelectOptionsError::JsonDocumentScanRange); + } + let is_noop = scan_range.start == Some(0) && scan_range.end.is_none(); + if !matches!(self.compression, SelectCompression::None) && !is_noop { + return Err(SelectOptionsError::CompressedInputScanRange); + } + if scan_range.start.is_some_and(|start| start < 0) + || scan_range.end.is_some_and(|end| end < 0) + { + return Err(SelectOptionsError::NegativeScanRange); + } + if let (Some(start), Some(end)) = (scan_range.start, scan_range.end) + && start > end + { + return Err(SelectOptionsError::ReversedScanRange); + } + Ok(()) + } +} + +fn validate_single_byte( + field: &'static str, + value: Option<&str>, +) -> std::result::Result<(), SelectOptionsError> { + if value.is_some_and(|value| value.len() != 1) { + return Err(SelectOptionsError::InvalidSingleByte { field }); + } + Ok(()) +} + +fn validate_input_record_delimiter( + value: Option<&str>, +) -> std::result::Result<(), SelectOptionsError> { + if value.is_some_and(|value| !(1..=2).contains(&value.len())) { + return Err(SelectOptionsError::InvalidCsvInputRecordDelimiter); + } + Ok(()) +} + +fn validate_output_record_delimiter( + value: Option<&str>, +) -> std::result::Result<(), SelectOptionsError> { + if value.is_some_and(|value| value.len() != 1 && value != "\r\n") { + return Err(SelectOptionsError::InvalidCsvOutputRecordDelimiter); + } + Ok(()) +} + impl Default for SelectOptions { fn default() -> Self { Self { @@ -130,3 +258,93 @@ impl Default for SelectOptions { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn validation_allows_parquet_scan_range() { + let options = SelectOptions { + input_format: SelectInputFormat::Parquet, + scan_range: SelectScanRangeOptions { + start: Some(1024), + end: Some(2047), + }, + ..SelectOptions::default() + }; + + options + .validate() + .expect("Parquet scan range should be supported"); + } + + #[test] + fn validation_rejects_non_noop_scan_range_for_compressed_input() { + let options = SelectOptions { + compression: SelectCompression::Gzip, + scan_range: SelectScanRangeOptions { + start: Some(1), + end: None, + }, + ..SelectOptions::default() + }; + + assert_eq!( + options.validate(), + Err(SelectOptionsError::CompressedInputScanRange) + ); + } + + #[test] + fn validation_allows_two_byte_csv_input_record_delimiter() { + let options = SelectOptions { + csv_input: SelectCsvInputOptions { + record_delimiter: Some("\r\n".to_string()), + ..SelectCsvInputOptions::default() + }, + ..SelectOptions::default() + }; + + options + .validate() + .expect("two-byte CSV input record delimiter should be supported"); + } + + #[test] + fn validation_rejects_empty_csv_input_record_delimiter() { + let options = SelectOptions { + csv_input: SelectCsvInputOptions { + record_delimiter: Some(String::new()), + ..SelectCsvInputOptions::default() + }, + ..SelectOptions::default() + }; + + assert_eq!( + options.validate(), + Err(SelectOptionsError::InvalidCsvInputRecordDelimiter) + ); + } + + #[test] + fn validation_ignores_csv_options_for_non_csv_formats() { + let options = SelectOptions { + input_format: SelectInputFormat::Json, + output_format: SelectOutputFormat::Json, + csv_input: SelectCsvInputOptions { + record_delimiter: Some(String::new()), + ..SelectCsvInputOptions::default() + }, + csv_output: SelectCsvOutputOptions { + field_delimiter: Some("||".to_string()), + ..SelectCsvOutputOptions::default() + }, + ..SelectOptions::default() + }; + + options + .validate() + .expect("inactive CSV options should not affect JSON requests"); + } +} diff --git a/crates/s3/src/admin/catalog.rs b/crates/s3/src/admin/catalog.rs index c021a84..3202332 100644 --- a/crates/s3/src/admin/catalog.rs +++ b/crates/s3/src/admin/catalog.rs @@ -302,8 +302,7 @@ impl AdminClient { serde_json::from_slice(&bytes) .map_err(|_| Error::General("Invalid catalog JSON response".into()))? }; - if !value.is_object() - && !(request.operation == Op::MaintenanceConfigShow && value.is_null()) + if !(value.is_object() || request.operation == Op::MaintenanceConfigShow && value.is_null()) { return Err(Error::General("Catalog response must be an object".into())); } diff --git a/crates/s3/src/select.rs b/crates/s3/src/select.rs index c46dac6..f7b09f0 100644 --- a/crates/s3/src/select.rs +++ b/crates/s3/src/select.rs @@ -23,8 +23,11 @@ pub async fn select_object_content( options: &SelectOptions, writer: &mut (dyn AsyncWrite + Send + Unpin), ) -> Result<()> { - let input = build_input_serialization(options)?; - let output = build_output_serialization(options)?; + options + .validate() + .map_err(|error| Error::General(error.to_string()))?; + let input = build_input_serialization(options); + let output = build_output_serialization(options); // aws-sdk-s3 `SelectObjectContent` does not expose object `VersionId`; the current object is used. let mut request = client @@ -35,7 +38,7 @@ pub async fn select_object_content( .expression_type(ExpressionType::Sql) .input_serialization(input) .output_serialization(output); - if let Some(scan_range) = build_scan_range(options)? { + if let Some(scan_range) = build_scan_range(options) { request = request.scan_range(scan_range); } if let Some(algorithm) = options.sse_customer.algorithm.as_deref() { @@ -96,99 +99,32 @@ fn quote_fields(quote_fields: RcSelectQuoteFields) -> QuoteFields { } } -fn build_scan_range(options: &SelectOptions) -> Result> { +fn build_scan_range(options: &SelectOptions) -> Option { let scan_range = &options.scan_range; if scan_range.start.is_none() && scan_range.end.is_none() { - return Ok(None); - } - if matches!(options.input_format, SelectInputFormat::Parquet) { - return Err(Error::General( - "ScanRange is not supported for Parquet input.".to_string(), - )); - } - if matches!(options.input_format, SelectInputFormat::Json) - && matches!(options.json_input.input_type, SelectJsonInputType::Document) - { - return Err(Error::General( - "ScanRange is not supported for JSON document input.".to_string(), - )); - } - if scan_range.start.is_some_and(|start| start < 0) || scan_range.end.is_some_and(|end| end < 0) - { - return Err(Error::General( - "ScanRange start and end must be non-negative.".to_string(), - )); - } - if let (Some(start), Some(end)) = (scan_range.start, scan_range.end) - && start > end - { - return Err(Error::General( - "ScanRange start must not be greater than end.".to_string(), - )); - } - Ok(Some( + return None; + } + Some( ScanRange::builder() .set_start(scan_range.start) .set_end(scan_range.end) .build(), - )) -} - -fn validate_single_byte(name: &str, value: Option<&str>) -> Result<()> { - if let Some(value) = value - && value.len() != 1 - { - return Err(Error::General(format!("{name} must be exactly one byte."))); - } - Ok(()) + ) } -fn validate_record_delimiter(name: &str, value: Option<&str>) -> Result<()> { - if let Some(value) = value - && value.len() != 1 - && value != "\r\n" - { - return Err(Error::General(format!( - "{name} must be exactly one byte or CRLF." - ))); - } - Ok(()) -} - -fn build_input_serialization(options: &SelectOptions) -> Result { - if matches!(options.input_format, SelectInputFormat::Parquet) - && !matches!(options.compression, SelectCompression::None) - { - return Err(Error::General( - "Parquet input does not support whole-object GZIP or BZIP2 compression.".to_string(), - )); - } - +fn build_input_serialization(options: &SelectOptions) -> InputSerialization { let compression = compression_type(options.compression); let mut b = InputSerialization::builder().compression_type(compression); match options.input_format { SelectInputFormat::Csv => { - validate_single_byte( - "CSV input field delimiter", - options.csv_input.field_delimiter.as_deref(), - )?; - validate_single_byte( - "CSV input quote character", - options.csv_input.quote_character.as_deref(), - )?; - validate_single_byte( - "CSV input quote escape character", - options.csv_input.quote_escape_character.as_deref(), - )?; - validate_single_byte( - "CSV input comment character", - options.csv_input.comments.as_deref(), - )?; let mut csv = CsvInput::builder() .file_header_info(csv_file_header_info(options.csv_input.file_header_info)); if let Some(delimiter) = options.csv_input.field_delimiter.as_deref() { csv = csv.field_delimiter(delimiter); } + if let Some(delimiter) = options.csv_input.record_delimiter.as_deref() { + csv = csv.record_delimiter(delimiter); + } if let Some(quote) = options.csv_input.quote_character.as_deref() { csv = csv.quote_character(quote); } @@ -212,29 +148,13 @@ fn build_input_serialization(options: &SelectOptions) -> Result Result { +fn build_output_serialization(options: &SelectOptions) -> OutputSerialization { let mut b = OutputSerialization::builder(); match options.output_format { SelectOutputFormat::Csv => { - validate_single_byte( - "CSV output field delimiter", - options.csv_output.field_delimiter.as_deref(), - )?; - validate_record_delimiter( - "CSV output record delimiter", - options.csv_output.record_delimiter.as_deref(), - )?; - validate_single_byte( - "CSV output quote character", - options.csv_output.quote_character.as_deref(), - )?; - validate_single_byte( - "CSV output quote escape character", - options.csv_output.quote_escape_character.as_deref(), - )?; let mut csv = CsvOutput::builder().quote_fields(quote_fields(options.csv_output.quote_fields)); if let Some(delimiter) = options.csv_output.field_delimiter.as_deref() { @@ -261,7 +181,7 @@ fn build_output_serialization(options: &SelectOptions) -> Result( @@ -295,7 +215,9 @@ fn map_select_initial_error( match &err { SdkError::ServiceError(se) => { let code = resolve_http_service_error_code(se.err(), se.raw()); - classify_aws_code(code, &err.to_string()) + let fallback = err.to_string(); + let message = se.err().message().unwrap_or(&fallback); + classify_aws_code(code, message) } SdkError::TimeoutError(_) => Error::Network("Request timeout".to_string()), SdkError::DispatchFailure(e) => Error::Network(format!("Network dispatch error: {e:?}")), @@ -312,7 +234,9 @@ fn map_select_stream_error( match &err { SdkError::ServiceError(se) => { let code = resolve_event_stream_error_code(se.err(), se.raw()); - classify_aws_code(code, &err.to_string()) + let fallback = err.to_string(); + let message = se.err().message().unwrap_or(&fallback); + classify_aws_code(code, message) } SdkError::TimeoutError(_) => Error::Network("Request timeout".to_string()), SdkError::DispatchFailure(e) => Error::Network(format!("Network dispatch error: {e:?}")), @@ -324,6 +248,9 @@ fn map_select_stream_error( fn classify_aws_code(code: Option<&str>, text: &str) -> Error { let c = code.filter(|s| !s.is_empty()); + if text.contains("NotImplemented") && c != Some("NotImplemented") { + return Error::UnsupportedFeature("The backend does not support S3 Select.".to_string()); + } match c { Some("NoSuchKey") => Error::NotFound("Object not found".to_string()), Some("NoSuchBucket") => Error::NotFound("Bucket not found".to_string()), @@ -331,15 +258,29 @@ fn classify_aws_code(code: Option<&str>, text: &str) -> Error { Some("NotImplemented") => { Error::UnsupportedFeature("The backend does not support S3 Select.".to_string()) } - Some("InvalidArgument") => Error::General(format!("Invalid S3 Select request: {text}")), - Some(_) if text.contains("NotImplemented") => { - Error::UnsupportedFeature("The backend does not support S3 Select.".to_string()) - } - Some(_) => Error::General(text.to_string()), + Some("SlowDown" | "Busy") => Error::Network(service_error_detail(c, text)), + Some("InvalidArgument") => Error::General(format!( + "Invalid S3 Select request: {}", + service_error_detail(c, text) + )), + Some("UnsupportedScanRangeInput") => Error::General(service_error_detail(c, text)), + Some(_) => Error::General(service_error_detail(c, text)), None => classify_aws_code_missing_metadata(text), } } +fn service_error_detail(code: Option<&str>, text: &str) -> String { + let text = text.trim(); + match ( + code, + text.is_empty() || text.eq_ignore_ascii_case("service error"), + ) { + (Some(code), true) => code.to_string(), + (Some(code), false) => format!("{code}: {text}"), + (None, _) => text.to_string(), + } +} + /// When the SDK did not surface `x-amz-error-code` / metadata, use minimal substring checks. fn classify_aws_code_missing_metadata(text: &str) -> Error { if text.contains("NotImplemented") { @@ -367,7 +308,7 @@ mod tests { use rc_core::{ SelectCompression, SelectCsvInputOptions, SelectCsvOutputOptions, SelectInputFormat, SelectJsonInputOptions, SelectJsonInputType, SelectJsonOutputOptions, SelectOptions, - SelectOutputFormat, SelectScanRangeOptions, + SelectOptionsError, SelectOutputFormat, SelectScanRangeOptions, }; #[test] @@ -385,7 +326,34 @@ mod tests { #[test] fn classify_fallback_network() { let e = classify_aws_code(Some("SlowDown"), "rate limited"); - assert!(matches!(e, Error::General(_))); + assert!( + matches!(e, Error::Network(msg) if msg.contains("SlowDown") && msg.contains("rate limited")) + ); + } + + #[test] + fn classify_busy_preserves_service_context() { + let e = classify_aws_code(Some("Busy"), "The service is unavailable. Try again later."); + assert!( + matches!(e, Error::Network(msg) if msg.contains("Busy") && msg.contains("unavailable")) + ); + } + + #[test] + fn classify_unsupported_scan_range_preserves_service_context() { + let e = classify_aws_code( + Some("UnsupportedScanRangeInput"), + "Scan range queries are not supported on this type of object.", + ); + assert!( + matches!(e, Error::General(msg) if msg.contains("UnsupportedScanRangeInput") && msg.contains("not supported")) + ); + } + + #[test] + fn classify_unsupported_scan_range_replaces_generic_service_text() { + let e = classify_aws_code(Some("UnsupportedScanRangeInput"), "service error"); + assert!(matches!(e, Error::General(msg) if msg == "UnsupportedScanRangeInput")); } #[test] @@ -464,9 +432,10 @@ mod tests { ..SelectOptions::default() }; - let error = build_input_serialization(&options) - .expect_err("parquet should reject whole-object compression"); - assert!(matches!(error, Error::General(_))); + assert_eq!( + options.validate(), + Err(SelectOptionsError::CompressedParquetInput) + ); } #[test] @@ -479,7 +448,10 @@ mod tests { ..SelectOptions::default() }; - build_input_serialization(&options).expect("parquet without whole-object compression"); + options + .validate() + .expect("parquet without whole-object compression should be valid"); + assert!(build_input_serialization(&options).parquet().is_some()); } #[test] @@ -492,7 +464,7 @@ mod tests { ..SelectOptions::default() }; - let input = build_input_serialization(&options).expect("csv input serialization"); + let input = build_input_serialization(&options); let csv = input.csv().expect("csv input is configured"); assert_eq!(input.compression_type(), Some(&CompressionType::Bzip2)); @@ -501,6 +473,25 @@ mod tests { assert!(input.parquet().is_none()); } + #[test] + fn csv_input_serialization_sets_record_delimiter() { + let options = SelectOptions { + expression: "SELECT * FROM S3Object".to_string(), + csv_input: SelectCsvInputOptions { + record_delimiter: Some("\r\n".to_string()), + ..SelectCsvInputOptions::default() + }, + ..SelectOptions::default() + }; + + options + .validate() + .expect("CSV input options should be valid"); + let input = build_input_serialization(&options); + let csv = input.csv().expect("CSV input is configured"); + assert_eq!(csv.record_delimiter(), Some("\r\n")); + } + #[test] fn json_input_serialization_uses_lines_mode() { let options = SelectOptions { @@ -511,7 +502,7 @@ mod tests { ..SelectOptions::default() }; - let input = build_input_serialization(&options).expect("json input serialization"); + let input = build_input_serialization(&options); let json = input.json().expect("json input is configured"); assert_eq!(input.compression_type(), Some(&CompressionType::Gzip)); @@ -529,7 +520,7 @@ mod tests { compression: SelectCompression::None, ..SelectOptions::default() }; - let csv_output = build_output_serialization(&csv_options).expect("csv output"); + let csv_output = build_output_serialization(&csv_options); let csv = csv_output.csv().expect("csv output is configured"); assert_eq!(csv.quote_fields(), Some(&QuoteFields::Asneeded)); assert!(csv_output.json().is_none()); @@ -541,7 +532,7 @@ mod tests { compression: SelectCompression::None, ..SelectOptions::default() }; - let json_output = build_output_serialization(&json_options).expect("json output"); + let json_output = build_output_serialization(&json_options); assert!(json_output.json().is_some()); assert!(json_output.csv().is_none()); } @@ -557,9 +548,10 @@ mod tests { ..SelectOptions::default() }; - let error = build_input_serialization(&options) + let error = options + .validate() .expect_err("multi-byte CSV input delimiter should be rejected"); - assert!(matches!(error, Error::General(msg) if msg.contains("field delimiter"))); + assert!(error.to_string().contains("field delimiter")); } #[test] @@ -573,7 +565,10 @@ mod tests { ..SelectOptions::default() }; - let output = build_output_serialization(&options).expect("CRLF record delimiter"); + options + .validate() + .expect("CRLF record delimiter should be valid"); + let output = build_output_serialization(&options); let csv = output.csv().expect("csv output is configured"); assert_eq!(csv.record_delimiter(), Some("\r\n")); } @@ -589,9 +584,10 @@ mod tests { ..SelectOptions::default() }; - let error = build_output_serialization(&options) + let error = options + .validate() .expect_err("multi-byte CSV output record delimiter should be rejected"); - assert!(matches!(error, Error::General(msg) if msg.contains("record delimiter"))); + assert!(error.to_string().contains("record delimiter")); } #[test] @@ -605,7 +601,7 @@ mod tests { ..SelectOptions::default() }; - let output = build_output_serialization(&options).expect("json output serialization"); + let output = build_output_serialization(&options); let json = output.json().expect("json output is configured"); assert_eq!(json.record_delimiter(), Some("\n")); } @@ -625,9 +621,66 @@ mod tests { ..SelectOptions::default() }; - let error = - build_scan_range(&options).expect_err("scan range should reject JSON document input"); - assert!(matches!(error, Error::General(msg) if msg.contains("JSON document"))); + let error = options + .validate() + .expect_err("scan range should reject JSON document input"); + assert_eq!(error, SelectOptionsError::JsonDocumentScanRange); + } + + #[test] + fn scan_range_allows_parquet_input() { + let options = SelectOptions { + expression: "SELECT * FROM S3Object".to_string(), + input_format: SelectInputFormat::Parquet, + scan_range: SelectScanRangeOptions { + start: Some(1024), + end: Some(2047), + }, + ..SelectOptions::default() + }; + + options + .validate() + .expect("RustFS supports Parquet scan ranges"); + let scan_range = build_scan_range(&options).expect("scan range should be configured"); + assert_eq!(scan_range.start(), Some(1024)); + assert_eq!(scan_range.end(), Some(2047)); + } + + #[test] + fn scan_range_rejects_compressed_input() { + let options = SelectOptions { + expression: "SELECT * FROM S3Object".to_string(), + compression: SelectCompression::Gzip, + scan_range: SelectScanRangeOptions { + start: Some(1), + end: None, + }, + ..SelectOptions::default() + }; + + let error = options + .validate() + .expect_err("compressed input should reject a non-noop scan range"); + assert_eq!(error, SelectOptionsError::CompressedInputScanRange); + } + + #[test] + fn scan_range_allows_noop_for_compressed_input() { + let options = SelectOptions { + expression: "SELECT * FROM S3Object".to_string(), + compression: SelectCompression::Bzip2, + scan_range: SelectScanRangeOptions { + start: Some(0), + end: None, + }, + ..SelectOptions::default() + }; + + options + .validate() + .expect("RustFS accepts a no-op compressed scan range"); + assert!(build_scan_range(&options).is_some()); } #[test] @@ -641,7 +694,9 @@ mod tests { ..SelectOptions::default() }; - let error = build_scan_range(&options).expect_err("start after end should be rejected"); - assert!(matches!(error, Error::General(msg) if msg.contains("greater than end"))); + let error = options + .validate() + .expect_err("start after end should be rejected"); + assert_eq!(error, SelectOptionsError::ReversedScanRange); } }