[SPARK-53991][SQL] Add SQL support for KLL quantiles functions based on DataSketches by dtenedor · Pull Request #52800 · apache/spark

dtenedor · 2025-10-30T19:47:05Z

What changes were proposed in this pull request?

This PR adds support for KLL (K-Linear-Logarithmic) quantile sketches to Spark SQL, based on the Apache DataSketches KLL library. KLL sketches provide a compact, approximate representation of data distributions, enabling efficient quantile estimation and rank queries on large datasets with bounded memory usage and strong accuracy guarantees.

Jira: https://issues.apache.org/jira/browse/SPARK-53991.

It introduces 18 new SQL functions organized into six categories:

Aggregation Functions
Creates a KLL sketch from input values (k is optional).

kll_sketch_agg_bigint(col, k)
kll_sketch_agg_float(col, k)
kll_sketch_agg_double(col, k)

Sketch Inspection Functions
Returns a human-readable string representation for debugging purposes.

kll_sketch_to_string_bigint(sketch)
kll_sketch_to_string_float(sketch)
kll_sketch_to_string_double(sketch)

Sketch Merging Functions
Merges two compatible sketches.

kll_sketch_merge_bigint(sketch1, sketch2)
kll_sketch_merge_float(sketch1, sketch2)
kll_sketch_merge_double(sketch1, sketch2)

Quantile Estimation Functions
Estimates the value at a given rank, supporting both single rank values and arrays of ranks for batch quantile queries.

kll_sketch_get_quantile_bigint(sketch, rank)
kll_sketch_get_quantile_float(sketch, rank)
kll_sketch_get_quantile_double(sketch, rank)

Rank Estimation Functions
Estimates the rank of a given value, supporting both single values and arrays of values for batch rank queries.

kll_sketch_get_rank_bigint(sketch, value)
kll_sketch_get_rank_float(sketch, value)
kll_sketch_get_rank_double(sketch, value)

Sketch Item Count Functions
Counts the number of items collected in the sketch so far.

kll_sketch_get_n_bigint(sketch)
kll_sketch_get_n_float(sketch)
kll_sketch_get_n_double(sketch)

This PR only includes SQL language support; Dataframe API support will follow in a separate PR.

Key Features:

Type Safety: Separate implementations for BIGINT (covering TINYINT/SMALLINT/INT), FLOAT, and DOUBLE types ensure type-safe operations
Array Support: Quantile and rank functions accept arrays for efficient batch operations
Memory Efficient: Sketches are serialized to BINARY type for compact storage and efficient shuffling
NULL Handling: All aggregate functions properly ignore NULL input values, consistent with standard SQL aggregate behavior
Error Handling: Comprehensive validation with structured error messages for: invalid quantile ranges (must be 0.0-1.0), incompatible sketch merges, invalid binary sketch data, type mismatches

Why are the changes needed?

KLL sketches enable approximate quantile and rank queries on large datasets with:

O(1) space complexity - Bounded memory usage regardless of data size
High accuracy - Configurable error bounds with proven theoretical guarantees
Fast queries - O(log n) query time for quantile/rank estimation
Mergeable - Sketches can be combined for distributed aggregation

Use cases include:

Approximate median/percentile calculations on massive datasets
Distribution analysis for monitoring and analytics
SLA compliance checking (e.g., p95, p99 latency)
Efficient histogram generation

Does this PR introduce any user-facing change?

Yes, this PR introduces 15 new SQL functions available in Spark SQL.

How was this patch tested?

SQL Golden File Tests: Added kllquantiles.sql with test queries covering:

All three data types (BIGINT, FLOAT, DOUBLE)
Multiple input sizes (empty, single value, multiple values)
NULL value handling (verified NULLs are ignored)
Quantile estimation (single and array inputs)
Rank estimation (single and array inputs)
Sketch merging
Approximate result validation using tolerance-based comparisons
Negative tests for error conditions (invalid quantiles, type mismatches, incompatible merges)

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

Yes, code assistance with claude-4.5-sonnet in combination with manual editing by the author.

dtenedor · 2025-10-30T20:21:40Z

cc @mkaravel @cboumalh

cboumalh · 2025-10-30T20:28:56Z

Hi @dtenedor, will take the time to review this. Thanks for the work!

cboumalh · 2025-10-31T23:03:01Z

