[SPARK-54199][SQL] Add DataFrame API support for new KLL quantiles sketch functions

### What changes were proposed in this pull request?

This PR adds DataFrame API support for the KLL quantile sketch functions that were previously added to Spark SQL in #52800. This lets users leverage KLL sketches through both Scala and Python DataFrame APIs in addition to the existing SQL interface.

**Key additions:**

1. **Scala DataFrame API** (`sql/api/src/main/scala/org/apache/spark/sql/functions.scala`):
   - 18 new functions covering aggregate, merge, quantile, and rank operations
   - Multiple overloads for each function supporting:
     - `Column` parameters for computed values
     - `String` parameters for column names
     - `Int` parameters for literal k values
     - Optional k parameters with sensible defaults
   - Functions for all three data type variants: bigint, float, double

2. **Python DataFrame API** (`python/pyspark/sql/functions/builtin.py`):
   - 18 corresponding Python functions with:
     - Comprehensive docstrings with usage examples
     - Proper type hints (`ColumnOrName`, `Optional[Union[int, Column]]`)
     - Support for both column objects and column name strings
   - Added to PySpark documentation reference

3. **Python Spark Connect Support** (`python/pyspark/sql/connect/functions/builtin.py`):
   - Full compatibility with Spark Connect architecture
   - All 18 functions properly registered

### Why are the changes needed?

While the SQL API for KLL sketches was previously added, DataFrame API support is essential for full usability. Without DataFrame API support, users would be forced to use SQL expressions via `expr()` or `selectExpr()`, which is less ergonomic and type-safe.

### Does this PR introduce any user-facing change?

Yes, this PR adds DataFrame API support for the 18 KLL sketch functions:

**Scala DataFrame API Example:**
```scala
import org.apache.spark.sql.functions._

// Create sketch with default k
val df = Seq(1, 2, 3, 4, 5).toDF("value")
val sketch = df.agg(kll_sketch_agg_bigint($"value"))

// Create sketch with custom k value
val sketch2 = df.agg(kll_sketch_agg_bigint("value", 400))

// Get median (0.5 quantile)
val sketchDf = df.agg(kll_sketch_agg_bigint($"value").alias("sketch"))
val median = sketchDf.select(kll_sketch_get_quantile_bigint($"sketch", lit(0.5)))

// Get multiple quantiles
val quantiles = sketchDf.select(
  kll_sketch_get_quantile_bigint($"sketch", array(lit(0.25), lit(0.5), lit(0.75)))
)

// Merge sketches
val merged = sketchDf.select(
  kll_sketch_merge_bigint($"sketch", $"sketch").alias("merged")
)

// Get count of items
val count = sketchDf.select(kll_sketch_get_n_bigint($"sketch"))
```

**Python DataFrame API Example:**
```python
from pyspark.sql import functions as sf

# Create sketch with default k
df = spark.createDataFrame([1, 2, 3, 4, 5], "INT")
sketch = df.agg(sf.kll_sketch_agg_bigint("value"))

# Create sketch with custom k value
sketch2 = df.agg(sf.kll_sketch_agg_bigint("value", 400))

# Get median (0.5 quantile)
sketch_df = df.agg(sf.kll_sketch_agg_bigint("value").alias("sketch"))
median = sketch_df.select(sf.kll_sketch_get_quantile_bigint("sketch", sf.lit(0.5)))

# Get multiple quantiles
quantiles = sketch_df.select(
    sf.kll_sketch_get_quantile_bigint("sketch", sf.array(sf.lit(0.25), sf.lit(0.5), sf.lit(0.75)))
)

# Merge sketches
merged = sketch_df.select(
    sf.kll_sketch_merge_bigint("sketch", "sketch").alias("merged")
)

# Get count of items
count = sketch_df.select(sf.kll_sketch_get_n_bigint("sketch"))
```

### How was this patch tested?

1. **Scala Unit Tests** (`DataFrameAggregateSuite`):
   - `kll_sketch_agg_{bigint,float,double}` with default and explicit k values
   - `kll_sketch_to_string` functions for all data types
   - `kll_sketch_get_n` functions for all data types
   - `kll_sketch_merge` operations
   - `kll_sketch_get_quantile` with single rank and array of ranks
   - `kll_sketch_get_rank` operations
   - Null value handling tests

2. **Python Unit Tests** (`test_functions.py`):
   - Comprehensive tests mirroring Scala tests
   - Tests for Column object and string column name overloads
   - Tests for optional k parameter
   - Array input tests for quantile/rank functions
   - Null handling validation
   - Type checking (bytes/bytearray for sketches, str for to_string, int/float for values)

