@@ -40,6 +40,24 @@ pub enum SchemaSource {
     Specified,
 }
 
+#[cfg(test)]
+mod compression_extension_tests {
+    use super::ListingTableConfig;
+
+    #[test]
+    fn new_codec_suffixes_preserve_csv_extension() {
+        for suffix in ["br", "zlib", "raw_deflate"] {
+            assert_eq!(
+                ListingTableConfig::infer_file_extension_and_compression_type(&format!(
+                    "data.csv.{suffix}"
+                ))
+                .unwrap(),
+                ("csv".to_owned(), Some(suffix.to_owned()))
+            );
+        }
+    }
+}
+
 /// Configuration for creating a [`crate::ListingTable`]
 ///
 /// # Schema Evolution Support
 
@@ -389,6 +389,28 @@ pub async fn pruned_partition_list<'a>(
     filters: &'a [Expr],
     file_extension: &'a str,
     partition_cols: &'a [(String, DataType)],
+) -> Result<BoxStream<'a, Result<PartitionedFile>>> {
+    pruned_partition_list_with_suffixes(
+        ctx,
+        store,
+        table_path,
+        filters,
+        file_extension,
+        None,
+        partition_cols,
+    )
+    .await
+}
+
+/// As [`pruned_partition_list`], but accepts several filename suffixes.
+pub async fn pruned_partition_list_with_suffixes<'a>(
+    ctx: &'a dyn Session,
+    store: &'a dyn ObjectStore,
+    table_path: &'a ListingTableUrl,
+    filters: &'a [Expr],
+    file_extension: &'a str,
+    file_suffixes: Option<&'a [String]>,
+    partition_cols: &'a [(String, DataType)],
 ) -> Result<BoxStream<'a, Result<PartitionedFile>>> {
     let prefix = if !partition_cols.is_empty() {
         evaluate_partition_prefix(partition_cols, filters)
@@ -397,7 +419,13 @@ pub async fn pruned_partition_list<'a>(
     };
 
     let objects = table_path
-        .list_prefixed_files(ctx, store, prefix, file_extension)
+        .list_prefixed_files_with_suffixes(
+            ctx,
+            store,
+            prefix,
+            file_extension,
+            file_suffixes,
+        )
         .await?
         .try_filter(|object_meta| futures::future::ready(object_meta.size > 0));
 
 
@@ -17,8 +17,9 @@
 
 use arrow::datatypes::{DataType, SchemaRef};
 use datafusion_catalog::Session;
-use datafusion_common::plan_err;
+use datafusion_common::{GetExt, plan_err};
 use datafusion_datasource::ListingTableUrl;
+use datafusion_datasource::file_compression_type::FileCompressionType;
 use datafusion_datasource::file_format::FileFormat;
 use datafusion_expr::{Partitioning, SortExpr};
 use futures::StreamExt;
