LuciferYang commented on code in PR #46095:
URL: https://github.com/apache/spark/pull/46095#discussion_r1568353748
##########
core/src/main/scala/org/apache/spark/util/collection/Utils.scala:
##########
@@ -42,6 +42,23 @@ private[spark] object Utils extends SparkCollectionUtils {
ordering.leastOf(input.asJava, num).iterator.asScala
}
+ /**
+ * Returns the last K elements from the input.
+ */
+ def takeLast[T](input: Iterator[T], num: Int): Iterator[T] = {
+ assert(num >= 0)
+ if (input.isEmpty || num == 0) {
+ Iterator.empty[T]
+ } else {
+ var last = Seq.empty[T]
+ val grouped = input.grouped(num)
+ while (grouped.hasNext) {
+ last = grouped.next()
+ }
+ last.iterator
+ }
Review Comment:
It seems that there is an error in this logic. `takeLast(Range(0,
100).iterator, 3).toSeq` returns `List(99)` instead of `List(97, 98, 99)`.
Perhaps a `Queue` should be used to temporarily store the result data？
```scala
assert(num >= 0)
if (input.isEmpty || num == 0) {
Iterator.empty[T]
} else {
val queue = mutable.Queue.empty[T]
input.foreach { item =>
if (queue.size >= num) {
queue.dequeue()
}
queue.enqueue(item)
}
queue.iterator
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
