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
