feat: add match_columns_by_header option for CSV

datafusion/common/src/config.rs

@@ -4087,6 +4087,17 @@ config_namespace! {
         /// of columns, it creates a union schema containing all columns found across the files, and will
         /// pad any files missing columns with null values for their rows.
         pub truncated_rows: Option<bool>, default = None
+        /// Whether to map file columns to the table schema by header name instead of by position.
+        ///
+        /// Only applies when `has_header` is true. When enabled, each file's header row is read
+        /// and columns are matched by name: table columns missing from a file are filled with
+        /// nulls, and file columns not declared in the table are ignored. Names are matched exactly
+        /// (case-sensitive) and duplicate header names resolve to the first occurrence. Header
+        /// names containing quoted newlines are not supported. The header is located the same
+        /// way the CSV parser does: a leading UTF-8 BOM, blank lines, and comment lines are skipped.
+        ///
+        /// Defaults to `false` when `None`.
+        pub match_columns_by_header: Option<bool>, default = None
     }
 }
 
@@ -4212,6 +4223,13 @@ impl CsvOptions {
         self
     }
 
+    /// Whether to map file columns to the table schema by header name instead of by position.
+    /// Only applies when the file has a header.
+    pub fn with_match_columns_by_header(mut self, match_by_header: bool) -> Self {
+        self.match_columns_by_header = Some(match_by_header);
+        self
+    }
+
     /// Set the compression level for the output file.
     /// The valid range depends on the compression algorithm.
     /// If not specified, the default level for the algorithm is used.

datafusion/core/src/datasource/file_format/options.rs

@@ -86,6 +86,12 @@ pub struct CsvReadOptions<'a> {
     /// When set to true then it will allow records with less than the expected number of columns and fill the missing columns with nulls.
     /// If the record’s schema is not nullable, then it will still return an error.
     pub truncated_rows: bool,
+    /// Whether to map file columns to the table schema by header name instead of by position.
+    ///
+    /// Only applies when `has_header` is true. Table columns missing from a file are
+    /// filled with nulls and file columns not declared in the table are ignored.
+    /// Defaults to `false`.
+    pub match_columns_by_header: bool,
 }
 
 impl Default for CsvReadOptions<'_> {
@@ -113,6 +119,7 @@ impl<'a> CsvReadOptions<'a> {
             comment: None,
             null_regex: None,
             truncated_rows: false,
+            match_columns_by_header: false,
         }
     }
 
@@ -228,6 +235,13 @@ impl<'a> CsvReadOptions<'a> {
         self.truncated_rows = truncated_rows;
         self
     }
+
+    /// Configure whether file columns are matched to the table schema by header name
+    /// instead of by position. Only applies when `has_header` is true.
+    pub fn match_columns_by_header(mut self, match_columns_by_header: bool) -> Self {
+        self.match_columns_by_header = match_columns_by_header;
+        self
+    }
 }
 
 /// Options that control the reading of Parquet files.
@@ -643,7 +657,8 @@ impl ReadOptions<'_> for CsvReadOptions<'_> {
             .with_schema_infer_max_rec(self.schema_infer_max_records)
             .with_file_compression_type(self.file_compression_type.to_owned())
             .with_null_regex(self.null_regex.clone())
-            .with_truncated_rows(self.truncated_rows);
+            .with_truncated_rows(self.truncated_rows)
+            .with_match_columns_by_header(self.match_columns_by_header);
 
         ListingOptions::new(Arc::new(file_format))
             .with_file_extension(self.file_extension)