Fixes missing nth_value UDAF expr function

datafusion/functions-aggregate/src/lib.rs

@@ -113,6 +113,7 @@ pub mod expr_fn {
     pub use super::median::median;
     pub use super::min_max::max;
     pub use super::min_max::min;
+    pub use super::nth_value::nth_value;
     pub use super::regr::regr_avgx;
     pub use super::regr::regr_avgy;
     pub use super::regr::regr_count;

datafusion/functions-aggregate/src/nth_value.rs

@@ -30,19 +30,33 @@ use datafusion_common::{exec_err, internal_err, not_impl_err, Result, ScalarValu
 use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs};
 use datafusion_expr::utils::format_state_name;
 use datafusion_expr::{
-    Accumulator, AggregateUDFImpl, ReversedUDAF, Signature, Volatility,
+    lit, Accumulator, AggregateUDFImpl, ExprFunctionExt, ReversedUDAF, Signature,
+    SortExpr, Volatility,
 };
 use datafusion_functions_aggregate_common::merge_arrays::merge_ordered_arrays;
 use datafusion_functions_aggregate_common::utils::ordering_fields;
 use datafusion_physical_expr::expressions::Literal;
 use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
 
-make_udaf_expr_and_func!(
-    NthValueAgg,
-    nth_value,
-    "Returns the nth value in a group of values.",
-    nth_value_udaf
-);
+create_func!(NthValueAgg, nth_value_udaf);
+
+/// Returns the nth value in a group of values.
+pub fn nth_value(
+    expr: datafusion_expr::Expr,
+    n: i64,
+    order_by: Vec<SortExpr>,
+) -> datafusion_expr::Expr {
+    let args = vec![expr, lit(n)];
+    if !order_by.is_empty() {
+        nth_value_udaf()
+            .call(args)
+            .order_by(order_by)
+            .build()
+            .unwrap()
+    } else {
+        nth_value_udaf().call(args)
+    }
+}
 
 /// Expression for a `NTH_VALUE(... ORDER BY ..., ...)` aggregation. In a multi
 /// partition setting, partial aggregations are computed for every partition,

datafusion/proto/tests/cases/roundtrip_logical_plan.rs

@@ -71,6 +71,7 @@ use datafusion_expr::{
 use datafusion_functions_aggregate::average::avg_udaf;
 use datafusion_functions_aggregate::expr_fn::{
     approx_distinct, array_agg, avg, bit_and, bit_or, bit_xor, bool_and, bool_or, corr,
+    nth_value,
 };
 use datafusion_functions_aggregate::string_agg::string_agg;
 use datafusion_proto::bytes::{
@@ -903,6 +904,18 @@ async fn roundtrip_expr_api() -> Result<()> {
             vec![lit(10), lit(20), lit(30)],
         ),
         row_number(),
+        nth_value(col("b"), 1, vec![]),
+        nth_value(
+            col("b"),
+            1,
+            vec![col("a").sort(false, false), col("b").sort(true, false)],
+        ),
+        nth_value(col("b"), -1, vec![]),
+        nth_value(
+            col("b"),
+            -1,
+            vec![col("a").sort(false, false), col("b").sort(true, false)],
+        ),
     ];
 
     // ensure expressions created with the expr api can be round tripped