You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by ueshin <gi...@git.apache.org> on 2018/07/18 12:50:17 UTC
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
GitHub user ueshin opened a pull request:
https://github.com/apache/spark/pull/21802
[SPARK-23928][SQL] Add shuffle collection function.
## What changes were proposed in this pull request?
This PR adds a new collection function: shuffle. It generates a random permutation of the given array.
## How was this patch tested?
New tests are added to CollectionExpressionsSuite.scala and DataFrameFunctionsSuite.scala.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/ueshin/apache-spark issues/SPARK-23928/shuffle
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/21802.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #21802
----
commit a3dbd93c0acbb2a3f3fb50574ae1e126c66c4d2d
Author: pkuwm <ih...@...>
Date: 2018-07-17T23:18:03Z
Add shuffle collection function.
commit b4cbb5558088356fe6be1cda053c9f91fbe7c538
Author: Takuya UESHIN <ue...@...>
Date: 2018-07-18T12:17:59Z
Refactor Shuffle function.
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r205483672
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RandomIndicesGenerator.scala ---
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.commons.math3.random.MersenneTwister
+
+/**
+ * This class is used to generate a random indices of given length.
+ *
+ * This implementation uses the "inside-out" version of Fisher-Yates algorithm.
+ * Reference:
+ * https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#The_%22inside-out%22_algorithm
+ */
+case class RandomIndicesGenerator(randomSeed: Long) {
+ private val random = new MersenneTwister(randomSeed)
+
+ def getNextIndices(length: Int): Array[Int] = {
--- End diff --
it should take a `Array[Int]`, to save the array creation.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r205703558
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/functions.scala ---
@@ -3545,6 +3545,14 @@ object functions {
*/
def array_max(e: Column): Column = withExpr { ArrayMax(e.expr) }
+ /**
+ * Returns a random permutation of the given array.
+ *
+ * @group collection_funcs
+ * @since 2.4.0
--- End diff --
Shall we match the documentation here as well?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r204087168
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -2086,6 +2087,20 @@ class Analyzer(
}
}
+ /**
+ * Set the seed for random number generation in Shuffle expressions.
+ */
+ object ResolvedShuffleExpressions extends Rule[LogicalPlan] {
+ private lazy val random = new Random()
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.transformUp {
+ case p if p.resolved => p
+ case p => p transformExpressionsUp {
+ case Shuffle(child, None) => Shuffle(child, Some(random.nextLong()))
--- End diff --
Yeah, in `Uuid` we want to make sure the same query plan can return the same result. It is more deterministic between retries.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on the issue:
https://github.com/apache/spark/pull/21802
cc @pkuwm
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on the issue:
https://github.com/apache/spark/pull/21802
Jenkins, retest this please.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution-unified/1193/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by kiszk <gi...@git.apache.org>.
Github user kiszk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r204270437
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala ---
@@ -1184,6 +1186,137 @@ case class ArraySort(child: Expression) extends UnaryExpression with ArraySortLi
override def prettyName: String = "array_sort"
}
+/**
+ * Returns a random permutation of the given array.
+ *
+ * This implementation uses the modern version of Fisher-Yates algorithm.
+ * Reference: https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#Modern_method
--- End diff --
In the latest commit, "inside-out" looks simpler without swapping (using just an assignment).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93373 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93373/testReport)** for PR 21802 at commit [`2ca1230`](https://github.com/apache/spark/commit/2ca12302e08d60ab9534d7d65fad9854fe1d6f28).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class RandomIndicesGenerator(randomSeed: Long) `
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203963010
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -2086,6 +2087,20 @@ class Analyzer(
}
}
+ /**
+ * Set the seed for random number generation in Shuffle expressions.
+ */
+ object ResolvedShuffleExpressions extends Rule[LogicalPlan] {
+ private lazy val random = new Random()
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.transformUp {
+ case p if p.resolved => p
+ case p => p transformExpressionsUp {
+ case Shuffle(child, None) => Shuffle(child, Some(random.nextLong()))
--- End diff --
I followed the implementation of `Uuid` and when we started doing this was at #20861. cc @viirya
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93279/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/21802
Do we really need full codegen for all of these collection functions? They seem pretty slow and specialization with full codegen won't help perf that much (and might even hurt by blowing up the code size) right?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203954268
--- Diff: python/pyspark/sql/functions.py ---
@@ -2382,6 +2382,20 @@ def array_sort(col):
return Column(sc._jvm.functions.array_sort(_to_java_column(col)))
+@since(2.4)
+def shuffle(col):
+ """
+ Collection function: Generates a random permutation of the given array.
+
+ .. note:: The function is non-deterministic because its results depends on order of rows which
--- End diff --
given a same input sequence, will this function always return the same permutation?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93261/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93268/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203955095
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -2086,6 +2087,20 @@ class Analyzer(
}
}
+ /**
+ * Set the seed for random number generation in Shuffle expressions.
+ */
+ object ResolvedShuffleExpressions extends Rule[LogicalPlan] {
+ private lazy val random = new Random()
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.transformUp {
+ case p if p.resolved => p
+ case p => p transformExpressionsUp {
+ case Shuffle(child, None) => Shuffle(child, Some(random.nextLong()))
--- End diff --
this looks reasonable, do we have any context about why we start doing this?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203560440
--- Diff: python/pyspark/sql/functions.py ---
@@ -2382,6 +2382,20 @@ def array_sort(col):
return Column(sc._jvm.functions.array_sort(_to_java_column(col)))
+@since(2.4)
+def shuffle(col):
+ """
+ Collection function: Generates a random permutation of the given array.
+
+ .. note:: The function is non-deterministic because its results depends on order of rows which
--- End diff --
The permutation is determined randomly but it is determined for the same query plan if the order of rows is determined, because the analyzer will assign a random seed for it.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on the issue:
https://github.com/apache/spark/pull/21802
Thanks! merging to master.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r205534846
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala ---
@@ -1184,6 +1184,110 @@ case class ArraySort(child: Expression) extends UnaryExpression with ArraySortLi
override def prettyName: String = "array_sort"
}
+/**
+ * Returns a random permutation of the given array.
+ */
+@ExpressionDescription(
+ usage = "_FUNC_(array) - Returns a random permutation of the given array.",
+ examples = """
+ Examples:
+ > SELECT _FUNC_(array(1, 20, 3, 5));
+ [3, 1, 5, 20]
+ > SELECT _FUNC_(array(1, 20, null, 3));
+ [20, null, 3, 1]
+ """, since = "2.4.0")
+case class Shuffle(child: Expression, randomSeed: Option[Long] = None)
+ extends UnaryExpression with ExpectsInputTypes with Stateful {
+
+ def this(child: Expression) = this(child, None)
+
+ override lazy val resolved: Boolean =
+ childrenResolved && checkInputDataTypes().isSuccess && randomSeed.isDefined
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(ArrayType)
+
+ override def dataType: DataType = child.dataType
+
+ @transient lazy val elementType: DataType = dataType.asInstanceOf[ArrayType].elementType
+
+ @transient private[this] var random: RandomIndicesGenerator = _
+
+ override protected def initializeInternal(partitionIndex: Int): Unit = {
+ random = RandomIndicesGenerator(randomSeed.get + partitionIndex)
+ }
+
+ override protected def evalInternal(input: InternalRow): Any = {
+ val value = child.eval(input)
+ if (value == null) {
+ null
+ } else {
+ val source = value.asInstanceOf[ArrayData]
+ val numElements = source.numElements()
+ val indices = random.getNextIndices(numElements)
+ new GenericArrayData(indices.map(source.get(_, elementType)))
+ }
+ }
+
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+ nullSafeCodeGen(ctx, ev, c => shuffleArrayCodeGen(ctx, ev, c))
+ }
+
+ private def shuffleArrayCodeGen(ctx: CodegenContext, ev: ExprCode, childName: String): String = {
+ val randomClass = classOf[RandomIndicesGenerator].getName
+
+ val rand = ctx.addMutableState(randomClass, "rand", forceInline = true)
+ ctx.addPartitionInitializationStatement(
+ s"$rand = new $randomClass(${randomSeed.get}L + partitionIndex);")
+
+ val isPrimitiveType = CodeGenerator.isPrimitiveType(elementType)
+
+ val numElements = ctx.freshName("numElements")
+ val arrayData = ctx.freshName("arrayData")
--- End diff --
Actually we need a new variable to use `ctx.createUnsafeArray()` which declares a new variable in it for now whereas `ev.value` is already declared.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by pkuwm <gi...@git.apache.org>.
Github user pkuwm commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203453412
--- Diff: python/pyspark/sql/functions.py ---
@@ -2382,6 +2382,20 @@ def array_sort(col):
return Column(sc._jvm.functions.array_sort(_to_java_column(col)))
+@since(2.4)
+def shuffle(col):
+ """
+ Collection function: Generates a random permutation of the given array.
+
+ .. note:: The function is non-deterministic because its results depends on order of rows which
--- End diff --
Maybe this one would be better? "The function is non-deterministic because it produces
an unbiased permutation: every permutation is equally likely."
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93661 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93661/testReport)** for PR 21802 at commit [`4135690`](https://github.com/apache/spark/commit/4135690f2cf1eea375a1a4f1697c0ffdb7436627).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93279 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93279/testReport)** for PR 21802 at commit [`f38b698`](https://github.com/apache/spark/commit/f38b6980875cc09fa15071439b3e1904c3722970).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution-unified/1126/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/21802
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r205702898
--- Diff: python/pyspark/sql/functions.py ---
@@ -2382,6 +2382,20 @@ def array_sort(col):
return Column(sc._jvm.functions.array_sort(_to_java_column(col)))
+@since(2.4)
+def shuffle(col):
+ """
+ Collection function: Generates a random permutation of the given array.
+
+ .. note:: The function is non-deterministic because its results depends on order of rows which
--- End diff --
typo: `results depends` found while reading this one.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93661 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93661/testReport)** for PR 21802 at commit [`4135690`](https://github.com/apache/spark/commit/4135690f2cf1eea375a1a4f1697c0ffdb7436627).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r204275498
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala ---
@@ -1419,4 +1421,71 @@ class CollectionExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper
assert(ArrayUnion(a20, a21).dataType.asInstanceOf[ArrayType].containsNull === false)
assert(ArrayUnion(a20, a22).dataType.asInstanceOf[ArrayType].containsNull === true)
}
+
+ test("Shuffle") {
+ // Primitive-type elements
+ val ai0 = Literal.create(Seq(1, 2, 3, 4, 5), ArrayType(IntegerType, containsNull = false))
+ val ai1 = Literal.create(Seq(1, 2, 3), ArrayType(IntegerType, containsNull = false))
+ val ai2 = Literal.create(Seq(null, 1, null, 3), ArrayType(IntegerType, containsNull = true))
+ val ai3 = Literal.create(Seq(2, null, 4, null), ArrayType(IntegerType, containsNull = true))
+ val ai4 = Literal.create(Seq(null, null, null), ArrayType(IntegerType, containsNull = true))
+ val ai5 = Literal.create(Seq(1), ArrayType(IntegerType, containsNull = false))
+ val ai6 = Literal.create(Seq.empty, ArrayType(IntegerType, containsNull = false))
+ val ai7 = Literal.create(null, ArrayType(IntegerType, containsNull = true))
+
+ checkEvaluation(Shuffle(ai0, Some(0)), Seq(4, 1, 2, 3, 5))
+ checkEvaluation(Shuffle(ai1, Some(0)), Seq(3, 1, 2))
+ checkEvaluation(Shuffle(ai2, Some(0)), Seq(3, null, 1, null))
+ checkEvaluation(Shuffle(ai3, Some(0)), Seq(null, 2, null, 4))
+ checkEvaluation(Shuffle(ai4, Some(0)), Seq(null, null, null))
+ checkEvaluation(Shuffle(ai5, Some(0)), Seq(1))
+ checkEvaluation(Shuffle(ai6, Some(0)), Seq.empty)
+ checkEvaluation(Shuffle(ai7, Some(0)), null)
+
+ // Non-primitive-type elements
+ val as0 = Literal.create(Seq("a", "b", "c", "d"), ArrayType(StringType, containsNull = false))
+ val as1 = Literal.create(Seq("a", "b", "c"), ArrayType(StringType, containsNull = false))
+ val as2 = Literal.create(Seq(null, "a", null, "c"), ArrayType(StringType, containsNull = true))
+ val as3 = Literal.create(Seq("b", null, "d", null), ArrayType(StringType, containsNull = true))
+ val as4 = Literal.create(Seq(null, null, null), ArrayType(StringType, containsNull = true))
+ val as5 = Literal.create(Seq("a"), ArrayType(StringType, containsNull = false))
+ val as6 = Literal.create(Seq.empty, ArrayType(StringType, containsNull = false))
+ val as7 = Literal.create(null, ArrayType(StringType, containsNull = true))
+ val aa = Literal.create(
+ Seq(Seq("a", "b"), Seq("c", "d"), Seq("e")),
+ ArrayType(ArrayType(StringType)))
+
+ checkEvaluation(Shuffle(as0, Some(0)), Seq("d", "a", "b", "c"))
+ checkEvaluation(Shuffle(as1, Some(0)), Seq("c", "a", "b"))
+ checkEvaluation(Shuffle(as2, Some(0)), Seq("c", null, "a", null))
+ checkEvaluation(Shuffle(as3, Some(0)), Seq(null, "b", null, "d"))
+ checkEvaluation(Shuffle(as4, Some(0)), Seq(null, null, null))
+ checkEvaluation(Shuffle(as5, Some(0)), Seq("a"))
+ checkEvaluation(Shuffle(as6, Some(0)), Seq.empty)
+ checkEvaluation(Shuffle(as7, Some(0)), null)
+ checkEvaluation(Shuffle(aa, Some(0)), Seq(Seq("e"), Seq("a", "b"), Seq("c", "d")))
+
+ val r = new Random()
+ val seed1 = Some(r.nextLong())
+ assert(evaluateWithoutCodegen(Shuffle(ai0, seed1)) ===
+ evaluateWithoutCodegen(Shuffle(ai0, seed1)))
+ assert(evaluateWithGeneratedMutableProjection(Shuffle(ai0, seed1)) ===
+ evaluateWithGeneratedMutableProjection(Shuffle(ai0, seed1)))
+ assert(evaluateWithUnsafeProjection(Shuffle(ai0, seed1)) ===
+ evaluateWithUnsafeProjection(Shuffle(ai0, seed1)))
+
+ val seed2 = Some(r.nextLong())
--- End diff --
I also followed a test for `Uuid` [MiscExpressionsSuite.scala#L46-L68](https://github.com/ueshin/apache-spark/blob/2ca12302e08d60ab9534d7d65fad9854fe1d6f28/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/MiscExpressionsSuite.scala#L46-L68) here.
@viirya WDYT about this?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203968608
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala ---
@@ -1184,6 +1186,137 @@ case class ArraySort(child: Expression) extends UnaryExpression with ArraySortLi
override def prettyName: String = "array_sort"
}
+/**
+ * Returns a random permutation of the given array.
+ *
+ * This implementation uses the modern version of Fisher-Yates algorithm.
+ * Reference: https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#Modern_method
--- End diff --
if we create a new array, I guess there should be some simpler algorithms without swapping...
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93493 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93493/testReport)** for PR 21802 at commit [`c56ecc5`](https://github.com/apache/spark/commit/c56ecc5b727b03734a5bd7917bae14b07d09ad7d).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r205534867
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RandomIndicesGenerator.scala ---
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.commons.math3.random.MersenneTwister
+
+/**
+ * This class is used to generate a random indices of given length.
+ *
+ * This implementation uses the "inside-out" version of Fisher-Yates algorithm.
+ * Reference:
+ * https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#The_%22inside-out%22_algorithm
+ */
+case class RandomIndicesGenerator(randomSeed: Long) {
+ private val random = new MersenneTwister(randomSeed)
+
+ def getNextIndices(length: Int): Array[Int] = {
--- End diff --
We need to create an array to store the shuffled indices anyway. If we want to pass an array to be shuffled, we need to create the array and fill it with 0 until n before we call this. But with this implementation, we don't need to fill the numbers prior to shuffle due to the "inside-out" version of Fisher-Yates algorithm. WDYT?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r205703459
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala ---
@@ -1184,6 +1184,110 @@ case class ArraySort(child: Expression) extends UnaryExpression with ArraySortLi
override def prettyName: String = "array_sort"
}
+/**
+ * Returns a random permutation of the given array.
+ */
+@ExpressionDescription(
+ usage = "_FUNC_(array) - Returns a random permutation of the given array.",
+ examples = """
+ Examples:
+ > SELECT _FUNC_(array(1, 20, 3, 5));
+ [3, 1, 5, 20]
+ > SELECT _FUNC_(array(1, 20, null, 3));
+ [20, null, 3, 1]
+ """, since = "2.4.0")
--- End diff --
We could add `note` here too.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93279 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93279/testReport)** for PR 21802 at commit [`f38b698`](https://github.com/apache/spark/commit/f38b6980875cc09fa15071439b3e1904c3722970).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93268 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93268/testReport)** for PR 21802 at commit [`9081e2f`](https://github.com/apache/spark/commit/9081e2f0b0371630aa315842f14931ee490ff461).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by kiszk <gi...@git.apache.org>.
Github user kiszk commented on the issue:
https://github.com/apache/spark/pull/21802
LGTM
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r204276502
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala ---
@@ -1419,4 +1421,71 @@ class CollectionExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper
assert(ArrayUnion(a20, a21).dataType.asInstanceOf[ArrayType].containsNull === false)
assert(ArrayUnion(a20, a22).dataType.asInstanceOf[ArrayType].containsNull === true)
}
+
+ test("Shuffle") {
+ // Primitive-type elements
+ val ai0 = Literal.create(Seq(1, 2, 3, 4, 5), ArrayType(IntegerType, containsNull = false))
+ val ai1 = Literal.create(Seq(1, 2, 3), ArrayType(IntegerType, containsNull = false))
+ val ai2 = Literal.create(Seq(null, 1, null, 3), ArrayType(IntegerType, containsNull = true))
+ val ai3 = Literal.create(Seq(2, null, 4, null), ArrayType(IntegerType, containsNull = true))
+ val ai4 = Literal.create(Seq(null, null, null), ArrayType(IntegerType, containsNull = true))
+ val ai5 = Literal.create(Seq(1), ArrayType(IntegerType, containsNull = false))
+ val ai6 = Literal.create(Seq.empty, ArrayType(IntegerType, containsNull = false))
+ val ai7 = Literal.create(null, ArrayType(IntegerType, containsNull = true))
+
+ checkEvaluation(Shuffle(ai0, Some(0)), Seq(4, 1, 2, 3, 5))
+ checkEvaluation(Shuffle(ai1, Some(0)), Seq(3, 1, 2))
+ checkEvaluation(Shuffle(ai2, Some(0)), Seq(3, null, 1, null))
+ checkEvaluation(Shuffle(ai3, Some(0)), Seq(null, 2, null, 4))
+ checkEvaluation(Shuffle(ai4, Some(0)), Seq(null, null, null))
+ checkEvaluation(Shuffle(ai5, Some(0)), Seq(1))
+ checkEvaluation(Shuffle(ai6, Some(0)), Seq.empty)
+ checkEvaluation(Shuffle(ai7, Some(0)), null)
+
+ // Non-primitive-type elements
+ val as0 = Literal.create(Seq("a", "b", "c", "d"), ArrayType(StringType, containsNull = false))
+ val as1 = Literal.create(Seq("a", "b", "c"), ArrayType(StringType, containsNull = false))
+ val as2 = Literal.create(Seq(null, "a", null, "c"), ArrayType(StringType, containsNull = true))
+ val as3 = Literal.create(Seq("b", null, "d", null), ArrayType(StringType, containsNull = true))
+ val as4 = Literal.create(Seq(null, null, null), ArrayType(StringType, containsNull = true))
+ val as5 = Literal.create(Seq("a"), ArrayType(StringType, containsNull = false))
+ val as6 = Literal.create(Seq.empty, ArrayType(StringType, containsNull = false))
+ val as7 = Literal.create(null, ArrayType(StringType, containsNull = true))
+ val aa = Literal.create(
+ Seq(Seq("a", "b"), Seq("c", "d"), Seq("e")),
+ ArrayType(ArrayType(StringType)))
+
+ checkEvaluation(Shuffle(as0, Some(0)), Seq("d", "a", "b", "c"))
+ checkEvaluation(Shuffle(as1, Some(0)), Seq("c", "a", "b"))
+ checkEvaluation(Shuffle(as2, Some(0)), Seq("c", null, "a", null))
+ checkEvaluation(Shuffle(as3, Some(0)), Seq(null, "b", null, "d"))
+ checkEvaluation(Shuffle(as4, Some(0)), Seq(null, null, null))
+ checkEvaluation(Shuffle(as5, Some(0)), Seq("a"))
+ checkEvaluation(Shuffle(as6, Some(0)), Seq.empty)
+ checkEvaluation(Shuffle(as7, Some(0)), null)
+ checkEvaluation(Shuffle(aa, Some(0)), Seq(Seq("e"), Seq("a", "b"), Seq("c", "d")))
+
+ val r = new Random()
+ val seed1 = Some(r.nextLong())
+ assert(evaluateWithoutCodegen(Shuffle(ai0, seed1)) ===
+ evaluateWithoutCodegen(Shuffle(ai0, seed1)))
+ assert(evaluateWithGeneratedMutableProjection(Shuffle(ai0, seed1)) ===
+ evaluateWithGeneratedMutableProjection(Shuffle(ai0, seed1)))
+ assert(evaluateWithUnsafeProjection(Shuffle(ai0, seed1)) ===
+ evaluateWithUnsafeProjection(Shuffle(ai0, seed1)))
+
+ val seed2 = Some(r.nextLong())
--- End diff --
I think this is what we expect. The result is decided by the random seed. So if using different random seeds, I think the results should be different.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93661/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93373 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93373/testReport)** for PR 21802 at commit [`2ca1230`](https://github.com/apache/spark/commit/2ca12302e08d60ab9534d7d65fad9854fe1d6f28).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93231 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93231/testReport)** for PR 21802 at commit [`b4cbb55`](https://github.com/apache/spark/commit/b4cbb5558088356fe6be1cda053c9f91fbe7c538).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r205483502
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala ---
@@ -1184,6 +1184,110 @@ case class ArraySort(child: Expression) extends UnaryExpression with ArraySortLi
override def prettyName: String = "array_sort"
}
+/**
+ * Returns a random permutation of the given array.
+ */
+@ExpressionDescription(
+ usage = "_FUNC_(array) - Returns a random permutation of the given array.",
+ examples = """
+ Examples:
+ > SELECT _FUNC_(array(1, 20, 3, 5));
+ [3, 1, 5, 20]
+ > SELECT _FUNC_(array(1, 20, null, 3));
+ [20, null, 3, 1]
+ """, since = "2.4.0")
+case class Shuffle(child: Expression, randomSeed: Option[Long] = None)
+ extends UnaryExpression with ExpectsInputTypes with Stateful {
+
+ def this(child: Expression) = this(child, None)
+
+ override lazy val resolved: Boolean =
+ childrenResolved && checkInputDataTypes().isSuccess && randomSeed.isDefined
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(ArrayType)
+
+ override def dataType: DataType = child.dataType
+
+ @transient lazy val elementType: DataType = dataType.asInstanceOf[ArrayType].elementType
+
+ @transient private[this] var random: RandomIndicesGenerator = _
+
+ override protected def initializeInternal(partitionIndex: Int): Unit = {
+ random = RandomIndicesGenerator(randomSeed.get + partitionIndex)
+ }
+
+ override protected def evalInternal(input: InternalRow): Any = {
+ val value = child.eval(input)
+ if (value == null) {
+ null
+ } else {
+ val source = value.asInstanceOf[ArrayData]
+ val numElements = source.numElements()
+ val indices = random.getNextIndices(numElements)
+ new GenericArrayData(indices.map(source.get(_, elementType)))
+ }
+ }
+
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+ nullSafeCodeGen(ctx, ev, c => shuffleArrayCodeGen(ctx, ev, c))
+ }
+
+ private def shuffleArrayCodeGen(ctx: CodegenContext, ev: ExprCode, childName: String): String = {
+ val randomClass = classOf[RandomIndicesGenerator].getName
+
+ val rand = ctx.addMutableState(randomClass, "rand", forceInline = true)
+ ctx.addPartitionInitializationStatement(
+ s"$rand = new $randomClass(${randomSeed.get}L + partitionIndex);")
+
+ val isPrimitiveType = CodeGenerator.isPrimitiveType(elementType)
+
+ val numElements = ctx.freshName("numElements")
+ val arrayData = ctx.freshName("arrayData")
--- End diff --
nit: we don't need the `arrayData` variable, we can assign `ev.value` directly.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93493/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21802
LGTM
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93231/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on the issue:
https://github.com/apache/spark/pull/21802
cc @cloud-fan @gatorsmile
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21802
@rxin Generally I agree with you, but currently the whole-stage-codegen doesn't support `CodegenFallback`, which means, if we don't implement codegen, this expression will stop WSC and hurt perf a lot.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution-unified/1112/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93373/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution-unified/1093/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93231 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93231/testReport)** for PR 21802 at commit [`b4cbb55`](https://github.com/apache/spark/commit/b4cbb5558088356fe6be1cda053c9f91fbe7c538).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class Shuffle(child: Expression, randomSeed: Option[Long] = None)`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203963022
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala ---
@@ -1184,6 +1186,137 @@ case class ArraySort(child: Expression) extends UnaryExpression with ArraySortLi
override def prettyName: String = "array_sort"
}
+/**
+ * Returns a random permutation of the given array.
+ *
+ * This implementation uses the modern version of Fisher-Yates algorithm.
+ * Reference: https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#Modern_method
--- End diff --
I copied the input array before starting shuffle not to change the input value. Isn't it safe?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution-unified/1398/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93268 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93268/testReport)** for PR 21802 at commit [`9081e2f`](https://github.com/apache/spark/commit/9081e2f0b0371630aa315842f14931ee490ff461).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution-unified/1269/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93493 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93493/testReport)** for PR 21802 at commit [`c56ecc5`](https://github.com/apache/spark/commit/c56ecc5b727b03734a5bd7917bae14b07d09ad7d).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203968939
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -2086,6 +2087,20 @@ class Analyzer(
}
}
+ /**
+ * Set the seed for random number generation in Shuffle expressions.
+ */
+ object ResolvedShuffleExpressions extends Rule[LogicalPlan] {
+ private lazy val random = new Random()
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.transformUp {
+ case p if p.resolved => p
+ case p => p transformExpressionsUp {
+ case Shuffle(child, None) => Shuffle(child, Some(random.nextLong()))
--- End diff --
then can we use a single rule to assign seed to these randomized functions?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93261 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93261/testReport)** for PR 21802 at commit [`9081e2f`](https://github.com/apache/spark/commit/9081e2f0b0371630aa315842f14931ee490ff461).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by mn-mikke <gi...@git.apache.org>.
Github user mn-mikke commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203407122
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala ---
@@ -1444,6 +1444,51 @@ class DataFrameFunctionsSuite extends QueryTest with SharedSQLContext {
)
}
+ test("shuffle function") {
+ // Shuffle expressions should produce same results at retries in the same DataFrame.
+ def checkResult(df: DataFrame): Unit = {
+ checkAnswer(df, df.collect())
+ }
+
+ // primitive-type elements
+ val idf = Seq(
+ Seq(1, 9, 8, 7),
+ Seq(5, 8, 9, 7, 2),
+ Seq.empty,
+ null
+ ).toDF("i")
+
+ def checkResult1(): Unit = {
--- End diff --
Maybe a different name for the method?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203974038
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala ---
@@ -1184,6 +1186,137 @@ case class ArraySort(child: Expression) extends UnaryExpression with ArraySortLi
override def prettyName: String = "array_sort"
}
+/**
+ * Returns a random permutation of the given array.
+ *
+ * This implementation uses the modern version of Fisher-Yates algorithm.
+ * Reference: https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#Modern_method
--- End diff --
Oh, I see. Let me try.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21802
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution-unified/1117/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on the issue:
https://github.com/apache/spark/pull/21802
Looks good to me too
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203956752
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala ---
@@ -1184,6 +1186,137 @@ case class ArraySort(child: Expression) extends UnaryExpression with ArraySortLi
override def prettyName: String = "array_sort"
}
+/**
+ * Returns a random permutation of the given array.
+ *
+ * This implementation uses the modern version of Fisher-Yates algorithm.
+ * Reference: https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#Modern_method
--- End diff --
is it safe to use Fisher-Yates here? AFAIK we should not change the input value, in case it's used by other expressions.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21802: [SPARK-23928][SQL] Add shuffle collection function.
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21802
**[Test build #93261 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93261/testReport)** for PR 21802 at commit [`9081e2f`](https://github.com/apache/spark/commit/9081e2f0b0371630aa315842f14931ee490ff461).
* This patch **fails due to an unknown error code, -9**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by mn-mikke <gi...@git.apache.org>.
Github user mn-mikke commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203388798
--- Diff: python/pyspark/sql/functions.py ---
@@ -2382,6 +2382,20 @@ def array_sort(col):
return Column(sc._jvm.functions.array_sort(_to_java_column(col)))
+@since(2.4)
+def shuffle(col):
+ """
+ Collection function: Generates a random permutation of the given array.
+
+ .. note:: The function is non-deterministic because its results depends on order of rows which
--- End diff --
Isn't it non-deterministic rather for the fact that the permutation is determined randomly?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by kiszk <gi...@git.apache.org>.
Github user kiszk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r204249027
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala ---
@@ -1419,4 +1421,71 @@ class CollectionExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper
assert(ArrayUnion(a20, a21).dataType.asInstanceOf[ArrayType].containsNull === false)
assert(ArrayUnion(a20, a22).dataType.asInstanceOf[ArrayType].containsNull === true)
}
+
+ test("Shuffle") {
+ // Primitive-type elements
+ val ai0 = Literal.create(Seq(1, 2, 3, 4, 5), ArrayType(IntegerType, containsNull = false))
+ val ai1 = Literal.create(Seq(1, 2, 3), ArrayType(IntegerType, containsNull = false))
+ val ai2 = Literal.create(Seq(null, 1, null, 3), ArrayType(IntegerType, containsNull = true))
+ val ai3 = Literal.create(Seq(2, null, 4, null), ArrayType(IntegerType, containsNull = true))
+ val ai4 = Literal.create(Seq(null, null, null), ArrayType(IntegerType, containsNull = true))
+ val ai5 = Literal.create(Seq(1), ArrayType(IntegerType, containsNull = false))
+ val ai6 = Literal.create(Seq.empty, ArrayType(IntegerType, containsNull = false))
+ val ai7 = Literal.create(null, ArrayType(IntegerType, containsNull = true))
+
+ checkEvaluation(Shuffle(ai0, Some(0)), Seq(4, 1, 2, 3, 5))
+ checkEvaluation(Shuffle(ai1, Some(0)), Seq(3, 1, 2))
+ checkEvaluation(Shuffle(ai2, Some(0)), Seq(3, null, 1, null))
+ checkEvaluation(Shuffle(ai3, Some(0)), Seq(null, 2, null, 4))
+ checkEvaluation(Shuffle(ai4, Some(0)), Seq(null, null, null))
+ checkEvaluation(Shuffle(ai5, Some(0)), Seq(1))
+ checkEvaluation(Shuffle(ai6, Some(0)), Seq.empty)
+ checkEvaluation(Shuffle(ai7, Some(0)), null)
+
+ // Non-primitive-type elements
+ val as0 = Literal.create(Seq("a", "b", "c", "d"), ArrayType(StringType, containsNull = false))
+ val as1 = Literal.create(Seq("a", "b", "c"), ArrayType(StringType, containsNull = false))
+ val as2 = Literal.create(Seq(null, "a", null, "c"), ArrayType(StringType, containsNull = true))
+ val as3 = Literal.create(Seq("b", null, "d", null), ArrayType(StringType, containsNull = true))
+ val as4 = Literal.create(Seq(null, null, null), ArrayType(StringType, containsNull = true))
+ val as5 = Literal.create(Seq("a"), ArrayType(StringType, containsNull = false))
+ val as6 = Literal.create(Seq.empty, ArrayType(StringType, containsNull = false))
+ val as7 = Literal.create(null, ArrayType(StringType, containsNull = true))
+ val aa = Literal.create(
+ Seq(Seq("a", "b"), Seq("c", "d"), Seq("e")),
+ ArrayType(ArrayType(StringType)))
+
+ checkEvaluation(Shuffle(as0, Some(0)), Seq("d", "a", "b", "c"))
+ checkEvaluation(Shuffle(as1, Some(0)), Seq("c", "a", "b"))
+ checkEvaluation(Shuffle(as2, Some(0)), Seq("c", null, "a", null))
+ checkEvaluation(Shuffle(as3, Some(0)), Seq(null, "b", null, "d"))
+ checkEvaluation(Shuffle(as4, Some(0)), Seq(null, null, null))
+ checkEvaluation(Shuffle(as5, Some(0)), Seq("a"))
+ checkEvaluation(Shuffle(as6, Some(0)), Seq.empty)
+ checkEvaluation(Shuffle(as7, Some(0)), null)
+ checkEvaluation(Shuffle(aa, Some(0)), Seq(Seq("e"), Seq("a", "b"), Seq("c", "d")))
+
+ val r = new Random()
+ val seed1 = Some(r.nextLong())
+ assert(evaluateWithoutCodegen(Shuffle(ai0, seed1)) ===
+ evaluateWithoutCodegen(Shuffle(ai0, seed1)))
+ assert(evaluateWithGeneratedMutableProjection(Shuffle(ai0, seed1)) ===
+ evaluateWithGeneratedMutableProjection(Shuffle(ai0, seed1)))
+ assert(evaluateWithUnsafeProjection(Shuffle(ai0, seed1)) ===
+ evaluateWithUnsafeProjection(Shuffle(ai0, seed1)))
+
+ val seed2 = Some(r.nextLong())
--- End diff --
Do we need to ensure this property (different seeds must generate different result)?
We likely expect this property. However, I think that this test is too strict.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r205704109
--- Diff: python/pyspark/sql/functions.py ---
@@ -2382,6 +2382,20 @@ def array_sort(col):
return Column(sc._jvm.functions.array_sort(_to_java_column(col)))
+@since(2.4)
+def shuffle(col):
+ """
+ Collection function: Generates a random permutation of the given array.
+
+ .. note:: The function is non-deterministic because its results depends on order of rows which
+ may be non-deterministic after a shuffle.
+
+ :param col: name of column or expression
--- End diff --
Python doctest looks missing.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21802: [SPARK-23928][SQL] Add shuffle collection functio...
Posted by ueshin <gi...@git.apache.org>.
Github user ueshin commented on a diff in the pull request:
https://github.com/apache/spark/pull/21802#discussion_r203962951
--- Diff: python/pyspark/sql/functions.py ---
@@ -2382,6 +2382,20 @@ def array_sort(col):
return Column(sc._jvm.functions.array_sort(_to_java_column(col)))
+@since(2.4)
+def shuffle(col):
+ """
+ Collection function: Generates a random permutation of the given array.
+
+ .. note:: The function is non-deterministic because its results depends on order of rows which
--- End diff --
The seed is fixed when analysis phase, so if we, say, `collect()` twice or more from the same DataFrame, we will get the same result:
```scala
val df = .. .select(shuffle('arr))
df.collect() == df.collect()
```
but if we create another DataFrame from the same input, we will get different results:
```scala
val df1 = .. .select(shuffle('arr))
val df2 = .. .select(shuffle('arr))
df1.collect() != df2.collect()
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org