feat(sql): support CSV input record delimiters

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",
+            "^Y",
+        ])
+        .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("^Y"));
+            }
+            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>,
@@ -254,6 +258,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,
@@ -301,6 +306,7 @@ fn validate_select_args(args: &SqlArgs) -> std::result::Result<(), String> {
         "--csv-input-field-delimiter",
         args.csv_input_field_delimiter.as_deref(),
     )?;
+    validate_input_record_delimiter(args.csv_input_record_delimiter.as_deref())?;
     validate_single_byte("--csv-input-quote", args.csv_input_quote.as_deref())?;
     validate_single_byte(
         "--csv-input-quote-escape",
@@ -332,6 +338,13 @@ fn validate_single_byte(name: &str, value: Option<&str>) -> std::result::Result<
     Ok(())
 }
 
+fn validate_input_record_delimiter(value: Option<&str>) -> std::result::Result<(), String> {
+    if value.is_some_and(|value| !(1..=2).contains(&value.len())) {
+        return Err("--csv-input-record-delimiter must be one or two bytes".to_string());
+    }
+    Ok(())
+}
+
 fn validate_record_delimiter(name: &str, value: Option<&str>) -> std::result::Result<(), String> {
     if let Some(value) = value
         && value.len() != 1
@@ -384,6 +397,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,
@@ -424,6 +438,14 @@ mod tests {
         assert_eq!(code, ExitCode::UsageError);
     }
 
+    #[tokio::test]
+    async fn sql_rejects_invalid_csv_input_record_delimiter() {
+        let mut args = base_args("a/b/c", "SELECT * FROM S3Object");
+        args.csv_input_record_delimiter = Some(String::new());
+        let code = execute(args, OutputConfig::default()).await;
+        assert_eq!(code, ExitCode::UsageError);
+    }
+
     #[tokio::test]
     async fn sql_rejects_scan_range_for_json_document() {
         let mut args = base_args("a/b/c", "SELECT * FROM S3Object");

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/select.rs

@@ -56,6 +56,7 @@ pub enum SelectQuoteFields {
 pub struct SelectCsvInputOptions {
     pub file_header_info: SelectCsvFileHeaderInfo,
     pub field_delimiter: Option<String>,
+    pub record_delimiter: Option<String>,
     pub quote_character: Option<String>,
     pub quote_escape_character: Option<String>,
     pub comments: Option<String>,

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()));
         }

crates/s3/src/select.rs

@@ -143,6 +143,15 @@ fn validate_single_byte(name: &str, value: Option<&str>) -> Result<()> {
     Ok(())
 }
 
+fn validate_input_record_delimiter(value: Option<&str>) -> Result<()> {
+    if value.is_some_and(|value| !(1..=2).contains(&value.len())) {
+        return Err(Error::General(
+            "CSV input record delimiter must be one or two bytes.".to_string(),
+        ));
+    }
+    Ok(())
+}
+
 fn validate_record_delimiter(name: &str, value: Option<&str>) -> Result<()> {
     if let Some(value) = value
         && value.len() != 1
@@ -184,11 +193,15 @@ fn build_input_serialization(options: &SelectOptions) -> Result<InputSerializati
                 "CSV input comment character",
                 options.csv_input.comments.as_deref(),
             )?;
+            validate_input_record_delimiter(options.csv_input.record_delimiter.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);
             }
@@ -501,6 +514,38 @@ 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("^Y".to_string()),
+                ..SelectCsvInputOptions::default()
+            },
+            ..SelectOptions::default()
+        };
+
+        let input = build_input_serialization(&options).expect("CSV input serialization");
+        let csv = input.csv().expect("CSV input is configured");
+        assert_eq!(csv.record_delimiter(), Some("^Y"));
+    }
+
+    #[test]
+    fn csv_input_rejects_invalid_record_delimiter() {
+        let options = SelectOptions {
+            expression: "SELECT * FROM S3Object".to_string(),
+            csv_input: SelectCsvInputOptions {
+                record_delimiter: Some("|||".to_string()),
+                ..SelectCsvInputOptions::default()
+            },
+            ..SelectOptions::default()
+        };
+
+        let error = build_input_serialization(&options)
+            .expect_err("three-byte CSV input record delimiter should be rejected");
+        assert!(matches!(error, Error::General(message) if message.contains("record delimiter")));
+    }
+
     #[test]
     fn json_input_serialization_uses_lines_mode() {
         let options = SelectOptions {