Skip to content

[SPARK-24896][SQL] Uuid should produce different values for each execution in streaming query #21854

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
viirya wants to merge 4 commits into from
Closed

[SPARK-24896][SQL] Uuid should produce different values for each execution in streaming query #21854

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8ef299f
Uuid should produce different values in streaming query.
viirya Jul 24, 2018
c1ce69c
Address comment.
viirya Jul 24, 2018
1d629dc
Merge remote-tracking branch 'upstream/master' into uuid_in_streaming
viirya Jul 24, 2018
c127053
Move added rule to IncrementalExecution.
viirya Jul 31, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,11 @@ package org.apache.spark.sql.execution.streaming
import java.util.UUID
import java.util.concurrent.atomic.AtomicInteger

import scala.util.Random

import org.apache.spark.internal.Logging
import org.apache.spark.sql.{AnalysisException, SparkSession, Strategy}
import org.apache.spark.sql.catalyst.expressions.CurrentBatchTimestamp
import org.apache.spark.sql.catalyst.expressions.{CurrentBatchTimestamp, Uuid}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.plans.physical.{AllTuples, ClusteredDistribution, HashPartitioning, SinglePartition}
import org.apache.spark.sql.catalyst.rules.Rule
Expand Down Expand Up @@ -73,10 +75,14 @@ class IncrementalExecution(
* with the desired literal
*/
override lazy val optimizedPlan: LogicalPlan = {
val random = new Random()

sparkSession.sessionState.optimizer.execute(withCachedData) transformAllExpressions {
case ts @ CurrentBatchTimestamp(timestamp, _, _) =>
logInfo(s"Current batch timestamp = $timestamp")
ts.toLiteral
// SPARK-24896: Set the seed for random number generation in Uuid expressions.
case _: Uuid => Uuid(Some(random.nextLong()))
}
}

Expand Down
22 changes: 21 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ import java.{util => ju}
import java.util.Optional
import java.util.concurrent.CountDownLatch

import scala.collection.mutable

import org.apache.commons.lang3.RandomStringUtils
import org.scalactic.TolerantNumerics
import org.scalatest.BeforeAndAfter
Expand All @@ -29,8 +31,9 @@ import org.scalatest.mockito.MockitoSugar

import org.apache.spark.SparkException
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, Dataset}
import org.apache.spark.sql.{Column, DataFrame, Dataset, Row}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.Uuid
import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.execution.streaming.sources.TestForeachWriter
import org.apache.spark.sql.functions._
Expand Down Expand Up @@ -834,6 +837,23 @@ class StreamingQuerySuite extends StreamTest with BeforeAndAfter with Logging wi
CheckLastBatch(("A", 1)))
}

test("Uuid in streaming query should not produce same uuids in each execution") {
val uuids = mutable.ArrayBuffer[String]()
def collectUuid: Seq[Row] => Unit = { rows: Seq[Row] =>
rows.foreach(r => uuids += r.getString(0))
}

val stream = MemoryStream[Int]
val df = stream.toDF().select(new Column(Uuid()))
testStream(df)(
AddData(stream, 1),
CheckAnswer(collectUuid),
AddData(stream, 2),
CheckAnswer(collectUuid)
)
assert(uuids.distinct.size == 2)
}

test("StreamingRelationV2/StreamingExecutionRelation/ContinuousExecutionRelation.toJSON " +
"should not fail") {
val df = spark.readStream.format("rate").load()
Expand Down