Merge remote-tracking branch 'apache/main' into alamb/tablefactory

Cargo.toml

@@ -31,6 +31,7 @@ members = [
     "datafusion/functions-aggregate-common",
     "datafusion/functions-nested",
     "datafusion/functions-window",
+    "datafusion/functions-window-common",
     "datafusion/optimizer",
     "datafusion/physical-expr",
     "datafusion/physical-expr-common",
@@ -103,6 +104,7 @@ datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", vers
 datafusion-functions-aggregate-common = { path = "datafusion/functions-aggregate-common", version = "42.0.0" }
 datafusion-functions-nested = { path = "datafusion/functions-nested", version = "42.0.0" }
 datafusion-functions-window = { path = "datafusion/functions-window", version = "42.0.0" }
+datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "42.0.0" }
 datafusion-optimizer = { path = "datafusion/optimizer", version = "42.0.0", default-features = false }
 datafusion-physical-expr = { path = "datafusion/physical-expr", version = "42.0.0", default-features = false }
 datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "42.0.0", default-features = false }
@@ -167,4 +169,3 @@ large_futures = "warn"
 
 [workspace.lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ["cfg(tarpaulin)"] }
-unused_imports = "deny"

datafusion-cli/src/main.rs

@@ -232,7 +232,7 @@ async fn main_inner() -> Result<()> {
 }
 
 fn create_runtime_env(rn_config: RuntimeConfig) -> Result<RuntimeEnv> {
-    RuntimeEnv::new(rn_config)
+    RuntimeEnv::try_new(rn_config)
 }
 
 fn parse_valid_file(dir: &str) -> Result<String, String> {

datafusion-examples/examples/advanced_udwf.rs

@@ -22,9 +22,11 @@ use arrow::{
     array::{ArrayRef, AsArray, Float64Array},
     datatypes::Float64Type,
 };
+use arrow_schema::Field;
 use datafusion::error::Result;
 use datafusion::prelude::*;
 use datafusion_common::ScalarValue;
+use datafusion_expr::function::WindowUDFFieldArgs;
 use datafusion_expr::{
     PartitionEvaluator, Signature, WindowFrame, WindowUDF, WindowUDFImpl,
 };
@@ -70,16 +72,15 @@ impl WindowUDFImpl for SmoothItUdf {
         &self.signature
     }
 
-    /// What is the type of value that will be returned by this function.
-    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
-        Ok(DataType::Float64)
-    }
-
     /// Create a `PartitionEvaluator` to evaluate this function on a new
     /// partition.
     fn partition_evaluator(&self) -> Result<Box<dyn PartitionEvaluator>> {
         Ok(Box::new(MyPartitionEvaluator::new()))
     }
+
+    fn field(&self, field_args: WindowUDFFieldArgs) -> Result<Field> {
+        Ok(Field::new(field_args.name(), DataType::Float64, true))
+    }
 }
 
 /// This implements the lowest level evaluation for a window function

datafusion-examples/examples/external_dependency/query-aws-s3.rs

