fix merge conflict

Cargo.toml

@@ -23,6 +23,7 @@ members = [
     "datafusion/core",
     "datafusion/expr",
     "datafusion/execution",
+    "datafusion/functions-aggregate",
     "datafusion/functions",
     "datafusion/functions-array",
     "datafusion/optimizer",
@@ -49,7 +50,7 @@ homepage = "https://github.com/apache/arrow-datafusion"
 license = "Apache-2.0"
 readme = "README.md"
 repository = "https://github.com/apache/arrow-datafusion"
-rust-version = "1.72"
+rust-version = "1.73"
 version = "37.0.0"
 
 [workspace.dependencies]
@@ -78,6 +79,7 @@ datafusion-common-runtime = { path = "datafusion/common-runtime", version = "37.
 datafusion-execution = { path = "datafusion/execution", version = "37.0.0" }
 datafusion-expr = { path = "datafusion/expr", version = "37.0.0" }
 datafusion-functions = { path = "datafusion/functions", version = "37.0.0" }
+datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "37.0.0" }
 datafusion-functions-array = { path = "datafusion/functions-array", version = "37.0.0" }
 datafusion-optimizer = { path = "datafusion/optimizer", version = "37.0.0", default-features = false }
 datafusion-physical-expr = { path = "datafusion/physical-expr", version = "37.0.0", default-features = false }

datafusion-cli/Cargo.toml

@@ -26,7 +26,7 @@ license = "Apache-2.0"
 homepage = "https://github.com/apache/arrow-datafusion"
 repository = "https://github.com/apache/arrow-datafusion"
 # Specify MSRV here as `cargo msrv` doesn't support workspace version
-rust-version = "1.72"
+rust-version = "1.73"
 readme = "README.md"
 
 [dependencies]

datafusion-cli/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-FROM rust:1.72-bullseye as builder
+FROM rust:1.73-bullseye as builder
 
 COPY . /usr/src/arrow-datafusion
 COPY ./datafusion /usr/src/arrow-datafusion/datafusion

datafusion/common/src/dfschema.rs

@@ -319,7 +319,7 @@ impl DFSchema {
         &self,
         qualifier: Option<&TableReference>,
         name: &str,
-    ) -> Result<Option<usize>> {
+    ) -> Option<usize> {
         let mut matches = self
             .iter()
             .enumerate()
@@ -345,19 +345,19 @@ impl DFSchema {
                 (None, Some(_)) | (None, None) => f.name() == name,
             })
             .map(|(idx, _)| idx);
-        Ok(matches.next())
+        matches.next()
     }
 
     /// Find the index of the column with the given qualifier and name
     pub fn index_of_column(&self, col: &Column) -> Result<usize> {
-        self.index_of_column_by_name(col.relation.as_ref(), &col.name)?
+        self.index_of_column_by_name(col.relation.as_ref(), &col.name)
             .ok_or_else(|| field_not_found(col.relation.clone(), &col.name, self))
     }
 
     /// Check if the column is in the current schema
