Github user jose-torres commented on a diff in the pull request:
https://github.com/apache/spark/pull/21559#discussion_r195516859
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/memoryV2.scala
---
@@ -110,40 +126,61 @@ class MemorySinkV2 extends DataSourceV2 with
StreamWriteSupport with MemorySinkB
def clear(): Unit = synchronized {
batches.clear()
+ numRows = 0
+ }
+
+ private def truncateRowsIfNeeded(rows: Array[Row], maxRows: Int,
batchId: Long): Array[Row] = {
--- End diff --
Can this go in MemorySinkBase?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
