You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2017/06/30 06:46:17 UTC
spark git commit: Revert "[SPARK-21258][SQL] Fix WindowExec complex
object aggregation with spilling"
Repository: spark
Updated Branches:
refs/heads/branch-2.1 d995dac1c -> 3ecef2491
Revert "[SPARK-21258][SQL] Fix WindowExec complex object aggregation with spilling"
This reverts commit d995dac1cdeec940364453675f59ce5cf2b53684.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/3ecef249
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/3ecef249
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/3ecef249
Branch: refs/heads/branch-2.1
Commit: 3ecef249184cdcf74765070d3ea39cf180214976
Parents: d995dac
Author: Wenchen Fan <we...@databricks.com>
Authored: Fri Jun 30 14:45:55 2017 +0800
Committer: Wenchen Fan <we...@databricks.com>
Committed: Fri Jun 30 14:45:55 2017 +0800
----------------------------------------------------------------------
.../execution/window/AggregateProcessor.scala | 7 +--
.../sql/DataFrameWindowFunctionsSuite.scala | 47 +-------------------
2 files changed, 3 insertions(+), 51 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/3ecef249/sql/core/src/main/scala/org/apache/spark/sql/execution/window/AggregateProcessor.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/AggregateProcessor.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/AggregateProcessor.scala
index dfa1100..c9f5d3b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/AggregateProcessor.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/AggregateProcessor.scala
@@ -145,13 +145,10 @@ private[window] final class AggregateProcessor(
/** Update the buffer. */
def update(input: InternalRow): Unit = {
- // TODO(hvanhovell) this sacrifices performance for correctness. We should make sure that
- // MutableProjection makes copies of the complex input objects it buffer.
- val copy = input.copy()
- updateProjection(join(buffer, copy))
+ updateProjection(join(buffer, input))
var i = 0
while (i < numImperatives) {
- imperatives(i).update(buffer, copy)
+ imperatives(i).update(buffer, input)
i += 1
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/3ecef249/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
index 204858f..1255c49 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
@@ -19,9 +19,8 @@ package org.apache.spark.sql
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction, Window}
import org.apache.spark.sql.functions._
-import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSQLContext
-import org.apache.spark.sql.types._
+import org.apache.spark.sql.types.{DataType, LongType, StructType}
/**
* Window function testing for DataFrame API.
@@ -424,48 +423,4 @@ class DataFrameWindowFunctionsSuite extends QueryTest with SharedSQLContext {
df.select(selectList: _*).where($"value" < 2),
Seq(Row(3, "1", null, 3.0, 4.0, 3.0), Row(5, "1", false, 4.0, 5.0, 5.0)))
}
-
- test("SPARK-21258: complex object in combination with spilling") {
- // Make sure we trigger the spilling path.
- withSQLConf(SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "17") {
- val sampleSchema = new StructType().
- add("f0", StringType).
- add("f1", LongType).
- add("f2", ArrayType(new StructType().
- add("f20", StringType))).
- add("f3", ArrayType(new StructType().
- add("f30", StringType)))
-
- val w0 = Window.partitionBy("f0").orderBy("f1")
- val w1 = w0.rowsBetween(Long.MinValue, Long.MaxValue)
-
- val c0 = first(struct($"f2", $"f3")).over(w0) as "c0"
- val c1 = last(struct($"f2", $"f3")).over(w1) as "c1"
-
- val input =
- """{"f1":1497820153720,"f2":[{"f20":"x","f21":0}],"f3":[{"f30":"x","f31":0}]}
- |{"f1":1497802179638}
- |{"f1":1497802189347}
- |{"f1":1497802189593}
- |{"f1":1497802189597}
- |{"f1":1497802189599}
- |{"f1":1497802192103}
- |{"f1":1497802193414}
- |{"f1":1497802193577}
- |{"f1":1497802193709}
- |{"f1":1497802202883}
- |{"f1":1497802203006}
- |{"f1":1497802203743}
- |{"f1":1497802203834}
- |{"f1":1497802203887}
- |{"f1":1497802203893}
- |{"f1":1497802203976}
- |{"f1":1497820168098}
- |""".stripMargin.split("\n").toSeq
-
- import testImplicits._
-
- spark.read.schema(sampleSchema).json(input.toDS()).select(c0, c1).foreach { _ => () }
- }
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org