@@ -64,7 +64,7 @@ async fn main() -> Result<()> {
     df.show().await?;
 
     // dynamic query by the file path
-    ctx.enable_url_table();
+    let ctx = ctx.enable_url_table();
     let df = ctx
         .sql(format!(r#"SELECT * FROM '{}' LIMIT 10"#, &path).as_str())
         .await?;

datafusion-examples/examples/optimizer_rule.rs

@@ -100,6 +100,7 @@ pub async fn main() -> Result<()> {
 
 /// An example OptimizerRule that replaces all `col = <const>` predicates with a
 /// user defined function
+#[derive(Default, Debug)]
 struct MyOptimizerRule {}
 
 impl OptimizerRule for MyOptimizerRule {

datafusion-examples/examples/simplify_udwf_expression.rs

@@ -17,12 +17,12 @@
 
 use std::any::Any;
 
-use arrow_schema::DataType;
+use arrow_schema::{DataType, Field};
 
 use datafusion::execution::context::SessionContext;
 use datafusion::functions_aggregate::average::avg_udaf;
 use datafusion::{error::Result, execution::options::CsvReadOptions};
-use datafusion_expr::function::WindowFunctionSimplification;
+use datafusion_expr::function::{WindowFunctionSimplification, WindowUDFFieldArgs};
 use datafusion_expr::{
     expr::WindowFunction, simplify::SimplifyInfo, Expr, PartitionEvaluator, Signature,
     Volatility, WindowUDF, WindowUDFImpl,
@@ -60,10 +60,6 @@ impl WindowUDFImpl for SimplifySmoothItUdf {
         &self.signature
     }
 
-    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
-        Ok(DataType::Float64)
-    }
-
     fn partition_evaluator(&self) -> Result<Box<dyn PartitionEvaluator>> {
         todo!()
     }
@@ -84,6 +80,10 @@ impl WindowUDFImpl for SimplifySmoothItUdf {
 
         Some(Box::new(simplify))
     }
+
+    fn field(&self, field_args: WindowUDFFieldArgs) -> Result<Field> {
+        Ok(Field::new(field_args.name(), DataType::Float64, true))
+    }
 }
 
 // create local execution context with `cars.csv` registered as a table named `cars`

datafusion/common/src/dfschema.rs

@@ -355,18 +355,7 @@ impl DFSchema {
                 // qualifier and name.
                 (Some(q), Some(field_q)) => q.resolved_eq(field_q) && f.name() == name,
                 // field to lookup is qualified but current field is unqualified.
-                (Some(qq), None) => {
-                    // the original field may now be aliased with a name that matches the
-                    // original qualified name
-                    let column = Column::from_qualified_name(f.name());
-                    match column {
-                        Column {
-                            relation: Some(r),
-                            name: column_name,
-                        } => &r == qq && column_name == name,
-                        _ => false,
-                    }
-                }
+                (Some(_), None) => false,
                 // field to lookup is unqualified, no need to compare qualifier
                 (None, Some(_)) | (None, None) => f.name() == name,
             })

datafusion/common/src/display/mod.rs

@@ -27,7 +27,7 @@ use std::{
 
 /// Represents which type of plan, when storing multiple
 /// for use in EXPLAIN plans
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
 pub enum PlanType {
     /// The initial LogicalPlan provided to DataFusion
     InitialLogicalPlan,
@@ -96,7 +96,7 @@ impl Display for PlanType {
 }
 
 /// Represents some sort of execution plan, in String form
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
 pub struct StringifiedPlan {
     /// An identifier of what type of plan this string represents
     pub plan_type: PlanType,

datafusion/common/src/functional_dependencies.rs

@@ -30,7 +30,7 @@ use crate::{DFSchema, DFSchemaRef, DataFusionError, JoinType, Result};
 use sqlparser::ast::TableConstraint;
 
 /// This object defines a constraint on a table.
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
 pub enum Constraint {
     /// Columns with the given indices form a composite primary key (they are
     /// jointly unique and not nullable):
@@ -40,7 +40,7 @@ pub enum Constraint {
 }
 
 /// This object encapsulates a list of functional constraints:
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
 pub struct Constraints {
     inner: Vec<Constraint>,
 }

datafusion/common/src/join_type.rs

@@ -26,7 +26,7 @@ use crate::error::_not_impl_err;
 use crate::{DataFusionError, Result};
 
 /// Join type
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
 pub enum JoinType {
     /// Inner Join
     Inner,
@@ -88,7 +88,7 @@ impl FromStr for JoinType {
 }
 
 /// Join constraint
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
 pub enum JoinConstraint {
     /// Join ON
     On,

datafusion/common/src/test_util.rs

@@ -29,6 +29,27 @@ use std::{error::Error, path::PathBuf};
 /// Expects to be called about like this:
 ///
 /// `assert_batch_eq!(expected_lines: &[&str], batches: &[RecordBatch])`
+///
+/// # Example
+/// ```
+/// # use std::sync::Arc;
+/// # use arrow::record_batch::RecordBatch;
+/// # use arrow_array::{ArrayRef, Int32Array};
+/// # use datafusion_common::assert_batches_eq;
+/// let col: ArrayRef = Arc::new(Int32Array::from(vec![1, 2]));
+///  let batch = RecordBatch::try_from_iter([("column", col)]).unwrap();
+/// // Expected output is a vec of strings
+/// let expected = vec![
+///     "+--------+",
+///     "| column |",
+///     "+--------+",
+///     "| 1      |",
+///     "| 2      |",
+///     "+--------+",
+/// ];
+/// // compare the formatted output of the record batch with the expected output
+/// assert_batches_eq!(expected, &[batch]);
+/// ```
 #[macro_export]
 macro_rules! assert_batches_eq {
     ($EXPECTED_LINES: expr, $CHUNKS: expr) => {
@@ -56,8 +77,7 @@ macro_rules! assert_batches_eq {
 /// vector of strings in a way that order does not matter.
 /// This is a macro so errors appear on the correct line
 ///
-/// Designed so that failure output can be directly copy/pasted
-/// into the test code as expected results.
+/// See [`assert_batches_eq`] for more details and example.
 ///
 /// Expects to be called about like this:
 ///

datafusion/core/Cargo.toml

@@ -145,6 +145,7 @@ bigdecimal = { workspace = true }
 criterion = { version = "0.5", features = ["async_tokio"] }
 csv = "1.1.6"
 ctor = { workspace = true }
+datafusion-functions-window-common = { workspace = true }
 doc-comment = { workspace = true }
 env_logger = { workspace = true }
 half = { workspace = true, default-features = true }