WindowFunction works with SQL but not with DataFrame

[aguspdana]

Query with WindowFunction return NotImplemented error, but the equivalent SQL query return the correct result.

use datafusion::error::Result;    
use datafusion::prelude::*;    
use datafusion::physical_plan::window_functions; 
   
#[tokio::main]    
async fn main() -> Result<()> {    
    let mut ctx = sigmo_df::utils::create_ctx("log", vec![
        sigmo_df::column::Column::String("id", vec![    
            Some("a"),    
            Some("b"),    
            Some("c"),    
            Some("c")    
        ]),    
        sigmo_df::column::Column::Numeric("value", vec![    
            Some(1.0),    
            Some(2.0),    
            Some(3.0),    
            Some(4.0)    
        ])    
    ])?;    
    
    // SQL                                                                   
    let res = ctx.sql("
        SELECT                                                               
            id,
            first_value(value) over (partition by id)
        FROM log;
    ")?;
    println!("SQL Plan:\n{}", res.to_logical_plan().display());
    res.show().await?;

    // DataFrame
    let df = ctx.table("log")?;
    let first_row = datafusion::logical_plan::Expr::WindowFunction {
        fun: window_functions::WindowFunction::BuiltInWindowFunction(
            window_functions::BuiltInWindowFunction::FirstValue
        ),
        args: vec![col("value")],
        partition_by: vec![col("id")],
        order_by: vec![],
        window_frame: None
    };
    let query = df
       .select(vec![col("id"), first_row])
        .unwrap();                          
    println!("\nDataFrame Plan:\n{}", query.to_logical_plan().display());
    query.show().await?;

    Ok(())
}

SQL Plan:
Projection: #log.id, #FIRST_VALUE(log.value) PARTITION BY [#log.id]
+----+------------------------+
| id | FIRST_VALUE(log.value) |
+----+------------------------+
| a  | 1                      |
| b  | 2                      |
| c  | 3                      |
| c  | 3                      |
+----+------------------------+

DataFrame Plan:
Projection: #log.id, FIRST_VALUE(#log.value) PARTITION BY [#log.id]
Error: NotImplemented("Physical plan does not support logical expression FIRST_VALUE(#log.value) PARTITION BY [#log.id]")

I've also tried using datafusion::physical_plan::aggregates::AggregateFunction::Sum, and it also returned NotImplemented error.

The error maybe triggered at https://github.com/apache/arrow-datafusion/blob/master/datafusion/src/physical_plan/planner.rs#L1105. But I'm not sure why the SQL query succeeded.

My project dependencies:

[dependencies]           
# arrow = "^5.1"
datafusion = { git = "https://github.com/apache/arrow-datafusion" }
tokio = { version = "1.12.0", features = ["full"] }
chrono = "0.4"

[jimexist]

yes i can confirm it was because in https://github.com/apache/arrow-datafusion/blob/4b577f374ce0922f61608be25d8d91c59a65c2cf/datafusion/src/execution/dataframe_impl.rs#L77-L82 we need to handle window expressions like in https://github.com/apache/arrow-datafusion/blob/4b577f374ce0922f61608be25d8d91c59a65c2cf/datafusion/src/sql/planner.rs#L842-L860 and it's not done yet

[xudong963]

Hi, @jimexist, are you fixing it? If not, I think I can help fix it. I'm getting familiar with the code of window functions.