tdas commented on a change in pull request #33093:
URL: https://github.com/apache/spark/pull/33093#discussion_r661566959
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala
##########
@@ -250,6 +370,63 @@ case class FlatMapGroupsWithStateExec(
}
}
- override protected def withNewChildInternal(newChild: SparkPlan):
FlatMapGroupsWithStateExec =
- copy(child = newChild)
+ override protected def withNewChildrenInternal(
+ newLeft: SparkPlan, newRight: SparkPlan): FlatMapGroupsWithStateExec =
+ copy(child = newLeft, initialState = newRight)
+}
+
+object FlatMapGroupsWithStateExec {
+
+ /**
+ * This method merges two grouped iterators such that we have a combined
iterator
+ * Iterator (K, Iterator(v), Iterator(s))
+ * @param leftGrpItr - Iterator (K, Iterator(V))
+ * @param rightGrpItr - Iterator (K, Iterator(S))
+ * @param comparator - BaseOrdering that can compare the items from both the
iterators.
+ */
+ def mergeGroupedIters(
+ leftGrpItr: Iterator[(InternalRow, Iterator[InternalRow])],
+ rightGrpItr: Iterator[(InternalRow, Iterator[InternalRow])],
+ comparator: BaseOrdering
+ ): Iterator[(InternalRow, Iterator[InternalRow], Iterator[InternalRow])] =
{
+ new NextIterator[(InternalRow, Iterator[InternalRow],
Iterator[InternalRow])] {
+
+ var leftRow: (InternalRow, Iterator[InternalRow]) = null
+ var rightRow: (InternalRow, Iterator[InternalRow]) = null
+
+ override def getNext(): (InternalRow, Iterator[InternalRow],
Iterator[InternalRow]) = {
+ if (leftRow == null && leftGrpItr.hasNext) {
+ leftRow = leftGrpItr.next()
+ }
+ if (rightRow == null && rightGrpItr.hasNext) {
+ rightRow = rightGrpItr.next()
+ }
+ if (leftRow == null && rightRow == null) {
+ finished = true
+ return null
+ }
+ val comparison =
Review comment:
explain this logic with inline comments.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
