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