fix(sql): align S3 Select behavior with RustFS

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([

crates/cli/src/commands/sql.rs

@@ -42,6 +42,10 @@ pub struct SqlArgs {
     #[arg(long)]
     pub csv_input_field_delimiter: Option<String>,
 
+    /// CSV input record delimiter (one or two bytes)
+    #[arg(long)]
+    pub csv_input_record_delimiter: Option<String>,
+
     /// CSV input quote character
     #[arg(long)]
     pub csv_input_quote: Option<String>,
@@ -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,14 +251,30 @@ 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(),
         compression: args.compression.into(),
         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 = [

crates/cli/src/commands/table/mod.rs

@@ -406,10 +406,10 @@ fn properties(values: Vec<String>) -> Result<BTreeMap<String, String>> {
     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<Prepared> {
                 {
                     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(),

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,