@@ -33,6 +34,9 @@ pub struct ListingOptions {
     /// A suffix on which files should be filtered (leave empty to
     /// keep all files on the path)
     pub file_extension: String,
+    /// Accepted filename suffixes for header-detected compression. An explicit
+    /// file-extension override disables this filter.
+    pub auto_file_suffixes: Option<Vec<String>>,
     /// The file format
     pub format: Arc<dyn FileFormat>,
     /// The expected partition column names in the folder structure.
@@ -96,14 +100,38 @@ pub struct ListingOptions {
     pub output_partitioning: Option<Partitioning>,
 }
 
+fn auto_compression_suffixes(base_extension: &str) -> Vec<String> {
+    let base = format!(".{}", base_extension.trim_start_matches('.'));
+    [
+        FileCompressionType::UNCOMPRESSED,
+        FileCompressionType::GZIP,
+        FileCompressionType::BZIP2,
+        FileCompressionType::XZ,
+        FileCompressionType::ZSTD,
+        FileCompressionType::DEFLATE,
+    ]
+    .into_iter()
+    .map(|compression| format!("{base}{}", compression.get_ext()))
+    .collect()
+}
+
 impl ListingOptions {
     /// Creates an options instance with the given format
     /// Default values:
     /// - use default file extension filter
     /// - no input partition to discover
     pub fn new(format: Arc<dyn FileFormat>) -> Self {
+        let auto = format.compression_type() == Some(FileCompressionType::AUTO);
+        let file_extension = if auto {
+            String::new()
+        } else {
+            format.get_ext()
+        };
+        let auto_file_suffixes =
+            auto.then(|| auto_compression_suffixes(&format.get_ext()));
         Self {
-            file_extension: format.get_ext(),
+            file_extension,
+            auto_file_suffixes,
             format,
             table_partition_cols: vec![],
             file_sort_order: vec![],
@@ -126,6 +154,15 @@ impl ListingOptions {
     /// ```
     pub fn with_file_extension(mut self, file_extension: impl Into<String>) -> Self {
         self.file_extension = file_extension.into();
+        self.auto_file_suffixes = None;
+        self
+    }
+
+    /// Detect compression while restricting a listing to files of this format.
+    pub fn with_auto_compression_file_extension(mut self, base_extension: &str) -> Self {
+        self.file_extension.clear();
+        self.auto_file_suffixes = (!base_extension.is_empty())
+            .then(|| auto_compression_suffixes(base_extension));
         self
     }
 
@@ -150,7 +187,7 @@ impl ListingOptions {
         S: Into<String>,
     {
         if let Some(file_extension) = file_extension {
-            self.file_extension = file_extension.into();
+            self = self.with_file_extension(file_extension);
         }
         self
     }
@@ -280,7 +317,12 @@ impl ListingOptions {
         let store = state.runtime_env().object_store(table_path)?;
 
         let all_files: Vec<_> = table_path
-            .list_all_files(state, store.as_ref(), &self.file_extension)
+            .list_all_files_with_suffixes(
+                state,
+                store.as_ref(),
+                &self.file_extension,
+                self.auto_file_suffixes.as_deref(),
+            )
             .await?
             .try_collect()
             .await?;
@@ -375,7 +417,12 @@ impl ListingOptions {
         // This can fail to detect inconsistent partition keys
         // A DFS traversal approach of the store can help here
         let files: Vec<_> = table_path
-            .list_all_files(state, store.as_ref(), &self.file_extension)
+            .list_all_files_with_suffixes(
+                state,
+                store.as_ref(),
+                &self.file_extension,
+                self.auto_file_suffixes.as_deref(),
+            )
             .await?
             .take(10)
             .try_collect()
 
@@ -17,7 +17,8 @@
 
 use crate::config::SchemaSource;
 use crate::helpers::{
-    expr_applicable_for_cols, filter_partitioned_file, pruned_partition_list,
+    expr_applicable_for_cols, filter_partitioned_file,
+    pruned_partition_list_with_suffixes as pruned_partition_list,
 };
 use crate::{ListingOptions, ListingTableConfig};
 use arrow::datatypes::{Field, Schema, SchemaBuilder, SchemaRef};
@@ -725,6 +726,7 @@ impl TableProvider for ListingTable {
             table_path,
             &[],
             &self.options.file_extension,
+            self.options.auto_file_suffixes.as_deref(),
             &self.options.table_partition_cols,
         )
         .await?;
@@ -812,6 +814,7 @@ impl ListingTable {
                 table_path,
                 listing_time_filters,
                 &self.options.file_extension,
+                self.options.auto_file_suffixes.as_deref(),
                 &self.options.table_partition_cols,
             )
         }))
 
@@ -35,6 +35,15 @@ pub enum CompressionTypeVariant {
     ZSTD,
     /// Uncompressed file
     UNCOMPRESSED,
+    /// Detect file compression from its header when reading
+    AUTO,
+    /// Brotli-compressed file
+    BROTLI,
+    /// Deflate-compressed file with a zlib header
+    DEFLATE,
+    /// Deflate-compressed file without a zlib header
+    #[expect(non_camel_case_types, reason = "match the external codec name")]
+    RAW_DEFLATE,
 }
 
 impl FromStr for CompressionTypeVariant {
@@ -48,6 +57,10 @@ impl FromStr for CompressionTypeVariant {
             "XZ" => Ok(Self::XZ),
             "ZST" | "ZSTD" => Ok(Self::ZSTD),
             "" | "UNCOMPRESSED" => Ok(Self::UNCOMPRESSED),
+            "AUTO" => Ok(Self::AUTO),
+            "BROTLI" | "BR" => Ok(Self::BROTLI),
+            "DEFLATE" | "ZLIB" => Ok(Self::DEFLATE),
+            "RAW_DEFLATE" => Ok(Self::RAW_DEFLATE),
             _ => Err(DataFusionError::NotImplemented(format!(
                 "Unsupported file compression type {s}"
             ))),
@@ -63,6 +76,10 @@ impl Display for CompressionTypeVariant {
             Self::XZ => "XZ",
             Self::ZSTD => "ZSTD",
             Self::UNCOMPRESSED => "",
+            Self::AUTO => "AUTO",
+            Self::BROTLI => "BROTLI",
+            Self::DEFLATE => "DEFLATE",
+            Self::RAW_DEFLATE => "RAW_DEFLATE",
         };
         write!(f, "{str}")
     }
 
@@ -643,8 +643,13 @@ impl ReadOptions<'_> for CsvReadOptions<'_> {
             .with_null_regex(self.null_regex.clone())
             .with_truncated_rows(self.truncated_rows);
 
-        ListingOptions::new(Arc::new(file_format))
-            .with_file_extension(self.file_extension)
+        let listing = ListingOptions::new(Arc::new(file_format));
+        let listing = if self.file_compression_type == FileCompressionType::AUTO {
+            listing.with_auto_compression_file_extension(self.file_extension)
+        } else {
+            listing.with_file_extension(self.file_extension)
+        };
+        listing
             .with_table_partition_cols(self.table_partition_cols.clone())
             .with_file_sort_order(self.file_sort_order.clone())
     }
@@ -724,8 +729,13 @@ impl ReadOptions<'_> for JsonReadOptions<'_> {
             .with_file_compression_type(self.file_compression_type.to_owned())
             .with_newline_delimited(self.newline_delimited);
 
-        ListingOptions::new(Arc::new(file_format))
-            .with_file_extension(self.file_extension)
+        let listing = ListingOptions::new(Arc::new(file_format));
+        let listing = if self.file_compression_type == FileCompressionType::AUTO {
+            listing.with_auto_compression_file_extension(self.file_extension)
+        } else {
+            listing.with_file_extension(self.file_extension)
+        };
+        listing
             .with_table_partition_cols(self.table_partition_cols.clone())
             .with_file_sort_order(self.file_sort_order.clone())
     }