[
https://issues.apache.org/jira/browse/SPARK-28067?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16870438#comment-16870438
]
Mark Sirek commented on SPARK-28067:
------------------------------------
[~mgaido] Here is the physical plan I'm getting. Maybe yours is different? I
tried on master this time...
{code:java}
msirek@skylake16:~/IdeaProjects/spark$ git status
On branch master
Your branch is up to date with 'origin/master'.
nothing to commit, working tree clean
msirek@skylake16:~/IdeaProjects/spark$ git log -5
--pretty=format:"%h%x09%an%x09%ad%x09%s"
870f972dcc Yuming Wang Sat Jun 22 09:15:07 2019 -0700 [SPARK-28104][SQL]
Implement Spark's own GetColumnsOperation
5ad1053f3e Bryan Cutler Sat Jun 22 11:20:35 2019 +0900
[SPARK-28128][PYTHON][SQL] Pandas Grouped UDFs skip empty partitions
113f8c8d13 HyukjinKwon Fri Jun 21 10:47:54 2019 -0700 [SPARK-28132][PYTHON]
Update document type conversion for Pandas UDFs (pyarrow 0.13.0, pandas 0.24.2,
Python 3.7)
9b9d81b821 HyukjinKwon Fri Jun 21 10:27:18 2019 -0700 [SPARK-28131][PYTHON]
Update document type conversion between Python data and SQL types in normal
UDFs (Python 3.7)
54da3bbfb2 Yesheng Ma Thu Jun 20 19:45:59 2019 -0700 [SPARK-28127][SQL] Micro
optimization on TreeNode's mapChildren method
msirek@skylake16:~/IdeaProjects/spark$ ./bin/spark-shell
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in
[jar:file:/home/msirek/IdeaProjects/spark/assembly/target/scala-2.12/jars/slf4j-log4j12-1.7.16.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in
[jar:file:/opt/cloudera/parcels/CDH-5.16.1-1.cdh5.16.1.p0.3/jars/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use
setLogLevel(newLevel).
19/06/22 22:13:39 WARN util.Utils: Service 'SparkUI' could not bind on port
4040. Attempting port 4041.
Spark context Web UI available at http://skylake16.home.colo:4041
Spark context available as 'sc' (master = local[*], app id =
local-1561266819220).
Spark session available as 'spark'.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 3.0.0-SNAPSHOT
/_/
Using Scala version 2.12.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_201)
Type in expressions to have them evaluated.
Type :help for more information.
scala> val df = Seq(
| (BigDecimal("10000000000000000000"), 1),
| (BigDecimal("10000000000000000000"), 1),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2),
| (BigDecimal("10000000000000000000"), 2)).toDF("decNum", "intNum")
df: org.apache.spark.sql.DataFrame = [decNum: decimal(38,18), intNum: int]
scala> val df2 = df.withColumnRenamed("decNum", "decNum2").join(df,
"intNum").agg(sum("decNum"))
df2: org.apache.spark.sql.DataFrame = [sum(decNum): decimal(38,18)]
scala> df2.explain
== Physical Plan ==
*(2) HashAggregate(keys=[], functions=[sum(decNum#14)])
+- Exchange SinglePartition
+- *(1) HashAggregate(keys=[], functions=[partial_sum(decNum#14)])
+- *(1) Project [decNum#14]
+- *(1) BroadcastHashJoin [intNum#8], [intNum#15], Inner, BuildLeft
:- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int,
false] as bigint)))
: +- LocalTableScan [intNum#8]
+- LocalTableScan [decNum#14, intNum#15]
scala> df2.show(40,false)
+---------------------------------------+
|sum(decNum) |
+---------------------------------------+
|40000000000000000000.000000000000000000|
+---------------------------------------+
{code}
> Incorrect results in decimal aggregation with whole-stage code gen enabled
> --------------------------------------------------------------------------
>
> Key: SPARK-28067
> URL: https://issues.apache.org/jira/browse/SPARK-28067
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 2.3.0, 2.4.0
> Environment: Ubuntu LTS 16.04
> Oracle Java 1.8.0_201
> spark-2.4.3-bin-without-hadoop
> spark-shell
> Reporter: Mark Sirek
> Priority: Minor
> Labels: correctness
>
> The following test case involving a join followed by a sum aggregation
> returns the wrong answer for the sum:
>
> {code:java}
> val df = Seq(
> (BigDecimal("10000000000000000000"), 1),
> (BigDecimal("10000000000000000000"), 1),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2),
> (BigDecimal("10000000000000000000"), 2)).toDF("decNum", "intNum")
> val df2 = df.withColumnRenamed("decNum", "decNum2").join(df,
> "intNum").agg(sum("decNum"))
> scala> df2.show(40,false)
> ---------------------------------------
> sum(decNum)
> ---------------------------------------
> 40000000000000000000.000000000000000000
> ---------------------------------------
>
> {code}
>
> The result should be 1040000000000000000000.0000000000000000.
> It appears a partial sum is computed for each join key, as the result
> returned would be the answer for all rows matching intNum === 1.
> If only the rows with intNum === 2 are included, the answer given is null:
>
> {code:java}
> scala> val df3 = df.filter($"intNum" === lit(2))
> df3: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [decNum:
> decimal(38,18), intNum: int]
> scala> val df4 = df3.withColumnRenamed("decNum", "decNum2").join(df3,
> "intNum").agg(sum("decNum"))
> df4: org.apache.spark.sql.DataFrame = [sum(decNum): decimal(38,18)]
> scala> df4.show(40,false)
> -----------
> sum(decNum)
> -----------
> null
> -----------
>
> {code}
>
> The correct answer, 1000000000000000000000.0000000000000000, doesn't fit in
> the DataType picked for the result, decimal(38,18), so an overflow occurs,
> which Spark then converts to null.
> The first example, which doesn't filter out the intNum === 1 values should
> also return null, indicating overflow, but it doesn't. This may mislead the
> user to think a valid sum was computed.
> If whole-stage code gen is turned off:
> spark.conf.set("spark.sql.codegen.wholeStage", false)
> ... incorrect results are not returned because the overflow is caught as an
> exception:
> java.lang.IllegalArgumentException: requirement failed: Decimal precision 39
> exceeds max precision 38
>
>
>
>
>
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
