Github user tejasapatil commented on a diff in the pull request:
https://github.com/apache/spark/pull/16909#discussion_r101150862
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/window/RowBuffer.scala
---
@@ -17,99 +17,33 @@
package org.apache.spark.sql.execution.window
-import scala.collection.mutable.ArrayBuffer
-
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.UnsafeRow
-import org.apache.spark.util.collection.unsafe.sort.{UnsafeExternalSorter,
UnsafeSorterIterator}
-
+import org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray
/**
- * The interface of row buffer for a partition. In absence of a buffer
pool (with locking), the
+ * Represents row buffer for a partition. In absence of a buffer pool
(with locking), the
* row buffer is used to materialize a partition of rows since we need to
repeatedly scan these
* rows in window function processing.
*/
-private[window] abstract class RowBuffer {
-
- /** Number of rows. */
- def size: Int
-
- /** Return next row in the buffer, null if no more left. */
- def next(): InternalRow
-
- /** Skip the next `n` rows. */
- def skip(n: Int): Unit
-
- /** Return a new RowBuffer that has the same rows. */
- def copy(): RowBuffer
-}
-
-/**
- * A row buffer based on ArrayBuffer (the number of rows is limited).
- */
-private[window] class ArrayRowBuffer(buffer: ArrayBuffer[UnsafeRow])
extends RowBuffer {
-
- private[this] var cursor: Int = -1
-
- /** Number of rows. */
- override def size: Int = buffer.length
-
- /** Return next row in the buffer, null if no more left. */
- override def next(): InternalRow = {
- cursor += 1
- if (cursor < buffer.length) {
- buffer(cursor)
- } else {
- null
- }
- }
-
- /** Skip the next `n` rows. */
- override def skip(n: Int): Unit = {
- cursor += n
- }
-
- /** Return a new RowBuffer that has the same rows. */
- override def copy(): RowBuffer = {
- new ArrayRowBuffer(buffer)
- }
-}
-
-/**
- * An external buffer of rows based on UnsafeExternalSorter.
- */
-private[window] class ExternalRowBuffer(sorter: UnsafeExternalSorter,
numFields: Int)
- extends RowBuffer {
-
- private[this] val iter: UnsafeSorterIterator = sorter.getIterator
-
- private[this] val currentRow = new UnsafeRow(numFields)
+private[window] class RowBuffer(appendOnlyExternalArray:
ExternalAppendOnlyUnsafeRowArray) {
--- End diff --
Removed `RowBuffer`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
