fix: Validate filter selectivity at SET time

datafusion/common/src/config.rs

@@ -21,7 +21,7 @@ use arrow_ipc::CompressionType;
 
 #[cfg(feature = "parquet_encryption")]
 use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
-use crate::error::{_config_datafusion_err, _config_err};
+use crate::error::_config_err;
 use crate::format::{ExplainAnalyzeCategories, ExplainFormat, MetricType};
 use crate::parquet_config::DFParquetWriterVersion;
 use crate::parsers::{CompressionTypeVariant, CsvQuoteStyle};
@@ -33,7 +33,6 @@ use std::any::Any;
 use std::collections::{BTreeMap, HashMap};
 use std::error::Error;
 use std::fmt::{self, Display};
-use std::num::NonZeroUsize;
 use std::str::FromStr;
 #[cfg(feature = "parquet_encryption")]
 use std::sync::Arc;
@@ -583,29 +582,123 @@ impl Display for SpillCompression {
     }
 }
 
+/// A reusable bounded `usize` configuration value.
+///
+/// This stores the configured value together with its inclusive lower bound and
+/// optional inclusive upper bound so wrapper types such as
+/// [`ConfigNonZeroUsize`] and [`ConfigMinTwoUsize`] can share the same range
+/// validation logic.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ConfigRangeUsize {
+    value: usize,
+    inclusive_min: usize,
+    /// `None` if there is no maximum limit.
+    inclusive_max: Option<usize>,
+}
+
+impl ConfigRangeUsize {
+    const fn in_range(
+        value: usize,
+        inclusive_min: usize,
+        inclusive_max: Option<usize>,
+    ) -> bool {
+        if value < inclusive_min {
+            return false;
+        }
+
+        match inclusive_max {
+            Some(inclusive_max) => value <= inclusive_max,
+            None => true,
+        }
+    }
+
+    const fn new_for_default(
+        value: usize,
+        inclusive_min: usize,
+        inclusive_max: Option<usize>,
+    ) -> Self {
+        if Self::in_range(value, inclusive_min, inclusive_max) {
+            Self {
+                value,
+                inclusive_min,
+                inclusive_max,
+            }
+        } else {
+            panic!("value out of allowed range")
+        }
+    }
+
+    /// Creates a [`ConfigRangeUsize`], returning a configuration error if the
+    /// value falls outside the provided inclusive bounds.
+    pub fn try_new(
+        value: usize,
+        inclusive_min: usize,
+        inclusive_max: Option<usize>,
+    ) -> Result<Self> {
+        if Self::in_range(value, inclusive_min, inclusive_max) {
+            Ok(Self {
+                value,
+                inclusive_min,
+                inclusive_max,
+            })
+        } else {
+            match inclusive_max {
+                Some(inclusive_max) => _config_err!(
+                    "value must be between {inclusive_min} and {inclusive_max}, got {value}"
+                ),
+                None => _config_err!("value must be at least {inclusive_min}"),
+            }
+        }
+    }
+
+    /// Returns the configured value.
+    pub const fn get(self) -> usize {
+        self.value
+    }
+
+    /// Returns the inclusive minimum bound.
+    pub const fn inclusive_min(self) -> usize {
+        self.inclusive_min
+    }
+
+    /// Returns the inclusive maximum bound, if any.
+    pub const fn inclusive_max(self) -> Option<usize> {
+        self.inclusive_max
+    }
+}
+
+impl Display for ConfigRangeUsize {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.get())
+    }
+}
+
 /// A `usize` configuration value that rejects zero when set from strings.
 ///
 /// Use this for options where zero is never a meaningful runtime value.
 /// Invalid values return a configuration error through [`ConfigField`].
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub struct ConfigNonZeroUsize(NonZeroUsize);
+pub struct ConfigNonZeroUsize(ConfigRangeUsize);
 
 /// Private helper for hard-coded defaults in `config_namespace!`, which cannot
 /// use `?`. All external construction should use [`ConfigNonZeroUsize::try_new`].
 const fn non_zero_usize_default(value: usize) -> ConfigNonZeroUsize {
-    match NonZeroUsize::new(value) {
-        Some(value) => ConfigNonZeroUsize(value),
-        None => panic!("value must be greater than 0"),
+    if value > 0 {
+        ConfigNonZeroUsize(ConfigRangeUsize::new_for_default(value, 1, None))
+    } else {
+        panic!("value must be greater than 0")
     }
 }
 
 impl ConfigNonZeroUsize {
     /// Creates a [`ConfigNonZeroUsize`], returning a configuration error if
     /// `value` is zero.
     pub fn try_new(value: usize) -> Result<Self> {
-        NonZeroUsize::new(value)
-            .map(Self)
-            .ok_or_else(|| _config_datafusion_err!("value must be greater than 0"))
+        if value > 0 {
+            Ok(Self(ConfigRangeUsize::new_for_default(value, 1, None)))
+        } else {
+            _config_err!("value must be greater than 0")
+        }
     }
 
     /// Returns the wrapped `usize`.
@@ -670,13 +763,13 @@ impl Display for ConfigNonZeroUsize {
 /// round down to a zero-capacity buffer and panic. Invalid values return a
 /// configuration error through [`ConfigField`] instead.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub struct ConfigMinTwoUsize(usize);
+pub struct ConfigMinTwoUsize(ConfigRangeUsize);
 
 /// Private helper for hard-coded defaults in `config_namespace!`, which cannot
 /// use `?`. All external construction should use [`ConfigMinTwoUsize::try_new`].
 const fn min_two_usize_default(value: usize) -> ConfigMinTwoUsize {
     if value >= 2 {
-        ConfigMinTwoUsize(value)
+        ConfigMinTwoUsize(ConfigRangeUsize::new_for_default(value, 2, None))
     } else {
         panic!("value must be at least 2")
     }
@@ -687,15 +780,15 @@ impl ConfigMinTwoUsize {
     /// `value` is less than 2.
     pub fn try_new(value: usize) -> Result<Self> {
         if value >= 2 {
-            Ok(Self(value))
+            Ok(Self(ConfigRangeUsize::new_for_default(value, 2, None)))
         } else {
             _config_err!("value must be at least 2")
         }
     }
 
     /// Returns the wrapped `usize`.
     pub const fn get(self) -> usize {
-        self.0
+        self.0.get()
     }
 }
 
@@ -748,6 +841,88 @@ impl Display for ConfigMinTwoUsize {
     }
 }
 
+/// Used for [`OptimizerOptions::default_filter_selectivity`] to represent
+/// an integer percentage value, when valid values are 0 to 100 inclusive.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ConfigFilterSelectivity(u8);
+
+/// Private helper for hard-coded defaults in `config_namespace!`, which cannot
+/// use `?`. All external construction should use
+/// [`ConfigFilterSelectivity::try_new`].
+const fn filter_selectivity_default(value: u8) -> ConfigFilterSelectivity {
+    if value <= 100 {
+        ConfigFilterSelectivity(value)
+    } else {
+        panic!("value must be between 0 and 100")
+    }
+}
+
+impl ConfigFilterSelectivity {
+    /// Creates a [`ConfigFilterSelectivity`], returning a configuration error
+    /// if `value` is greater than 100.
+    pub fn try_new(value: u8) -> Result<Self> {
+        if value <= 100 {
+            Ok(Self(value))
+        } else {
+            _config_err!("value must be between 0 and 100, got {value}")
+        }
+    }
+
+    /// Returns the wrapped `u8`.
+    pub const fn get(self) -> u8 {
+        self.0
+    }
+}
+
+impl From<ConfigFilterSelectivity> for u8 {
+    fn from(value: ConfigFilterSelectivity) -> Self {
+        value.get()
+    }
+}
+
+impl FromStr for ConfigFilterSelectivity {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Self::try_new(default_config_transform(s)?)
+    }
+}
+
+impl ConfigField for ConfigFilterSelectivity {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        v.some(key, self, description)
+    }
+
+    fn set(&mut self, key: &str, value: &str) -> Result<()> {
+        if !key.is_empty() {
+            return _config_err!(
+                "Config field default_filter_selectivity is a scalar ConfigFilterSelectivity and does not have nested field \"{}\"",
+                key
+            );
+        }
+
+        *self = ConfigFilterSelectivity::from_str(value)?;
+        Ok(())
+    }
+
+    fn reset(&mut self, key: &str) -> Result<()> {
+        if key.is_empty() {
+            Ok(())
+        } else {
+            _config_err!(
+                "Config field default_filter_selectivity is a scalar ConfigFilterSelectivity and does not have nested field \"{}\"",
+                key
+            )
+        }
+    }
+}
+
+impl Display for ConfigFilterSelectivity {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.get())
+    }
+}
+
 /// Policy for handling duplicate keys in Spark-compatible map-construction
 /// functions (`map_from_arrays`, `map_from_entries`, `str_to_map`). Mirrors
 /// Spark's [`spark.sql.mapKeyDedupPolicy`](https://github.com/apache/spark/blob/cf3a34e19dfcf70e2d679217ff1ba21302212472/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L4961).
