Pass opt-in numeric CSV Boolean mode through scans and plan protobuf

Cargo.toml

@@ -89,30 +89,30 @@ version = "55.1.0"
 #
 # See for more details: https://github.com/rust-lang/cargo/issues/11329
 apache-avro = { version = "0.21", default-features = false }
-arrow = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", features = [
+arrow = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", features = [
     "prettyprint",
     "chrono-tz",
 ] }
-arrow-avro = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", default-features = false, features = [
+arrow-avro = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", default-features = false, features = [
     "deflate",
     "snappy",
     "zstd",
     "bzip2",
     "xz",
 ] }
-arrow-buffer = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", default-features = false }
-arrow-data = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", default-features = false }
-arrow-flight = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", features = [
+arrow-buffer = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", default-features = false }
+arrow-data = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", default-features = false }
+arrow-flight = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", features = [
     "flight-sql-experimental",
 ] }
 # Both codecs are required here to make sure that code paths like
 # file-spilling have access to all compression codecs.
-arrow-ipc = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", default-features = false, features = [
+arrow-ipc = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", default-features = false, features = [
     "lz4",
     "zstd",
 ] }
-arrow-ord = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", default-features = false }
-arrow-schema = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", default-features = false }
+arrow-ord = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", default-features = false }
+arrow-schema = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", default-features = false }
 async-trait = "0.1.89"
 bigdecimal = "0.4.8"
 bytes = "1.11"
@@ -178,7 +178,7 @@ memchr = "2.8.1"
 num-traits = { version = "0.2" }
 object_store = { version = "0.13.2", default-features = false }
 parking_lot = "0.12"
