You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by dt...@apache.org on 2026/02/11 21:34:25 UTC
(spark) branch master updated: [SPARK-54179][SQL][FOLLOW-UP] Add Dataframe API support for Tuple sketches
This is an automated email from the ASF dual-hosted git repository.
dtenedor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 77980546e305 [SPARK-54179][SQL][FOLLOW-UP] Add Dataframe API support for Tuple sketches
77980546e305 is described below
commit 77980546e305ebdee89bd5fef40e8a4532a0a1b7
Author: Chris Boumalhab <cb...@amazon.com>
AuthorDate: Wed Feb 11 13:34:13 2026 -0800
[SPARK-54179][SQL][FOLLOW-UP] Add Dataframe API support for Tuple sketches
### What changes were proposed in this pull request?
This PR adds DataFrame API support for Tuple sketch functions that were previously only available in Spark SQL. The following functions are now accessible through the DataFrame API:
- `tuple_sketch_agg_*` - Aggregation functions for creating tuple sketches
- `tuple_union_agg_*` - Union aggregation functions
- `tuple_intersection_agg_*` - Intersection aggregation functions
- `tuple_sketch_estimate_*` - Estimation functions for tuple sketches
- `tuple_sketch_summary_*` - Summary functions for tuple sketches
- `tuple_sketch_theta_*` - Theta extraction functions
- `tuple_union_*` - Union operations
- `tuple_intersection_*` - Intersection operations
- `tuple_difference_*` - Difference operations
This is a follow-up to SPARK-54179 that completes the DataFrame API parity for tuple sketch operations.
### Why are the changes needed?
This PR expands the sketch feature's adoption in DataFrame cases.
### Does this PR introduce _any_ user-facing change?
yes, it provides the DataFrame API for Tuple sketches.
```
* tuple_sketch_agg_*
* tuple_union_agg_*
* tuple_intersection_agg_*
* tuple_sketch_estimate_*
* tuple_sketch_summary_*
* tuple_sketch_theta_*
* tuple_union_*
* tuple_intersection_*
* tuple_difference_*
```
### How was this patch tested?
Added tests to DataFrameAggregateSuite.scala
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 4.5
Closes #54041 from cboumalh/cboumalh-tuple-dataframe-followup.
Lead-authored-by: Chris Boumalhab <cb...@amazon.com>
Co-authored-by: Chris Boumalhab <84...@users.noreply.github.com>
Signed-off-by: Daniel Tenedorio <da...@databricks.com>
---
docs/sql-ref-sketch-aggregates.md | 410 +++++++-
.../source/reference/pyspark.sql/functions.rst | 18 +
python/pyspark/sql/connect/functions/builtin.py | 224 +++++
python/pyspark/sql/functions/__init__.py | 18 +
python/pyspark/sql/functions/builtin.py | 864 +++++++++++++++++
python/pyspark/sql/tests/test_functions.py | 536 +++++++++++
.../scala/org/apache/spark/sql/functions.scala | 1011 ++++++++++++++++++++
.../apache/spark/sql/DataFrameAggregateSuite.scala | 605 ++++++++++++
8 files changed, 3679 insertions(+), 7 deletions(-)
diff --git a/docs/sql-ref-sketch-aggregates.md b/docs/sql-ref-sketch-aggregates.md
index 6b92ba7b3c9e..f516ecf24601 100644
--- a/docs/sql-ref-sketch-aggregates.md
+++ b/docs/sql-ref-sketch-aggregates.md
@@ -22,10 +22,11 @@ license: |
Spark's SQL and DataFrame APIs provide a collection of sketch-based approximate functions powered by the [Apache DataSketches](https://datasketches.apache.org/) library. These functions enable efficient probabilistic computations on large datasets with bounded memory usage and accuracy guarantees.
Sketches are compact data structures that summarize large datasets, supporting distributed aggregation through serialization and merging. This makes them ideal for use cases including (so far):
-- **Approximate count distinct** (HLL and Theta sketches)
+- **Approximate count distinct** (HLL, Theta, and Tuple sketches)
- **Approximate quantile estimation** (KLL sketches)
- **Approximate frequent items** (Top-K sketches)
-- **Set operations** on distinct counts (Theta sketches)
+- **Set operations** on distinct counts (Theta and Tuple sketches)
+- **Distinct counting with aggregated summaries** (Tuple sketches)
### Table of Contents
@@ -42,6 +43,16 @@ Sketches are compact data structures that summarize large datasets, supporting d
* [theta_union](#theta_union)
* [theta_intersection](#theta_intersection)
* [theta_difference](#theta_difference)
+* [Tuple Sketch Functions](#tuple-sketch-functions)
+ * [tuple_sketch_agg_*](#tuple_sketch_agg_)
+ * [tuple_union_agg_*](#tuple_union_agg_)
+ * [tuple_intersection_agg_*](#tuple_intersection_agg_)
+ * [tuple_sketch_estimate_*](#tuple_sketch_estimate_)
+ * [tuple_sketch_summary_*](#tuple_sketch_summary_)
+ * [tuple_sketch_theta_*](#tuple_sketch_theta_)
+ * [tuple_union_*](#tuple_union_)
+ * [tuple_intersection_*](#tuple_intersection_)
+ * [tuple_difference_*](#tuple_difference_)
* [KLL Quantile Sketch Functions](#kll-quantile-sketch-functions)
* [kll_sketch_agg_*](#kll_sketch_agg_)
* [kll_merge_agg_*](#kll_merge_agg_)
@@ -63,6 +74,7 @@ Sketches are compact data structures that summarize large datasets, supporting d
* [Example: Computing Percentiles Over Time with KLL Sketches](#example-computing-percentiles-over-time-with-kll-sketches)
* [Example: Set Operations with Theta Sketches](#example-set-operations-with-theta-sketches)
* [Example: Finding Trending Items with Top-K Sketches](#example-finding-trending-items-with-top-k-sketches)
+ * [Example: Distinct Users with Aggregated Metrics Using Tuple Sketches](#example-distinct-users-with-aggregated-metrics-using-tuple-sketches)
---
@@ -220,6 +232,8 @@ FROM VALUES (1, 4), (1, 4), (2, 5), (2, 5), (3, 6) tab(col1, col2);
Theta sketches provide approximate count distinct with support for set operations (union, intersection, and difference). This makes them ideal for computing unique counts across overlapping datasets.
+**Note:** If you need to track both distinct counts AND aggregated metrics (like sum, min, or max of associated values), consider using [Tuple Sketch Functions](#tuple-sketch-functions) instead, which extend Theta sketches with numeric summary values.
+
See the [Apache DataSketches Theta documentation](https://datasketches.apache.org/docs/Theta/ThetaSketches.html) for more information.
### theta_sketch_agg
@@ -433,6 +447,307 @@ FROM VALUES (5, 4), (1, 4), (2, 5), (2, 5), (3, 1) tab(col1, col2);
---
+## Tuple Sketch Functions
+
+Tuple sketches extend Theta sketches by associating a numeric summary value with each distinct key. They provide approximate distinct counting with set operations (union, intersection, difference) while also aggregating summary values using configurable modes (sum, min, max, etc.). This makes them useful for scenarios where you need both cardinality estimation and aggregated metrics.
+
+**Note:** If you only need distinct counting without aggregated metrics, consider using [Theta Sketch Functions](#theta-sketch-functions) instead, which are more memory-efficient for pure cardinality estimation with set operations.
+
+See the [Apache DataSketches Tuple documentation](https://datasketches.apache.org/docs/Tuple/TupleOverview.html) for more information.
+
+Tuple functions are type-specific:
+- **DOUBLE** variants: For double-precision floating-point summary values
+- **INTEGER** variants: For integer summary values
+
+**Summary Aggregation Modes:**
+When multiple values are associated with the same key (during sketch creation or set operations), the mode parameter determines how summary values are combined:
+- **sum**: Adds all summary values together (default)
+- **min**: Keeps the minimum summary value
+- **max**: Keeps the maximum summary value
+- **alwaysone**: Sets all summary values to 1, effectively treating the sketch like a Theta sketch (useful for converting Tuple sketches to cardinality-only analysis)
+
+### tuple_sketch_agg_*
+
+Creates a Tuple sketch from key-value pairs, aggregating summary values for distinct keys.
+
+**Syntax:**
+```sql
+tuple_sketch_agg_double(key, summary [, lgNomEntries] [, mode])
+tuple_sketch_agg_integer(key, summary [, lgNomEntries] [, mode])
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `key` | INT, BIGINT, FLOAT, DOUBLE, STRING, BINARY, ARRAY<INT>, or ARRAY<BIGINT> | The key column for distinct counting |
+| `summary` | DOUBLE (for _double) or INT (for _integer) | The numeric value to aggregate for each key |
+| `lgNomEntries` | INT (optional) | Log-base-2 of nominal entries. Range: 4-26. Default: 12. |
+| `mode` | STRING (optional) | Summary aggregation mode: "sum" (default), "min", "max", "alwaysone" (sets all summary values to 1) |
+
+Returns a BINARY containing the Tuple sketch in compact binary representation.
+
+**Examples:**
+```sql
+-- Basic usage: count distinct keys and sum their values
+SELECT
+ tuple_sketch_estimate_integer(tuple_sketch_agg_integer(key, value)) as distinct_keys,
+ tuple_sketch_summary_integer(tuple_sketch_agg_integer(key, value)) as total_value
+FROM VALUES (1, 10), (2, 20), (2, 30) tab(key, value);
+-- Result: distinct_keys=2, total_value=60
+
+-- With custom lgNomEntries and mode
+SELECT tuple_sketch_summary_double(
+ tuple_sketch_agg_double(key, value, 16, 'sum'), 'max')
+FROM VALUES (1, 10.0), (2, 20.0), (2, 30.0) tab(key, value);
+-- Result: 50.0 (max value for distinct keys)
+```
+
+**Notes:**
+- NULL keys are ignored during aggregation.
+- When the same key appears multiple times, summary values are aggregated according to the mode parameter.
+- The sketch can be stored and later merged using tuple union/intersection functions.
+
+---
+
+### tuple_union_agg_*
+
+Aggregates multiple Tuple sketches using union operation, combining distinct keys from all sketches.
+
+**Syntax:**
+```sql
+tuple_union_agg_double(sketch [, lgNomEntries] [, mode])
+tuple_union_agg_integer(sketch [, lgNomEntries] [, mode])
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `sketch` | BINARY | A Tuple sketch in binary format |
+| `lgNomEntries` | INT (optional) | Log-base-2 of nominal entries. Range: 4-26. Default: 12. |
+| `mode` | STRING (optional) | Summary aggregation mode: "sum" (default), "min", "max", "alwaysone" (sets all summary values to 1) |
+
+Returns a BINARY containing the merged Tuple sketch.
+
+**Examples:**
+```sql
+-- Merge sketches from different data sources
+SELECT tuple_sketch_estimate_double(tuple_union_agg_double(sketch))
+FROM (
+ SELECT tuple_sketch_agg_double(key, value) as sketch
+ FROM VALUES (1, 10.0), (2, 20.0) tab(key, value)
+ UNION ALL
+ SELECT tuple_sketch_agg_double(key, value) as sketch
+ FROM VALUES (3, 30.0), (4, 40.0) tab(key, value)
+);
+-- Result: 4.0 (union of keys 1,2,3,4)
+```
+
+---
+
+### tuple_intersection_agg_*
+
+Aggregates multiple Tuple sketches using intersection operation, finding common distinct keys.
+
+**Syntax:**
+```sql
+tuple_intersection_agg_double(sketch [, mode])
+tuple_intersection_agg_integer(sketch [, mode])
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `sketch` | BINARY | A Tuple sketch in binary format |
+| `mode` | STRING (optional) | Summary aggregation mode: "sum" (default), "min", "max", "alwaysone" (sets all summary values to 1) |
+
+Returns a BINARY containing the intersected Tuple sketch.
+
+**Examples:**
+```sql
+-- Find common keys across sketches
+SELECT tuple_sketch_estimate_integer(tuple_intersection_agg_integer(sketch))
+FROM (
+ SELECT tuple_sketch_agg_integer(key, value) as sketch
+ FROM VALUES (1, 10), (2, 20), (3, 30) tab(key, value)
+ UNION ALL
+ SELECT tuple_sketch_agg_integer(key, value) as sketch
+ FROM VALUES (2, 40), (3, 50), (4, 60) tab(key, value)
+);
+-- Result: 2.0 (keys 2 and 3 are common)
+```
+
+---
+
+### tuple_sketch_estimate_*
+
+Estimates the number of distinct keys from a Tuple sketch.
+
+**Syntax:**
+```sql
+tuple_sketch_estimate_double(sketch)
+tuple_sketch_estimate_integer(sketch)
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `sketch` | BINARY | A Tuple sketch in binary format |
+
+Returns a DOUBLE representing the estimated count of distinct keys.
+
+**Examples:**
+```sql
+SELECT tuple_sketch_estimate_double(
+ tuple_sketch_agg_double(key, value))
+FROM VALUES (1, 10.0), (2, 20.0), (2, 30.0) tab(key, value);
+-- Result: 2.0
+```
+
+---
+
+### tuple_sketch_summary_*
+
+Returns the aggregated summary value from a Tuple sketch.
+
+**Syntax:**
+```sql
+tuple_sketch_summary_double(sketch [, mode])
+tuple_sketch_summary_integer(sketch [, mode])
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `sketch` | BINARY | A Tuple sketch in binary format |
+| `mode` | STRING (optional) | Summary aggregation mode: "sum" (default), "min", "max", "alwaysone" (sets all summary values to 1) |
+
+Returns the aggregated summary value (DOUBLE or BIGINT depending on variant).
+
+**Examples:**
+```sql
+-- Get the sum of all summary values
+SELECT tuple_sketch_summary_integer(
+ tuple_sketch_agg_integer(key, value))
+FROM VALUES (1, 10), (2, 20), (2, 30) tab(key, value);
+-- Result: 60 (10 + 20 + 30, with key 2's values summed)
+```
+
+---
+
+### tuple_sketch_theta_*
+
+Returns the theta value from a Tuple sketch, indicating sampling rate.
+
+**Syntax:**
+```sql
+tuple_sketch_theta_double(sketch)
+tuple_sketch_theta_integer(sketch)
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `sketch` | BINARY | A Tuple sketch in binary format |
+
+Returns a DOUBLE between 0.0 and 1.0 representing the theta value (1.0 means no sampling).
+
+**Examples:**
+```sql
+SELECT tuple_sketch_theta_double(
+ tuple_sketch_agg_double(key, value))
+FROM VALUES (1, 10.0), (2, 20.0) tab(key, value);
+-- Result: 1.0
+```
+
+---
+
+### tuple_union_*
+
+Merges two Tuple sketches using union (scalar function).
+
+**Syntax:**
+```sql
+tuple_union_double(first, second [, lgNomEntries] [, mode])
+tuple_union_integer(first, second [, lgNomEntries] [, mode])
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `first` | BINARY | First Tuple sketch |
+| `second` | BINARY | Second Tuple sketch |
+| `lgNomEntries` | INT (optional) | Log-base-2 of nominal entries. Range: 4-26. Default: 12. |
+| `mode` | STRING (optional) | Summary aggregation mode: "sum" (default), "min", "max", "alwaysone" (sets all summary values to 1) |
+
+Returns a BINARY containing the merged Tuple sketch.
+
+**Examples:**
+```sql
+-- Union of two sketches from different columns
+SELECT tuple_sketch_estimate_double(
+ tuple_union_double(
+ tuple_sketch_agg_double(key1, v1),
+ tuple_sketch_agg_double(key2, v2)))
+FROM VALUES (1, 10.0, 3, 30.0), (2, 20.0, 4, 40.0) tab(key1, v1, key2, v2);
+-- Result: 4.0 (keys 1,2,3,4 are in the union)
+```
+
+---
+
+### tuple_intersection_*
+
+Computes the intersection of two Tuple sketches (scalar function).
+
+**Syntax:**
+```sql
+tuple_intersection_double(first, second [, mode])
+tuple_intersection_integer(first, second [, mode])
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `first` | BINARY | First Tuple sketch |
+| `second` | BINARY | Second Tuple sketch |
+| `mode` | STRING (optional) | Summary aggregation mode: "sum" (default), "min", "max", "alwaysone" (sets all summary values to 1) |
+
+Returns a BINARY containing the intersected Tuple sketch.
+
+**Examples:**
+```sql
+-- Find keys present in both sketches
+SELECT tuple_sketch_estimate_integer(
+ tuple_intersection_integer(
+ tuple_sketch_agg_integer(key1, v1),
+ tuple_sketch_agg_integer(key2, v2)))
+FROM VALUES (1, 10, 2, 20), (2, 20, 3, 30), (3, 30, 4, 40) tab(key1, v1, key2, v2);
+-- Result: 2.0 (keys 2 and 3 are common)
+```
+
+---
+
+### tuple_difference_*
+
+Computes the set difference of two Tuple sketches (A - B), returning keys in the first sketch but not in the second.
+
+**Syntax:**
+```sql
+tuple_difference_double(first, second)
+tuple_difference_integer(first, second)
+```
+
+| Argument | Type | Description |
+|----------|------|-------------|
+| `first` | BINARY | First Tuple sketch (A) |
+| `second` | BINARY | Second Tuple sketch (B) |
+
+Returns a BINARY containing a Tuple sketch representing keys in A but not in B.
+
+**Examples:**
+```sql
+-- Find keys only in first sketch
+SELECT tuple_sketch_estimate_double(
+ tuple_difference_double(
+ tuple_sketch_agg_double(key1, v1),
+ tuple_sketch_agg_double(key2, v2)))
+FROM VALUES (1, 10.0, 4, 40.0), (2, 20.0, 4, 40.0), (3, 30.0, 5, 50.0), (4, 40.0, 5, 50.0) tab(key1, v1, key2, v2);
+-- Result: 3.0 (keys 1, 2, 3 are in key1 but not key2)
+```
+
+---
+
## KLL Quantile Sketch Functions
KLL (K-Linear-Logarithmic) sketches provide approximate quantile estimation. They are useful for computing percentiles, medians, and other order statistics on large datasets without sorting.
@@ -782,14 +1097,16 @@ FROM VALUES 'a', 'b', 'c', 'c', 'c', 'c', 'd', 'd' tab(expr);
## Best Practices
-### Choosing Between HLL and Theta Sketches
+### Choosing Between HLL, Theta, and Tuple Sketches
| Use Case | Recommended Sketch |
|----------|-------------------|
-| Simple count distinct | HLL (more memory efficient) |
-| Set operations (union, intersection, difference) | Theta |
+| Simple count distinct | HLL (most memory efficient) |
+| Set operations (union, intersection, difference) | Theta or Tuple |
+| Distinct count with aggregated metrics (sum, min, max) | Tuple |
| Very high cardinality with moderate accuracy | HLL with higher lgConfigK |
-| Need to compute overlaps between datasets | Theta |
+| Need to compute overlaps between datasets | Theta or Tuple |
+| Counting distinct users with total spend | Tuple (key=user_id, summary=spend) |
### Accuracy vs. Memory Trade-offs
@@ -797,6 +1114,7 @@ FROM VALUES 'a', 'b', 'c', 'c', 'c', 'c', 'd', 'd' tab(expr);
|-------------|-----------|---------------------|
| HLL | lgConfigK | Higher accuracy, more memory (2^lgConfigK bytes) |
| Theta | lgNomEntries | Higher accuracy, more memory (8 * 2^lgNomEntries bytes) |
+| Tuple | lgNomEntries | Higher accuracy, more memory (16 * 2^lgNomEntries bytes for double, 12 * 2^lgNomEntries bytes for integer) |
| KLL | k | Higher accuracy, more memory |
| Top-K | maxItemsTracked | Better heavy-hitter detection, more memory |
@@ -980,9 +1298,87 @@ WHERE hour_ts = TIMESTAMP'2024-01-15 14:00:00';
-- Query: Get top 10 searches across the full day by combining sketches
SELECT approx_top_k_estimate(
- approx_top_k_combine(search_sketch, 10000),
+ approx_top_k_combine(search_sketch, 10000),
10
) as daily_top_searches
FROM hourly_search_sketches
WHERE DATE(hour_ts) = DATE'2024-01-15';
```
+
+### Example: Distinct Users with Aggregated Metrics Using Tuple Sketches
+
+Tuple sketches enable tracking both distinct counts and aggregated metrics simultaneously. This is useful for scenarios like counting unique users while also summing their activity metrics.
+
+```sql
+-- Create a table to store daily tuple sketches tracking users and their spend
+CREATE TABLE daily_user_spend_sketches (
+ event_date DATE,
+ user_spend_sketch BINARY
+) USING PARQUET;
+
+-- Day 1: Track distinct users with their total spend
+INSERT INTO daily_user_spend_sketches
+SELECT
+ DATE'2024-01-01' as event_date,
+ tuple_sketch_agg_double(user_id, purchase_amount) as user_spend_sketch
+FROM purchases_day1;
+
+-- Day 2: Track next day's users and spend
+INSERT INTO daily_user_spend_sketches
+SELECT
+ DATE'2024-01-02' as event_date,
+ tuple_sketch_agg_double(user_id, purchase_amount) as user_spend_sketch
+FROM purchases_day2;
+
+-- Query: Get distinct users and total spend for a single day
+SELECT
+ event_date,
+ tuple_sketch_estimate_double(user_spend_sketch) as unique_users,
+ tuple_sketch_summary_double(user_spend_sketch) as total_spend
+FROM daily_user_spend_sketches
+WHERE event_date = DATE'2024-01-01';
+
+-- Query: Get unique users and total spend across a week (merging sketches)
+SELECT
+ tuple_sketch_estimate_double(tuple_union_agg_double(user_spend_sketch)) as weekly_unique_users,
+ tuple_sketch_summary_double(tuple_union_agg_double(user_spend_sketch)) as weekly_total_spend
+FROM daily_user_spend_sketches
+WHERE event_date BETWEEN DATE'2024-01-01' AND DATE'2024-01-07';
+
+-- Query: Find users who made purchases in week 1 but not week 2
+WITH week1_sketch AS (
+ SELECT tuple_union_agg_double(user_spend_sketch) as sketch
+ FROM daily_user_spend_sketches
+ WHERE event_date BETWEEN DATE'2024-01-01' AND DATE'2024-01-07'
+),
+week2_sketch AS (
+ SELECT tuple_union_agg_double(user_spend_sketch) as sketch
+ FROM daily_user_spend_sketches
+ WHERE event_date BETWEEN DATE'2024-01-08' AND DATE'2024-01-14'
+)
+SELECT tuple_sketch_estimate_double(
+ tuple_difference_double(
+ (SELECT sketch FROM week1_sketch),
+ (SELECT sketch FROM week2_sketch)
+ )
+) as churned_users;
+
+-- Query: Find users who purchased in both weeks (intersection)
+WITH week1_sketch AS (
+ SELECT tuple_union_agg_double(user_spend_sketch) as sketch
+ FROM daily_user_spend_sketches
+ WHERE event_date BETWEEN DATE'2024-01-01' AND DATE'2024-01-07'
+),
+week2_sketch AS (
+ SELECT tuple_union_agg_double(user_spend_sketch) as sketch
+ FROM daily_user_spend_sketches
+ WHERE event_date BETWEEN DATE'2024-01-08' AND DATE'2024-01-14'
+)
+SELECT
+ tuple_sketch_estimate_double(
+ tuple_intersection_double(
+ (SELECT sketch FROM week1_sketch),
+ (SELECT sketch FROM week2_sketch)
+ )
+ ) as retained_users;
+```
diff --git a/python/docs/source/reference/pyspark.sql/functions.rst b/python/docs/source/reference/pyspark.sql/functions.rst
index 9fcdac38e7d5..58cbd92cfd63 100644
--- a/python/docs/source/reference/pyspark.sql/functions.rst
+++ b/python/docs/source/reference/pyspark.sql/functions.rst
@@ -501,6 +501,12 @@ Aggregate Functions
theta_intersection_agg
theta_sketch_agg
theta_union_agg
+ tuple_intersection_agg_double
+ tuple_intersection_agg_integer
+ tuple_sketch_agg_double
+ tuple_sketch_agg_integer
+ tuple_union_agg_double
+ tuple_union_agg_integer
try_avg
try_sum
var_pop
@@ -665,6 +671,18 @@ Misc Functions
theta_intersection
theta_sketch_estimate
theta_union
+ tuple_difference_double
+ tuple_difference_integer
+ tuple_intersection_double
+ tuple_intersection_integer
+ tuple_sketch_estimate_double
+ tuple_sketch_estimate_integer
+ tuple_sketch_summary_double
+ tuple_sketch_summary_integer
+ tuple_sketch_theta_double
+ tuple_sketch_theta_integer
+ tuple_union_double
+ tuple_union_integer
try_aes_decrypt
try_reflect
typeof
diff --git a/python/pyspark/sql/connect/functions/builtin.py b/python/pyspark/sql/connect/functions/builtin.py
index 136db27b66ff..e48ca48c2b5e 100644
--- a/python/pyspark/sql/connect/functions/builtin.py
+++ b/python/pyspark/sql/connect/functions/builtin.py
@@ -4575,6 +4575,96 @@ def theta_intersection_agg(
theta_intersection_agg.__doc__ = pysparkfuncs.theta_intersection_agg.__doc__
+def tuple_sketch_agg_double(
+ key: "ColumnOrName",
+ summary: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_sketch_agg_double"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, key, summary, _lgNomEntries, _mode)
+
+
+tuple_sketch_agg_double.__doc__ = pysparkfuncs.tuple_sketch_agg_double.__doc__
+
+
+def tuple_sketch_agg_integer(
+ key: "ColumnOrName",
+ summary: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_sketch_agg_integer"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, key, summary, _lgNomEntries, _mode)
+
+
+tuple_sketch_agg_integer.__doc__ = pysparkfuncs.tuple_sketch_agg_integer.__doc__
+
+
+def tuple_union_agg_double(
+ col: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_union_agg_double"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, col, _lgNomEntries, _mode)
+
+
+tuple_union_agg_double.__doc__ = pysparkfuncs.tuple_union_agg_double.__doc__
+
+
+def tuple_union_agg_integer(
+ col: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_union_agg_integer"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, col, _lgNomEntries, _mode)
+
+
+tuple_union_agg_integer.__doc__ = pysparkfuncs.tuple_union_agg_integer.__doc__
+
+
+def tuple_intersection_agg_double(
+ col: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_intersection_agg_double"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col)
+ else:
+ return _invoke_function_over_columns(fn, col, lit(mode))
+
+
+tuple_intersection_agg_double.__doc__ = pysparkfuncs.tuple_intersection_agg_double.__doc__
+
+
+def tuple_intersection_agg_integer(
+ col: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_intersection_agg_integer"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col)
+ else:
+ return _invoke_function_over_columns(fn, col, lit(mode))
+
+
+tuple_intersection_agg_integer.__doc__ = pysparkfuncs.tuple_intersection_agg_integer.__doc__
+
+
def kll_sketch_agg_bigint(
col: "ColumnOrName",
k: Optional[Union[int, Column]] = None,
@@ -4816,6 +4906,140 @@ def theta_difference(col1: "ColumnOrName", col2: "ColumnOrName") -> Column:
theta_difference.__doc__ = pysparkfuncs.theta_difference.__doc__
+def tuple_sketch_estimate_double(col: "ColumnOrName") -> Column:
+ fn = "tuple_sketch_estimate_double"
+ return _invoke_function_over_columns(fn, col)
+
+
+tuple_sketch_estimate_double.__doc__ = pysparkfuncs.tuple_sketch_estimate_double.__doc__
+
+
+def tuple_sketch_estimate_integer(col: "ColumnOrName") -> Column:
+ fn = "tuple_sketch_estimate_integer"
+ return _invoke_function_over_columns(fn, col)
+
+
+tuple_sketch_estimate_integer.__doc__ = pysparkfuncs.tuple_sketch_estimate_integer.__doc__
+
+
+def tuple_sketch_summary_double(
+ col: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_sketch_summary_double"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col)
+ else:
+ return _invoke_function_over_columns(fn, col, lit(mode))
+
+
+tuple_sketch_summary_double.__doc__ = pysparkfuncs.tuple_sketch_summary_double.__doc__
+
+
+def tuple_sketch_summary_integer(
+ col: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_sketch_summary_integer"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col)
+ else:
+ return _invoke_function_over_columns(fn, col, lit(mode))
+
+
+tuple_sketch_summary_integer.__doc__ = pysparkfuncs.tuple_sketch_summary_integer.__doc__
+
+
+def tuple_sketch_theta_double(col: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("tuple_sketch_theta_double", col)
+
+
+tuple_sketch_theta_double.__doc__ = pysparkfuncs.tuple_sketch_theta_double.__doc__
+
+
+def tuple_sketch_theta_integer(col: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("tuple_sketch_theta_integer", col)
+
+
+tuple_sketch_theta_integer.__doc__ = pysparkfuncs.tuple_sketch_theta_integer.__doc__
+
+
+def tuple_union_double(
+ col1: "ColumnOrName",
+ col2: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_union_double"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, col1, col2, _lgNomEntries, _mode)
+
+
+tuple_union_double.__doc__ = pysparkfuncs.tuple_union_double.__doc__
+
+
+def tuple_union_integer(
+ col1: "ColumnOrName",
+ col2: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_union_integer"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, col1, col2, _lgNomEntries, _mode)
+
+
+tuple_union_integer.__doc__ = pysparkfuncs.tuple_union_integer.__doc__
+
+
+def tuple_intersection_double(
+ col1: "ColumnOrName",
+ col2: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_intersection_double"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col1, col2)
+ else:
+ return _invoke_function_over_columns(fn, col1, col2, lit(mode))
+
+
+tuple_intersection_double.__doc__ = pysparkfuncs.tuple_intersection_double.__doc__
+
+
+def tuple_intersection_integer(
+ col1: "ColumnOrName",
+ col2: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ fn = "tuple_intersection_integer"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col1, col2)
+ else:
+ return _invoke_function_over_columns(fn, col1, col2, lit(mode))
+
+
+tuple_intersection_integer.__doc__ = pysparkfuncs.tuple_intersection_integer.__doc__
+
+
+def tuple_difference_double(col1: "ColumnOrName", col2: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("tuple_difference_double", col1, col2)
+
+
+tuple_difference_double.__doc__ = pysparkfuncs.tuple_difference_double.__doc__
+
+
+def tuple_difference_integer(col1: "ColumnOrName", col2: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("tuple_difference_integer", col1, col2)
+
+
+tuple_difference_integer.__doc__ = pysparkfuncs.tuple_difference_integer.__doc__
+
+
# Predicates Function
diff --git a/python/pyspark/sql/functions/__init__.py b/python/pyspark/sql/functions/__init__.py
index ab563c18fbf1..3ad43e5e34cb 100644
--- a/python/pyspark/sql/functions/__init__.py
+++ b/python/pyspark/sql/functions/__init__.py
@@ -421,6 +421,12 @@ __all__ = [ # noqa: F405
"theta_intersection_agg",
"theta_sketch_agg",
"theta_union_agg",
+ "tuple_intersection_agg_double",
+ "tuple_intersection_agg_integer",
+ "tuple_sketch_agg_double",
+ "tuple_sketch_agg_integer",
+ "tuple_union_agg_double",
+ "tuple_union_agg_integer",
"try_avg",
"try_sum",
"var_pop",
@@ -531,6 +537,18 @@ __all__ = [ # noqa: F405
"theta_intersection",
"theta_sketch_estimate",
"theta_union",
+ "tuple_difference_double",
+ "tuple_difference_integer",
+ "tuple_intersection_double",
+ "tuple_intersection_integer",
+ "tuple_sketch_estimate_double",
+ "tuple_sketch_estimate_integer",
+ "tuple_sketch_summary_double",
+ "tuple_sketch_summary_integer",
+ "tuple_sketch_theta_double",
+ "tuple_sketch_theta_integer",
+ "tuple_union_double",
+ "tuple_union_integer",
"try_aes_decrypt",
"try_reflect",
"typeof",
diff --git a/python/pyspark/sql/functions/builtin.py b/python/pyspark/sql/functions/builtin.py
index 81f1bd4d02c9..c12ee65c2d63 100644
--- a/python/pyspark/sql/functions/builtin.py
+++ b/python/pyspark/sql/functions/builtin.py
@@ -26863,6 +26863,320 @@ def theta_intersection_agg(col: "ColumnOrName") -> Column:
return _invoke_function_over_columns(fn, col)
+@_try_remote_functions
+def tuple_sketch_agg_double(
+ key: "ColumnOrName",
+ summary: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Aggregate function: returns the compact binary representation of the Datasketches
+ TupleSketch with double summaries built from the key and summary columns.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ key : :class:`~pyspark.sql.Column` or column name
+ The column containing key values
+ summary : :class:`~pyspark.sql.Column` or column name
+ The column containing double summary values
+ lgNomEntries : :class:`~pyspark.sql.Column` or int, optional
+ The log-base-2 of nominal entries (must be between 4 and 26, defaults to 12)
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_estimate_double`
+ :meth:`pyspark.sql.functions.tuple_sketch_summary_double`
+ :meth:`pyspark.sql.functions.tuple_union_agg_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10.0), (2, 20.0), (2, 30.0)], ["key", "value"])
+ >>> df.agg(sf.tuple_sketch_estimate_double(
+ ... sf.tuple_sketch_agg_double("key", "value"))).show()
+ +--------------------------------------------------------------------------+
+ |tuple_sketch_estimate_double(tuple_sketch_agg_double(key, value, 12, sum))|
+ +--------------------------------------------------------------------------+
+ | 2.0|
+ +--------------------------------------------------------------------------+
+ """
+ fn = "tuple_sketch_agg_double"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, key, summary, _lgNomEntries, _mode)
+
+
+@_try_remote_functions
+def tuple_sketch_agg_integer(
+ key: "ColumnOrName",
+ summary: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Aggregate function: returns the compact binary representation of the Datasketches
+ TupleSketch with integer summaries built from the key and summary columns.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ key : :class:`~pyspark.sql.Column` or column name
+ The column containing key values
+ summary : :class:`~pyspark.sql.Column` or column name
+ The column containing integer summary values
+ lgNomEntries : :class:`~pyspark.sql.Column` or int, optional
+ The log-base-2 of nominal entries (must be between 4 and 26, defaults to 12)
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_estimate_integer`
+ :meth:`pyspark.sql.functions.tuple_sketch_summary_integer`
+ :meth:`pyspark.sql.functions.tuple_union_agg_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10), (2, 20), (2, 30)], ["key", "value"])
+ >>> df.agg(sf.tuple_sketch_estimate_integer(
+ ... sf.tuple_sketch_agg_integer("key", "value"))).show()
+ +----------------------------------------------------------------------------+
+ |tuple_sketch_estimate_integer(tuple_sketch_agg_integer(key, value, 12, sum))|
+ +----------------------------------------------------------------------------+
+ | 2.0|
+ +----------------------------------------------------------------------------+
+ """
+ fn = "tuple_sketch_agg_integer"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, key, summary, _lgNomEntries, _mode)
+
+
+@_try_remote_functions
+def tuple_union_agg_double(
+ col: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Aggregate function: returns the compact binary representation of the Datasketches
+ TupleSketch that is the union of the double TupleSketch objects in the input column.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing binary TupleSketch representations
+ lgNomEntries : :class:`~pyspark.sql.Column` or int, optional
+ The log-base-2 of nominal entries (must be between 4 and 26, defaults to 12)
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the merged TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_double`
+ :meth:`pyspark.sql.functions.tuple_union_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df1 = spark.createDataFrame([(1, 10.0), (2, 20.0)], ["key", "value"])
+ >>> df1 = df1.agg(sf.tuple_sketch_agg_double("key", "value").alias("sketch"))
+ >>> df2 = spark.createDataFrame([(3, 30.0), (4, 40.0)], ["key", "value"])
+ >>> df2 = df2.agg(sf.tuple_sketch_agg_double("key", "value").alias("sketch"))
+ >>> df3 = df1.union(df2)
+ >>> df3.agg(sf.tuple_sketch_estimate_double(sf.tuple_union_agg_double("sketch"))).show()
+ +---------------------------------------------------------------------+
+ |tuple_sketch_estimate_double(tuple_union_agg_double(sketch, 12, sum))|
+ +---------------------------------------------------------------------+
+ | 4.0|
+ +---------------------------------------------------------------------+
+ """
+ fn = "tuple_union_agg_double"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, col, _lgNomEntries, _mode)
+
+
+@_try_remote_functions
+def tuple_union_agg_integer(
+ col: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Aggregate function: returns the compact binary representation of the Datasketches
+ TupleSketch that is the union of the integer TupleSketch objects in the input column.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing binary TupleSketch representations
+ lgNomEntries : :class:`~pyspark.sql.Column` or int, optional
+ The log-base-2 of nominal entries (must be between 4 and 26, defaults to 12)
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the merged TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_union_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df1 = spark.createDataFrame([(1, 10), (2, 20)], ["key", "value"])
+ >>> df1 = df1.agg(sf.tuple_sketch_agg_integer("key", "value").alias("sketch"))
+ >>> df2 = spark.createDataFrame([(3, 30), (4, 40)], ["key", "value"])
+ >>> df2 = df2.agg(sf.tuple_sketch_agg_integer("key", "value").alias("sketch"))
+ >>> df3 = df1.union(df2)
+ >>> df3.agg(sf.tuple_sketch_estimate_integer(sf.tuple_union_agg_integer("sketch"))).show()
+ +-----------------------------------------------------------------------+
+ |tuple_sketch_estimate_integer(tuple_union_agg_integer(sketch, 12, sum))|
+ +-----------------------------------------------------------------------+
+ | 4.0|
+ +-----------------------------------------------------------------------+
+ """
+ fn = "tuple_union_agg_integer"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, col, _lgNomEntries, _mode)
+
+
+@_try_remote_functions
+def tuple_intersection_agg_double(
+ col: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Aggregate function: returns the compact binary representation of the Datasketches
+ TupleSketch that is the intersection of the double TupleSketch objects in the input column.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing binary TupleSketch representations
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the intersected TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_double`
+ :meth:`pyspark.sql.functions.tuple_intersection_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df1 = spark.createDataFrame([(1, 10.0), (2, 20.0), (3, 30.0)], ["key", "value"])
+ >>> df1 = df1.agg(sf.tuple_sketch_agg_double("key", "value").alias("sketch"))
+ >>> df2 = spark.createDataFrame([(2, 40.0), (3, 50.0), (4, 60.0)], ["key", "value"])
+ >>> df2 = df2.agg(sf.tuple_sketch_agg_double("key", "value").alias("sketch"))
+ >>> df3 = df1.union(df2)
+ >>> df3.agg(sf.tuple_sketch_estimate_double(sf.tuple_intersection_agg_double("sketch"))).show()
+ +------------------------------------------------------------------------+
+ |tuple_sketch_estimate_double(tuple_intersection_agg_double(sketch, sum))|
+ +------------------------------------------------------------------------+
+ | 2.0|
+ +------------------------------------------------------------------------+
+ """
+ fn = "tuple_intersection_agg_double"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col)
+ else:
+ return _invoke_function_over_columns(fn, col, lit(mode))
+
+
+@_try_remote_functions
+def tuple_intersection_agg_integer(
+ col: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Aggregate function: returns the compact binary representation of the Datasketches
+ TupleSketch that is the intersection of the integer TupleSketch objects in the input column.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing binary TupleSketch representations
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the intersected TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_intersection_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df1 = spark.createDataFrame([(1, 10), (2, 20), (3, 30)], ["key", "value"])
+ >>> df1 = df1.agg(sf.tuple_sketch_agg_integer("key", "value").alias("sketch"))
+ >>> df2 = spark.createDataFrame([(2, 40), (3, 50), (4, 60)], ["key", "value"])
+ >>> df2 = df2.agg(sf.tuple_sketch_agg_integer("key", "value").alias("sketch"))
+ >>> df3 = df1.union(df2)
+ >>> df3.agg(sf.tuple_sketch_estimate_integer(sf.tuple_intersection_agg_integer("sketch"))).show()
+ +--------------------------------------------------------------------------+
+ |tuple_sketch_estimate_integer(tuple_intersection_agg_integer(sketch, sum))|
+ +--------------------------------------------------------------------------+
+ | 2.0|
+ +--------------------------------------------------------------------------+
+ """
+ fn = "tuple_intersection_agg_integer"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col)
+ else:
+ return _invoke_function_over_columns(fn, col, lit(mode))
+
+
@_try_remote_functions
def kll_sketch_agg_bigint(
col: "ColumnOrName",
@@ -27801,6 +28115,556 @@ def theta_difference(col1: "ColumnOrName", col2: "ColumnOrName") -> Column:
return _invoke_function_over_columns(fn, col1, col2)
+@_try_remote_functions
+def tuple_sketch_estimate_double(col: "ColumnOrName") -> Column:
+ """
+ Returns the estimated number of distinct keys from a Datasketches TupleSketch
+ with double summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing a binary TupleSketch representation
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The estimated cardinality.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_double`
+ :meth:`pyspark.sql.functions.tuple_sketch_summary_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10.0), (2, 20.0), (2, 30.0)], ["key", "value"])
+ >>> df.agg(sf.tuple_sketch_estimate_double(
+ ... sf.tuple_sketch_agg_double("key", "value"))).show()
+ +--------------------------------------------------------------------------+
+ |tuple_sketch_estimate_double(tuple_sketch_agg_double(key, value, 12, sum))|
+ +--------------------------------------------------------------------------+
+ | 2.0|
+ +--------------------------------------------------------------------------+
+ """
+ fn = "tuple_sketch_estimate_double"
+ return _invoke_function_over_columns(fn, col)
+
+
+@_try_remote_functions
+def tuple_sketch_estimate_integer(col: "ColumnOrName") -> Column:
+ """
+ Returns the estimated number of distinct keys from a Datasketches TupleSketch
+ with integer summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing a binary TupleSketch representation
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The estimated cardinality.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_sketch_summary_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10), (2, 20), (2, 30)], ["key", "value"])
+ >>> df.agg(sf.tuple_sketch_estimate_integer(
+ ... sf.tuple_sketch_agg_integer("key", "value"))).show()
+ +----------------------------------------------------------------------------+
+ |tuple_sketch_estimate_integer(tuple_sketch_agg_integer(key, value, 12, sum))|
+ +----------------------------------------------------------------------------+
+ | 2.0|
+ +----------------------------------------------------------------------------+
+ """
+ fn = "tuple_sketch_estimate_integer"
+ return _invoke_function_over_columns(fn, col)
+
+
+@_try_remote_functions
+def tuple_sketch_summary_double(
+ col: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Returns the aggregated summary value from a Datasketches TupleSketch with double summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing a binary TupleSketch representation
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The aggregated summary value.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_double`
+ :meth:`pyspark.sql.functions.tuple_sketch_estimate_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10.0), (2, 20.0), (2, 30.0)], ["key", "value"])
+ >>> df.agg(sf.tuple_sketch_summary_double(
+ ... sf.tuple_sketch_agg_double("key", "value"))).show()
+ +------------------------------------------------------------------------------+
+ |tuple_sketch_summary_double(tuple_sketch_agg_double(key, value, 12, sum), sum)|
+ +------------------------------------------------------------------------------+
+ | 60.0|
+ +------------------------------------------------------------------------------+
+ """
+ fn = "tuple_sketch_summary_double"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col)
+ else:
+ return _invoke_function_over_columns(fn, col, lit(mode))
+
+
+@_try_remote_functions
+def tuple_sketch_summary_integer(
+ col: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Returns the aggregated summary value from a Datasketches TupleSketch with integer summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing a binary TupleSketch representation
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The aggregated summary value.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_sketch_estimate_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10), (2, 20), (2, 30)], ["key", "value"])
+ >>> df.agg(sf.tuple_sketch_summary_integer(
+ ... sf.tuple_sketch_agg_integer("key", "value"))).show()
+ +--------------------------------------------------------------------------------+
+ |tuple_sketch_summary_integer(tuple_sketch_agg_integer(key, value, 12, sum), sum)|
+ +--------------------------------------------------------------------------------+
+ | 60|
+ +--------------------------------------------------------------------------------+
+ """
+ fn = "tuple_sketch_summary_integer"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col)
+ else:
+ return _invoke_function_over_columns(fn, col, lit(mode))
+
+
+@_try_remote_functions
+def tuple_sketch_theta_double(col: "ColumnOrName") -> Column:
+ """
+ Returns the theta value from a Datasketches TupleSketch with double summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing a binary TupleSketch representation
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The theta value (between 0.0 and 1.0).
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_double`
+ :meth:`pyspark.sql.functions.tuple_sketch_estimate_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10.0), (2, 20.0)], ["key", "value"])
+ >>> df.agg(sf.tuple_sketch_theta_double(
+ ... sf.tuple_sketch_agg_double("key", "value"))).show()
+ +-----------------------------------------------------------------------+
+ |tuple_sketch_theta_double(tuple_sketch_agg_double(key, value, 12, sum))|
+ +-----------------------------------------------------------------------+
+ | 1.0|
+ +-----------------------------------------------------------------------+
+ """
+ return _invoke_function_over_columns("tuple_sketch_theta_double", col)
+
+
+@_try_remote_functions
+def tuple_sketch_theta_integer(col: "ColumnOrName") -> Column:
+ """
+ Returns the theta value from a Datasketches TupleSketch with integer summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The column containing a binary TupleSketch representation
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The theta value (between 0.0 and 1.0).
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_sketch_estimate_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10), (2, 20)], ["key", "value"])
+ >>> df.agg(sf.tuple_sketch_theta_integer(
+ ... sf.tuple_sketch_agg_integer("key", "value"))).show()
+ +-------------------------------------------------------------------------+
+ |tuple_sketch_theta_integer(tuple_sketch_agg_integer(key, value, 12, sum))|
+ +-------------------------------------------------------------------------+
+ | 1.0|
+ +-------------------------------------------------------------------------+
+ """
+ return _invoke_function_over_columns("tuple_sketch_theta_integer", col)
+
+
+@_try_remote_functions
+def tuple_union_double(
+ col1: "ColumnOrName",
+ col2: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Returns the union of two Datasketches TupleSketch objects with double summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col1 : :class:`~pyspark.sql.Column` or column name
+ The first TupleSketch column
+ col2 : :class:`~pyspark.sql.Column` or column name
+ The second TupleSketch column
+ lgNomEntries : :class:`~pyspark.sql.Column` or int, optional
+ The log-base-2 of nominal entries (must be between 4 and 26, defaults to 12)
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the merged TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_double`
+ :meth:`pyspark.sql.functions.tuple_union_agg_double`
+ :meth:`pyspark.sql.functions.tuple_intersection_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10.0, 3, 30.0), (2, 20.0, 4, 40.0)], ["key1", "v1", "key2", "v2"]) # noqa
+ >>> df = df.agg(
+ ... sf.tuple_sketch_agg_double("key1", "v1").alias("sketch1"),
+ ... sf.tuple_sketch_agg_double("key2", "v2").alias("sketch2")
+ ... )
+ >>> df.select(sf.tuple_sketch_estimate_double(sf.tuple_union_double(df.sketch1, "sketch2"))).show() # noqa
+ +---------------------------------------------------------------------------+
+ |tuple_sketch_estimate_double(tuple_union_double(sketch1, sketch2, 12, sum))|
+ +---------------------------------------------------------------------------+
+ | 4.0|
+ +---------------------------------------------------------------------------+
+ """
+ fn = "tuple_union_double"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, col1, col2, _lgNomEntries, _mode)
+
+
+@_try_remote_functions
+def tuple_union_integer(
+ col1: "ColumnOrName",
+ col2: "ColumnOrName",
+ lgNomEntries: Optional[Union[int, Column]] = None,
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Returns the union of two Datasketches TupleSketch objects with integer summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col1 : :class:`~pyspark.sql.Column` or column name
+ The first TupleSketch column
+ col2 : :class:`~pyspark.sql.Column` or column name
+ The second TupleSketch column
+ lgNomEntries : :class:`~pyspark.sql.Column` or int, optional
+ The log-base-2 of nominal entries (must be between 4 and 26, defaults to 12)
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the merged TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_union_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_intersection_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10, 3, 30), (2, 20, 4, 40)], ["key1", "v1", "key2", "v2"]) # noqa
+ >>> df = df.agg(
+ ... sf.tuple_sketch_agg_integer("key1", "v1").alias("sketch1"),
+ ... sf.tuple_sketch_agg_integer("key2", "v2").alias("sketch2")
+ ... )
+ >>> df.select(sf.tuple_sketch_estimate_integer(sf.tuple_union_integer(df.sketch1, "sketch2"))).show() # noqa
+ +-----------------------------------------------------------------------------+
+ |tuple_sketch_estimate_integer(tuple_union_integer(sketch1, sketch2, 12, sum))|
+ +-----------------------------------------------------------------------------+
+ | 4.0|
+ +-----------------------------------------------------------------------------+
+ """
+ fn = "tuple_union_integer"
+ _lgNomEntries = lit(12) if lgNomEntries is None else lit(lgNomEntries)
+ _mode = lit("sum") if mode is None else lit(mode)
+
+ return _invoke_function_over_columns(fn, col1, col2, _lgNomEntries, _mode)
+
+
+@_try_remote_functions
+def tuple_intersection_double(
+ col1: "ColumnOrName",
+ col2: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Returns the intersection of two Datasketches TupleSketch objects with double summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col1 : :class:`~pyspark.sql.Column` or column name
+ The first TupleSketch column
+ col2 : :class:`~pyspark.sql.Column` or column name
+ The second TupleSketch column
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the intersected TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_double`
+ :meth:`pyspark.sql.functions.tuple_union_double`
+ :meth:`pyspark.sql.functions.tuple_intersection_agg_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10.0, 2, 20.0), (2, 20.0, 3, 30.0), (3, 30.0, 4, 40.0)], ["key1", "v1", "key2", "v2"]) # noqa
+ >>> df = df.agg(
+ ... sf.tuple_sketch_agg_double("key1", "v1").alias("sketch1"),
+ ... sf.tuple_sketch_agg_double("key2", "v2").alias("sketch2")
+ ... )
+ >>> df.select(sf.tuple_sketch_estimate_double(sf.tuple_intersection_double(df.sketch1, "sketch2"))).show() # noqa
+ +------------------------------------------------------------------------------+
+ |tuple_sketch_estimate_double(tuple_intersection_double(sketch1, sketch2, sum))|
+ +------------------------------------------------------------------------------+
+ | 2.0|
+ +------------------------------------------------------------------------------+
+ """
+ fn = "tuple_intersection_double"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col1, col2)
+ else:
+ return _invoke_function_over_columns(fn, col1, col2, lit(mode))
+
+
+@_try_remote_functions
+def tuple_intersection_integer(
+ col1: "ColumnOrName",
+ col2: "ColumnOrName",
+ mode: Optional[Union[str, Column]] = None,
+) -> Column:
+ """
+ Returns the intersection of two Datasketches TupleSketch objects with integer summaries.
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col1 : :class:`~pyspark.sql.Column` or column name
+ The first TupleSketch column
+ col2 : :class:`~pyspark.sql.Column` or column name
+ The second TupleSketch column
+ mode : :class:`~pyspark.sql.Column` or str, optional
+ The summary mode: "sum" (default), "min", "max", or "alwaysone"
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the intersected TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_union_integer`
+ :meth:`pyspark.sql.functions.tuple_intersection_agg_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10, 2, 20), (2, 20, 3, 30), (3, 30, 4, 40)], ["key1", "v1", "key2", "v2"]) # noqa
+ >>> df = df.agg(
+ ... sf.tuple_sketch_agg_integer("key1", "v1").alias("sketch1"),
+ ... sf.tuple_sketch_agg_integer("key2", "v2").alias("sketch2")
+ ... )
+ >>> df.select(sf.tuple_sketch_estimate_integer(sf.tuple_intersection_integer(df.sketch1, "sketch2"))).show() # noqa
+ +--------------------------------------------------------------------------------+
+ |tuple_sketch_estimate_integer(tuple_intersection_integer(sketch1, sketch2, sum))|
+ +--------------------------------------------------------------------------------+
+ | 2.0|
+ +--------------------------------------------------------------------------------+
+ """
+ fn = "tuple_intersection_integer"
+ if mode is None:
+ return _invoke_function_over_columns(fn, col1, col2)
+ else:
+ return _invoke_function_over_columns(fn, col1, col2, lit(mode))
+
+
+@_try_remote_functions
+def tuple_difference_double(col1: "ColumnOrName", col2: "ColumnOrName") -> Column:
+ """
+ Returns the set difference of two Datasketches TupleSketch objects with double summaries
+ (elements in first sketch but not in second).
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col1 : :class:`~pyspark.sql.Column` or column name
+ The first TupleSketch column
+ col2 : :class:`~pyspark.sql.Column` or column name
+ The second TupleSketch column
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the difference TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_double`
+ :meth:`pyspark.sql.functions.tuple_union_double`
+ :meth:`pyspark.sql.functions.tuple_intersection_double`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10.0, 4, 40.0), (2, 20.0, 4, 40.0), (3, 30.0, 5, 50.0), (4, 40.0, 5, 50.0)], ["key1", "v1", "key2", "v2"]) # noqa
+ >>> df = df.agg(
+ ... sf.tuple_sketch_agg_double("key1", "v1").alias("sketch1"),
+ ... sf.tuple_sketch_agg_double("key2", "v2").alias("sketch2")
+ ... )
+ >>> df.select(sf.tuple_sketch_estimate_double(sf.tuple_difference_double(df.sketch1, "sketch2"))).show() # noqa
+ +-----------------------------------------------------------------------+
+ |tuple_sketch_estimate_double(tuple_difference_double(sketch1, sketch2))|
+ +-----------------------------------------------------------------------+
+ | 3.0|
+ +-----------------------------------------------------------------------+
+ """
+ return _invoke_function_over_columns("tuple_difference_double", col1, col2)
+
+
+@_try_remote_functions
+def tuple_difference_integer(col1: "ColumnOrName", col2: "ColumnOrName") -> Column:
+ """
+ Returns the set difference of two Datasketches TupleSketch objects with integer summaries
+ (elements in first sketch but not in second).
+
+ .. versionadded:: 4.2.0
+
+ Parameters
+ ----------
+ col1 : :class:`~pyspark.sql.Column` or column name
+ The first TupleSketch column
+ col2 : :class:`~pyspark.sql.Column` or column name
+ The second TupleSketch column
+
+ Returns
+ -------
+ :class:`~pyspark.sql.Column`
+ The binary representation of the difference TupleSketch.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.tuple_sketch_agg_integer`
+ :meth:`pyspark.sql.functions.tuple_union_integer`
+ :meth:`pyspark.sql.functions.tuple_intersection_integer`
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([(1, 10, 4, 40), (2, 20, 4, 40), (3, 30, 5, 50), (4, 40, 5, 50)], ["key1", "v1", "key2", "v2"]) # noqa
+ >>> df = df.agg(
+ ... sf.tuple_sketch_agg_integer("key1", "v1").alias("sketch1"),
+ ... sf.tuple_sketch_agg_integer("key2", "v2").alias("sketch2")
+ ... )
+ >>> df.select(sf.tuple_sketch_estimate_integer(sf.tuple_difference_integer(df.sketch1, "sketch2"))).show() # noqa
+ +-------------------------------------------------------------------------+
+ |tuple_sketch_estimate_integer(tuple_difference_integer(sketch1, sketch2))|
+ +-------------------------------------------------------------------------+
+ | 3.0|
+ +-------------------------------------------------------------------------+
+ """
+ return _invoke_function_over_columns("tuple_difference_integer", col1, col2)
+
+
# ---------------------- Predicates functions ------------------------------
diff --git a/python/pyspark/sql/tests/test_functions.py b/python/pyspark/sql/tests/test_functions.py
index f3e629f6a4b3..bfad39d6babc 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -2361,6 +2361,542 @@ class FunctionsTestsMixin:
# Should have 5 values (1,2,3,4,6 - null is ignored)
self.assertEqual(n, 5)
+ def test_tuple_sketch_agg_double_basic(self):
+ """Test tuple_sketch_agg_double basic functionality"""
+ df = self.spark.createDataFrame(
+ [(1, 1.5), (2, 2.5), (3, 3.5), (1, 0.5), (2, 1.0)], ["key", "summary"]
+ )
+
+ # Test with default parameters
+ sketch1 = df.agg(F.tuple_sketch_agg_double("key", "summary")).first()[0]
+ self.assertIsNotNone(sketch1)
+ self.assertIsInstance(sketch1, (bytes, bytearray))
+
+ # Test with explicit lgNomEntries
+ sketch2 = df.agg(F.tuple_sketch_agg_double("key", "summary", 10)).first()[0]
+ self.assertIsNotNone(sketch2)
+
+ # Test with lgNomEntries and mode
+ sketch3 = df.agg(F.tuple_sketch_agg_double("key", "summary", 10, "sum")).first()[0]
+ self.assertIsNotNone(sketch3)
+
+ # Test with Column objects
+ sketch4 = df.agg(F.tuple_sketch_agg_double(F.col("key"), F.col("summary"))).first()[0]
+ self.assertIsNotNone(sketch4)
+
+ def test_tuple_sketch_agg_integer_basic(self):
+ """Test tuple_sketch_agg_integer basic functionality"""
+ df = self.spark.createDataFrame(
+ [(1, 10), (2, 20), (3, 30), (1, 5), (2, 15)], ["key", "summary"]
+ )
+
+ # Test with default parameters
+ sketch1 = df.agg(F.tuple_sketch_agg_integer("key", "summary")).first()[0]
+ self.assertIsNotNone(sketch1)
+ self.assertIsInstance(sketch1, (bytes, bytearray))
+
+ # Test with explicit lgNomEntries
+ sketch2 = df.agg(F.tuple_sketch_agg_integer("key", "summary", 10)).first()[0]
+ self.assertIsNotNone(sketch2)
+
+ # Test with lgNomEntries and mode
+ sketch3 = df.agg(F.tuple_sketch_agg_integer("key", "summary", 10, "max")).first()[0]
+ self.assertIsNotNone(sketch3)
+
+ # Test with different mode
+ sketch4 = df.agg(F.tuple_sketch_agg_integer("key", "summary", 10, "min")).first()[0]
+ self.assertIsNotNone(sketch4)
+
+ def test_tuple_sketch_estimate_and_summary_double(self):
+ """Test tuple_sketch_estimate and summary functions - double"""
+ df = self.spark.createDataFrame(
+ [(1, 1.5), (2, 2.5), (3, 3.5), (1, 0.5), (2, 1.0)], ["key", "summary"]
+ )
+ sketch_df = df.agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+
+ # Test estimate
+ estimate = sketch_df.select(F.tuple_sketch_estimate_double("sketch")).first()[0]
+ self.assertEqual(estimate, 3.0)
+
+ # Test summary with default mode (sum)
+ summary1 = sketch_df.select(F.tuple_sketch_summary_double("sketch")).first()[0]
+ self.assertEqual(summary1, 9.0)
+
+ # Test summary with explicit mode
+ summary2 = sketch_df.select(F.tuple_sketch_summary_double("sketch", "min")).first()[0]
+ self.assertEqual(summary2, 2.0)
+
+ # Test theta
+ theta = sketch_df.select(F.tuple_sketch_theta_double("sketch")).first()[0]
+ self.assertIsNotNone(theta)
+ self.assertGreater(theta, 0.0)
+ self.assertLessEqual(theta, 1.0)
+
+ # Test with Column objects
+ estimate2 = sketch_df.select(F.tuple_sketch_estimate_double(F.col("sketch"))).first()[0]
+ self.assertIsNotNone(estimate2)
+
+ def test_tuple_sketch_estimate_and_summary_integer(self):
+ """Test tuple_sketch_estimate and summary functions - integer"""
+ df = self.spark.createDataFrame(
+ [(1, 10), (2, 20), (3, 30), (1, 5), (2, 15)], ["key", "summary"]
+ )
+ sketch_df = df.agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+
+ # Test estimate
+ estimate = sketch_df.select(F.tuple_sketch_estimate_integer("sketch")).first()[0]
+ self.assertEqual(estimate, 3.0)
+
+ # Test summary with default mode (sum)
+ summary1 = sketch_df.select(F.tuple_sketch_summary_integer("sketch")).first()[0]
+ self.assertEqual(summary1, 80)
+
+ # Test summary with explicit mode
+ summary2 = sketch_df.select(F.tuple_sketch_summary_integer("sketch", "max")).first()[0]
+ self.assertEqual(summary2, 35)
+
+ # Test theta
+ theta = sketch_df.select(F.tuple_sketch_theta_integer("sketch")).first()[0]
+ self.assertIsNotNone(theta)
+ self.assertGreater(theta, 0.0)
+ self.assertLessEqual(theta, 1.0)
+
+ # Test with min mode
+ summary3 = sketch_df.select(F.tuple_sketch_summary_integer("sketch", "min")).first()[0]
+ self.assertEqual(summary3, 15)
+
+ def test_tuple_union_double_basic(self):
+ """Test tuple_union_double basic functionality"""
+ df1 = self.spark.createDataFrame([(1, 1.5), (2, 2.5)], ["key", "summary"])
+ df2 = self.spark.createDataFrame([(3, 3.5), (4, 4.5)], ["key", "summary"])
+
+ sketch1_df = df1.agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+ sketch2_df = df2.agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+
+ joined = sketch1_df.crossJoin(sketch2_df.withColumnRenamed("sketch", "sketch2"))
+
+ # Test union with default parameters
+ union1 = joined.select(F.tuple_union_double("sketch", "sketch2")).first()[0]
+ self.assertIsNotNone(union1)
+ self.assertIsInstance(union1, (bytes, bytearray))
+
+ # Test union with lgNomEntries
+ union2 = joined.select(F.tuple_union_double("sketch", "sketch2", 10)).first()[0]
+ self.assertIsNotNone(union2)
+
+ # Test union with lgNomEntries and mode
+ union3 = joined.select(F.tuple_union_double("sketch", "sketch2", 10, "sum")).first()[0]
+ self.assertIsNotNone(union3)
+
+ # Verify estimate from union
+ estimate = joined.select(
+ F.tuple_sketch_estimate_double(F.tuple_union_double("sketch", "sketch2"))
+ ).first()[0]
+ self.assertEqual(estimate, 4.0)
+
+ def test_tuple_union_integer_basic(self):
+ """Test tuple_union_integer basic functionality"""
+ df1 = self.spark.createDataFrame([(1, 10), (2, 20)], ["key", "summary"])
+ df2 = self.spark.createDataFrame([(3, 30), (4, 40)], ["key", "summary"])
+
+ sketch1_df = df1.agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+ sketch2_df = df2.agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+
+ joined = sketch1_df.crossJoin(sketch2_df.withColumnRenamed("sketch", "sketch2"))
+
+ # Test union with default parameters
+ union1 = joined.select(F.tuple_union_integer("sketch", "sketch2")).first()[0]
+ self.assertIsNotNone(union1)
+
+ # Test union with lgNomEntries and mode
+ union2 = joined.select(F.tuple_union_integer("sketch", "sketch2", 10, "max")).first()[0]
+ self.assertIsNotNone(union2)
+
+ # Test with column names
+ union3 = joined.select(F.tuple_union_integer("sketch", "sketch2", 10)).first()[0]
+ self.assertIsNotNone(union3)
+
+ # Verify estimate from union
+ estimate = joined.select(
+ F.tuple_sketch_estimate_integer(F.tuple_union_integer("sketch", "sketch2"))
+ ).first()[0]
+ self.assertEqual(estimate, 4.0)
+
+ def test_tuple_intersection_double_basic(self):
+ """Test tuple_intersection_double basic functionality"""
+ df1 = self.spark.createDataFrame([(1, 1.5), (2, 2.5), (3, 3.5)], ["key", "summary"])
+ df2 = self.spark.createDataFrame([(2, 1.0), (3, 2.0), (4, 3.0)], ["key", "summary"])
+
+ sketch1_df = df1.agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+ sketch2_df = df2.agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+
+ joined = sketch1_df.crossJoin(sketch2_df.withColumnRenamed("sketch", "sketch2"))
+
+ # Test intersection with default mode
+ intersection1 = joined.select(F.tuple_intersection_double("sketch", "sketch2")).first()[0]
+ self.assertIsNotNone(intersection1)
+ self.assertIsInstance(intersection1, (bytes, bytearray))
+
+ # Test intersection with mode
+ intersection2 = joined.select(
+ F.tuple_intersection_double("sketch", "sketch2", "sum")
+ ).first()[0]
+ self.assertIsNotNone(intersection2)
+
+ # Test with min mode
+ intersection3 = joined.select(
+ F.tuple_intersection_double("sketch", "sketch2", "min")
+ ).first()[0]
+ self.assertIsNotNone(intersection3)
+
+ # Verify estimate from intersection (keys 2 and 3 are common)
+ estimate = joined.select(
+ F.tuple_sketch_estimate_double(F.tuple_intersection_double("sketch", "sketch2"))
+ ).first()[0]
+ self.assertEqual(estimate, 2.0)
+
+ def test_tuple_intersection_integer_basic(self):
+ """Test tuple_intersection_integer basic functionality"""
+ df1 = self.spark.createDataFrame([(1, 10), (2, 20), (3, 30)], ["key", "summary"])
+ df2 = self.spark.createDataFrame([(2, 15), (3, 25), (4, 35)], ["key", "summary"])
+
+ sketch1_df = df1.agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+ sketch2_df = df2.agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+
+ joined = sketch1_df.crossJoin(sketch2_df.withColumnRenamed("sketch", "sketch2"))
+
+ # Test intersection with default mode
+ intersection1 = joined.select(F.tuple_intersection_integer("sketch", "sketch2")).first()[0]
+ self.assertIsNotNone(intersection1)
+
+ # Test intersection with mode
+ intersection2 = joined.select(
+ F.tuple_intersection_integer("sketch", "sketch2", "max")
+ ).first()[0]
+ self.assertIsNotNone(intersection2)
+
+ # Verify estimate from intersection (keys 2 and 3 are common)
+ estimate = joined.select(
+ F.tuple_sketch_estimate_integer(F.tuple_intersection_integer("sketch", "sketch2"))
+ ).first()[0]
+ self.assertEqual(estimate, 2.0)
+
+ def test_tuple_difference_double_basic(self):
+ """Test tuple_difference_double basic functionality"""
+ df1 = self.spark.createDataFrame([(1, 1.5), (2, 2.5), (3, 3.5)], ["key", "summary"])
+ df2 = self.spark.createDataFrame([(2, 1.0), (3, 2.0)], ["key", "summary"])
+
+ sketch1_df = df1.agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+ sketch2_df = df2.agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+
+ joined = sketch1_df.crossJoin(sketch2_df.withColumnRenamed("sketch", "sketch2"))
+
+ # Test difference
+ difference = joined.select(F.tuple_difference_double("sketch", "sketch2")).first()[0]
+ self.assertIsNotNone(difference)
+ self.assertIsInstance(difference, (bytes, bytearray))
+
+ # Verify estimate from difference (only key 1 is unique to df1)
+ estimate = joined.select(
+ F.tuple_sketch_estimate_double(F.tuple_difference_double("sketch", "sketch2"))
+ ).first()[0]
+ self.assertEqual(estimate, 1.0)
+
+ def test_tuple_difference_integer_basic(self):
+ """Test tuple_difference_integer basic functionality"""
+ df1 = self.spark.createDataFrame([(1, 10), (2, 20), (3, 30)], ["key", "summary"])
+ df2 = self.spark.createDataFrame([(2, 15), (3, 25)], ["key", "summary"])
+
+ sketch1_df = df1.agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+ sketch2_df = df2.agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+
+ joined = sketch1_df.crossJoin(sketch2_df.withColumnRenamed("sketch", "sketch2"))
+
+ # Test difference
+ difference = joined.select(F.tuple_difference_integer("sketch", "sketch2")).first()[0]
+ self.assertIsNotNone(difference)
+
+ # Verify estimate from difference (only key 1 is unique to df1)
+ estimate = joined.select(
+ F.tuple_sketch_estimate_integer(F.tuple_difference_integer("sketch", "sketch2"))
+ ).first()[0]
+ self.assertEqual(estimate, 1.0)
+
+ def test_tuple_union_agg_double_basic(self):
+ """Test tuple_union_agg_double basic functionality"""
+ df1 = self.spark.createDataFrame([(1, 1, 1.5), (1, 2, 2.5)], ["id", "key", "summary"])
+ df2 = self.spark.createDataFrame([(1, 3, 3.5), (1, 4, 4.5)], ["id", "key", "summary"])
+
+ sketch1_df = df1.groupBy("id").agg(
+ F.tuple_sketch_agg_double("key", "summary").alias("sketch")
+ )
+ sketch2_df = df2.groupBy("id").agg(
+ F.tuple_sketch_agg_double("key", "summary").alias("sketch")
+ )
+
+ combined = sketch1_df.union(sketch2_df)
+
+ # Test union_agg with default parameters
+ union1 = combined.groupBy("id").agg(F.tuple_union_agg_double("sketch")).first()[1]
+ self.assertIsNotNone(union1)
+ self.assertIsInstance(union1, (bytes, bytearray))
+
+ # Test union_agg with lgNomEntries
+ union2 = combined.groupBy("id").agg(F.tuple_union_agg_double("sketch", 10)).first()[1]
+ self.assertIsNotNone(union2)
+
+ # Test union_agg with lgNomEntries and mode
+ union3 = (
+ combined.groupBy("id").agg(F.tuple_union_agg_double("sketch", 10, "sum")).first()[1]
+ )
+ self.assertIsNotNone(union3)
+
+ # Verify estimate from union_agg (keys 1, 2, 3, 4 should all be present)
+ estimate_result = (
+ combined.groupBy("id")
+ .agg(F.tuple_union_agg_double("sketch").alias("union_sketch"))
+ .select(F.tuple_sketch_estimate_double("union_sketch"))
+ .first()[0]
+ )
+ self.assertEqual(estimate_result, 4.0)
+
+ def test_tuple_union_agg_integer_basic(self):
+ """Test tuple_union_agg_integer basic functionality"""
+ df1 = self.spark.createDataFrame([(1, 1, 10), (1, 2, 20)], ["id", "key", "summary"])
+ df2 = self.spark.createDataFrame([(1, 3, 30), (1, 4, 40)], ["id", "key", "summary"])
+
+ sketch1_df = df1.groupBy("id").agg(
+ F.tuple_sketch_agg_integer("key", "summary").alias("sketch")
+ )
+ sketch2_df = df2.groupBy("id").agg(
+ F.tuple_sketch_agg_integer("key", "summary").alias("sketch")
+ )
+
+ combined = sketch1_df.union(sketch2_df)
+
+ # Test union_agg with default parameters
+ union1 = combined.groupBy("id").agg(F.tuple_union_agg_integer("sketch")).first()[1]
+ self.assertIsNotNone(union1)
+
+ # Test union_agg with lgNomEntries and mode
+ union2 = (
+ combined.groupBy("id").agg(F.tuple_union_agg_integer("sketch", 10, "max")).first()[1]
+ )
+ self.assertIsNotNone(union2)
+
+ # Test with lgNomEntries
+ union3 = combined.groupBy("id").agg(F.tuple_union_agg_integer("sketch", 10)).first()[1]
+ self.assertIsNotNone(union3)
+
+ # Verify estimate from union_agg (keys 1, 2, 3, 4 should all be present)
+ estimate_result = (
+ combined.groupBy("id")
+ .agg(F.tuple_union_agg_integer("sketch").alias("union_sketch"))
+ .select(F.tuple_sketch_estimate_integer("union_sketch"))
+ .first()[0]
+ )
+ self.assertEqual(estimate_result, 4.0)
+
+ def test_tuple_intersection_agg_double_basic(self):
+ """Test tuple_intersection_agg_double basic functionality"""
+ df1 = self.spark.createDataFrame(
+ [(1, 1, 1.5), (1, 2, 2.5), (1, 3, 3.5)], ["id", "key", "summary"]
+ )
+ df2 = self.spark.createDataFrame(
+ [(1, 2, 1.0), (1, 3, 2.0), (1, 4, 3.0)], ["id", "key", "summary"]
+ )
+
+ sketch1_df = df1.groupBy("id").agg(
+ F.tuple_sketch_agg_double("key", "summary").alias("sketch")
+ )
+ sketch2_df = df2.groupBy("id").agg(
+ F.tuple_sketch_agg_double("key", "summary").alias("sketch")
+ )
+
+ combined = sketch1_df.union(sketch2_df)
+
+ # Test intersection_agg with default mode
+ intersection1 = (
+ combined.groupBy("id").agg(F.tuple_intersection_agg_double("sketch")).first()[1]
+ )
+ self.assertIsNotNone(intersection1)
+ self.assertIsInstance(intersection1, (bytes, bytearray))
+
+ # Test intersection_agg with mode
+ intersection2 = (
+ combined.groupBy("id").agg(F.tuple_intersection_agg_double("sketch", "sum")).first()[1]
+ )
+ self.assertIsNotNone(intersection2)
+
+ # Test with min mode
+ intersection3 = (
+ combined.groupBy("id").agg(F.tuple_intersection_agg_double("sketch", "min")).first()[1]
+ )
+ self.assertIsNotNone(intersection3)
+
+ # Verify estimate from intersection_agg (keys 2 and 3 are common)
+ estimate_result = (
+ combined.groupBy("id")
+ .agg(F.tuple_intersection_agg_double("sketch").alias("intersection_sketch"))
+ .select(F.tuple_sketch_estimate_double("intersection_sketch"))
+ .first()[0]
+ )
+ self.assertEqual(estimate_result, 2.0)
+
+ def test_tuple_intersection_agg_integer_basic(self):
+ """Test tuple_intersection_agg_integer basic functionality"""
+ df1 = self.spark.createDataFrame(
+ [(1, 1, 10), (1, 2, 20), (1, 3, 30)], ["id", "key", "summary"]
+ )
+ df2 = self.spark.createDataFrame(
+ [(1, 2, 15), (1, 3, 25), (1, 4, 35)], ["id", "key", "summary"]
+ )
+
+ sketch1_df = df1.groupBy("id").agg(
+ F.tuple_sketch_agg_integer("key", "summary").alias("sketch")
+ )
+ sketch2_df = df2.groupBy("id").agg(
+ F.tuple_sketch_agg_integer("key", "summary").alias("sketch")
+ )
+
+ combined = sketch1_df.union(sketch2_df)
+
+ # Test intersection_agg with default mode
+ intersection1 = (
+ combined.groupBy("id").agg(F.tuple_intersection_agg_integer("sketch")).first()[1]
+ )
+ self.assertIsNotNone(intersection1)
+
+ # Test intersection_agg with mode
+ intersection2 = (
+ combined.groupBy("id").agg(F.tuple_intersection_agg_integer("sketch", "max")).first()[1]
+ )
+ self.assertIsNotNone(intersection2)
+
+ # Verify estimate from intersection_agg (keys 2 and 3 are common)
+ estimate_result = (
+ combined.groupBy("id")
+ .agg(F.tuple_intersection_agg_integer("sketch").alias("intersection_sketch"))
+ .select(F.tuple_sketch_estimate_integer("intersection_sketch"))
+ .first()[0]
+ )
+ self.assertEqual(estimate_result, 2.0)
+
+ def test_tuple_sketch_with_nulls(self):
+ """Test tuple sketch with null values"""
+ df = self.spark.createDataFrame([(1, 10), (2, 20), (3, None)], ["key", "summary"])
+
+ # Null summaries should be handled
+ sketch = df.agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+ estimate = sketch.select(F.tuple_sketch_estimate_integer("sketch")).first()[0]
+ self.assertIsNotNone(estimate)
+ self.assertEqual(estimate, 2.0)
+
+ def test_tuple_sketch_comprehensive_double(self):
+ """Test tuple_sketch_agg + operations + estimate comprehensive test - double"""
+ df1 = self.spark.createDataFrame(
+ [(1, "a", 1.0), (1, "a", 2.0), (1, "b", 3.0), (1, "c", 4.0)],
+ ["id", "key", "summary"],
+ )
+
+ df2 = self.spark.createDataFrame(
+ [(1, "a", 0.5), (1, "c", 1.5), (1, "d", 2.5), (1, "e", 3.5)],
+ ["id", "key", "summary"],
+ )
+
+ # Test tuple_sketch_agg and estimate
+ res1 = (
+ df1.groupBy("id")
+ .agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+ .withColumn("estimate", F.tuple_sketch_estimate_double("sketch"))
+ .withColumn("summary_total", F.tuple_sketch_summary_double("sketch"))
+ )
+
+ row1 = res1.first()
+ self.assertGreaterEqual(row1["estimate"], 3.0)
+ self.assertGreater(row1["summary_total"], 0.0)
+
+ # Test union operations
+ sketch1 = df1.groupBy("id").agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+ sketch2 = df2.groupBy("id").agg(F.tuple_sketch_agg_double("key", "summary").alias("sketch"))
+
+ union_result = (
+ sketch1.union(sketch2)
+ .groupBy("id")
+ .agg(F.tuple_union_agg_double("sketch").alias("union_sketch"))
+ .withColumn("union_estimate", F.tuple_sketch_estimate_double("union_sketch"))
+ )
+
+ union_row = union_result.first()
+ self.assertGreaterEqual(union_row["union_estimate"], 5.0)
+
+ # Test intersection
+ joined = sketch1.crossJoin(sketch2.withColumnRenamed("sketch", "sketch2"))
+ intersection_result = joined.withColumn(
+ "intersection_sketch", F.tuple_intersection_double("sketch", "sketch2")
+ ).withColumn("intersection_estimate", F.tuple_sketch_estimate_double("intersection_sketch"))
+
+ intersection_row = intersection_result.first()
+ self.assertGreaterEqual(intersection_row["intersection_estimate"], 2.0)
+
+ # Test difference
+ difference_result = joined.withColumn(
+ "difference_sketch", F.tuple_difference_double("sketch", "sketch2")
+ ).withColumn("difference_estimate", F.tuple_sketch_estimate_double("difference_sketch"))
+
+ difference_row = difference_result.first()
+ self.assertGreaterEqual(difference_row["difference_estimate"], 1.0)
+
+ def test_tuple_sketch_comprehensive_integer(self):
+ """Test tuple_sketch_agg + operations + estimate comprehensive test - integer"""
+ df1 = self.spark.createDataFrame(
+ [(1, "a", 10), (1, "a", 20), (1, "b", 30), (1, "c", 40)], ["id", "key", "summary"]
+ )
+
+ df2 = self.spark.createDataFrame(
+ [(1, "a", 5), (1, "c", 15), (1, "d", 25), (1, "e", 35)], ["id", "key", "summary"]
+ )
+
+ # Test tuple_sketch_agg and estimate
+ res1 = (
+ df1.groupBy("id")
+ .agg(F.tuple_sketch_agg_integer("key", "summary").alias("sketch"))
+ .withColumn("estimate", F.tuple_sketch_estimate_integer("sketch"))
+ .withColumn("summary_total", F.tuple_sketch_summary_integer("sketch"))
+ )
+
+ row1 = res1.first()
+ self.assertGreaterEqual(row1["estimate"], 3.0)
+ self.assertGreater(row1["summary_total"], 0)
+
+ # Test with different modes
+ res_max = (
+ df1.groupBy("id")
+ .agg(F.tuple_sketch_agg_integer("key", "summary", 12, "max").alias("sketch"))
+ .withColumn("summary_max", F.tuple_sketch_summary_integer("sketch", "max"))
+ )
+
+ row_max = res_max.first()
+ self.assertGreater(row_max["summary_max"], 0)
+
+ # Test union with mode
+ sketch1 = df1.groupBy("id").agg(
+ F.tuple_sketch_agg_integer("key", "summary").alias("sketch")
+ )
+ sketch2 = df2.groupBy("id").agg(
+ F.tuple_sketch_agg_integer("key", "summary").alias("sketch")
+ )
+
+ union_result = (
+ sketch1.union(sketch2)
+ .groupBy("id")
+ .agg(F.tuple_union_agg_integer("sketch", 12, "sum").alias("union_sketch"))
+ .withColumn("union_estimate", F.tuple_sketch_estimate_integer("union_sketch"))
+ )
+
+ union_row = union_result.first()
+ self.assertGreaterEqual(union_row["union_estimate"], 5.0)
+
def test_datetime_functions(self):
df = self.spark.range(1).selectExpr("'2017-01-22' as dateCol")
parse_result = df.select(F.to_date(F.col("dateCol"))).first()
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/functions.scala b/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
index 136b2535374b..1a261eae6a10 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
@@ -1304,6 +1304,493 @@ object functions {
def theta_union_agg(columnName: String): Column =
theta_union_agg(Column(columnName))
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by intersecting the Datasketches TupleSketch instances
+ * in the input column via a Datasketches Intersection instance. The mode parameter specifies
+ * the aggregation mode for numeric summaries during intersection (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_double(e: Column, mode: Column): Column =
+ Column.fn("tuple_intersection_agg_double", e, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by intersecting the Datasketches TupleSketch instances
+ * in the input column via a Datasketches Intersection instance. The mode parameter specifies
+ * the aggregation mode for numeric summaries during intersection (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_double(e: Column, mode: String): Column =
+ Column.fn("tuple_intersection_agg_double", e, lit(mode))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by intersecting the Datasketches TupleSketch instances
+ * in the input column via a Datasketches Intersection instance. The mode parameter specifies
+ * the aggregation mode for numeric summaries during intersection (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_double(columnName: String, mode: String): Column =
+ tuple_intersection_agg_double(Column(columnName), mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by intersecting the Datasketches TupleSketch instances
+ * in the input column via a Datasketches Intersection instance. It is configured with the
+ * default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_double(e: Column): Column =
+ Column.fn("tuple_intersection_agg_double", e)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by intersecting the Datasketches TupleSketch instances
+ * in the input column via a Datasketches Intersection instance. It is configured with the
+ * default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_double(columnName: String): Column =
+ tuple_intersection_agg_double(Column(columnName))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by intersecting the Datasketches TupleSketch
+ * instances in the input column via a Datasketches Intersection instance. The mode parameter
+ * specifies the aggregation mode for numeric summaries during intersection (sum, min, max,
+ * alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_integer(e: Column, mode: Column): Column =
+ Column.fn("tuple_intersection_agg_integer", e, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by intersecting the Datasketches TupleSketch
+ * instances in the input column via a Datasketches Intersection instance. The mode parameter
+ * specifies the aggregation mode for numeric summaries during intersection (sum, min, max,
+ * alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_integer(e: Column, mode: String): Column =
+ Column.fn("tuple_intersection_agg_integer", e, lit(mode))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by intersecting the Datasketches TupleSketch
+ * instances in the input column via a Datasketches Intersection instance. The mode parameter
+ * specifies the aggregation mode for numeric summaries during intersection (sum, min, max,
+ * alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_integer(columnName: String, mode: String): Column =
+ tuple_intersection_agg_integer(Column(columnName), mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by intersecting the Datasketches TupleSketch
+ * instances in the input column via a Datasketches Intersection instance. It is configured with
+ * the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_integer(e: Column): Column =
+ Column.fn("tuple_intersection_agg_integer", e)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by intersecting the Datasketches TupleSketch
+ * instances in the input column via a Datasketches Intersection instance. It is configured with
+ * the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_agg_integer(columnName: String): Column =
+ tuple_intersection_agg_integer(Column(columnName))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries and aggregation mode. The mode parameter
+ * specifies the aggregation mode for numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_double(
+ key: Column,
+ summary: Column,
+ lgNomEntries: Column,
+ mode: Column): Column =
+ Column.fn("tuple_sketch_agg_double", key, summary, lgNomEntries, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries and aggregation mode. The mode parameter
+ * specifies the aggregation mode for numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_double(
+ key: Column,
+ summary: Column,
+ lgNomEntries: Int,
+ mode: String): Column =
+ Column.fn("tuple_sketch_agg_double", key, summary, lit(lgNomEntries), lit(mode))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries and aggregation mode. The mode parameter
+ * specifies the aggregation mode for numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_double(
+ keyColumnName: String,
+ summaryColumnName: String,
+ lgNomEntries: Int,
+ mode: String): Column =
+ tuple_sketch_agg_double(Column(keyColumnName), Column(summaryColumnName), lgNomEntries, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries. It uses the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_double(key: Column, summary: Column, lgNomEntries: Int): Column =
+ Column.fn("tuple_sketch_agg_double", key, summary, lit(lgNomEntries))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries. It uses the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_double(
+ keyColumnName: String,
+ summaryColumnName: String,
+ lgNomEntries: Int): Column =
+ tuple_sketch_agg_double(Column(keyColumnName), Column(summaryColumnName), lgNomEntries)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary built with the key and summary values in the input columns. It
+ * uses the default values of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_double(key: Column, summary: Column): Column =
+ Column.fn("tuple_sketch_agg_double", key, summary)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary built with the key and summary values in the input columns. It
+ * uses the default values of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_double(keyColumnName: String, summaryColumnName: String): Column =
+ tuple_sketch_agg_double(Column(keyColumnName), Column(summaryColumnName))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries and aggregation mode. The mode parameter
+ * specifies the aggregation mode for numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_integer(
+ key: Column,
+ summary: Column,
+ lgNomEntries: Column,
+ mode: Column): Column =
+ Column.fn("tuple_sketch_agg_integer", key, summary, lgNomEntries, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries and aggregation mode. The mode parameter
+ * specifies the aggregation mode for numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_integer(
+ key: Column,
+ summary: Column,
+ lgNomEntries: Int,
+ mode: String): Column =
+ Column.fn("tuple_sketch_agg_integer", key, summary, lit(lgNomEntries), lit(mode))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries and aggregation mode. The mode parameter
+ * specifies the aggregation mode for numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_integer(
+ keyColumnName: String,
+ summaryColumnName: String,
+ lgNomEntries: Int,
+ mode: String): Column =
+ tuple_sketch_agg_integer(Column(keyColumnName), Column(summaryColumnName), lgNomEntries, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries. It uses the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_integer(key: Column, summary: Column, lgNomEntries: Int): Column =
+ Column.fn("tuple_sketch_agg_integer", key, summary, lit(lgNomEntries))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary built with the key and summary values in the input columns and
+ * configured with the `lgNomEntries` nominal entries. It uses the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_integer(
+ keyColumnName: String,
+ summaryColumnName: String,
+ lgNomEntries: Int): Column =
+ tuple_sketch_agg_integer(Column(keyColumnName), Column(summaryColumnName), lgNomEntries)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary built with the key and summary values in the input columns. It
+ * uses the default values of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_integer(key: Column, summary: Column): Column =
+ Column.fn("tuple_sketch_agg_integer", key, summary)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary built with the key and summary values in the input columns. It
+ * uses the default values of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_agg_integer(keyColumnName: String, summaryColumnName: String): Column =
+ tuple_sketch_agg_integer(Column(keyColumnName), Column(summaryColumnName))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for numeric
+ * summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_double(e: Column, lgNomEntries: Column, mode: Column): Column =
+ Column.fn("tuple_union_agg_double", e, lgNomEntries, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for numeric
+ * summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_double(e: Column, lgNomEntries: Int, mode: String): Column =
+ Column.fn("tuple_union_agg_double", e, lit(lgNomEntries), lit(mode))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for numeric
+ * summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_double(columnName: String, lgNomEntries: Int, mode: String): Column =
+ tuple_union_agg_double(Column(columnName), lgNomEntries, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer. It uses the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_double(e: Column, lgNomEntries: Int): Column =
+ Column.fn("tuple_union_agg_double", e, lit(lgNomEntries))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer. It uses the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_double(columnName: String, lgNomEntries: Int): Column =
+ tuple_union_agg_double(Column(columnName), lgNomEntries)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It is configured with the default values
+ * of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_double(e: Column): Column =
+ Column.fn("tuple_union_agg_double", e)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with a double type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It is configured with the default values
+ * of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_double(columnName: String): Column =
+ tuple_union_agg_double(Column(columnName))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for numeric
+ * summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_integer(e: Column, lgNomEntries: Column, mode: Column): Column =
+ Column.fn("tuple_union_agg_integer", e, lgNomEntries, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for numeric
+ * summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_integer(e: Column, lgNomEntries: Int, mode: String): Column =
+ Column.fn("tuple_union_agg_integer", e, lit(lgNomEntries), lit(mode))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for numeric
+ * summaries (sum, min, max, alwaysone).
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_integer(columnName: String, lgNomEntries: Int, mode: String): Column =
+ tuple_union_agg_integer(Column(columnName), lgNomEntries, mode)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer. It uses the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_integer(e: Column, lgNomEntries: Int): Column =
+ Column.fn("tuple_union_agg_integer", e, lit(lgNomEntries))
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It allows the configuration of
+ * `lgNomEntries` log nominal entries for the union buffer. It uses the default mode of 'sum'.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_integer(columnName: String, lgNomEntries: Int): Column =
+ tuple_union_agg_integer(Column(columnName), lgNomEntries)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It is configured with the default values
+ * of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_integer(e: Column): Column =
+ Column.fn("tuple_union_agg_integer", e)
+
+ /**
+ * Aggregate function: returns the compact binary representation of the Datasketches TupleSketch
+ * with an integer type summary, generated by the union of Datasketches TupleSketch instances in
+ * the input column via a Datasketches Union instance. It is configured with the default values
+ * of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group agg_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_agg_integer(columnName: String): Column =
+ tuple_union_agg_integer(Column(columnName))
+
/**
* Aggregate function: returns the compact binary representation of the Datasketches
* KllLongsSketch built with the values in the input column. The optional k parameter controls
@@ -4133,6 +4620,530 @@ object functions {
def theta_union(c1: Column, c2: Column, lgNomEntries: Column): Column =
Column.fn("theta_union", c1, c2, lgNomEntries)
+ /**
+ * Subtracts two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches AnotB object. Returns elements in the
+ * first sketch that are not in the second sketch.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_difference_double(c1: Column, c2: Column): Column =
+ Column.fn("tuple_difference_double", c1, c2)
+
+ /**
+ * Subtracts two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches AnotB object. Returns elements in the
+ * first sketch that are not in the second sketch.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_difference_double(columnName1: String, columnName2: String): Column =
+ tuple_difference_double(Column(columnName1), Column(columnName2))
+
+ /**
+ * Subtracts two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches AnotB object. Returns elements in the
+ * first sketch that are not in the second sketch.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_difference_integer(c1: Column, c2: Column): Column =
+ Column.fn("tuple_difference_integer", c1, c2)
+
+ /**
+ * Subtracts two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches AnotB object. Returns elements in the
+ * first sketch that are not in the second sketch.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_difference_integer(columnName1: String, columnName2: String): Column =
+ tuple_difference_integer(Column(columnName1), Column(columnName2))
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Intersection object. The mode parameter
+ * specifies the aggregation mode for numeric summaries during intersection (sum, min, max,
+ * alwaysone). It is configured with the default mode of 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_double(c1: Column, c2: Column): Column =
+ Column.fn("tuple_intersection_double", c1, c2)
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Intersection object. The mode parameter
+ * specifies the aggregation mode for numeric summaries during intersection (sum, min, max,
+ * alwaysone). It is configured with the default mode of 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_double(columnName1: String, columnName2: String): Column =
+ tuple_intersection_double(Column(columnName1), Column(columnName2))
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Intersection object. The mode parameter
+ * specifies the aggregation mode for numeric summaries during intersection (sum, min, max,
+ * alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_double(c1: Column, c2: Column, mode: String): Column =
+ Column.fn("tuple_intersection_double", c1, c2, lit(mode))
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Intersection object. The mode parameter
+ * specifies the aggregation mode for numeric summaries during intersection (sum, min, max,
+ * alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_double(columnName1: String, columnName2: String, mode: String): Column =
+ tuple_intersection_double(Column(columnName1), Column(columnName2), mode)
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Intersection object. The mode parameter
+ * specifies the aggregation mode for numeric summaries during intersection (sum, min, max,
+ * alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_double(c1: Column, c2: Column, mode: Column): Column =
+ Column.fn("tuple_intersection_double", c1, c2, mode)
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with integer
+ * summary data type in the input columns using a Datasketches Intersection object. The mode
+ * parameter specifies the aggregation mode for numeric summaries during intersection (sum, min,
+ * max, alwaysone). It is configured with the default mode of 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_integer(c1: Column, c2: Column): Column =
+ Column.fn("tuple_intersection_integer", c1, c2)
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with integer
+ * summary data type in the input columns using a Datasketches Intersection object. The mode
+ * parameter specifies the aggregation mode for numeric summaries during intersection (sum, min,
+ * max, alwaysone). It is configured with the default mode of 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_integer(columnName1: String, columnName2: String): Column =
+ tuple_intersection_integer(Column(columnName1), Column(columnName2))
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with integer
+ * summary data type in the input columns using a Datasketches Intersection object. The mode
+ * parameter specifies the aggregation mode for numeric summaries during intersection (sum, min,
+ * max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_integer(c1: Column, c2: Column, mode: String): Column =
+ Column.fn("tuple_intersection_integer", c1, c2, lit(mode))
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with integer
+ * summary data type in the input columns using a Datasketches Intersection object. The mode
+ * parameter specifies the aggregation mode for numeric summaries during intersection (sum, min,
+ * max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_integer(columnName1: String, columnName2: String, mode: String): Column =
+ tuple_intersection_integer(Column(columnName1), Column(columnName2), mode)
+
+ /**
+ * Intersects two binary representations of Datasketches TupleSketch objects with integer
+ * summary data type in the input columns using a Datasketches Intersection object. The mode
+ * parameter specifies the aggregation mode for numeric summaries during intersection (sum, min,
+ * max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_intersection_integer(c1: Column, c2: Column, mode: Column): Column =
+ Column.fn("tuple_intersection_integer", c1, c2, mode)
+
+ /**
+ * Returns the estimated number of unique values given the binary representation of a
+ * Datasketches TupleSketch with double summary data type.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_estimate_double(c: Column): Column =
+ Column.fn("tuple_sketch_estimate_double", c)
+
+ /**
+ * Returns the estimated number of unique values given the binary representation of a
+ * Datasketches TupleSketch with double summary data type.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_estimate_double(columnName: String): Column =
+ tuple_sketch_estimate_double(Column(columnName))
+
+ /**
+ * Returns the estimated number of unique values given the binary representation of a
+ * Datasketches TupleSketch with integer summary data type.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_estimate_integer(c: Column): Column =
+ Column.fn("tuple_sketch_estimate_integer", c)
+
+ /**
+ * Returns the estimated number of unique values given the binary representation of a
+ * Datasketches TupleSketch with integer summary data type.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_estimate_integer(columnName: String): Column =
+ tuple_sketch_estimate_integer(Column(columnName))
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with double summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone). It is
+ * configured with the default mode of 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_double(c: Column): Column =
+ Column.fn("tuple_sketch_summary_double", c)
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with double summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone). It is
+ * configured with the default mode of 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_double(columnName: String): Column =
+ tuple_sketch_summary_double(Column(columnName))
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with double summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_double(c: Column, mode: String): Column =
+ Column.fn("tuple_sketch_summary_double", c, lit(mode))
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with double summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_double(columnName: String, mode: String): Column =
+ tuple_sketch_summary_double(Column(columnName), mode)
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with double summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_double(c: Column, mode: Column): Column =
+ Column.fn("tuple_sketch_summary_double", c, mode)
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with integer summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone). It is
+ * configured with the default mode of 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_integer(c: Column): Column =
+ Column.fn("tuple_sketch_summary_integer", c)
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with integer summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone). It is
+ * configured with the default mode of 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_integer(columnName: String): Column =
+ tuple_sketch_summary_integer(Column(columnName))
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with integer summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_integer(c: Column, mode: String): Column =
+ Column.fn("tuple_sketch_summary_integer", c, lit(mode))
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with integer summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_integer(columnName: String, mode: String): Column =
+ tuple_sketch_summary_integer(Column(columnName), mode)
+
+ /**
+ * Aggregates the summary values from a Datasketches TupleSketch with integer summary data type.
+ * The mode parameter specifies the aggregation mode (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_summary_integer(c: Column, mode: Column): Column =
+ Column.fn("tuple_sketch_summary_integer", c, mode)
+
+ /**
+ * Returns the theta value (sampling rate) from a Datasketches TupleSketch with double summary
+ * data type. The theta value represents the effective sampling rate of the sketch, between 0.0
+ * and 1.0.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_theta_double(c: Column): Column =
+ Column.fn("tuple_sketch_theta_double", c)
+
+ /**
+ * Returns the theta value (sampling rate) from a Datasketches TupleSketch with double summary
+ * data type. The theta value represents the effective sampling rate of the sketch, between 0.0
+ * and 1.0.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_theta_double(columnName: String): Column =
+ tuple_sketch_theta_double(Column(columnName))
+
+ /**
+ * Returns the theta value (sampling rate) from a Datasketches TupleSketch with integer summary
+ * data type. The theta value represents the effective sampling rate of the sketch, between 0.0
+ * and 1.0.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_theta_integer(c: Column): Column =
+ Column.fn("tuple_sketch_theta_integer", c)
+
+ /**
+ * Returns the theta value (sampling rate) from a Datasketches TupleSketch with integer summary
+ * data type. The theta value represents the effective sampling rate of the sketch, between 0.0
+ * and 1.0.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_sketch_theta_integer(columnName: String): Column =
+ tuple_sketch_theta_integer(Column(columnName))
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Union object. It is configured with the
+ * default values of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_double(c1: Column, c2: Column): Column =
+ Column.fn("tuple_union_double", c1, c2)
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Union object. It is configured with the
+ * default values of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_double(columnName1: String, columnName2: String): Column =
+ tuple_union_double(Column(columnName1), Column(columnName2))
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer. It uses the default mode of
+ * 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_double(c1: Column, c2: Column, lgNomEntries: Int): Column =
+ Column.fn("tuple_union_double", c1, c2, lit(lgNomEntries))
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer. It uses the default mode of
+ * 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_double(columnName1: String, columnName2: String, lgNomEntries: Int): Column =
+ tuple_union_double(Column(columnName1), Column(columnName2), lgNomEntries)
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for
+ * numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_double(c1: Column, c2: Column, lgNomEntries: Int, mode: String): Column =
+ Column.fn("tuple_union_double", c1, c2, lit(lgNomEntries), lit(mode))
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for
+ * numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_double(
+ columnName1: String,
+ columnName2: String,
+ lgNomEntries: Int,
+ mode: String): Column =
+ tuple_union_double(Column(columnName1), Column(columnName2), lgNomEntries, mode)
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with double summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for
+ * numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_double(c1: Column, c2: Column, lgNomEntries: Column, mode: Column): Column =
+ Column.fn("tuple_union_double", c1, c2, lgNomEntries, mode)
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches Union object. It is configured with the
+ * default values of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_integer(c1: Column, c2: Column): Column =
+ Column.fn("tuple_union_integer", c1, c2)
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches Union object. It is configured with the
+ * default values of 12 for `lgNomEntries` and 'sum' for mode.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_integer(columnName1: String, columnName2: String): Column =
+ tuple_union_integer(Column(columnName1), Column(columnName2))
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer. It uses the default mode of
+ * 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_integer(c1: Column, c2: Column, lgNomEntries: Int): Column =
+ Column.fn("tuple_union_integer", c1, c2, lit(lgNomEntries))
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer. It uses the default mode of
+ * 'sum'.
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_integer(columnName1: String, columnName2: String, lgNomEntries: Int): Column =
+ tuple_union_integer(Column(columnName1), Column(columnName2), lgNomEntries)
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for
+ * numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_integer(c1: Column, c2: Column, lgNomEntries: Int, mode: String): Column =
+ Column.fn("tuple_union_integer", c1, c2, lit(lgNomEntries), lit(mode))
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for
+ * numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_integer(
+ columnName1: String,
+ columnName2: String,
+ lgNomEntries: Int,
+ mode: String): Column =
+ tuple_union_integer(Column(columnName1), Column(columnName2), lgNomEntries, mode)
+
+ /**
+ * Unions two binary representations of Datasketches TupleSketch objects with integer summary
+ * data type in the input columns using a Datasketches Union object. It allows the configuration
+ * of `lgNomEntries` log nominal entries for the union buffer and the aggregation mode for
+ * numeric summaries (sum, min, max, alwaysone).
+ *
+ * @group misc_funcs
+ * @since 4.2.0
+ */
+ def tuple_union_integer(c1: Column, c2: Column, lgNomEntries: Column, mode: Column): Column =
+ Column.fn("tuple_union_integer", c1, c2, lgNomEntries, mode)
+
/**
* Returns a string with human readable summary information about the KLL bigint sketch.
*
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
index 0a9e08f5c57e..ffc85fef1d10 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -3596,6 +3596,611 @@ class DataFrameAggregateSuite extends QueryTest
val n = merged.select(kll_sketch_get_n_bigint($"merged_sketch")).collect()(0)(0)
assert(n == 5L)
}
+
+ test("SPARK-54179: tuple_sketch_agg_double basic functionality") {
+ val df = Seq((1, 1.5), (2, 2.5), (3, 3.5), (1, 0.5), (2, 1.0)).toDF("key", "summary")
+
+ // Test with default parameters
+ val sketch1 = df.agg(tuple_sketch_agg_double($"key", $"summary")).collect()(0)(0)
+ assert(sketch1 != null)
+ assert(sketch1.asInstanceOf[Array[Byte]].length > 0)
+
+ // Test with explicit lgNomEntries
+ val sketch2 = df.agg(tuple_sketch_agg_double($"key", $"summary", 10)).collect()(0)(0)
+ assert(sketch2 != null)
+ assert(sketch2.asInstanceOf[Array[Byte]].length > 0)
+
+ // Test with lgNomEntries and mode
+ val sketch3 = df.agg(tuple_sketch_agg_double($"key", $"summary", 10, "sum")).collect()(0)(0)
+ assert(sketch3 != null)
+
+ // Test with column names
+ val sketch4 = df.agg(tuple_sketch_agg_double("key", "summary")).collect()(0)(0)
+ assert(sketch4 != null)
+ }
+
+ test("SPARK-54179: tuple_sketch_agg_integer basic functionality") {
+ val df = Seq((1, 10), (2, 20), (3, 30), (1, 5), (2, 15)).toDF("key", "summary")
+
+ // Test with default parameters
+ val sketch1 = df.agg(tuple_sketch_agg_integer($"key", $"summary")).collect()(0)(0)
+ assert(sketch1 != null)
+ assert(sketch1.asInstanceOf[Array[Byte]].length > 0)
+
+ // Test with explicit lgNomEntries
+ val sketch2 = df.agg(tuple_sketch_agg_integer($"key", $"summary", 10)).collect()(0)(0)
+ assert(sketch2 != null)
+
+ // Test with lgNomEntries and mode
+ val sketch3 = df.agg(tuple_sketch_agg_integer($"key", $"summary", 10, "max")).collect()(0)(0)
+ assert(sketch3 != null)
+
+ // Test with column names
+ val sketch4 = df.agg(tuple_sketch_agg_integer("key", "summary", 10, "min")).collect()(0)(0)
+ assert(sketch4 != null)
+ }
+
+ test("SPARK-54179: tuple_sketch_estimate and summary functions - double") {
+ val df = Seq((1, 1.5), (2, 2.5), (3, 3.5), (1, 0.5), (2, 1.0)).toDF("key", "summary")
+ val sketchDf = df.agg(tuple_sketch_agg_double($"key", $"summary").alias("sketch"))
+
+ // Test estimate
+ val estimate = sketchDf.select(tuple_sketch_estimate_double($"sketch")).collect()(0)(0)
+ assert(estimate == 3.0)
+
+ // Test summary with default mode (sum)
+ val summary1 = sketchDf.select(tuple_sketch_summary_double($"sketch")).collect()(0)(0)
+ assert(summary1 == 9.0)
+
+ // Test summary with explicit mode
+ val summary2 = sketchDf.select(tuple_sketch_summary_double($"sketch", "min")).collect()(0)(0)
+ assert(summary2 == 2.0)
+
+ // Test theta
+ val theta = sketchDf.select(tuple_sketch_theta_double($"sketch")).collect()(0)(0)
+ assert(theta != null)
+ assert(theta.asInstanceOf[Double] > 0.0 && theta.asInstanceOf[Double] <= 1.0)
+
+ // Test with column names
+ val estimate2 = sketchDf.select(tuple_sketch_estimate_double("sketch")).collect()(0)(0)
+ assert(estimate2 != null)
+ }
+
+ test("SPARK-54179: tuple_sketch_estimate and summary functions - integer") {
+ val df = Seq((1, 10), (2, 20), (3, 30), (1, 5), (2, 15)).toDF("key", "summary")
+ val sketchDf = df.agg(tuple_sketch_agg_integer($"key", $"summary").alias("sketch"))
+
+ // Test estimate
+ val estimate = sketchDf.select(tuple_sketch_estimate_integer($"sketch")).collect()(0)(0)
+ assert(estimate == 3.0)
+
+ // Test summary with default mode (sum)
+ val summary1 = sketchDf.select(tuple_sketch_summary_integer($"sketch")).collect()(0)(0)
+ assert(summary1 == 80)
+
+ // Test summary with explicit mode
+ val summary2 = sketchDf.select(tuple_sketch_summary_integer($"sketch", "max")).collect()(0)(0)
+ assert(summary2 == 35)
+
+ // Test theta
+ val theta = sketchDf.select(tuple_sketch_theta_integer($"sketch")).collect()(0)(0)
+ assert(theta != null)
+ assert(theta.asInstanceOf[Double] > 0.0 && theta.asInstanceOf[Double] <= 1.0)
+
+ // Test with column names
+ val summary3 = sketchDf.select(tuple_sketch_summary_integer("sketch", "min")).collect()(0)(0)
+ assert(summary3 == 15)
+ }
+
+ test("SPARK-54179: tuple_union_double basic functionality") {
+ val df1 = Seq((1, 1.5), (2, 2.5)).toDF("key", "summary")
+ val df2 = Seq((3, 3.5), (4, 4.5)).toDF("key", "summary")
+
+ val sketch1Df = df1.agg(tuple_sketch_agg_double($"key", $"summary").alias("sketch"))
+ val sketch2Df = df2.agg(tuple_sketch_agg_double($"key", $"summary").alias("sketch"))
+
+ val joined = sketch1Df.crossJoin(sketch2Df.withColumnRenamed("sketch", "sketch2"))
+
+ // Test union with default parameters
+ val union1 = joined.select(tuple_union_double($"sketch", $"sketch2")).collect()(0)(0)
+ assert(union1 != null)
+ assert(union1.asInstanceOf[Array[Byte]].length > 0)
+
+ // Test union with lgNomEntries
+ val union2 = joined.select(tuple_union_double($"sketch", $"sketch2", 10)).collect()(0)(0)
+ assert(union2 != null)
+
+ // Test union with lgNomEntries and mode
+ val union3 =
+ joined.select(tuple_union_double($"sketch", $"sketch2", 10, "sum")).collect()(0)(0)
+ assert(union3 != null)
+
+ // Test with column names
+ val union4 = joined.select(tuple_union_double("sketch", "sketch2")).collect()(0)(0)
+ assert(union4 != null)
+
+ // Verify estimate from union
+ val estimate = joined
+ .select(tuple_sketch_estimate_double(tuple_union_double($"sketch", $"sketch2")))
+ .collect()(0)(0)
+ assert(estimate == 4.0)
+ }
+
+ test("SPARK-54179: tuple_union_integer basic functionality") {
+ val df1 = Seq((1, 10), (2, 20)).toDF("key", "summary")
+ val df2 = Seq((3, 30), (4, 40)).toDF("key", "summary")
+
+ val sketch1Df = df1.agg(tuple_sketch_agg_integer($"key", $"summary").alias("sketch"))
+ val sketch2Df = df2.agg(tuple_sketch_agg_integer($"key", $"summary").alias("sketch"))
+
+ val joined = sketch1Df.crossJoin(sketch2Df.withColumnRenamed("sketch", "sketch2"))
+
+ // Test union with default parameters
+ val union1 = joined.select(tuple_union_integer($"sketch", $"sketch2")).collect()(0)(0)
+ assert(union1 != null)
+
+ // Test union with lgNomEntries and mode
+ val union2 = joined
+ .select(tuple_union_integer($"sketch", $"sketch2", 10, "max"))
+ .collect()(0)(0)
+ assert(union2 != null)
+
+ // Test with column names
+ val union3 = joined.select(tuple_union_integer("sketch", "sketch2", 10)).collect()(0)(0)
+ assert(union3 != null)
+
+ // Verify estimate from union
+ val estimate = joined
+ .select(tuple_sketch_estimate_integer(tuple_union_integer($"sketch", $"sketch2")))
+ .collect()(0)(0)
+ assert(estimate == 4.0)
+ }
+
+ test("SPARK-54179: tuple_intersection_double basic functionality") {
+ val df1 = Seq((1, 1.5), (2, 2.5), (3, 3.5)).toDF("key", "summary")
+ val df2 = Seq((2, 1.0), (3, 2.0), (4, 3.0)).toDF("key", "summary")
+
+ val sketch1Df = df1.agg(tuple_sketch_agg_double($"key", $"summary").alias("sketch"))
+ val sketch2Df = df2.agg(tuple_sketch_agg_double($"key", $"summary").alias("sketch"))
+
+ val joined = sketch1Df.crossJoin(sketch2Df.withColumnRenamed("sketch", "sketch2"))
+
+ // Test intersection with default mode
+ val intersection1 = joined
+ .select(tuple_intersection_double($"sketch", $"sketch2"))
+ .collect()(0)(0)
+ assert(intersection1 != null)
+ assert(intersection1.asInstanceOf[Array[Byte]].length > 0)
+
+ // Test intersection with mode
+ val intersection2 = joined
+ .select(tuple_intersection_double($"sketch", $"sketch2", "sum"))
+ .collect()(0)(0)
+ assert(intersection2 != null)
+
+ // Test with column names
+ val intersection3 = joined
+ .select(tuple_intersection_double("sketch", "sketch2", "min"))
+ .collect()(0)(0)
+ assert(intersection3 != null)
+
+ // Verify estimate from intersection (keys 2 and 3 are common)
+ val estimate = joined
+ .select(tuple_sketch_estimate_double(tuple_intersection_double($"sketch", $"sketch2")))
+ .collect()(0)(0)
+ assert(estimate == 2.0)
+ }
+
+ test("SPARK-54179: tuple_intersection_integer basic functionality") {
+ val df1 = Seq((1, 10), (2, 20), (3, 30)).toDF("key", "summary")
+ val df2 = Seq((2, 15), (3, 25), (4, 35)).toDF("key", "summary")
+
+ val sketch1Df = df1.agg(tuple_sketch_agg_integer($"key", $"summary").alias("sketch"))
+ val sketch2Df = df2.agg(tuple_sketch_agg_integer($"key", $"summary").alias("sketch"))
+
+ val joined = sketch1Df.crossJoin(sketch2Df.withColumnRenamed("sketch", "sketch2"))
+
+ // Test intersection with default mode
+ val intersection1 = joined
+ .select(tuple_intersection_integer($"sketch", $"sketch2"))
+ .collect()(0)(0)
+ assert(intersection1 != null)
+
+ // Test intersection with mode
+ val intersection2 = joined
+ .select(tuple_intersection_integer($"sketch", $"sketch2", "max"))
+ .collect()(0)(0)
+ assert(intersection2 != null)
+
+ // Test with column names
+ val intersection3 = joined
+ .select(tuple_intersection_integer("sketch", "sketch2"))
+ .collect()(0)(0)
+ assert(intersection3 != null)
+
+ // Verify estimate from intersection (keys 2 and 3 are common)
+ val estimate = joined
+ .select(tuple_sketch_estimate_integer(tuple_intersection_integer($"sketch", $"sketch2")))
+ .collect()(0)(0)
+ assert(estimate == 2.0)
+ }
+
+ test("SPARK-54179: tuple_difference_double basic functionality") {
+ val df1 = Seq((1, 1.5), (2, 2.5), (3, 3.5)).toDF("key", "summary")
+ val df2 = Seq((2, 1.0), (3, 2.0)).toDF("key", "summary")
+
+ val sketch1Df = df1.agg(tuple_sketch_agg_double($"key", $"summary").alias("sketch"))
+ val sketch2Df = df2.agg(tuple_sketch_agg_double($"key", $"summary").alias("sketch"))
+
+ val joined = sketch1Df.crossJoin(sketch2Df.withColumnRenamed("sketch", "sketch2"))
+
+ // Test difference
+ val difference = joined
+ .select(tuple_difference_double($"sketch", $"sketch2"))
+ .collect()(0)(0)
+ assert(difference != null)
+ assert(difference.asInstanceOf[Array[Byte]].length > 0)
+
+ // Test with column names
+ val difference2 = joined
+ .select(tuple_difference_double("sketch", "sketch2"))
+ .collect()(0)(0)
+ assert(difference2 != null)
+
+ // Verify estimate from difference (only key 1 is unique to df1)
+ val estimate = joined
+ .select(tuple_sketch_estimate_double(tuple_difference_double($"sketch", $"sketch2")))
+ .collect()(0)(0)
+ assert(estimate == 1.0)
+ }
+
+ test("SPARK-54179: tuple_difference_integer basic functionality") {
+ val df1 = Seq((1, 10), (2, 20), (3, 30)).toDF("key", "summary")
+ val df2 = Seq((2, 15), (3, 25)).toDF("key", "summary")
+
+ val sketch1Df = df1.agg(tuple_sketch_agg_integer($"key", $"summary").alias("sketch"))
+ val sketch2Df = df2.agg(tuple_sketch_agg_integer($"key", $"summary").alias("sketch"))
+
+ val joined = sketch1Df.crossJoin(sketch2Df.withColumnRenamed("sketch", "sketch2"))
+
+ // Test difference
+ val difference = joined
+ .select(tuple_difference_integer($"sketch", $"sketch2"))
+ .collect()(0)(0)
+ assert(difference != null)
+
+ // Test with column names
+ val difference2 = joined
+ .select(tuple_difference_integer("sketch", "sketch2"))
+ .collect()(0)(0)
+ assert(difference2 != null)
+
+ // Verify estimate from difference (only key 1 is unique to df1)
+ val estimate = joined
+ .select(tuple_sketch_estimate_integer(tuple_difference_integer($"sketch", $"sketch2")))
+ .collect()(0)(0)
+ assert(estimate == 1.0)
+ }
+
+ test("SPARK-54179: tuple_union_agg_double basic functionality") {
+ val df1 = Seq((1, 1, 1.5), (1, 2, 2.5)).toDF("id", "key", "summary")
+ val df2 = Seq((1, 3, 3.5), (1, 4, 4.5)).toDF("id", "key", "summary")
+
+ val sketch1Df = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_double($"key", $"summary")
+ .alias("sketch"))
+ val sketch2Df = df2
+ .groupBy("id")
+ .agg(tuple_sketch_agg_double($"key", $"summary")
+ .alias("sketch"))
+
+ val combined = sketch1Df.union(sketch2Df)
+
+ // Test union_agg with default parameters
+ val union1 = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_double($"sketch"))
+ .collect()(0)(1)
+ assert(union1 != null)
+ assert(union1.asInstanceOf[Array[Byte]].length > 0)
+
+ // Test union_agg with lgNomEntries
+ val union2 = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_double($"sketch", 10))
+ .collect()(0)(1)
+ assert(union2 != null)
+
+ // Test union_agg with lgNomEntries and mode
+ val union3 = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_double($"sketch", 10, "sum"))
+ .collect()(0)(1)
+ assert(union3 != null)
+
+ // Test with column name
+ val union4 = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_double("sketch"))
+ .collect()(0)(1)
+ assert(union4 != null)
+
+ // Verify estimate from union_agg (keys 1, 2, 3, 4 should all be present)
+ val estimateResult = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_double($"sketch").alias("union_sketch"))
+ .select(tuple_sketch_estimate_double($"union_sketch"))
+ .collect()(0)(0)
+ assert(estimateResult == 4.0)
+ }
+
+ test("SPARK-54179: tuple_union_agg_integer basic functionality") {
+ val df1 = Seq((1, 1, 10), (1, 2, 20)).toDF("id", "key", "summary")
+ val df2 = Seq((1, 3, 30), (1, 4, 40)).toDF("id", "key", "summary")
+
+ val sketch1Df = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_integer($"key", $"summary")
+ .alias("sketch"))
+ val sketch2Df = df2
+ .groupBy("id")
+ .agg(tuple_sketch_agg_integer($"key", $"summary")
+ .alias("sketch"))
+
+ val combined = sketch1Df.union(sketch2Df)
+
+ // Test union_agg with default parameters
+ val union1 = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_integer($"sketch"))
+ .collect()(0)(1)
+ assert(union1 != null)
+
+ // Test union_agg with lgNomEntries and mode
+ val union2 = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_integer($"sketch", 10, "max"))
+ .collect()(0)(1)
+ assert(union2 != null)
+
+ // Test with column name
+ val union3 = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_integer("sketch", 10))
+ .collect()(0)(1)
+ assert(union3 != null)
+
+ // Verify estimate from union_agg (keys 1, 2, 3, 4 should all be present)
+ val estimateResult = combined
+ .groupBy("id")
+ .agg(tuple_union_agg_integer($"sketch").alias("union_sketch"))
+ .select(tuple_sketch_estimate_integer($"union_sketch"))
+ .collect()(0)(0)
+ assert(estimateResult == 4.0)
+ }
+
+ test("SPARK-54179: tuple_intersection_agg_double basic functionality") {
+ val df1 = Seq((1, 1, 1.5), (1, 2, 2.5), (1, 3, 3.5)).toDF("id", "key", "summary")
+ val df2 = Seq((1, 2, 1.0), (1, 3, 2.0), (1, 4, 3.0)).toDF("id", "key", "summary")
+
+ val sketch1Df = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_double($"key", $"summary")
+ .alias("sketch"))
+ val sketch2Df = df2
+ .groupBy("id")
+ .agg(tuple_sketch_agg_double($"key", $"summary")
+ .alias("sketch"))
+
+ val combined = sketch1Df.union(sketch2Df)
+
+ // Test intersection_agg with default mode
+ val intersection1 = combined
+ .groupBy("id")
+ .agg(tuple_intersection_agg_double($"sketch"))
+ .collect()(0)(1)
+ assert(intersection1 != null)
+ assert(intersection1.asInstanceOf[Array[Byte]].length > 0)
+
+ // Test intersection_agg with mode
+ val intersection2 = combined
+ .groupBy("id")
+ .agg(tuple_intersection_agg_double($"sketch", "sum"))
+ .collect()(0)(1)
+ assert(intersection2 != null)
+
+ // Test with column name
+ val intersection3 = combined
+ .groupBy("id")
+ .agg(tuple_intersection_agg_double("sketch", "min"))
+ .collect()(0)(1)
+ assert(intersection3 != null)
+
+ // Verify estimate from intersection_agg (keys 2 and 3 are common)
+ val estimateResult = combined
+ .groupBy("id")
+ .agg(tuple_intersection_agg_double($"sketch").alias("intersection_sketch"))
+ .select(tuple_sketch_estimate_double($"intersection_sketch"))
+ .collect()(0)(0)
+ assert(estimateResult == 2.0)
+ }
+
+ test("SPARK-54179: tuple_intersection_agg_integer basic functionality") {
+ val df1 = Seq((1, 1, 10), (1, 2, 20), (1, 3, 30)).toDF("id", "key", "summary")
+ val df2 = Seq((1, 2, 15), (1, 3, 25), (1, 4, 35)).toDF("id", "key", "summary")
+
+ val sketch1Df = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_integer($"key", $"summary")
+ .alias("sketch"))
+ val sketch2Df = df2
+ .groupBy("id")
+ .agg(tuple_sketch_agg_integer($"key", $"summary")
+ .alias("sketch"))
+
+ val combined = sketch1Df.union(sketch2Df)
+
+ // Test intersection_agg with default mode
+ val intersection1 = combined
+ .groupBy("id")
+ .agg(tuple_intersection_agg_integer($"sketch"))
+ .collect()(0)(1)
+ assert(intersection1 != null)
+
+ // Test intersection_agg with mode
+ val intersection2 = combined
+ .groupBy("id")
+ .agg(tuple_intersection_agg_integer($"sketch", "max"))
+ .collect()(0)(1)
+ assert(intersection2 != null)
+
+ // Test with column name
+ val intersection3 = combined
+ .groupBy("id")
+ .agg(tuple_intersection_agg_integer("sketch"))
+ .collect()(0)(1)
+ assert(intersection3 != null)
+
+ // Verify estimate from intersection_agg (keys 2 and 3 are common)
+ val estimateResult = combined
+ .groupBy("id")
+ .agg(tuple_intersection_agg_integer($"sketch").alias("intersection_sketch"))
+ .select(tuple_sketch_estimate_integer($"intersection_sketch"))
+ .collect()(0)(0)
+ assert(estimateResult == 2.0)
+ }
+
+ test("SPARK-54179: tuple_sketch_agg + operations + estimate comprehensive test - double") {
+ val df1 = Seq((1, "a", 1.0), (1, "a", 2.0), (1, "b", 3.0), (1, "c", 4.0))
+ .toDF("id", "key", "summary")
+ df1.createOrReplaceTempView("tuple_df1")
+
+ val df2 = Seq((1, "a", 0.5), (1, "c", 1.5), (1, "d", 2.5), (1, "e", 3.5))
+ .toDF("id", "key", "summary")
+ df2.createOrReplaceTempView("tuple_df2")
+
+ // Test tuple_sketch_agg and estimate via DataFrame
+ val res1 = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_double($"key", $"summary").as("sketch"))
+ .withColumn("estimate", tuple_sketch_estimate_double($"sketch"))
+ .withColumn("summary_total", tuple_sketch_summary_double($"sketch"))
+
+ val row1 = res1.collect()(0)
+ assert(row1.getAs[Double]("estimate") >= 3.0) // at least 3 distinct keys
+ assert(row1.getAs[Double]("summary_total") > 0.0)
+
+ // Test via SQL
+ val res2 = sql("""
+ SELECT
+ id,
+ tuple_sketch_agg_double(key, summary) as sketch,
+ tuple_sketch_estimate_double(tuple_sketch_agg_double(key, summary)) as estimate,
+ tuple_sketch_summary_double(tuple_sketch_agg_double(key, summary)) as summary_total
+ FROM tuple_df1
+ GROUP BY id
+ """)
+
+ val row2 = res2.collect()(0)
+ assert(row2.getAs[Double]("estimate") >= 3.0)
+ assert(row2.getAs[Double]("summary_total") > 0.0)
+
+ // Test union operations
+ val sketch1 = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_double($"key", $"summary")
+ .as("sketch"))
+ val sketch2 = df2
+ .groupBy("id")
+ .agg(tuple_sketch_agg_double($"key", $"summary")
+ .as("sketch"))
+
+ val unionResult = sketch1
+ .union(sketch2)
+ .groupBy("id")
+ .agg(tuple_union_agg_double($"sketch").as("union_sketch"))
+ .withColumn("union_estimate", tuple_sketch_estimate_double($"union_sketch"))
+
+ val unionRow = unionResult.collect()(0)
+ assert(unionRow.getAs[Double]("union_estimate") >= 5.0) // union should have 5 distinct keys
+
+ // Test intersection
+ val joined = sketch1.crossJoin(sketch2.withColumnRenamed("sketch", "sketch2"))
+ val intersectionResult = joined
+ .withColumn("intersection_sketch", tuple_intersection_double($"sketch", $"sketch2"))
+ .withColumn("intersection_estimate", tuple_sketch_estimate_double($"intersection_sketch"))
+
+ val intersectionRow = intersectionResult.collect()(0)
+ assert(intersectionRow.getAs[Double]("intersection_estimate") >= 2.0) // a and c are common
+
+ // Test difference
+ val differenceResult = joined
+ .withColumn("difference_sketch", tuple_difference_double($"sketch", $"sketch2"))
+ .withColumn("difference_estimate", tuple_sketch_estimate_double($"difference_sketch"))
+
+ val differenceRow = differenceResult.collect()(0)
+ assert(differenceRow.getAs[Double]("difference_estimate") >= 1.0) // b is unique to df1
+ }
+
+ test("SPARK-54179: tuple_sketch_agg + operations + estimate comprehensive test - integer") {
+ val df1 = Seq((1, "a", 10), (1, "a", 20), (1, "b", 30), (1, "c", 40))
+ .toDF("id", "key", "summary")
+
+ val df2 = Seq((1, "a", 5), (1, "c", 15), (1, "d", 25), (1, "e", 35))
+ .toDF("id", "key", "summary")
+
+ // Test tuple_sketch_agg and estimate
+ val res1 = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_integer($"key", $"summary").as("sketch"))
+ .withColumn("estimate", tuple_sketch_estimate_integer($"sketch"))
+ .withColumn("summary_total", tuple_sketch_summary_integer($"sketch"))
+
+ val row1 = res1.collect()(0)
+ assert(row1.getAs[Double]("estimate") >= 3.0)
+ assert(row1.getAs[Long]("summary_total") > 0)
+
+ // Test with different modes
+ val resMax = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_integer($"key", $"summary", 12, "max").as("sketch"))
+ .withColumn("summary_max", tuple_sketch_summary_integer($"sketch", "max"))
+
+ val rowMax = resMax.collect()(0)
+ assert(rowMax.getAs[Long]("summary_max") > 0)
+
+ // Test union with mode
+ val sketch1 = df1
+ .groupBy("id")
+ .agg(tuple_sketch_agg_integer($"key", $"summary")
+ .as("sketch"))
+ val sketch2 = df2
+ .groupBy("id")
+ .agg(tuple_sketch_agg_integer($"key", $"summary")
+ .as("sketch"))
+
+ val unionResult = sketch1
+ .union(sketch2)
+ .groupBy("id")
+ .agg(tuple_union_agg_integer($"sketch", 12, "sum").as("union_sketch"))
+ .withColumn("union_estimate", tuple_sketch_estimate_integer($"union_sketch"))
+
+ val unionRow = unionResult.collect()(0)
+ assert(unionRow.getAs[Double]("union_estimate") >= 5.0)
+ }
+
+ test("SPARK-54179: tuple_sketch with null values") {
+ val df = Seq((1, Some(10)), (2, Some(20)), (3, None)).toDF("key", "summary")
+
+ // Null summaries should be handled
+ val sketch = df.agg(tuple_sketch_agg_integer($"key", $"summary").alias("sketch"))
+ val estimate = sketch
+ .select(tuple_sketch_estimate_integer($"sketch"))
+ .collect()(0)(0)
+ assert(estimate != null)
+ assert(estimate.asInstanceOf[Double] == 2.0)
+ }
}
case class B(c: Option[Double])
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org