### Was this patch authored or co-authored using generative AI tooling?

Yes, IDE assistance used `claude-4.5-sonnet` with manual validation and integration.

Closes #52900 from dtenedor/dataframe-api-kll-functions.

Authored-by: Daniel Tenedorio <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>

Author: dtenedor

python/docs/source/reference/pyspark.sql/functions.rst

@@ -458,6 +458,9 @@ Aggregate Functions
     histogram_numeric
     hll_sketch_agg
     hll_union_agg
+    kll_sketch_agg_bigint
+    kll_sketch_agg_double
+    kll_sketch_agg_float
     kurtosis
     last
     last_value
@@ -631,6 +634,21 @@ Misc Functions
     current_user
     hll_sketch_estimate
     hll_union
+    kll_sketch_get_n_bigint
+    kll_sketch_get_n_double
+    kll_sketch_get_n_float
+    kll_sketch_get_quantile_bigint
+    kll_sketch_get_quantile_double
+    kll_sketch_get_quantile_float
+    kll_sketch_get_rank_bigint
+    kll_sketch_get_rank_double
+    kll_sketch_get_rank_float
+    kll_sketch_merge_bigint
+    kll_sketch_merge_double
+    kll_sketch_merge_float
+    kll_sketch_to_string_bigint
+    kll_sketch_to_string_double
+    kll_sketch_to_string_float
     input_file_block_length
     input_file_block_start
     input_file_name

python/pyspark/sql/connect/functions/builtin.py

