tuplesPerSecond -> rowsPerSecond

Author: zsxwing

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/RateSourceProvider.scala

@@ -38,12 +38,12 @@ import org.apache.spark.util.{ManualClock, SystemClock}
  *  with 0L.
  *
  *  This source supports the following options:
- *  - `tuplesPerSecond` (e.g. 100, default: 1): How many tuples should be generated per second.
+ *  - `rowsPerSecond` (e.g. 100, default: 1): How many tuples should be generated per second.
  *  - `rampUpTime` (e.g. 5s, default: 0s): How long to ramp up before the generating speed
- *    becomes `tuplesPerSecond`. Using finer granularities than seconds will be truncated to integer
+ *    becomes `rowsPerSecond`. Using finer granularities than seconds will be truncated to integer
  *    seconds.
  *  - `numPartitions` (e.g. 10, default: Spark's default parallelism): The partition number for the
- *    generated tuples. The source will try its best to reach `tuplesPerSecond`, but the query may
+ *    generated tuples. The source will try its best to reach `rowsPerSecond`, but the query may
  *    be resource constrained, and `numPartitions` can be tweaked to help reach the desired speed.
  */
 class RateSourceProvider extends StreamSourceProvider with DataSourceRegister {
@@ -63,10 +63,10 @@ class RateSourceProvider extends StreamSourceProvider with DataSourceRegister {
       parameters: Map[String, String]): Source = {
     val params = CaseInsensitiveMap(parameters)
 
-    val tuplesPerSecond = params.get("tuplesPerSecond").map(_.toLong).getOrElse(1L)
-    if (tuplesPerSecond <= 0) {
+    val rowsPerSecond = params.get("rowsPerSecond").map(_.toLong).getOrElse(1L)
+    if (rowsPerSecond <= 0) {
       throw new IllegalArgumentException(
-        s"Invalid value '${params("tuplesPerSecond")}'. The option 'tuplesPerSecond' " +
+        s"Invalid value '${params("rowsPerSecond")}'. The option 'rowsPerSecond' " +
           "must be positive")
     }
 
@@ -89,7 +89,7 @@ class RateSourceProvider extends StreamSourceProvider with DataSourceRegister {
     new RateStreamSource(
       sqlContext,
       metadataPath,
-      tuplesPerSecond,
+      rowsPerSecond,
       rampUpTimeSeconds,
       numPartitions,
       params.get("useManualClock").map(_.toBoolean).getOrElse(false) // Only for testing
@@ -108,7 +108,7 @@ object RateSourceProvider {
 class RateStreamSource(
     sqlContext: SQLContext,
     metadataPath: String,
-    tuplesPerSecond: Long,
+    rowsPerSecond: Long,
     rampUpTimeSeconds: Long,
     numPartitions: Int,
     useManualClock: Boolean) extends Source with Logging {
@@ -118,11 +118,11 @@ class RateStreamSource(
 
   val clock = if (useManualClock) new ManualClock else new SystemClock
 
-  private val maxSeconds = Long.MaxValue / tuplesPerSecond
+  private val maxSeconds = Long.MaxValue / rowsPerSecond
 
   if (rampUpTimeSeconds > maxSeconds) {
     throw new ArithmeticException(
-      s"Integer overflow. Max offset with $tuplesPerSecond tuplesPerSecond" +
+      s"Integer overflow. Max offset with $rowsPerSecond rowsPerSecond" +
         s" is $maxSeconds, but 'rampUpTimeSeconds' is $rampUpTimeSeconds.")
   }
 
@@ -183,14 +183,14 @@ class RateStreamSource(
     assert(startSeconds <= endSeconds, s"startSeconds($startSeconds) > endSeconds($endSeconds)")
     if (endSeconds > maxSeconds) {
       throw new ArithmeticException("Integer overflow. Max offset with " +
-        s"$tuplesPerSecond tuplesPerSecond is $maxSeconds, but it's $endSeconds now.")
+        s"$rowsPerSecond rowsPerSecond is $maxSeconds, but it's $endSeconds now.")
     }
     // Fix "lastTimeMs" for recovery
     if (lastTimeMs < TimeUnit.SECONDS.toMillis(endSeconds) + startTimeMs) {
       lastTimeMs = TimeUnit.SECONDS.toMillis(endSeconds) + startTimeMs
     }
-    val rangeStart = valueAtSecond(startSeconds, tuplesPerSecond, rampUpTimeSeconds)
-    val rangeEnd = valueAtSecond(endSeconds, tuplesPerSecond, rampUpTimeSeconds)
+    val rangeStart = valueAtSecond(startSeconds, rowsPerSecond, rampUpTimeSeconds)
+    val rangeEnd = valueAtSecond(endSeconds, rowsPerSecond, rampUpTimeSeconds)
     logDebug(s"startSeconds: $startSeconds, endSeconds: $endSeconds, " +
       s"rangeStart: $rangeStart, rangeEnd: $rangeEnd")
 
@@ -254,14 +254,14 @@ class RateStreamSource(
 object RateStreamSource {
 
   /** Calculate the end value we will emit at the time `seconds`. */
-  def valueAtSecond(seconds: Long, tuplesPerSecond: Long, rampUpTimeSeconds: Long): Long = {
-    // E.g., rampUpTimeSeconds = 4, tuplesPerSecond = 10
+  def valueAtSecond(seconds: Long, rowsPerSecond: Long, rampUpTimeSeconds: Long): Long = {
+    // E.g., rampUpTimeSeconds = 4, rowsPerSecond = 10
     // Then speedDeltaPerSecond = 2
     //
     // seconds   = 0 1 2  3  4  5  6
     // speed     = 0 2 4  6  8 10 10 (speedDeltaPerSecond * seconds)
     // end value = 0 2 6 12 20 30 40 (0 + speedDeltaPerSecond * seconds) * (seconds + 1) / 2
-    val speedDeltaPerSecond = tuplesPerSecond / (rampUpTimeSeconds + 1)
+    val speedDeltaPerSecond = rowsPerSecond / (rampUpTimeSeconds + 1)
     if (seconds <= rampUpTimeSeconds) {
       // Calculate "(0 + speedDeltaPerSecond * seconds) * (seconds + 1) / 2" in a special way to
       // avoid overflow
@@ -272,8 +272,8 @@ object RateStreamSource {
       }
     } else {
       // rampUpPart is just a special case of the above formula: rampUpTimeSeconds == seconds
-      val rampUpPart = valueAtSecond(rampUpTimeSeconds, tuplesPerSecond, rampUpTimeSeconds)
-      rampUpPart + (seconds - rampUpTimeSeconds) * tuplesPerSecond
+      val rampUpPart = valueAtSecond(rampUpTimeSeconds, rowsPerSecond, rampUpTimeSeconds)
+      rampUpPart + (seconds - rampUpTimeSeconds) * rowsPerSecond
     }
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/RateSourceSuite.scala

@@ -42,7 +42,7 @@ class RateSourceSuite extends StreamTest {
   test("basic") {
     val input = spark.readStream
       .format("rate")
-      .option("tuplesPerSecond", "10")
+      .option("rowsPerSecond", "10")
       .option("useManualClock", "true")
       .load()
     testStream(input)(
@@ -56,10 +56,10 @@ class RateSourceSuite extends StreamTest {
     )
   }
 
-  test("uniform distribution of event timestamps: tuplesPerSecond > 1000") {
+  test("uniform distribution of event timestamps: rowsPerSecond > 1000") {
     val input = spark.readStream
       .format("rate")
-      .option("tuplesPerSecond", "1500")
+      .option("rowsPerSecond", "1500")
       .option("useManualClock", "true")
       .load()
       .as[(java.sql.Timestamp, Long)]
@@ -73,10 +73,10 @@ class RateSourceSuite extends StreamTest {
     )
   }
 
-  test("uniform distribution of event timestamps: tuplesPerSecond < 1000") {
+  test("uniform distribution of event timestamps: rowsPerSecond < 1000") {
     val input = spark.readStream
       .format("rate")
-      .option("tuplesPerSecond", "400")
+      .option("rowsPerSecond", "400")
       .option("useManualClock", "true")
       .load()
       .as[(java.sql.Timestamp, Long)]
@@ -92,23 +92,23 @@ class RateSourceSuite extends StreamTest {
   test("valueAtSecond") {
     import RateStreamSource._
 
-    assert(valueAtSecond(seconds = 0, tuplesPerSecond = 5, rampUpTimeSeconds = 2) === 0)
-    assert(valueAtSecond(seconds = 1, tuplesPerSecond = 5, rampUpTimeSeconds = 2) === 1)
-    assert(valueAtSecond(seconds = 2, tuplesPerSecond = 5, rampUpTimeSeconds = 2) === 3)
-    assert(valueAtSecond(seconds = 3, tuplesPerSecond = 5, rampUpTimeSeconds = 2) === 8)
-
-    assert(valueAtSecond(seconds = 0, tuplesPerSecond = 10, rampUpTimeSeconds = 4) === 0)
-    assert(valueAtSecond(seconds = 1, tuplesPerSecond = 10, rampUpTimeSeconds = 4) === 2)
-    assert(valueAtSecond(seconds = 2, tuplesPerSecond = 10, rampUpTimeSeconds = 4) === 6)
-    assert(valueAtSecond(seconds = 3, tuplesPerSecond = 10, rampUpTimeSeconds = 4) === 12)
-    assert(valueAtSecond(seconds = 4, tuplesPerSecond = 10, rampUpTimeSeconds = 4) === 20)
-    assert(valueAtSecond(seconds = 5, tuplesPerSecond = 10, rampUpTimeSeconds = 4) === 30)
+    assert(valueAtSecond(seconds = 0, rowsPerSecond = 5, rampUpTimeSeconds = 2) === 0)
+    assert(valueAtSecond(seconds = 1, rowsPerSecond = 5, rampUpTimeSeconds = 2) === 1)
+    assert(valueAtSecond(seconds = 2, rowsPerSecond = 5, rampUpTimeSeconds = 2) === 3)
+    assert(valueAtSecond(seconds = 3, rowsPerSecond = 5, rampUpTimeSeconds = 2) === 8)
+
+    assert(valueAtSecond(seconds = 0, rowsPerSecond = 10, rampUpTimeSeconds = 4) === 0)
+    assert(valueAtSecond(seconds = 1, rowsPerSecond = 10, rampUpTimeSeconds = 4) === 2)
+    assert(valueAtSecond(seconds = 2, rowsPerSecond = 10, rampUpTimeSeconds = 4) === 6)
+    assert(valueAtSecond(seconds = 3, rowsPerSecond = 10, rampUpTimeSeconds = 4) === 12)
+    assert(valueAtSecond(seconds = 4, rowsPerSecond = 10, rampUpTimeSeconds = 4) === 20)
+    assert(valueAtSecond(seconds = 5, rowsPerSecond = 10, rampUpTimeSeconds = 4) === 30)
   }
 
   test("rampUpTime") {
     val input = spark.readStream
       .format("rate")
-      .option("tuplesPerSecond", "10")
+      .option("rowsPerSecond", "10")
       .option("rampUpTime", "4s")
       .option("useManualClock", "true")
       .load()
@@ -138,7 +138,7 @@ class RateSourceSuite extends StreamTest {
   test("numPartitions") {
     val input = spark.readStream
       .format("rate")
-      .option("tuplesPerSecond", "10")
+      .option("rowsPerSecond", "10")
       .option("numPartitions", "6")
       .option("useManualClock", "true")
       .load()
@@ -153,15 +153,15 @@ class RateSourceSuite extends StreamTest {
   testQuietly("overflow") {
     val input = spark.readStream
       .format("rate")
-      .option("tuplesPerSecond", Long.MaxValue.toString)
+      .option("rowsPerSecond", Long.MaxValue.toString)
       .option("useManualClock", "true")
       .load()
       .select(spark_partition_id())
       .distinct()
     testStream(input)(
       AdvanceRateManualClock(2),
       ExpectFailure[ArithmeticException](t => {
-        Seq("overflow", "tuplesPerSecond").foreach { msg =>
+        Seq("overflow", "rowsPerSecond").foreach { msg =>
           assert(t.getMessage.contains(msg))
         }
       })
@@ -189,7 +189,7 @@ class RateSourceSuite extends StreamTest {
       }
     }
 
-    testIllegalOptionValue("tuplesPerSecond", "-1", Seq("-1", "tuplesPerSecond", "positive"))
+    testIllegalOptionValue("rowsPerSecond", "-1", Seq("-1", "rowsPerSecond", "positive"))
     testIllegalOptionValue("numPartitions", "-1", Seq("-1", "numPartitions", "positive"))
   }
 }