[SPARK-7965] Wrong answers for queries with multiple window specs in the same expression - ASF JIRA

Rank to Top

Rank to Bottom

Attach files

Attach Screenshot

Bulk Copy Attachments

Bulk Move Attachments

Voters

Watch issue

Watchers

Create sub-task

Convert to sub-task

Link

Clone

Labels

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Bug
Status: Resolved
Priority: Major
Resolution: Fixed
Affects Version/s: 1.4.0
Fix Version/s: 1.4.0
Component/s: SQL
Labels:
- correctness

Description

I think that Spark SQL may be returning incorrect answers for queries that use multiple window specifications within the same expression. Here's an example that illustrates the problem.

Say that I have a table with a single numeric column and that I want to compute a cumulative distribution function over this column. Let's call this table nums:

val nums = sc.parallelize(1 to 10).map(x => (x)).toDF("x")
nums.registerTempTable("nums")

It's easy to compute a running sum over this column:

sqlContext.sql("""
    select sum(x) over (rows between unbounded preceding and current row) from nums
""").collect()

nums: org.apache.spark.sql.DataFrame = [x: int]
res29: Array[org.apache.spark.sql.Row] = Array([1], [3], [6], [10], [15], [21], [28], [36], [45], [55])

It's also easy to compute a total sum over all rows:

sqlContext.sql("""
    select sum(x) over (rows between unbounded preceding and unbounded following) from nums
""").collect()

res34: Array[org.apache.spark.sql.Row] = Array([55], [55], [55], [55], [55], [55], [55], [55], [55], [55])

Let's say that I combine these expressions to compute a CDF:

sqlContext.sql("""
	select (sum(x) over (rows between unbounded preceding and current row))
    /
    (sum(x) over (rows between unbounded preceding and unbounded following)) from nums
""").collect()

res31: Array[org.apache.spark.sql.Row] = Array([1.0], [1.0], [1.0], [1.0], [1.0], [1.0], [1.0], [1.0], [1.0], [1.0])

This seems wrong. Note that if we combine the running total, global total, and combined expression in the same query, then we see that the first two values are computed correctly / but the combined expression seems to be incorrect:

sqlContext.sql("""
    select
    sum(x) over (rows between unbounded preceding and current row) as running_sum,
    (sum(x) over (rows between unbounded preceding and unbounded following)) as total_sum,
    ((sum(x) over (rows between unbounded preceding and current row))
    /
    (sum(x) over (rows between unbounded preceding and unbounded following))) as combined
    from nums 
""").collect()

res40: Array[org.apache.spark.sql.Row] = Array([1,55,1.0], [3,55,1.0], [6,55,1.0], [10,55,1.0], [15,55,1.0], [21,55,1.0], [28,55,1.0], [36,55,1.0], [45,55,1.0], [55,55,1.0])

/cc Yin Huai