+
+ def kllSketchInvalidQuantileRangeError(function: String, quantile: Double): Throwable = {
+ new SparkRuntimeException(
+ errorClass = "KLL_SKETCH_INVALID_QUANTILE_RANGE",
+ messageParameters = Map(
+ "quantile" -> toSQLValue(quantile, DoubleType)))
+ }
+
+ def kllSketchInvalidInputError(function: String, reason: String): Throwable = {
+ new SparkRuntimeException(
+ errorClass = "KLL_SKETCH_INVALID_INPUT",
+ messageParameters = Map("reason" -> reason))
+ }
+
+ def kllSketchIncompatibleMergeError(function: String, reason: String): Throwable = {
+ new SparkRuntimeException(
+ errorClass = "KLL_SKETCH_INCOMPATIBLE_MERGE",
+ messageParameters = Map("reason" -> reason))
+ }
+
+ def kllSketchKMustBeConstantError(function: String): Throwable = {
+ new SparkRuntimeException(
+ errorClass = "KLL_SKETCH_K_MUST_BE_CONSTANT",
+ messageParameters = Map.empty)
+ }
+
+ def kllSketchKOutOfRangeError(function: String, k: Int): Throwable = {
+ new SparkRuntimeException(
+ errorClass = "KLL_SKETCH_K_OUT_OF_RANGE",
+ messageParameters = Map("k" -> toSQLValue(k, IntegerType)))
+ }


It may be beneficial to users to add this "function" -> toSQLId(function) in the messageParameters to know where the exception is coming from

Sure, sounds good, we're already passing in the function anyway. Done.

cboumalh · 2025-10-31T23:04:20Z

Just one more thing with QueryExecutionErrors and should be good on my end. Thank you again!

…sages

cboumalh · 2025-11-03T19:47:58Z

LGTM

cboumalh · 2025-11-03T19:50:17Z

Also this is the PR to check the foldable expression in HLL and Theta: #52836

dtenedor · 2025-11-04T20:17:23Z

The CI failure appears unrelated.

dtenedor · 2025-11-04T20:17:49Z

Thank you for your review, merging to master!

…etch 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 <daniel.tenedorio@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

…on DataSketches ### What changes were proposed in this pull request? This PR adds support for KLL (K-Linear-Logarithmic) quantile sketches to Spark SQL, based on the Apache DataSketches KLL library. KLL sketches provide a compact, approximate representation of data distributions, enabling efficient quantile estimation and rank queries on large datasets with bounded memory usage and strong accuracy guarantees. Jira: https://issues.apache.org/jira/browse/SPARK-53991. It introduces 18 new SQL functions organized into six categories: 1. Aggregation Functions Creates a KLL sketch from input values. ``` kll_sketch_agg_bigint(col) kll_sketch_agg_float(col) kll_sketch_agg_double(col) ``` 2. Sketch Inspection Functions Returns a human-readable string representation for debugging purposes. ``` kll_sketch_to_string_bigint(sketch) kll_sketch_to_string_float(sketch) kll_sketch_to_string_double(sketch) ``` 3. Sketch Merging Functions Merges two compatible sketches. ``` kll_sketch_merge_bigint(sketch1, sketch2) kll_sketch_merge_float(sketch1, sketch2) kll_sketch_merge_double(sketch1, sketch2) ``` 4. Quantile Estimation Functions Estimates the value at a given rank, supporting both single rank values and arrays of ranks for batch quantile queries. ``` kll_sketch_get_quantile_bigint(sketch, rank) kll_sketch_get_quantile_float(sketch, rank) kll_sketch_get_quantile_double(sketch, rank) ``` 5. Rank Estimation Functions Estimates the rank of a given value, supporting both single values and arrays of values for batch rank queries. ``` kll_sketch_get_rank_bigint(sketch, value) kll_sketch_get_rank_float(sketch, value) kll_sketch_get_rank_double(sketch, value) ``` 6. Sketch Item Count Functions Counts the number of items collected in the sketch so far. ``` kll_sketch_get_n_bigint(sketch) kll_sketch_get_n_float(sketch) kll_sketch_get_n_double(sketch) ``` This PR only includes SQL language support; Dataframe API support will follow in a separate PR. Key Features: * Type Safety: Separate implementations for BIGINT (covering TINYINT/SMALLINT/INT), FLOAT, and DOUBLE types ensure type-safe operations * Array Support: Quantile and rank functions accept arrays for efficient batch operations * Memory Efficient: Sketches are serialized to BINARY type for compact storage and efficient shuffling * NULL Handling: All aggregate functions properly ignore NULL input values, consistent with standard SQL aggregate behavior * Error Handling: Comprehensive validation with structured error messages for: invalid quantile ranges (must be 0.0-1.0), incompatible sketch merges, invalid binary sketch data, type mismatches ### Why are the changes needed? KLL sketches enable approximate quantile and rank queries on large datasets with: * O(1) space complexity - Bounded memory usage regardless of data size * High accuracy - Configurable error bounds with proven theoretical guarantees * Fast queries - O(log n) query time for quantile/rank estimation * Mergeable - Sketches can be combined for distributed aggregation Use cases include: * Approximate median/percentile calculations on massive datasets * Distribution analysis for monitoring and analytics * SLA compliance checking (e.g., p95, p99 latency) * Efficient histogram generation ### Does this PR introduce _any_ user-facing change? Yes, this PR introduces 15 new SQL functions available in Spark SQL. ### How was this patch tested? SQL Golden File Tests: Added `kllquantiles.sql` with test queries covering: * All three data types (BIGINT, FLOAT, DOUBLE) * Multiple input sizes (empty, single value, multiple values) * NULL value handling (verified NULLs are ignored) * Quantile estimation (single and array inputs) * Rank estimation (single and array inputs) * Sketch merging * Approximate result validation using tolerance-based comparisons * Negative tests for error conditions (invalid quantiles, type mismatches, incompatible merges) ### Was this patch authored or co-authored using generative AI tooling? Yes, code assistance with `claude-4.5-sonnet` in combination with manual editing by the author. Closes#52800 from dtenedor/kll-quantiles-functions. Authored-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