-parquet = { git = "https://github.com/Embucket/arrow-rs.git", rev = "5d211125b8e8a5784f2caf606cdf88684dec30a5", default-features = false, features = [
+parquet = { git = "https://github.com/Embucket/arrow-rs.git", rev = "25a9c23d7b9706238b00b1288d27f036c186cd8e", default-features = false, features = [
     "arrow",
     "async",
     "object_store",

datafusion/datasource-csv/src/file_format.rs

@@ -136,6 +136,7 @@ impl GetExt for CsvFormatFactory {
 pub struct CsvFormat {
     options: CsvOptions,
     record_error_handler: Option<Arc<dyn CsvRecordErrorHandler>>,
+    numeric_boolean_values: bool,
 }
 
 impl CsvFormat {
@@ -202,6 +203,12 @@ impl CsvFormat {
         self
     }
 
+    /// Accept exact `0` and `1` values in Boolean CSV columns.
+    pub fn with_numeric_boolean_values(mut self, allow: bool) -> Self {
+        self.numeric_boolean_values = allow;
+        self
+    }
+
     /// Skip malformed field-count records and report them to `handler`.
     ///
     /// This recovery path retains source record bytes and disables file range
@@ -492,7 +499,9 @@ impl FileFormat for CsvFormat {
         if csv_options.has_header.is_none() {
             csv_options.has_header = Some(true);
         }
-        let mut source = CsvSource::new(table_schema).with_csv_options(csv_options);
+        let mut source = CsvSource::new(table_schema)
+            .with_csv_options(csv_options)
+            .with_numeric_boolean_values(self.numeric_boolean_values);
         if let Some(handler) = &self.record_error_handler {
             source = source.with_record_error_handler(Arc::clone(handler));
         }

datafusion/datasource-csv/src/source.rs

@@ -87,6 +87,7 @@ use tokio::io::AsyncWriteExt;
 #[derive(Debug, Clone)]
 pub struct CsvSource {
     options: CsvOptions,
+    numeric_boolean_values: bool,
     record_error_handler: Option<Arc<dyn csv::CsvRecordErrorHandler>>,
     batch_size: Option<usize>,
     table_schema: TableSchema,
@@ -100,6 +101,7 @@ impl CsvSource {
         let table_schema = table_schema.into();
         Self {
             options: CsvOptions::default(),
+            numeric_boolean_values: false,
             record_error_handler: None,
             projection: SplitProjection::unprojected(&table_schema),
             table_schema,
@@ -114,6 +116,17 @@ impl CsvSource {
         self
     }
 
+    /// Accept exact numeric Boolean literals (`0` and `1`) while scanning CSV.
+    pub fn with_numeric_boolean_values(mut self, allow: bool) -> Self {
+        self.numeric_boolean_values = allow;
+        self
+    }
+
+    /// Whether exact `0` and `1` values are accepted in Boolean CSV columns.
+    pub fn numeric_boolean_values(&self) -> bool {
+        self.numeric_boolean_values
+    }
+
     /// Skip malformed field-count records and report them to `handler`.
     pub fn with_record_error_handler(
         mut self,
@@ -217,6 +230,7 @@ impl CsvSource {
                 )
                 .with_header(self.has_header())
                 .with_quote(self.quote())
+                .with_numeric_boolean_values(self.numeric_boolean_values)
                 .with_preserve_quoted_empty(self.preserve_quoted_empty())
                 .with_truncated_rows(self.truncate_rows());
         if let Some(terminator) = self.terminator() {
@@ -404,6 +418,7 @@ impl FileSource for CsvSource {
             truncate_rows: self.truncate_rows(),
             null_regex: self.options.null_regex.clone(),
             preserve_quoted_empty: self.preserve_quoted_empty(),
+            numeric_boolean_values: self.numeric_boolean_values(),
         };
         Ok(Some(protobuf::PhysicalPlanNode {
             physical_plan_type: Some(PhysicalPlanType::CsvScan(node)),
@@ -684,6 +699,7 @@ impl CsvSource {
         let source = Arc::new(
             CsvSource::new(table_schema)
                 .with_csv_options(csv_options)
+                .with_numeric_boolean_values(scan.numeric_boolean_values)
                 .with_escape(escape)
                 .with_comment(comment),
         );
@@ -702,13 +718,50 @@ impl CsvSource {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use arrow::array::{Int64Array, StringArray};
+    use arrow::array::{AsArray, Int64Array, StringArray};
     use arrow::csv::{CsvRecordError, CsvRecordErrorHandler};
     use arrow::datatypes::{DataType, Field, Schema};
     use arrow::error::ArrowError;
     use std::io::Cursor;
     use std::sync::Mutex;
 
+    #[test]
+    fn csv_source_numeric_boolean_values_are_opt_in() {
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "flag",
+            DataType::Boolean,
+            false,
+        )]));
+        let mut source = CsvSource::new(schema).with_csv_options(CsvOptions {
+            has_header: Some(false),
+            ..Default::default()
+        });
+        source.batch_size = Some(1024);
+        let strict_error = source
+            .open(Cursor::new(b"true\n1\n0\nfalse\n"))
+            .unwrap()
+            .next()
+            .unwrap()
+            .unwrap_err();
+        assert!(
+            strict_error
+                .to_string()
+                .contains("value '1' as type 'Boolean'")
+        );
+
+        let batches = source
+            .with_numeric_boolean_values(true)
+            .open(Cursor::new(b"true\n1\n0\nfalse\n"))
+            .unwrap()
+            .collect::<std::result::Result<Vec<_>, _>>()
+            .unwrap();
+        let values = batches[0].column(0).as_boolean();
+        assert_eq!(
+            values.values().iter().collect::<Vec<_>>(),
+            [true, true, false, false]
+        );
+    }
+
     #[test]
     fn csv_source_applies_null_regex_to_loaded_rows() {
         let schema = Arc::new(Schema::new(vec![

datafusion/proto-models/proto/datafusion.proto

@@ -1288,6 +1288,7 @@ message CsvScanExecNode {
   bool truncate_rows = 8;
   optional string null_regex = 9;
   bool preserve_quoted_empty = 10;
+  bool numeric_boolean_values = 11;
 }
 
 message JsonScanExecNode {