tdas commented on a change in pull request #33336:
URL: https://github.com/apache/spark/pull/33336#discussion_r669979360
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/objects.scala
##########
@@ -389,6 +389,102 @@ case class AppendColumnsWithObjectExec(
copy(child = newChild)
}
+/**
+ * Groups the input rows together and calls the function with each group and
an iterator containing
+ * all elements in the group. The result of this function is flattened before
being output. This
+ * version of the Physical operator takes a user provided initial state.
+ */
+case class MapGroupsWithInitialStateExec(
+ func: (Any, Iterator[Any], LogicalGroupState[Any]) => TraversableOnce[Any],
+ keyDeserializer: Expression,
+ valueDeserializer: Expression,
+ initialStateDeserializer: Expression,
+ groupingAttributes: Seq[Attribute],
+ initialStateGroupingAttributes: Seq[Attribute],
+ dataAttributes: Seq[Attribute],
+ initialStateDataAttrs: Seq[Attribute],
+ outputObjAttr: Attribute,
+ initialState: SparkPlan,
+ timeoutConf: GroupStateTimeout,
+ child: SparkPlan) extends BinaryExecNode with ObjectProducerExec {
+
+ override def left: SparkPlan = child
+
+ override def right: SparkPlan = initialState
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
Review comment:
what is the output partitioning in MapGroupsExec... it should be the same
--
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]