…etch 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 <daniel.tenedorio@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

…etch 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 apache#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. Closesapache#52900 from dtenedor/dataframe-api-kll-functions. Authored-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

…on DataSketches ### What changes were proposed in this pull request? This PR adds support for KLL (K-Linear-Logarithmic) quantile sketches to Spark SQL, based on the Apache DataSketches KLL library. KLL sketches provide a compact, approximate representation of data distributions, enabling efficient quantile estimation and rank queries on large datasets with bounded memory usage and strong accuracy guarantees. Jira: https://issues.apache.org/jira/browse/SPARK-53991. It introduces 18 new SQL functions organized into six categories: 1. Aggregation Functions Creates a KLL sketch from input values. ``` kll_sketch_agg_bigint(col) kll_sketch_agg_float(col) kll_sketch_agg_double(col) ``` 2. Sketch Inspection Functions Returns a human-readable string representation for debugging purposes. ``` kll_sketch_to_string_bigint(sketch) kll_sketch_to_string_float(sketch) kll_sketch_to_string_double(sketch) ``` 3. Sketch Merging Functions Merges two compatible sketches. ``` kll_sketch_merge_bigint(sketch1, sketch2) kll_sketch_merge_float(sketch1, sketch2) kll_sketch_merge_double(sketch1, sketch2) ``` 4. Quantile Estimation Functions Estimates the value at a given rank, supporting both single rank values and arrays of ranks for batch quantile queries. ``` kll_sketch_get_quantile_bigint(sketch, rank) kll_sketch_get_quantile_float(sketch, rank) kll_sketch_get_quantile_double(sketch, rank) ``` 5. Rank Estimation Functions Estimates the rank of a given value, supporting both single values and arrays of values for batch rank queries. ``` kll_sketch_get_rank_bigint(sketch, value) kll_sketch_get_rank_float(sketch, value) kll_sketch_get_rank_double(sketch, value) ``` 6. Sketch Item Count Functions Counts the number of items collected in the sketch so far. ``` kll_sketch_get_n_bigint(sketch) kll_sketch_get_n_float(sketch) kll_sketch_get_n_double(sketch) ``` This PR only includes SQL language support; Dataframe API support will follow in a separate PR. Key Features: * Type Safety: Separate implementations for BIGINT (covering TINYINT/SMALLINT/INT), FLOAT, and DOUBLE types ensure type-safe operations * Array Support: Quantile and rank functions accept arrays for efficient batch operations * Memory Efficient: Sketches are serialized to BINARY type for compact storage and efficient shuffling * NULL Handling: All aggregate functions properly ignore NULL input values, consistent with standard SQL aggregate behavior * Error Handling: Comprehensive validation with structured error messages for: invalid quantile ranges (must be 0.0-1.0), incompatible sketch merges, invalid binary sketch data, type mismatches ### Why are the changes needed? KLL sketches enable approximate quantile and rank queries on large datasets with: * O(1) space complexity - Bounded memory usage regardless of data size * High accuracy - Configurable error bounds with proven theoretical guarantees * Fast queries - O(log n) query time for quantile/rank estimation * Mergeable - Sketches can be combined for distributed aggregation Use cases include: * Approximate median/percentile calculations on massive datasets * Distribution analysis for monitoring and analytics * SLA compliance checking (e.g., p95, p99 latency) * Efficient histogram generation ### Does this PR introduce _any_ user-facing change? Yes, this PR introduces 15 new SQL functions available in Spark SQL. ### How was this patch tested? SQL Golden File Tests: Added `kllquantiles.sql` with test queries covering: * All three data types (BIGINT, FLOAT, DOUBLE) * Multiple input sizes (empty, single value, multiple values) * NULL value handling (verified NULLs are ignored) * Quantile estimation (single and array inputs) * Rank estimation (single and array inputs) * Sketch merging * Approximate result validation using tolerance-based comparisons * Negative tests for error conditions (invalid quantiles, type mismatches, incompatible merges) ### Was this patch authored or co-authored using generative AI tooling? Yes, code assistance with `claude-4.5-sonnet` in combination with manual editing by the author. Closesapache#52800 from dtenedor/kll-quantiles-functions. Authored-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

