feat: add approx_top_k aggregate function

datafusion/core/tests/dataframe/dataframe_functions.rs

@@ -32,7 +32,9 @@ use datafusion_common::test_util::batches_to_string;
 use datafusion_common::{DFSchema, ScalarValue};
 use datafusion_expr::expr::Alias;
 use datafusion_expr::{ExprSchemable, LogicalPlanBuilder, table_scan};
-use datafusion_functions_aggregate::expr_fn::{approx_median, approx_percentile_cont};
+use datafusion_functions_aggregate::expr_fn::{
+    approx_median, approx_percentile_cont, approx_top_k,
+};
 use datafusion_functions_nested::map::map;
 use insta::assert_snapshot;
 
@@ -409,6 +411,27 @@ async fn test_fn_approx_median() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_fn_approx_top_k() -> Result<()> {
+    // Column b has values [1, 10, 10, 100] -- 10 appears twice, others once.
+    let expr = approx_top_k(vec![col("b"), lit(2)]);
+
+    let df = create_test_table().await?;
+    let batches = df.aggregate(vec![], vec![expr]).unwrap().collect().await?;
+
+    assert_snapshot!(
+        batches_to_string(&batches),
+        @r"
+    +-----------------------------------------------+
+    | approx_top_k(test.b,Int32(2))                 |
+    +-----------------------------------------------+
+    | [{value: 10, count: 2}, {value: 1, count: 1}] |
+    +-----------------------------------------------+
+    ");
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_fn_approx_percentile_cont() -> Result<()> {
     let expr = approx_percentile_cont(col("b").sort(true, false), lit(0.5), None);