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

#[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?;