peay created SPARK-18473: ---------------------------- Summary: Correctness issue in INNER join result with window functions Key: SPARK-18473 URL: https://issues.apache.org/jira/browse/SPARK-18473 Project: Spark Issue Type: Bug Components: PySpark, Spark Core, SQL Affects Versions: 2.0.1 Reporter: peay
I have stumbled onto a corner case where an INNER join appears to return incorrect results. I believe the join should behave as the identity, but instead, some values are shuffled around, and some are just plain wrong. This can be reproduced as follows: joining {code} +-----+---------+------+--------+--------+----------+------+ |index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|sessId| +-----+---------+------+--------+--------+----------+------+ | 1| 1| 1| 0| 1| 0| 1| | 2| 2| 0| 0| 1| 0| 1| | 1| 3| 1| 0| 2| 0| 2| +-----+---------+------+--------+--------+----------+------+ {code} with {code} +------+ |sessId| +------+ | 1| | 2| +------+ {code} The result is {code} +------+-----+---------+------+--------+--------+----------+ |sessId|index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount| +------+-----+---------+------+--------+--------+----------+ | 1| 2| 2| 0| 0| 1| 0| | 2| 1| 1| 1| 0| 1| -1| | 2| 1| 3| 1| 0| 2| 0| +------+-----+---------+------+--------+--------+----------+ {code} Note how rows have a sessId of 2 (instead of one row as expected), and how `fiftyCount` can now be negative while always zero in the original dataframe. The first dataframe uses two windows: - `hasOne` uses a `window.rowsBetween(-10, 0)`. - `hasFifty` uses a `window.rowsBetween(-10, -1)`. The result is **correct** if: - `hasFifty` is changed to `window.rowsBetween(-10, 0)` instead of `window.rowsBetween(-10, -1)`. - I add `.fillna({ 'numOnesBefore': 0 })`, although there are no visible effect on the dataframe as shown by `show` as far as I can tell. - I use a LEFT OUTER join instead of INNER JOIN. - I write both dataframes to Parquet, read them back and join these. This can be reproduced in pyspark using: {code:python} import pyspark.sql.functions as F from pyspark.sql.functions import col from pyspark.sql.window import Window df1 = sql_context.createDataFrame( pd.DataFrame({"index": [1, 2, 1], "timeStamp": [1, 2, 3]}) ) window = Window.partitionBy(F.lit(1)).orderBy("timeStamp", "index") df2 = ( df1 .withColumn("hasOne", (col("index") == 1).cast("int")) .withColumn("hasFifty", (col("index") == 50).cast("int")) .withColumn("numOnesBefore", F.sum(col("hasOne")).over(window.rowsBetween(-10, 0))) .withColumn("numFiftyStrictlyBefore", F.sum(col("hasFifty")).over(window.rowsBetween(-10, -1))) .fillna({ 'numFiftyStrictlyBefore': 0 }) .withColumn("sessId", col("numOnesBefore") - col("numFiftyStrictlyBefore")) ) df_selector = sql_context.createDataFrame(pd.DataFrame({"sessId": [1, 2]})) df_joined = df_selector.join(df2, "sessId", how="inner") df2.show() df_selector.show() df_joined.show() {code} -- This message was sent by Atlassian JIRA (v6.3.4#6332) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org