[GitHub] [spark] HeartSaVioR commented on a diff in pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait

[GitBox <gi...@apache.org>]

HeartSaVioR commented on code in PR #37859:
URL: https://github.com/apache/spark/pull/37859#discussion_r969132744


##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala:
##########
@@ -33,59 +34,35 @@ import org.apache.spark.util.{CompletionIterator, SerializableConfiguration}
 
 /**
  * Physical operator for executing `FlatMapGroupsWithState`
- *
- * @param func function called on each group
- * @param keyDeserializer used to extract the key object for each group.
- * @param valueDeserializer used to extract the items in the iterator from an input row.
- * @param initialStateDeserializer used to extract the state object from the initialState dataset
- * @param groupingAttributes used to group the data
- * @param dataAttributes used to read the data
- * @param outputObjAttr Defines the output object
- * @param stateEncoder used to serialize/deserialize state before calling `func`
- * @param outputMode the output mode of `func`
- * @param timeoutConf used to timeout groups that have not received data in a while
- * @param batchTimestampMs processing timestamp of the current batch.
- * @param eventTimeWatermark event time watermark for the current batch
- * @param initialState the user specified initial state
- * @param hasInitialState indicates whether the initial state is provided or not
- * @param child the physical plan for the underlying data
  */
-case class FlatMapGroupsWithStateExec(
-    func: (Any, Iterator[Any], LogicalGroupState[Any]) => Iterator[Any],
-    keyDeserializer: Expression,
-    valueDeserializer: Expression,
-    initialStateDeserializer: Expression,
-    groupingAttributes: Seq[Attribute],
-    initialStateGroupAttrs: Seq[Attribute],
-    dataAttributes: Seq[Attribute],
-    initialStateDataAttrs: Seq[Attribute],
-    outputObjAttr: Attribute,
-    stateInfo: Option[StatefulOperatorStateInfo],
-    stateEncoder: ExpressionEncoder[Any],
-    stateFormatVersion: Int,
-    outputMode: OutputMode,
-    timeoutConf: GroupStateTimeout,
-    batchTimestampMs: Option[Long],
-    eventTimeWatermark: Option[Long],
-    initialState: SparkPlan,
-    hasInitialState: Boolean,
-    child: SparkPlan
-  ) extends BinaryExecNode with ObjectProducerExec with StateStoreWriter with WatermarkSupport {
-
-  import FlatMapGroupsWithStateExecHelper._
+trait FlatMapGroupsWithStateExecBase
+    extends StateStoreWriter with WatermarkSupport {
   import GroupStateImpl._
+  import FlatMapGroupsWithStateExecHelper._
 
-  override def left: SparkPlan = child
+  protected val groupingAttributes: Seq[Attribute]
 
-  override def right: SparkPlan = initialState
+  protected val initialStateDeserializer: Expression
+  protected val initialStateGroupAttrs: Seq[Attribute]
+  protected val initialStateDataAttrs: Seq[Attribute]
+  protected val initialState: SparkPlan
+  protected val hasInitialState: Boolean
 
-  private val isTimeoutEnabled = timeoutConf != NoTimeout
-  private val watermarkPresent = child.output.exists {
+  val stateInfo: Option[StatefulOperatorStateInfo]

Review Comment:
   Ah OK that was from another trait. Thanks for explanation.



-- 
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-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org