brkyvz commented on code in PR #48401:
URL: https://github.com/apache/spark/pull/48401#discussion_r1844378796


##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StatefulProcessorHandleImpl.scala:
##########
@@ -343,7 +497,8 @@ class StatefulProcessorHandleImpl(
  * actually done. We need this class because we can only collect the schemas 
after
  * the StatefulProcessor is initialized.
  */
-class DriverStatefulProcessorHandleImpl(timeMode: TimeMode, keyExprEnc: 
ExpressionEncoder[Any])
+class DriverStatefulProcessorHandleImpl(
+    timeMode: TimeMode, keyExprEnc: ExpressionEncoder[Any], initializeAvroEnc: 
Boolean)

Review Comment:
   nit: one line per parameter please



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala:
##########
@@ -718,6 +733,22 @@ object TransformWithStateExec {
       stateStoreCkptIds = None
     )
 
+    val stateStoreEncoding = child.session.sessionState.conf.getConf(
+      SQLConf.STREAMING_STATE_STORE_ENCODING_FORMAT
+    )
+
+    def getDriverProcessorHandle(): DriverStatefulProcessorHandleImpl = {
+      val driverProcessorHandle = new DriverStatefulProcessorHandleImpl(
+        timeMode, keyEncoder, initializeAvroEnc =
+         avroEncodingEnabled(stateStoreEncoding))
+      
driverProcessorHandle.setHandleState(StatefulProcessorHandleState.PRE_INIT)
+      statefulProcessor.setHandle(driverProcessorHandle)
+      statefulProcessor.init(outputMode, timeMode)
+      driverProcessorHandle
+    }
+
+    val columnFamilySchemas = getDriverProcessorHandle().getColumnFamilySchemas

Review Comment:
   nit: maybe if you add a `withColumnFamilySchema` method, you can remove the 
need for this duplication and can just call it below after creating the class



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala:
##########
@@ -104,7 +106,9 @@ case class TransformWithStateExec(
    * @return a new instance of the driver processor handle
    */
   private def getDriverProcessorHandle(): DriverStatefulProcessorHandleImpl = {

Review Comment:
   nit: should this method have an assertion that it is being called on the 
Driver?



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala:
##########
@@ -596,7 +601,8 @@ case class TransformWithStateExec(
   private def initNewStateStoreAndProcessData(
       partitionId: Int,
       hadoopConfBroadcast: Broadcast[SerializableConfiguration])
-    (f: StateStore => CompletionIterator[InternalRow, Iterator[InternalRow]]):
+    (f: StateStore =>
+      CompletionIterator[InternalRow, Iterator[InternalRow]]):

Review Comment:
   also nit: should we use type aliasing to shorten this 
`CompletionIterator[InternalRow, Iterator[InternalRow]]`? Like `type 
ResultIterator = CompletionIterator[InternalRow, Iterator[InternalRow]]`



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StatefulProcessorHandleImpl.scala:
##########
@@ -118,6 +122,14 @@ class StatefulProcessorHandleImpl(
 
   currState = CREATED
 
+  private def getAvroEnc(stateName: String): Option[AvroEncoder] = {
+    if (!schemas.contains(stateName)) {
+      None
+    } else {
+      schemas(stateName).avroEnc
+    }

Review Comment:
   `schemas.get(stateName).map(_.avroEnc)`?



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala:
##########
@@ -596,7 +601,8 @@ case class TransformWithStateExec(
   private def initNewStateStoreAndProcessData(
       partitionId: Int,
       hadoopConfBroadcast: Broadcast[SerializableConfiguration])
-    (f: StateStore => CompletionIterator[InternalRow, Iterator[InternalRow]]):
+    (f: StateStore =>
+      CompletionIterator[InternalRow, Iterator[InternalRow]]):

Review Comment:
   uber nit: change necessary?



-- 
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

Reply via email to