-    pub fn is_column_from_schema(&self, col: &Column) -> Result<bool> {
+    pub fn is_column_from_schema(&self, col: &Column) -> bool {
         self.index_of_column_by_name(col.relation.as_ref(), &col.name)
-            .map(|idx| idx.is_some())
+            .is_some()
     }
 
     /// Find the field with the given name
@@ -381,7 +381,7 @@ impl DFSchema {
     ) -> Result<(Option<&TableReference>, &Field)> {
         if let Some(qualifier) = qualifier {
             let idx = self
-                .index_of_column_by_name(Some(qualifier), name)?
+                .index_of_column_by_name(Some(qualifier), name)
                 .ok_or_else(|| field_not_found(Some(qualifier.clone()), name, self))?;
             Ok((self.field_qualifiers[idx].as_ref(), self.field(idx)))
         } else {
@@ -519,7 +519,7 @@ impl DFSchema {
         name: &str,
     ) -> Result<&Field> {
         let idx = self
-            .index_of_column_by_name(Some(qualifier), name)?
+            .index_of_column_by_name(Some(qualifier), name)
             .ok_or_else(|| field_not_found(Some(qualifier.clone()), name, self))?;
 
         Ok(self.field(idx))
@@ -1190,11 +1190,8 @@ mod tests {
                 .to_string(),
             expected_help
         );
-        assert!(schema.index_of_column_by_name(None, "y").unwrap().is_none());
-        assert!(schema
-            .index_of_column_by_name(None, "t1.c0")
-            .unwrap()
-            .is_none());
+        assert!(schema.index_of_column_by_name(None, "y").is_none());
+        assert!(schema.index_of_column_by_name(None, "t1.c0").is_none());
 
         Ok(())
     }
@@ -1284,28 +1281,28 @@ mod tests {
         {
             let col = Column::from_qualified_name("t1.c0");
             let schema = DFSchema::try_from_qualified_schema("t1", &test_schema_1())?;
-            assert!(schema.is_column_from_schema(&col)?);
+            assert!(schema.is_column_from_schema(&col));
         }
 
         // qualified not exists
         {
             let col = Column::from_qualified_name("t1.c2");
             let schema = DFSchema::try_from_qualified_schema("t1", &test_schema_1())?;
-            assert!(!schema.is_column_from_schema(&col)?);
+            assert!(!schema.is_column_from_schema(&col));
         }
 
         // unqualified exists
         {
             let col = Column::from_name("c0");
             let schema = DFSchema::try_from_qualified_schema("t1", &test_schema_1())?;
-            assert!(schema.is_column_from_schema(&col)?);
+            assert!(schema.is_column_from_schema(&col));
         }
 
         // unqualified not exists
         {
             let col = Column::from_name("c2");
             let schema = DFSchema::try_from_qualified_schema("t1", &test_schema_1())?;
-            assert!(!schema.is_column_from_schema(&col)?);
+            assert!(!schema.is_column_from_schema(&col));
         }
 
         Ok(())

datafusion/common/src/tree_node.rs

@@ -25,10 +25,9 @@ use crate::Result;
 /// These macros are used to determine continuation during transforming traversals.
 macro_rules! handle_transform_recursion {
     ($F_DOWN:expr, $F_CHILD:expr, $F_UP:expr) => {{
-        #[allow(clippy::redundant_closure_call)]
         $F_DOWN?
             .transform_children(|n| n.map_children($F_CHILD))?
-            .transform_parent(|n| $F_UP(n))
+            .transform_parent($F_UP)
     }};
 }
 

datafusion/core/Cargo.toml

@@ -30,7 +30,7 @@ authors = { workspace = true }
 # Specify MSRV here as `cargo msrv` doesn't support workspace version and fails with
 # "Unable to find key 'package.rust-version' (or 'package.metadata.msrv') in 'arrow-datafusion/Cargo.toml'"
 # https://github.com/foresterre/cargo-msrv/issues/590
-rust-version = "1.72"
+rust-version = "1.73"
 
 [lib]
 name = "datafusion"
@@ -98,6 +98,7 @@ datafusion-common-runtime = { workspace = true }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
 datafusion-functions = { workspace = true }
+datafusion-functions-aggregate = { workspace = true }
 datafusion-functions-array = { workspace = true, optional = true }
 datafusion-optimizer = { workspace = true }
 datafusion-physical-expr = { workspace = true }

datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs

@@ -232,12 +232,8 @@ impl PruningStatistics for BloomFilterStatistics {
                     ScalarValue::Float32(Some(v)) => sbbf.check(v),
                     ScalarValue::Int64(Some(v)) => sbbf.check(v),
                     ScalarValue::Int32(Some(v)) => sbbf.check(v),
-                    ScalarValue::Int16(Some(v)) => sbbf.check(v),
-                    ScalarValue::Int8(Some(v)) => sbbf.check(v),
                     ScalarValue::UInt64(Some(v)) => sbbf.check(v),
                     ScalarValue::UInt32(Some(v)) => sbbf.check(v),
-                    ScalarValue::UInt16(Some(v)) => sbbf.check(v),
-                    ScalarValue::UInt8(Some(v)) => sbbf.check(v),
                     ScalarValue::Decimal128(Some(v), p, s) => match parquet_type {
                         Type::INT32 => {
                             //https://github.com/apache/parquet-format/blob/eb4b31c1d64a01088d02a2f9aefc6c17c54cc6fc/Encodings.md?plain=1#L35-L42

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

@@ -44,6 +44,7 @@ use crate::{
     datasource::{provider_as_source, MemTable, TableProvider, ViewTable},
     error::{DataFusionError, Result},
     execution::{options::ArrowReadOptions, runtime_env::RuntimeEnv, FunctionRegistry},
+    logical_expr::AggregateUDF,
     logical_expr::{
         CreateCatalog, CreateCatalogSchema, CreateExternalTable, CreateFunction,
         CreateMemoryTable, CreateView, DropCatalogSchema, DropFunction, DropTable,
@@ -53,10 +54,11 @@ use crate::{
     optimizer::analyzer::{Analyzer, AnalyzerRule},
     optimizer::optimizer::{Optimizer, OptimizerConfig, OptimizerRule},
     physical_optimizer::optimizer::{PhysicalOptimizer, PhysicalOptimizerRule},
-    physical_plan::{udaf::AggregateUDF, udf::ScalarUDF, ExecutionPlan},
+    physical_plan::{udf::ScalarUDF, ExecutionPlan},
     physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner},
     variable::{VarProvider, VarType},
 };
+use crate::{functions, functions_aggregate, functions_array};
 
 use arrow::datatypes::{DataType, SchemaRef};
 use arrow::record_batch::RecordBatch;
@@ -65,18 +67,15 @@ use datafusion_common::{
     alias::AliasGenerator,
     config::{ConfigExtension, TableOptions},
     exec_err, not_impl_err, plan_datafusion_err, plan_err,
-    tree_node::{TreeNode, TreeNodeRecursion, TreeNodeVisitor},
+    tree_node::{TreeNodeRecursion, TreeNodeVisitor},
     SchemaReference, TableReference,
 };
 use datafusion_execution::registry::SerializerRegistry;
-use datafusion_expr::type_coercion::aggregates::NUMERICS;
-use datafusion_expr::{create_first_value, Signature, Volatility};
 use datafusion_expr::{
     logical_plan::{DdlStatement, Statement},
     var_provider::is_system_variables,
     Expr, StringifiedPlan, UserDefinedLogicalNode, WindowUDF,
 };
-use datafusion_physical_expr::create_first_value_accumulator;
 use datafusion_sql::{
     parser::{CopyToSource, CopyToStatement, DFParser},
     planner::{object_name_to_table_reference, ContextProvider, ParserOptions, SqlToRel},
@@ -85,7 +84,6 @@ use datafusion_sql::{
 
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
-use log::debug;
 use parking_lot::RwLock;
 use sqlparser::dialect::dialect_from_str;
 use url::Url;
@@ -1452,29 +1450,16 @@ impl SessionState {
         };
 
         // register built in functions
-        datafusion_functions::register_all(&mut new_self)
+        functions::register_all(&mut new_self)
             .expect("can not register built in functions");
 
         // register crate of array expressions (if enabled)
         #[cfg(feature = "array_expressions")]
-        datafusion_functions_array::register_all(&mut new_self)
+        functions_array::register_all(&mut new_self)
             .expect("can not register array expressions");
 
-        let first_value = create_first_value(
-            "FIRST_VALUE",
-            Signature::uniform(1, NUMERICS.to_vec(), Volatility::Immutable),
-            Arc::new(create_first_value_accumulator),
-        );
-
-        match new_self.register_udaf(Arc::new(first_value)) {
-            Ok(Some(existing_udaf)) => {
-                debug!("Overwrite existing UDAF: {}", existing_udaf.name());
-            }
-            Ok(None) => {}
-            Err(err) => {
-                panic!("Failed to register UDAF: {}", err);
-            }
-        }
+        functions_aggregate::register_all(&mut new_self)
+            .expect("can not register aggregate functions");
 
         new_self
     }
@@ -2313,7 +2298,7 @@ impl SQLOptions {
     /// Return an error if the [`LogicalPlan`] has any nodes that are
     /// incompatible with this [`SQLOptions`].
     pub fn verify_plan(&self, plan: &LogicalPlan) -> Result<()> {
-        plan.visit(&mut BadPlanVisitor::new(self))?;
+        plan.visit_with_subqueries(&mut BadPlanVisitor::new(self))?;
         Ok(())
     }
 }

datafusion/core/src/lib.rs

@@ -541,6 +541,11 @@ pub mod functions_array {
     pub use datafusion_functions_array::*;
 }
 
+/// re-export of [`datafusion_functions_aggregate`] crate
+pub mod functions_aggregate {
+    pub use datafusion_functions_aggregate::*;
+}
+
 #[cfg(test)]
 pub mod test;
 pub mod test_util;

datafusion/core/src/physical_optimizer/pruning.rs

@@ -330,7 +330,7 @@ pub trait PruningStatistics {
 /// `x = 5` | `CASE WHEN x_null_count = x_row_count THEN false ELSE x_min <= 5 AND 5 <= x_max END`
 /// `x < 5` | `CASE WHEN x_null_count = x_row_count THEN false ELSE x_max < 5 END`
 /// `x = 5 AND y = 10` | `CASE WHEN x_null_count = x_row_count THEN false ELSE x_min <= 5 AND 5 <= x_max END AND CASE WHEN y_null_count = y_row_count THEN false ELSE y_min <= 10 AND 10 <= y_max END`
-/// `x IS NULL`  | `CASE WHEN x_null_count = x_row_count THEN false ELSE x_null_count > 0 END`
+/// `x IS NULL`  | `x_null_count > 0`
 /// `CAST(x as int) = 5` | `CASE WHEN x_null_count = x_row_count THEN false ELSE CAST(x_min as int) <= 5 AND 5 <= CAST(x_max as int) END`
 ///
 /// ## Predicate Evaluation