Fix doc placement, import path, extract helper, defer selectivity computation

- Fix expression_analyzer_registry doc comment misplaced between
  function_factory's doc comment and field declaration
- Fix module doc example import path (physical_plan -> physical_expr)
- Extract expression_analyzer_registry() helper in planner to avoid
  repeating the config check 4 times
- Defer left_sel/right_sel computation to AND/OR arms only, avoiding
  unnecessary sub-expression selectivity estimation for comparison
  operators

Author: asolimando

datafusion/core/src/execution/session_state.rs

@@ -186,12 +186,12 @@ pub struct SessionState {
     table_factories: HashMap<String, Arc<dyn TableProviderFactory>>,
     /// Runtime environment
     runtime_env: Arc<RuntimeEnv>,
+    /// Registry for expression-level statistics analyzers (NDV, selectivity, etc.)
+    expression_analyzer_registry: Arc<ExpressionAnalyzerRegistry>,
     /// [FunctionFactory] to support pluggable user defined function handler.
     ///
     /// It will be invoked on `CREATE FUNCTION` statements.
     /// thus, changing dialect o PostgreSql is required
-    /// Registry for expression-level statistics analyzers (NDV, selectivity, etc.)
-    expression_analyzer_registry: Arc<ExpressionAnalyzerRegistry>,
     function_factory: Option<Arc<dyn FunctionFactory>>,
     cache_factory: Option<Arc<dyn CacheFactory>>,
     /// Cache logical plans of prepared statements for later execution.

datafusion/core/src/physical_planner.rs

@@ -56,6 +56,7 @@ use crate::physical_plan::{
     WindowExpr, displayable, windows,
 };
 use crate::schema_equivalence::schema_satisfied_by;
+use datafusion_physical_expr::expression_analyzer::ExpressionAnalyzerRegistry;
 
 use arrow::array::{RecordBatch, builder::StringBuilder};
 use arrow::compute::SortOptions;
@@ -1110,17 +1111,11 @@ impl DefaultPhysicalPlanner {
                             physical_input,
                         )
                         .with_batch_size(session_state.config().batch_size());
-                        let builder = if session_state
-                            .config_options()
-                            .optimizer
-                            .enable_expression_analyzer
-                        {
-                            builder.with_expression_analyzer_registry(Arc::clone(
-                                session_state.expression_analyzer_registry(),
-                            ))
-                        } else {
-                            builder
-                        };
+                        let builder =
+                            match Self::expression_analyzer_registry(session_state) {
+                                Some(r) => builder.with_expression_analyzer_registry(r),
+                                None => builder,
+                            };
                         builder.build()?
                     }
                     PlanAsyncExpr::Async(
@@ -1141,17 +1136,11 @@ impl DefaultPhysicalPlanner {
                             (0..input.schema().fields().len()).collect::<Vec<_>>(),
                         ))?
                         .with_batch_size(session_state.config().batch_size());
-                        let builder = if session_state
-                            .config_options()
-                            .optimizer
-                            .enable_expression_analyzer
-                        {
-                            builder.with_expression_analyzer_registry(Arc::clone(
-                                session_state.expression_analyzer_registry(),
-                            ))
-                        } else {
-                            builder
-                        };
+                        let builder =
+                            match Self::expression_analyzer_registry(session_state) {
+                                Some(r) => builder.with_expression_analyzer_registry(r),
+                                None => builder,
+                            };
                         builder.build()?
                     }
                     _ => {
@@ -2861,6 +2850,17 @@ impl DefaultPhysicalPlanner {
         Ok(mem_exec)
     }
 
+    /// Returns the expression analyzer registry if the config option is enabled.
+    fn expression_analyzer_registry(
+        session_state: &SessionState,
+    ) -> Option<Arc<ExpressionAnalyzerRegistry>> {
+        session_state
+            .config_options()
+            .optimizer
+            .enable_expression_analyzer
+            .then(|| Arc::clone(session_state.expression_analyzer_registry()))
+    }
+
     fn create_project_physical_exec(
         &self,
         session_state: &SessionState,
@@ -2921,14 +2921,8 @@ impl DefaultPhysicalPlanner {
                     .map(|(expr, alias)| ProjectionExpr { expr, alias })
                     .collect();
                 let mut proj_exec = ProjectionExec::try_new(proj_exprs, input_exec)?;
-                if session_state
-                    .config_options()
-                    .optimizer
-                    .enable_expression_analyzer
-                {
-                    proj_exec = proj_exec.with_expression_analyzer_registry(Arc::clone(
-                        session_state.expression_analyzer_registry(),
-                    ));
+                if let Some(r) = Self::expression_analyzer_registry(session_state) {
+                    proj_exec = proj_exec.with_expression_analyzer_registry(r);
                 }
                 Ok(Arc::new(proj_exec))
             }

datafusion/physical-expr/src/expression_analyzer/default.rs

@@ -72,18 +72,34 @@ impl ExpressionAnalyzer for DefaultExpressionAnalyzer {
     ) -> AnalysisResult<f64> {
         // Binary expressions: AND, OR, comparisons
         if let Some(binary) = expr.as_any().downcast_ref::<BinaryExpr>() {
-            let left_sel =
-                self.estimate_selectivity_recursive(binary.left(), input_stats, registry);
-            let right_sel = self.estimate_selectivity_recursive(
-                binary.right(),
-                input_stats,
-                registry,
-            );
-
             let sel = match binary.op() {
-                // Logical operators
-                Operator::And => left_sel * right_sel,
-                Operator::Or => left_sel + right_sel - (left_sel * right_sel),
+                // Logical operators: need child selectivities
+                Operator::And => {
+                    let left_sel = self.estimate_selectivity_recursive(
+                        binary.left(),
+                        input_stats,
+                        registry,
+                    );
+                    let right_sel = self.estimate_selectivity_recursive(
+                        binary.right(),
+                        input_stats,
+                        registry,
+                    );
+                    left_sel * right_sel
+                }
+                Operator::Or => {
+                    let left_sel = self.estimate_selectivity_recursive(
+                        binary.left(),
+                        input_stats,
+                        registry,
+                    );
+                    let right_sel = self.estimate_selectivity_recursive(
+                        binary.right(),
+                        input_stats,
+                        registry,
+                    );
+                    left_sel + right_sel - (left_sel * right_sel)
+                }
 
                 // Equality: selectivity = 1/NDV
                 Operator::Eq => {

datafusion/physical-expr/src/expression_analyzer/mod.rs

@@ -40,7 +40,7 @@
 //! # Example
 //!
 //! ```ignore
-//! use datafusion_physical_plan::expression_analyzer::*;
+//! use datafusion_physical_expr::expression_analyzer::*;
 //!
 //! // Create registry with default analyzer
 //! let mut registry = ExpressionAnalyzerRegistry::new();