[GitHub] [spark] c21 commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility

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

c21 commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r826649881



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StatefulOperatorPartitioning.scala
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, Distribution, StatefulOpClusteredDistribution}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.STATEFUL_OPERATOR_USE_STRICT_DISTRIBUTION
+
+/**
+ * This object is to provide clustered distribution for stateful operator with ensuring backward
+ * compatibility. Please read through the NOTE on the classdoc of
+ * [[StatefulOpClusteredDistribution]] before making any changes. Please refer SPARK-38204
+ * for details.
+ *
+ * Do not use methods in this object for stateful operators which already uses
+ * [[StatefulOpClusteredDistribution]] as its required child distribution.
+ */
+object StatefulOperatorPartitioning {
+
+  def getClusteredDistributionWithBackwardCompatibility(
+      expressions: Seq[Expression],
+      stateInfo: StatefulOperatorStateInfo,
+      conf: SQLConf): Distribution = {
+    getClusteredDistributionWithBackwardCompatibility(expressions, stateInfo.numPartitions, conf)
+  }
+
+  def getClusteredDistributionWithBackwardCompatibility(

Review comment:
       To me, it feels a little bit too long, how about something like `getCompatibleDistribution`, or just `getDistribution` given the class name `StatefulOperatorPartitioning` already has some context. Non-blocking comment. cc @cloud-fan as well.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/BaseAggregateExec.scala
##########
@@ -92,7 +95,20 @@ trait BaseAggregateExec extends UnaryExecNode with AliasAwareOutputPartitioning
   override def requiredChildDistribution: List[Distribution] = {
     requiredChildDistributionExpressions match {
       case Some(exprs) if exprs.isEmpty => AllTuples :: Nil
-      case Some(exprs) => ClusteredDistribution(exprs) :: Nil
+      case Some(exprs) =>
+        if (isStreaming) {
+          numShufflePartitions match {
+            case Some(parts) =>
+              StatefulOperatorPartitioning.getClusteredDistributionWithBackwardCompatibility(
+                exprs, parts, conf) :: Nil
+
+            case _ =>
+              throw new IllegalStateException("Expected to set the number of partitions before " +

Review comment:
       Maybe we can also add a `require` assertion at class level, e.g. at Line 38
   
   ```
   require(isStreaming == numShufflePartitions.isDefined, "Expected to set the number of partitions for streaming aggregate")
   ```
   
   Or we can define only one variable `numStatefulShufflePartitions: Option[Int]` instead of two: `isStreaming` and `numShufflePartitions`?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/OffsetSeq.scala
##########
@@ -118,7 +118,8 @@ object OffsetSeqMetadata extends Logging {
       StreamingAggregationStateManager.legacyVersion.toString,
     STREAMING_JOIN_STATE_FORMAT_VERSION.key ->
       SymmetricHashJoinStateManager.legacyVersion.toString,
-    STATE_STORE_COMPRESSION_CODEC.key -> "lz4"
+    STATE_STORE_COMPRESSION_CODEC.key -> "lz4",
+    STATEFUL_OPERATOR_USE_STRICT_DISTRIBUTION.key -> "false"

Review comment:
       neat workaround!




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