Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/21559#discussion_r195269434
--- 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] = {
+ if (rows.length > maxRows) {
+ logWarning(s"Truncating batch $batchId to $maxRows rows")
--- End diff --
How does take behave with negative rows? Printing a warning message with
negative values may be weird. I would also include the sink limit in the
warning.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
