Github user mukulmurthy commented on a diff in the pull request:
https://github.com/apache/spark/pull/21559#discussion_r195268130
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala
---
@@ -294,6 +333,16 @@ class MemorySink(val schema: StructType, outputMode:
OutputMode) extends Sink
def clear(): Unit = synchronized {
batches.clear()
+ numRows = 0
+ }
+
+ private def truncateRowsIfNeeded(rows: Array[Row], maxRows: Int,
batchId: Long): Array[Row] = {
+ if (rows.length > maxRows) {
--- End diff --
Also adding a check here to make sure maxRows >= 0. It shouldn't ever be
negative, but doesn't hurt to safeguard.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
