feat: add approx_top_k aggregate function

Add a new approx_top_k(expression, k) aggregate function that returns
the approximate top-k most frequent values with their estimated counts,
using the Filtered Space-Saving algorithm.

The implementation uses a capacity multiplier of 3 (matching ClickHouse's
default) and includes an alpha map for improved accuracy by filtering
low-frequency noise before it enters the main summary.

Return type is List(Struct({value: T, count: UInt64})) ordered by count
descending, where T matches the input column type.

Closes #20967

Author: sesteves

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,28 @@ 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.
+    // Use k=1 to avoid non-deterministic ordering among tied items.
+    let expr = approx_top_k(vec![col("b"), lit(1)]);
+
+    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(1)) |
+    +-------------------------------+
+    | [{value: 10, count: 2}]       |
+    +-------------------------------+
+    ");
+
+    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);