Merge remote-tracking branch 'apache/main' into arrow-avro

Author: alamb

Cargo.lock

@@ -175,9 +175,10 @@ fn simplify_demo() -> Result<()> {
     // the ExecutionProps carries information needed to simplify
     // expressions, such as the current time (to evaluate `now()`
     // correctly)
-    let context = SimplifyContext::default()
+    let context = SimplifyContext::builder()
         .with_schema(schema)
-        .with_current_time();
+        .with_current_time()
+        .build();
     let simplifier = ExprSimplifier::new(context);
 
     // And then call the simplify_expr function:
@@ -192,9 +193,10 @@ fn simplify_demo() -> Result<()> {
 
     // here are some other examples of what DataFusion is capable of
     let schema = Schema::new(vec![make_field("i", DataType::Int64)]).to_dfschema_ref()?;
-    let context = SimplifyContext::default()
+    let context = SimplifyContext::builder()
         .with_schema(Arc::clone(&schema))
-        .with_current_time();
+        .with_current_time()
+        .build();
     let simplifier = ExprSimplifier::new(context);
 
     // basic arithmetic simplification
@@ -554,9 +556,10 @@ fn type_coercion_demo() -> Result<()> {
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     // 2. Type coercion with `ExprSimplifier::coerce`.
-    let context = SimplifyContext::default()
+    let context = SimplifyContext::builder()
         .with_schema(Arc::new(df_schema.clone()))
-        .with_current_time();
+        .with_current_time()
+        .build();
     let simplifier = ExprSimplifier::new(context);
     let coerced_expr = simplifier.coerce(expr.clone(), &df_schema)?;
     let physical_expr = datafusion::physical_expr::create_physical_expr(

datafusion-examples/examples/query_planning/expr_api.rs

@@ -84,10 +84,10 @@ pub trait TableProvider: Debug + Sync + Send {
         None
     }
 
-    /// Create an [`ExecutionPlan`] for scanning the table with optionally
-    /// specified `projection`, `filter` and `limit`, described below.
+    /// Create an [`ExecutionPlan`] for scanning the table with optional
+    /// `projection`, `filter`, and `limit`, described below.
     ///
-    /// The `ExecutionPlan` is responsible scanning the datasource's
+    /// The returned `ExecutionPlan` is responsible for scanning the datasource's
     /// partitions in a streaming, parallelized fashion.
     ///
     /// # Projection
@@ -96,33 +96,30 @@ pub trait TableProvider: Debug + Sync + Send {
     /// specified. The projection is a set of indexes of the fields in
     /// [`Self::schema`].
     ///
-    /// DataFusion provides the projection to scan only the columns actually
-    /// used in the query to improve performance, an optimization  called
-    /// "Projection Pushdown". Some datasources, such as Parquet, can use this
-    /// information to go significantly faster when only a subset of columns is
-    /// required.
+    /// DataFusion provides the projection so the scan reads only the columns
+    /// actually used in the query, an optimization called "Projection
+    /// Pushdown". Some datasources, such as Parquet, can use this information
+    /// to go significantly faster when only a subset of columns is required.
     ///
     /// # Filters
     ///
     /// A list of boolean filter [`Expr`]s to evaluate *during* the scan, in the
     /// manner specified by [`Self::supports_filters_pushdown`]. Only rows for
-    /// which *all* of the `Expr`s evaluate to `true` must be returned (aka the
-    /// expressions are `AND`ed together).
+    /// which *all* of the `Expr`s evaluate to `true` must be returned (that is,
+    /// the expressions are `AND`ed together).
     ///
-    /// To enable filter pushdown you must override
-    /// [`Self::supports_filters_pushdown`] as the default implementation does
-    /// not and `filters` will be empty.
+    /// To enable filter pushdown, override
+    /// [`Self::supports_filters_pushdown`]. The default implementation does not
+    /// push down filters, and `filters` will be empty.
     ///
-    /// DataFusion pushes filtering into the scans whenever possible
-    /// ("Filter Pushdown"), and depending on the format and the
-    /// implementation of the format, evaluating the predicate during the scan
-    /// can increase performance significantly.
+    /// DataFusion pushes filters into scans whenever possible ("Filter
+    /// Pushdown"). Depending on the data format and implementation, evaluating
+    /// predicates during the scan can significantly improve performance.
     ///
     /// ## Note: Some columns may appear *only* in Filters
     ///
-    /// In certain cases, a query may only use a certain column in a Filter that
-    /// has been completely pushed down to the scan. In this case, the
-    /// projection will not contain all the columns found in the filter
+    /// In some cases, a query may use a column only in a filter and the
+    /// projection will not contain all columns referenced by the filter
     /// expressions.
     ///
     /// For example, given the query `SELECT t.a FROM t WHERE t.b > 5`,
@@ -154,15 +151,40 @@ pub trait TableProvider: Debug + Sync + Send {
     ///
     /// # Limit
     ///
-    /// If `limit` is specified,  must only produce *at least* this many rows,
-    /// (though it may return more).  Like Projection Pushdown and Filter
-    /// Pushdown, DataFusion pushes `LIMIT`s  as far down in the plan as
-    /// possible, called "Limit Pushdown" as some sources can use this
-    /// information to improve their performance. Note that if there are any
-    /// Inexact filters pushed down, the LIMIT cannot be pushed down. This is
-    /// because inexact filters do not guarantee that every filtered row is
-    /// removed, so applying the limit could lead to too few rows being available
-    /// to return as a final result.
+    /// If `limit` is specified, the scan must produce *at least* this many
+    /// rows, though it may return more. Like Projection Pushdown and Filter
+    /// Pushdown, DataFusion pushes `LIMIT`s as far down in the plan as
+    /// possible. This is called "Limit Pushdown", and some sources can use the
+    /// information to improve performance.
+    ///
+    /// Note: If any pushed-down filters are `Inexact`, the `LIMIT` cannot be
+    /// pushed down. Inexact filters do not guarantee that every filtered row is
+    /// removed, so applying the limit could leave too few rows to return in the
+    /// final result.
+    ///
+    /// # Evaluation Order
+    ///
+    /// The logical evaluation order is `filters`, then `limit`, then
+    /// `projection`.
+    ///
+    /// Note that `limit` applies to the filtered result, not to the unfiltered
+    /// input, and `projection` affects only which columns are returned, not
+    /// which rows qualify.
+    ///
+    /// For example, if a scan receives:
+    ///
+    /// - `projection = [a]`
+    /// - `filters = [b > 5]`
+    /// - `limit = Some(3)`
+    ///
+    /// It must logically produce results equivalent to:
+    ///
+    /// ```text
+    /// PROJECTION a (LIMIT 3 (SCAN WHERE b > 5))
+    /// ```
+    ///
+    /// As noted above, columns referenced only by pushed-down filters may be
+    /// absent from `projection`.
     async fn scan(
         &self,
         state: &dyn Session,

datafusion/catalog/src/table.rs

@@ -1488,12 +1488,13 @@ impl SessionContext {
         })?;
 
         let state = self.state.read();
-        let context = SimplifyContext::default()
+        let context = SimplifyContext::builder()
             .with_schema(Arc::clone(prepared.plan.schema()))
             .with_config_options(Arc::clone(state.config_options()))
             .with_query_execution_start_time(
                 state.execution_props().query_execution_start_time,
-            );
+            )
+            .build();
         let simplifier = ExprSimplifier::new(context);
 
         // Only allow literals as parameters for now.

datafusion/core/src/execution/context/mod.rs

@@ -743,12 +743,13 @@ impl SessionState {
         df_schema: &DFSchema,
     ) -> datafusion_common::Result<Arc<dyn PhysicalExpr>> {
         let config_options = self.config_options();
-        let simplify_context = SimplifyContext::default()
+        let simplify_context = SimplifyContext::builder()
             .with_schema(Arc::new(df_schema.clone()))
             .with_config_options(Arc::clone(config_options))
             .with_query_execution_start_time(
                 self.execution_props().query_execution_start_time,
-            );
+            )
+            .build();
         let simplifier = ExprSimplifier::new(simplify_context);
         // apply type coercion here to ensure types match
         let mut expr = simplifier.coerce(expr, df_schema)?;
@@ -1835,11 +1836,12 @@ impl ContextProvider for SessionContextProvider<'_> {
             .get(name)
             .cloned()
             .ok_or_else(|| plan_datafusion_err!("table function '{name}' not found"))?;
-        let simplify_context = SimplifyContext::default()
+        let simplify_context = SimplifyContext::builder()
             .with_config_options(Arc::clone(self.state.config_options()))
             .with_query_execution_start_time(
                 self.state.execution_props().query_execution_start_time,
-            );
+            )
+            .build();
         let simplifier = ExprSimplifier::new(simplify_context);
         let schema = DFSchema::empty();
         let args = args

datafusion/core/src/execution/session_state.rs

@@ -166,7 +166,9 @@ impl TestParquetFile {
         let df_schema = Arc::clone(&self.schema).to_dfschema_ref()?;
 
         // run coercion on the filters to coerce types etc.
-        let context = SimplifyContext::default().with_schema(Arc::clone(&df_schema));
+        let context = SimplifyContext::builder()
+            .with_schema(Arc::clone(&df_schema))
+            .build();
         if let Some(filter) = maybe_filter {
             let simplifier = ExprSimplifier::new(context);
             let filter = simplifier.coerce(filter, &df_schema).unwrap();