…etch 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 apache#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. Closesapache#52900 from dtenedor/dataframe-api-kll-functions. Authored-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

figure-shao · 2025-12-12T00:44:11Z

+ """,
+ group = "misc_funcs",
+ since = "4.1.0")
+case class KllSketchMergeDouble(left: Expression, right: Expression) extends KllSketchMergeBase {


Great feature! @dtenedor
What should I use to aggregate KLL sketches together in a group by? Being aggregated in a group by is the biggest advantage of having sketches, they act as partial aggregates and contributes to a complete aggregate at query time. But this is not supported.
SELECT kll_sketch_merge_double(kll) FROM quantile.table GROUP BY name
yields an error
[[WRONG_NUM_ARGS.WITHOUT_SUGGESTION](https://learn.microsoft.com/azure/databricks/error-messages/wrong-num-args-error-class#without_suggestion)] The `kll_sketch_merge_double` requires 2 parameters but the actual number is 1. Please, refer to 'https://spark.apache.org/docs/latest/sql-ref-functions.html' for a fix.
(Azure Databricks runtime 18.0, spark 2.13)
I believe it's a easy lift to make this usable as a group by aggregator. Please help!

Great point @figure-shao, agreed this would be valuable.
@dtenedor , if you're planning to extend the KLL functions to support partial aggregation semantics (allowing kll_sketch_merge_* to be used as proper aggregators), I'm happy to help with the implementation or take on the follow-up work if that’s useful.
No pressure at all, just let me know what would be most helpful.

@figure-shao that is a good point, we should add this.
@cboumalh thanks for adding the implementation! Let's review and merge your PR. I can take a look there next.

…s deterministic ### What changes were proposed in this pull request? In #52800, we added SQL support for KLL quantiles functions based on DataSketches. In this PR, we update some of the golden file tests to make them deterministic. ### Why are the changes needed? The previous tests generated string summaries of KLL quantile sketches and then split them by newlines and made case-sensitive checks for substrings. It turns out this was brittle, so this PR updates the tests to avoid the newline-splitting and makes the substring checks case-insenstiive. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? This PR updates test coverage only. ### Was this patch authored or co-authored using generative AI tooling? No Closes#53549 from dtenedor/kll-quantile-golden-files-fix. Authored-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…s deterministic ### What changes were proposed in this pull request? In #52800, we added SQL support for KLL quantiles functions based on DataSketches. In this PR, we update some of the golden file tests to make them deterministic. ### Why are the changes needed? The previous tests generated string summaries of KLL quantile sketches and then split them by newlines and made case-sensitive checks for substrings. It turns out this was brittle, so this PR updates the tests to avoid the newline-splitting and makes the substring checks case-insenstiive. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? This PR updates test coverage only. ### Was this patch authored or co-authored using generative AI tooling? No Closes#53549 from dtenedor/kll-quantile-golden-files-fix. Authored-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit e4b9993) Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…moving internal details ### What changes were proposed in this pull request? This PR improves error messages from the new KLL quantile sketch functions added in #52800. ### Why are the changes needed? The previous error messages reported internal DataSketches library state which was not meaningful for end users of the SQL/DF functions in Apache Spark. ### Does this PR introduce _any_ user-facing change? Yes, error messages are improved. For example, before this change, we observed the following: ``` SELECT kll_sketch_get_rank_bigint(agg, 5) AS wrong_type FROM ( SELECT kll_sketch_agg_float(col1) AS agg FROM t_float_1_5_through_7_11 ) > For function `kll_sketch_get_rank_bigint`, invalid KLL sketch binary data: reqOffset: 40, reqLength: 56, (reqOff + reqLen): 96, allocSize: 60" ``` Now the error message becomes: ``` > "Invalid call to `kll_sketch_get_rank_bigint`; only valid KLL sketch buffers are supported as inputs (such as those produced by the `kll_sketch_agg` function)." ``` ### How was this patch tested? This PR updates golden file test coverage to show the improved error messages. ### Was this patch authored or co-authored using generative AI tooling? No. Closes#53702 from dtenedor/kll-quantiles-golden-files. Authored-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

…moving internal details ### What changes were proposed in this pull request? This PR improves error messages from the new KLL quantile sketch functions added in #52800. ### Why are the changes needed? The previous error messages reported internal DataSketches library state which was not meaningful for end users of the SQL/DF functions in Apache Spark. ### Does this PR introduce _any_ user-facing change? Yes, error messages are improved. For example, before this change, we observed the following: ``` SELECT kll_sketch_get_rank_bigint(agg, 5) AS wrong_type FROM ( SELECT kll_sketch_agg_float(col1) AS agg FROM t_float_1_5_through_7_11 ) > For function `kll_sketch_get_rank_bigint`, invalid KLL sketch binary data: reqOffset: 40, reqLength: 56, (reqOff + reqLen): 96, allocSize: 60" ``` Now the error message becomes: ``` > "Invalid call to `kll_sketch_get_rank_bigint`; only valid KLL sketch buffers are supported as inputs (such as those produced by the `kll_sketch_agg` function)." ``` ### How was this patch tested? This PR updates golden file test coverage to show the improved error messages. ### Was this patch authored or co-authored using generative AI tooling? No. Closes#53702 from dtenedor/kll-quantiles-golden-files. Authored-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> (cherry picked from commit 9aea212) Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

### What changes were proposed in this pull request? This PR adds SQL aggregate functions with their tests for the KLL merge aggregate functions: - `kll_merge_agg_bigint` - `kll_merge_agg_float` - `kll_merge_agg_double` These aggregate functions merge multiple binary KLL sketch representations. Initial PRs: - #52900 - #52800 ### Why are the changes needed? The existing scalar `kll_sketch_merge_*` functions can only merge two sketches at a time. In distributed computing scenarios where sketches are pre-computed across multiple partitions, time windows, or datasets, users need to merge many sketches together. ### Does this PR introduce _any_ user-facing change? Yes, this PR adds 3 new aggregate functions. ### How was this patch tested? New SQL tests were added to `sql/core/src/test/resources/sql-tests/inputs/kllquantiles.sql`: **Positive tests:** - Merging bigint/float/double sketches from multiple rows - Merging with custom k parameters (400, 300, 500) - NULL value handling **Negative tests:** - Type mismatches (passing non-binary types) - Invalid binary data - k parameter validation (too small, too large, NULL, non-constant) ### Was this patch authored or co-authored using generative AI tooling? claude-4.5-sonnet and manual changes. Closes#53548 from cboumalh/cboumalh-kll-enhancement. Lead-authored-by: Chris Boumalhab <cboumalh@amazon.com> Co-authored-by: Chris Boumalhab <84485659+cboumalh@users.noreply.github.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

This PR adds SQL aggregate functions with their tests for the KLL merge aggregate functions: - `kll_merge_agg_bigint` - `kll_merge_agg_float` - `kll_merge_agg_double` These aggregate functions merge multiple binary KLL sketch representations. Initial PRs: - #52900 - #52800 The existing scalar `kll_sketch_merge_*` functions can only merge two sketches at a time. In distributed computing scenarios where sketches are pre-computed across multiple partitions, time windows, or datasets, users need to merge many sketches together. Yes, this PR adds 3 new aggregate functions. New SQL tests were added to `sql/core/src/test/resources/sql-tests/inputs/kllquantiles.sql`: **Positive tests:** - Merging bigint/float/double sketches from multiple rows - Merging with custom k parameters (400, 300, 500) - NULL value handling **Negative tests:** - Type mismatches (passing non-binary types) - Invalid binary data - k parameter validation (too small, too large, NULL, non-constant) claude-4.5-sonnet and manual changes. Closes#53548 from cboumalh/cboumalh-kll-enhancement. Lead-authored-by: Chris Boumalhab <cboumalh@amazon.com> Co-authored-by: Chris Boumalhab <84485659+cboumalh@users.noreply.github.com> Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com> (cherry picked from commit fc15f72) Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

commit
02b76d2

github-actionsBot added the SQL label Oct 30, 2025

improve performance
053f480

dtenedor requested a review from gengliangwang October 30, 2025 20:18

cboumalh reviewed Oct 30, 2025
View reviewed changes

Comment thread...alyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/kllAggregates.scala Outdated

dtenedor requested a review from cboumalh October 30, 2025 21:09

fix ExpressionSchemaSuite
bf0eb3a

cboumalh reviewed Oct 30, 2025
View reviewed changes

cboumalh reviewed Oct 31, 2025
View reviewed changes

Comment thread...alyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/kllAggregates.scala Outdated

cboumalh reviewed Oct 31, 2025
View reviewed changes

Comment threadsql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/kllExpressions.scala

cboumalh reviewed Oct 31, 2025
View reviewed changes

Comment thread...alyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/kllAggregates.scala Outdated

cboumalh reviewed Oct 31, 2025
View reviewed changes

Comment thread...alyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/kllAggregates.scala Outdated

dtenedor added 6 commits October 31, 2025 12:04

respond to some code review comments
b25f0a5

respond to code review comments about which types the functions accept
587c76f

respond to code reivew comments about breaking up import wildcards
6235b82

respond to code review comments to add the get_N functions
d217733

respond to code review comments about using heapify
4057c66

respond to code review comments about using heapify
7f4444f

dtenedor requested review from cboumalh and removed request for cboumalh and gengliangwang October 31, 2025 20:58

cboumalh reviewed Oct 31, 2025
View reviewed changes

respond to code review comments to include function name in error mes…
66ced6a
…sages

dtenedor requested a review from cboumalh November 3, 2025 19:46

cboumalh approved these changes Nov 3, 2025
View reviewed changes

sync
872c0b5

dtenedor closed this in e3a43baNov 4, 2025

dtenedor mentioned this pull request Nov 5, 2025
[SPARK-54199][SQL] Add DataFrame API support for new KLL quantiles sketch functions #52900
Closed

cboumalh mentioned this pull request Nov 15, 2025
[SPARK-54179][SQL] Add Native Support for Apache Tuple Sketches #52883
Closed

figure-shao reviewed Dec 12, 2025
View reviewed changes

cboumalh mentioned this pull request Dec 19, 2025
[SPARK-54785][SQL] Add support for binary sketch aggregations in KLL #53548
Closed

dtenedor mentioned this pull request Dec 19, 2025
[SPARK-53991][SQL][TEST][FOLLOWUP] Make KLL quantile golden file tests deterministic #53549
Closed

dtenedor mentioned this pull request Jan 6, 2026
[SPARK-53991][SQL][TEST][FOLLOW-UP] Improve KLL quantile errors by removing internal details #53702
Closed

Uh oh!

Conversation

dtenedor commented Oct 30, 2025• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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

Uh oh!

dtenedor commented Oct 30, 2025

Uh oh!

cboumalh commented Oct 30, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cboumalhOct 31, 2025• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dtenedorNov 3, 2025

Choose a reason for hiding this comment

Uh oh!

cboumalh commented Oct 31, 2025• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cboumalh commented Nov 3, 2025

Uh oh!

cboumalh commented Nov 3, 2025

Uh oh!

dtenedor commented Nov 4, 2025

Uh oh!

dtenedor commented Nov 4, 2025

Uh oh!

figure-shaoDec 12, 2025

Choose a reason for hiding this comment

Uh oh!

cboumalhDec 12, 2025

Choose a reason for hiding this comment

Uh oh!

cboumalhDec 19, 2025

Choose a reason for hiding this comment

Uh oh!

dtenedorJan 8, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

dtenedor commented Oct 30, 2025•
edited
Loading

cboumalhOct 31, 2025•
edited
Loading

cboumalh commented Oct 31, 2025•
edited
Loading