Async UDF returning a scalar fails with "all columns in a record batch must have the same length"

[hareshkh]

Describe the bug

If an async UDF returns ColumnarValue::Scalar (e.g. all arguments are literals), queries over more than one row fail with all columns in a record batch must have the same length. AsyncFuncExpr::invoke_with_args converts the scalar with to_array_of_size(1) instead of the number of rows.

To Reproduce

An async UDF that returns its argument, called with a literal on a table with more than one row:

#[derive(Debug, PartialEq, Eq, Hash)]
struct EchoUdf {
    signature: Signature,
}

impl ScalarUDFImpl for EchoUdf {
    fn name(&self) -> &str { "echo" }
    fn signature(&self) -> &Signature { &self.signature }
    fn return_type(&self, _: &[DataType]) -> Result<DataType> { Ok(DataType::Utf8) }
    fn invoke_with_args(&self, _: ScalarFunctionArgs) -> Result<ColumnarValue> { unreachable!() }
}

#[async_trait]
impl AsyncScalarUDFImpl for EchoUdf {
    async fn invoke_async_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
        // A literal argument arrives as a scalar, so this returns a scalar
        Ok(args.args[0].clone())
    }
}

let ctx = SessionContext::new();
let udf = EchoUdf {
    signature: Signature::exact(vec![DataType::Utf8], Volatility::Volatile),
};
ctx.register_udf(AsyncScalarUDF::new(Arc::new(udf)).into_scalar_udf());
ctx.sql("SELECT echo('constant') FROM (VALUES (1), (2), (3)) t(x)")
    .await?
    .show()
    .await?;

Expected behavior

The scalar result applies to every row, so the query returns three rows of constant, the same as a non-async UDF returning a scalar.

Additional context

No response

[hareshkh]

take