How to evaluate function's args one by one / implement short circuit evaluation

[haohuaijin]

When working on #8927, I find when evaluating a scalar function, we first evaluate all args and then execute the actual function; see below
https://github.com/apache/arrow-datafusion/blob/827668ab3d4659329ff30e6672f398c934b803be/datafusion/physical-expr/src/scalar_function.rs#L152-L161

But for a function like coalesce, it is a short-circuit function; we should evaluate its args one by one(like we do for CASE), otherwise, the below query will get the error:

select coalesce(1, 4/0);

---

To do this, I have two ideas

First

change the ScalarFunctionImplementation

Before
pub type ScalarFunctionImplementation =
    Arc<dyn Fn(&[ColumnarValue]) -> Result<ColumnarValue> + Send + Sync>;

After
pub type ScalarFunctionImplementation =
    Arc<dyn Fn(&[Arc<dyn PhysicalExpr], &RecordBatch) -> Result<ColumnarValue> + Send + Sync>;

Second

treat coalesce as a specific case in https://github.com/apache/arrow-datafusion/blob/827668ab3d4659329ff30e6672f398c934b803be/datafusion/physical-expr/src/scalar_function.rs#L152-L156
and implement coalesce like below

            (_, Ok(scalar_fun)) if scalar_fun == BuiltinScalarFunction::Coalesce => {
                return _coalesce(self.args.as_ref(), batch);
            }

Third

add a function to identify if the function is a short-circuit function, and create the function with different args for non-short-circuit and short-circuit functions(create_physical_expr), then add two functions to ScalarUDFImpl that make use can create themself short-circuit the function. just like

pub trait ScalarUDFImpl: Debug + Send + Sync {
......
    fn short_circuit(&self) -> bool {
        false
    }

    fn invoke_short_circuit(&self, args: &[Arc<dyn PhysicalExpr>], batch: &RecordBatch) -> Result<ColumnarValue>;
}

[haohuaijin]

cc @alamb @jackwener

[jackwener]

In my opinion, the overall solution is reasonable.

But I'm not sure about details like the design of the interface.

[alamb]

I agree this is a very reasonable usecase to try and support

One challenge is likely to be that PhysicalExpr is defined in a different crate and I don't think we can easily change ScalarFunctionImplementation without introducing a cyclic dependency

Maybe we can take advantage of having a trait for defining scalar functions now, ScalarUDFImpl: https://github.com/apache/arrow-datafusion/blob/bc0ba6a724aaaf312b770451718cbce696de6640/datafusion/expr/src/udf.rs#L237

For example, perhaps we could provide a version of invoke like invoke_raw or invoke_short_circut that got a callback it could use to get the argument values

Something like this perhaps 🤔

pub trait ScalarUDFImpl: Debug + Send + Sync {

    // existing API in terms of columnar value
    fn invoke(&self, args: &[ColumnarValue]) -> Result<ColumnarValue> {
       // calls new API invoke_short_circuit
    }

    /// Evaluate the function without first evaluating arguments. 
    /// if the function needs values of its arguments, it calls some 
    /// method on `ArgEvaluator`
    fn invoke_short_circuit(&self, args: &dyn ArgEvaluator)  {
    ...
    }
  
}

[GhostBoyBoy]

I think we can use SQL annotations to implement it, like this: -- set short circuited=true