@@ -35,6 +35,7 @@ use crate::filter_pushdown::{
     ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase,
     FilterPushdownPropagation, PushedDown,
 };
+use crate::joins::utils::max_distinct_count;
 use crate::limit::LocalLimitExec;
 use crate::metrics::{MetricBuilder, MetricType};
 use crate::projection::{
@@ -63,13 +64,13 @@ use datafusion_execution::TaskContext;
 use datafusion_expr::Operator;
 use datafusion_physical_expr::equivalence::ProjectionMapping;
 use datafusion_physical_expr::expressions::{
-    BinaryExpr, Column, IsNotNullExpr, Literal, lit,
+    BinaryExpr, Column, InListExpr, IsNotNullExpr, Literal, lit,
 };
 use datafusion_physical_expr::intervals::utils::check_support;
 use datafusion_physical_expr::utils::{collect_columns, reassign_expr_columns};
 use datafusion_physical_expr::{
     AcrossPartitions, AnalysisContext, ConstExpr, ExprBoundaries, PhysicalExpr, analyze,
-    conjunction, split_conjunction,
+    conjunction, conjunction_opt, split_conjunction,
 };
 
 use datafusion_physical_expr_common::physical_expr::fmt_sql;
@@ -359,29 +360,84 @@ impl FilterExec {
         } else {
             let null_rejecting_columns = collect_null_rejecting_columns(predicate);
 
-            if check_support(predicate, schema) {
+            // Estimate the predicate one top-level conjunct at a time. Interval
+            // analysis rejects a whole predicate if any part of it is out of
+            // reach, and an `IN` list is out of reach because the planner expands
+            // it into a chain of `OR`s. Splitting first means a query like
+            // `d_dom between 1 and 2 AND d_year IN (1999, 2000, 2001)` no longer
+            // falls back to the default selectivity for all of it: TPC-DS
+            // estimates that at 20% of `date_dim`, or 14,610 rows, where 72
+            // survive.
+            let (supported, rest): (Vec<_>, Vec<_>) = split_conjunction(predicate)
+                .into_iter()
+                .partition(|conjunct| check_support(conjunct, schema));
+            let split_anything = !rest.is_empty();
+
+            // Selectivity of the conjuncts interval analysis cannot see. An
+            // unrecognized one still contributes the default, once, as the whole
+            // predicate used to.
+            let mut unanalyzed_selectivity = 1.0;
+            let mut has_unknown_conjunct = false;
+            for conjunct in rest {
+                match in_list_selectivity(
+                    conjunct,
+                    &input_stats.column_statistics,
+                    &input_num_rows,
+                ) {
+                    Some(selectivity) => unanalyzed_selectivity *= selectivity,
+                    None => has_unknown_conjunct = true,
+                }
+            }
+            if has_unknown_conjunct {
+                unanalyzed_selectivity *= default_selectivity as f64 / 100.0;
+            }
+
+            // Rebuilding the conjunction re-associates it, and interval
+            // propagation is sensitive to the shape of the tree it walks, so the
+            // predicate is passed through untouched unless something was actually
+            // split off it. It must never be passed through when something was:
+            // interval analysis rejects the parts that were split off, and asking
+            // it to walk them anyway is an error rather than a fallback.
+            let analyzable = if split_anything {
+                conjunction_opt(supported.into_iter().cloned())
+            } else {
+                Some(Arc::clone(predicate))
+            };
+            if let Some(analyzable) = analyzable {
                 let input_analysis_ctx = AnalysisContext::try_from_statistics(
                     schema,
                     &input_stats.column_statistics,
                 )?;
-                let analysis_ctx = analyze(predicate, input_analysis_ctx, schema)?;
-                let selectivity = analysis_ctx.selectivity.unwrap_or(1.0);
+                let analysis_ctx = analyze(&analyzable, input_analysis_ctx, schema)?;
+                let selectivity =
+                    analysis_ctx.selectivity.unwrap_or(1.0) * unanalyzed_selectivity;
                 let filtered_num_rows =
                     input_num_rows.with_estimated_selectivity(selectivity);
-                let cs = collect_new_statistics(
+                let mut cs = collect_new_statistics(
                     schema,
                     &input_stats.column_statistics,
                     analysis_ctx.boundaries,
                     selectivity,
                     &null_rejecting_columns,
                     filtered_num_rows,
                 );
+                // A column pinned by an equality that was split off has one
+                // distinct value left, which interval analysis of the remaining
+                // conjuncts cannot know.
+                for idx in &eq_columns {
+                    if let Some(col_stat) = cs.get_mut(*idx)
+                        && col_stat.distinct_count != Precision::Exact(1)
+                    {
+                        col_stat.distinct_count =
+                            distinct_count_for_singleton_domain(filtered_num_rows);
+                    }
+                }
                 (selectivity, filtered_num_rows, cs)
             } else {
-                // Without interval boundaries, use the default selectivity and
-                // apply the row-count constraints that still follow from the
-                // filter predicate.
-                let selectivity = default_selectivity as f64 / 100.0;
+                // Nothing to derive boundaries from, so keep the input's value
+                // statistics and apply only the row-count constraints that still
+                // follow from the filter predicate.
+                let selectivity = unanalyzed_selectivity;
                 let filtered_num_rows =
                     input_num_rows.with_estimated_selectivity(selectivity);
                 let mut cs = input_stats.to_inexact().column_statistics;
@@ -959,6 +1015,91 @@ impl EmbeddedProjection for FilterExec {
 ///
 /// Only AND conjunctions are traversed; OR is intentionally skipped
 /// since `a = 1 OR a = 2` does not pin NDV to 1.
+/// Estimates the selectivity of `col IN (a, b, c)`, including the chain of
+/// `OR`ed equalities the planner expands a short list into.
+///
+/// Interval arithmetic cannot narrow a column from a disjunction, so without this
+/// such a conjunct would contribute nothing but the default selectivity. Here the
+/// fraction of values the list selects is `distinct literals / distinct values`,
+/// the same reasoning `col = literal` gets from `1 / NDV`.
+///
+/// Returns `None` when the conjunct is not a list of literals over one column.
+fn in_list_selectivity(
+    conjunct: &Arc<dyn PhysicalExpr>,
+    column_statistics: &[ColumnStatistics],
+    num_rows: &Precision<usize>,
+) -> Option<f64> {
+    let mut column = None;
+    let mut values = HashSet::new();
+    if let Some(in_list) = conjunct.downcast_ref::<InListExpr>() {
+        if in_list.negated() {
+            // `NOT IN` selects the complement, which is usually most of the
+            // column, and is left to the default rather than guessed at.
+            return None;
+        }
+        column = Some(in_list.expr().downcast_ref::<Column>()?);
+        for value in in_list.list() {
+            values.insert(value.downcast_ref::<Literal>()?.value().clone());
+        }
+    } else {
+        // Only a disjunction, so that a plain `col = literal` keeps whatever
+        // estimate it gets today rather than being re-derived here.
+        let binary = conjunct.downcast_ref::<BinaryExpr>()?;
+        if *binary.op() != Operator::Or {
+            return None;
+        }
+        collect_or_equalities(conjunct, &mut column, &mut values)?;
+    }
+
+    let column = column?;
+    if values.len() < 2 {
+        return None;
+    }
+    let stats = column_statistics.get(column.index())?;
+    let distinct = *max_distinct_count(num_rows, stats).get_value()?;
+    if distinct == 0 {
+        return None;
+    }
+    Some((values.len() as f64 / distinct as f64).min(1.0))
+}
+
+/// Collects the literals of an `OR` chain of equalities over a single column.
+///
+/// Returns `None` as soon as the expression is anything else, so a mixed
+/// disjunction such as `a = 1 OR b = 2` is not mistaken for a list.
+fn collect_or_equalities<'a>(
+    expr: &'a Arc<dyn PhysicalExpr>,
+    column: &mut Option<&'a Column>,
+    values: &mut HashSet<ScalarValue>,
+) -> Option<()> {
+    let binary = expr.downcast_ref::<BinaryExpr>()?;
+    match binary.op() {
+        Operator::Or => {
+            collect_or_equalities(binary.left(), column, values)?;
+            collect_or_equalities(binary.right(), column, values)
+        }
+        Operator::Eq => {
+            let (found, literal) = match (
+                binary.left().downcast_ref::<Column>(),
+                binary.right().downcast_ref::<Column>(),
+            ) {
+                (Some(col), None) => (col, binary.right()),
+                (None, Some(col)) => (col, binary.left()),
+                _ => return None,
+            };
+            // Every equality has to constrain the same column, or the fraction
+            // below would not describe the conjunct.
+            if column.is_some_and(|current| current != found) {
+                return None;
+            }
+            *column = Some(found);
+            values.insert(literal.downcast_ref::<Literal>()?.value().clone());
+            Some(())
+        }
+        _ => None,
+    }
+}
+
 fn collect_equality_columns(predicate: &Arc<dyn PhysicalExpr>) -> (HashSet<usize>, bool) {
     let mut eq_values: HashMap<usize, ScalarValue> = HashMap::new();
     let mut infeasible = false;
@@ -2747,9 +2888,10 @@ mod tests {
                         Arc::new(Literal::new(ScalarValue::Utf8(Some("b".to_string())))),
                     )),
                 )),
-                // Input NDV is 50, but the 20% default selectivity on 100 rows
-                // estimates 20 output rows, so NDV is capped at 20.
-                vec![Precision::Inexact(20)],
+                // The two listed values are 2 of the column's 50, so 4 of the
+                // 100 input rows are expected and NDV is capped at 4. Still not
+                // collapsed to 1, which is what this case guards.
+                vec![Precision::Inexact(4)],
             ),
             (
                 "AND with mixed types (Utf8 + Int32)",
 
@@ -934,10 +934,12 @@ logical_plan
 05)--SubqueryAlias: d
 06)----TableScan: department projection=[dept_name]
 physical_plan
-01)CrossJoinExec
-02)--FilterExec: name@1 = Alice OR name@1 = Bob
+01)ProjectionExec: expr=[emp_id@1 as emp_id, name@2 as name, dept_name@0 as dept_name]
+02)--CrossJoinExec
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
-04)--DataSourceExec: partitions=1, partition_sizes=[1]
+04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+05)------FilterExec: name@1 = Alice OR name@1 = Bob
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 # expect no row for Carol
 query ITT
@@ -947,10 +949,10 @@ JOIN department AS d
 ON (e.name = 'Alice' OR e.name = 'Bob');
 ----
 1 Alice HR
-1 Alice Engineering
-1 Alice Sales
 2 Bob HR
+1 Alice Engineering
 2 Bob Engineering
+1 Alice Sales
 2 Bob Sales
 
 # OR conditions on Filter (not join filter)