@@ -1728,7 +1903,7 @@ config_namespace! {
         /// The default filter selectivity used by Filter Statistics
         /// when an exact selectivity cannot be determined. Valid values are
         /// between 0 (no selectivity) and 100 (all rows are selected).
-        pub default_filter_selectivity: u8, default = 20
+        pub default_filter_selectivity: ConfigFilterSelectivity, default = filter_selectivity_default(20)
 
         /// When set to true, the optimizer will not attempt to convert Union to Interleave
         pub prefer_existing_union: bool, default = false

datafusion/core/src/physical_planner.rs

@@ -1221,7 +1221,8 @@ impl DefaultPhysicalPlanner {
                     .config()
                     .options()
                     .optimizer
-                    .default_filter_selectivity;
+                    .default_filter_selectivity
+                    .get();
                 let filter_exec: Arc<dyn ExecutionPlan> =
                     Arc::new(filter.with_default_selectivity(selectivity)?);
                 filter_exec

datafusion/sqllogictest/test_files/set_variable.slt

@@ -778,6 +778,27 @@ caused by
 Invalid or Unsupported Configuration: value must be at least 2
 
 
+# default_filter_selectivity is a percentage (0..=100); previously an
+# out-of-range value like 200 was silently accepted here and only rejected
+# later, when FilterExec was built.
+statement error
+SET datafusion.optimizer.default_filter_selectivity = 200
+----
+DataFusion error: Error setting config datafusion.optimizer.default_filter_selectivity
+caused by
+Invalid or Unsupported Configuration: value must be between 0 and 100, got 200
+
+
+statement ok
+SET datafusion.optimizer.default_filter_selectivity = 100
+
+statement ok
+SET datafusion.optimizer.default_filter_selectivity = 0
+
+statement ok
+SET datafusion.optimizer.default_filter_selectivity = 20
+
+
 # Config reset
 statement ok
 RESET datafusion.catalog.create_default_catalog_and_schema