@@ -4529,6 +4529,168 @@ def theta_intersection_agg(
 theta_intersection_agg.__doc__ = pysparkfuncs.theta_intersection_agg.__doc__
 
 
+def kll_sketch_agg_bigint(
+    col: "ColumnOrName",
+    k: Optional[Union[int, Column]] = None,
+) -> Column:
+    fn = "kll_sketch_agg_bigint"
+    if k is None:
+        return _invoke_function_over_columns(fn, col)
+    else:
+        return _invoke_function_over_columns(fn, col, lit(k))
+
+
+kll_sketch_agg_bigint.__doc__ = pysparkfuncs.kll_sketch_agg_bigint.__doc__
+
+
+def kll_sketch_agg_float(
+    col: "ColumnOrName",
+    k: Optional[Union[int, Column]] = None,
+) -> Column:
+    fn = "kll_sketch_agg_float"
+    if k is None:
+        return _invoke_function_over_columns(fn, col)
+    else:
+        return _invoke_function_over_columns(fn, col, lit(k))
+
+
+kll_sketch_agg_float.__doc__ = pysparkfuncs.kll_sketch_agg_float.__doc__
+
+
+def kll_sketch_agg_double(
+    col: "ColumnOrName",
+    k: Optional[Union[int, Column]] = None,
+) -> Column:
+    fn = "kll_sketch_agg_double"
+    if k is None:
+        return _invoke_function_over_columns(fn, col)
+    else:
+        return _invoke_function_over_columns(fn, col, lit(k))
+
+
+kll_sketch_agg_double.__doc__ = pysparkfuncs.kll_sketch_agg_double.__doc__
+
+
+def kll_sketch_to_string_bigint(col: "ColumnOrName") -> Column:
+    fn = "kll_sketch_to_string_bigint"
+    return _invoke_function_over_columns(fn, col)
+
+
+kll_sketch_to_string_bigint.__doc__ = pysparkfuncs.kll_sketch_to_string_bigint.__doc__
+
+
+def kll_sketch_to_string_float(col: "ColumnOrName") -> Column:
+    fn = "kll_sketch_to_string_float"
+    return _invoke_function_over_columns(fn, col)
+
+
+kll_sketch_to_string_float.__doc__ = pysparkfuncs.kll_sketch_to_string_float.__doc__
+
+
+def kll_sketch_to_string_double(col: "ColumnOrName") -> Column:
+    fn = "kll_sketch_to_string_double"
+    return _invoke_function_over_columns(fn, col)
+
+
+kll_sketch_to_string_double.__doc__ = pysparkfuncs.kll_sketch_to_string_double.__doc__
+
+
+def kll_sketch_get_n_bigint(col: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_n_bigint"
+    return _invoke_function_over_columns(fn, col)
+
+
+kll_sketch_get_n_bigint.__doc__ = pysparkfuncs.kll_sketch_get_n_bigint.__doc__
+
+
+def kll_sketch_get_n_float(col: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_n_float"
+    return _invoke_function_over_columns(fn, col)
+
+
+kll_sketch_get_n_float.__doc__ = pysparkfuncs.kll_sketch_get_n_float.__doc__
+
+
+def kll_sketch_get_n_double(col: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_n_double"
+    return _invoke_function_over_columns(fn, col)
+
+
+kll_sketch_get_n_double.__doc__ = pysparkfuncs.kll_sketch_get_n_double.__doc__
+
+
+def kll_sketch_merge_bigint(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+    fn = "kll_sketch_merge_bigint"
+    return _invoke_function_over_columns(fn, left, right)
+
+
+kll_sketch_merge_bigint.__doc__ = pysparkfuncs.kll_sketch_merge_bigint.__doc__
+
+
+def kll_sketch_merge_float(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+    fn = "kll_sketch_merge_float"
+    return _invoke_function_over_columns(fn, left, right)
+
+
+kll_sketch_merge_float.__doc__ = pysparkfuncs.kll_sketch_merge_float.__doc__
+
+
+def kll_sketch_merge_double(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+    fn = "kll_sketch_merge_double"
+    return _invoke_function_over_columns(fn, left, right)
+
+
+kll_sketch_merge_double.__doc__ = pysparkfuncs.kll_sketch_merge_double.__doc__
+
+
+def kll_sketch_get_quantile_bigint(sketch: "ColumnOrName", rank: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_quantile_bigint"
+    return _invoke_function_over_columns(fn, sketch, rank)
+
+
+kll_sketch_get_quantile_bigint.__doc__ = pysparkfuncs.kll_sketch_get_quantile_bigint.__doc__
+
+
+def kll_sketch_get_quantile_float(sketch: "ColumnOrName", rank: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_quantile_float"
+    return _invoke_function_over_columns(fn, sketch, rank)
+
+
+kll_sketch_get_quantile_float.__doc__ = pysparkfuncs.kll_sketch_get_quantile_float.__doc__
+
+
+def kll_sketch_get_quantile_double(sketch: "ColumnOrName", rank: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_quantile_double"
+    return _invoke_function_over_columns(fn, sketch, rank)
+
+
+kll_sketch_get_quantile_double.__doc__ = pysparkfuncs.kll_sketch_get_quantile_double.__doc__
+
+
+def kll_sketch_get_rank_bigint(sketch: "ColumnOrName", quantile: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_rank_bigint"
+    return _invoke_function_over_columns(fn, sketch, quantile)
+
+
+kll_sketch_get_rank_bigint.__doc__ = pysparkfuncs.kll_sketch_get_rank_bigint.__doc__
+
+
+def kll_sketch_get_rank_float(sketch: "ColumnOrName", quantile: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_rank_float"
+    return _invoke_function_over_columns(fn, sketch, quantile)
+
+
+kll_sketch_get_rank_float.__doc__ = pysparkfuncs.kll_sketch_get_rank_float.__doc__
+
+
+def kll_sketch_get_rank_double(sketch: "ColumnOrName", quantile: "ColumnOrName") -> Column:
+    fn = "kll_sketch_get_rank_double"
+    return _invoke_function_over_columns(fn, sketch, quantile)
+
+
+kll_sketch_get_rank_double.__doc__ = pysparkfuncs.kll_sketch_get_rank_double.__doc__
+
+
 def theta_sketch_estimate(col: "ColumnOrName") -> Column:
     fn = "theta_sketch_estimate"
     return _invoke_function_over_columns(fn, col)

python/pyspark/sql/functions/__init__.py

@@ -372,6 +372,9 @@
     "histogram_numeric",
     "hll_sketch_agg",
     "hll_union_agg",
+    "kll_sketch_agg_bigint",
+    "kll_sketch_agg_double",
+    "kll_sketch_agg_float",
     "kurtosis",
     "last",
     "last_value",
@@ -495,6 +498,21 @@
     "input_file_block_start",
     "input_file_name",
     "java_method",
+    "kll_sketch_get_n_bigint",
+    "kll_sketch_get_n_double",
+    "kll_sketch_get_n_float",
+    "kll_sketch_get_quantile_bigint",
+    "kll_sketch_get_quantile_double",
+    "kll_sketch_get_quantile_float",
+    "kll_sketch_get_rank_bigint",
+    "kll_sketch_get_rank_double",
+    "kll_sketch_get_rank_float",
+    "kll_sketch_merge_bigint",
+    "kll_sketch_merge_double",
+    "kll_sketch_merge_float",
+    "kll_sketch_to_string_bigint",
+    "kll_sketch_to_string_double",
+    "kll_sketch_to_string_float",
     "monotonically_increasing_id",
     "raise_error",
     "reflect",