You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/02/28 04:04:11 UTC
[GitHub] [spark] HeartSaVioR opened a new pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
HeartSaVioR opened a new pull request #35673:
URL: https://github.com/apache/spark/pull/35673
### What changes were proposed in this pull request?
This PR proposes to use StatefulOpClusteredDistribution for stateful operators which requires exact order of clustering keys without allowing sub-clustering keys, so that stateful operators will have consistent partitioning across lifetime of the query.
(For sure, it doesn't include the case grouping keys are changed. We have state schema checker verifying on the changes, but changing name is allowed so swapping keys with same data type is still allowed.)
The change will break the existing queries having checkpoint in prior to Spark 3.3 and bring silent correctness issues. To remedy the problem, we introduce a new internal config `spark.sql.streaming.statefulOperator.useStrictDistribution`, which defaults to true for new queries but defaults to false for queries starting from checkpoint in prior to Spark 3.3. If the new config is set to false, stateful operator will use ClusteredDistribution which retains the old requirement of child distribution.
Note that in this change we don't fix the root problem against old checkpoints. Long-term fix should be crafted carefully, after collecting evidence on the impact of SPARK-38204. (e.g. how many queries on end users would encounter SPARK-38204.)
This PR adds E2E tests for the cases which trigger SPARK-38204, and verify the behavior with new query (3.3) & old query (in prior to 3.3).
### Why are the changes needed?
Please refer the description of JIRA issue [SPARK-38024](https://issues.apache.org/jira/browse/SPARK-38204) for details, since the description is quite long to include here.
### Does this PR introduce _any_ user-facing change?
Yes, stateful operators no longer accept the child output partitioning having subset of grouping keys and trigger additional shuffle. This will ensure consistent partitioning with stateful operators across lifetime of the query.
### How was this patch tested?
New UTs including backward compatibility are added.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1068601041
Thanks all for reviewing and merging! Let's port back this to the next bugfix release(s) as this is about correctness. I'll craft a PR for 3.2 branch as well. Not sure we will plan to have another 3.1.x after having 3.3.0, hence let's do this for 3.2 only for now.
--
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
[GitHub] [spark] xuanyuanking closed pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
xuanyuanking closed pull request #35673:
URL: https://github.com/apache/spark/pull/35673
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1057570601
cc. friendly reminder, @cloud-fan @viirya @sunchao @c21
Worth noting that I marked this as a blocker for 3.3 - while this is not a regression, we should not leave this to break more queries silently once we identified a problem. Long-term fix can be done on demand. (e.g. let stateful operator have desired fixed output partitioning despite it's not full cluster keys)
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r826477929
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -447,8 +475,10 @@ object AggUtils {
val aggregateExpressions = functionsWithoutDistinct.map(_.copy(mode = PartialMerge))
val aggregateAttributes = aggregateExpressions.map(_.resultAttribute)
MergingSessionsExec(
- requiredChildDistributionExpressions = None,
- requiredChildDistributionOption = Some(restored.requiredChildDistribution),
+ requiredChildDistributionExpressions = Some(groupingWithoutSessionAttributes),
+ isStreaming = true,
+ // This will be replaced with actual value in state rule.
Review comment:
We don't create a dedicate class for state rule. See https://github.com/apache/spark/blob/a30575e91e82e9f4394add40f1ba9265eb1f7819/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala#L105-L209
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r826710332
##########
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:
The error framework is for user facing errors. This is something like "this should not be called, internal error". I just made the error message be general to make our developer life be easier.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1056158157
Failed test - org.apache.spark.sql.streaming.StreamingQueryListenerSuite.`single listener, check trigger events are generated correctly` - doesn't seem to be relevant.
Scala 2.13 failure seems to be relevant
```
[error] /home/runner/work/spark/spark/sql/core/src/test/scala/org/apache/spark/sql/streaming/util/StatefulOpClusteredDistributionTestHelper.scala:30:70: match may not be exhaustive.
[error] It would fail on the following inputs: (AllTuples, _), (BroadcastDistribution(_), _), (ClusteredDistribution(_, _, _), _), (OrderedDistribution(_), _), (StatefulOpClusteredDistribution(_, _), _), (UnspecifiedDistribution, _)
[error] plan.requiredChildDistribution.zip(desiredClusterColumns).forall {
[error] ^
[error] /home/runner/work/spark/spark/sql/core/src/test/scala/org/apache/spark/sql/streaming/util/StatefulOpClusteredDistributionTestHelper.scala:42:70: match may not be exhaustive.
[error] It would fail on the following inputs: (AllTuples, _), (BroadcastDistribution(_), _), (ClusteredDistribution(_, _, _), _), (OrderedDistribution(_), _), (StatefulOpClusteredDistribution(_, _), _), (UnspecifiedDistribution, _)
[error] plan.requiredChildDistribution.zip(desiredClusterColumns).forall {
[error] ^
```
It caught the missing cases - I may need to add these cases.
--
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
[GitHub] [spark] c21 commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by 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
[GitHub] [spark] LuciferYang commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r826712437
##########
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:
ok
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1053872194
cc. @cloud-fan @viirya @sunchao @c21 who have context of this fix (this is a continuous work of #35419)
cc. @tdas @zsxwing @brkyvz @xuanyuanking for seeking eyes of reviews on additional streaming experts
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r819293042
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSessionWindowDistributionSuite.scala
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.streaming
+
+import java.io.File
+
+import org.apache.commons.io.FileUtils
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.plans.physical.UnspecifiedDistribution
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.streaming.{MemoryStream, SessionWindowStateStoreRestoreExec, SessionWindowStateStoreSaveExec}
+import org.apache.spark.sql.functions.{count, session_window}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.streaming.util.StatefulOpClusteredDistributionTestHelper
+import org.apache.spark.util.Utils
+
+class StreamingSessionWindowDistributionSuite extends StreamTest
+ with StatefulOpClusteredDistributionTestHelper with Logging {
+
+ import testImplicits._
+
+ test("SPARK-38204: session window aggregation should require StatefulOpClusteredDistribution " +
+ "from children") {
+
+ withSQLConf(
+ // exclude partial merging session to simplify test
+ SQLConf.STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION.key -> "false") {
+
+ val inputData = MemoryStream[(String, String, Long)]
+
+ // Split the lines into words, treat words as sessionId of events
+ val events = inputData.toDF()
+ .select($"_1".as("value"), $"_2".as("userId"), $"_3".as("timestamp"))
+ .withColumn("eventTime", $"timestamp".cast("timestamp"))
+ .withWatermark("eventTime", "30 seconds")
+ .selectExpr("explode(split(value, ' ')) AS sessionId", "userId", "eventTime")
+
+ val sessionUpdates = events
+ .repartition($"userId")
+ .groupBy(session_window($"eventTime", "10 seconds") as 'session, 'sessionId, 'userId)
+ .agg(count("*").as("numEvents"))
+ .selectExpr("sessionId", "userId", "CAST(session.start AS LONG)",
+ "CAST(session.end AS LONG)",
+ "CAST(session.end AS LONG) - CAST(session.start AS LONG) AS durationMs",
+ "numEvents")
+
+ testStream(sessionUpdates, OutputMode.Append())(
+ AddData(inputData,
+ ("hello world spark streaming", "key1", 40L),
+ ("world hello structured streaming", "key2", 41L)
+ ),
+
+ // skip checking the result, since we focus to verify the physical plan
+ ProcessAllAvailable(),
+ Execute { query =>
+ val numPartitions = query.lastExecution.numStateStores
+
+ val operators = query.lastExecution.executedPlan.collect {
+ case s: SessionWindowStateStoreRestoreExec => s
+ case s: SessionWindowStateStoreSaveExec => s
+ }
+
+ assert(operators.nonEmpty)
+ operators.foreach { stateOp =>
+ assert(requireStatefulOpClusteredDistribution(stateOp, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ assert(hasDesiredHashPartitioningInChildren(stateOp, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ }
+
+ // Verify aggregations in between, except partial aggregation.
+ // This includes MergingSessionsExec.
+ val allAggregateExecs = query.lastExecution.executedPlan.collect {
+ case a: BaseAggregateExec => a
+ }
+
+ val aggregateExecsWithoutPartialAgg = allAggregateExecs.filter {
+ _.requiredChildDistribution.head != UnspecifiedDistribution
+ }
+
+ // We expect single partial aggregation since we disable partial merging sessions.
+ // Remaining agg execs should have child producing expected output partitioning.
+ assert(allAggregateExecs.length - 1 === aggregateExecsWithoutPartialAgg.length)
+
+ // For aggregate execs, we make sure output partitioning of the children is same as
+ // we expect, HashPartitioning with clustering keys & number of partitions.
+ aggregateExecsWithoutPartialAgg.foreach { aggr =>
+ assert(hasDesiredHashPartitioningInChildren(aggr, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ }
+ }
+ )
+ }
+ }
+
+ test("SPARK-38204: session window aggregation should require ClusteredDistribution " +
+ "from children if the query starts from checkpoint in 3.2") {
+
+ withSQLConf(
+ // exclude partial merging session to simplify test
+ SQLConf.STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION.key -> "false",
+ // explode(split(value, ' ')) with value as nullable field seems to give different
+ // nullability between Spark 3.2 and 3.3 (nullable in 3.2 and non-null in 3.3)
+ // This should be allowed, but unfortunately, nullability check in state store schema
+ // checker seems to have a bug (`from` and `to` are swapped). Applying temporary
+ // workaround for now.
+ // TODO: investigate the schema checker and remove this once it is resolved.
Review comment:
https://github.com/apache/spark/pull/35731
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSessionWindowDistributionSuite.scala
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.streaming
+
+import java.io.File
+
+import org.apache.commons.io.FileUtils
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.plans.physical.UnspecifiedDistribution
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.streaming.{MemoryStream, SessionWindowStateStoreRestoreExec, SessionWindowStateStoreSaveExec}
+import org.apache.spark.sql.functions.{count, session_window}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.streaming.util.StatefulOpClusteredDistributionTestHelper
+import org.apache.spark.util.Utils
+
+class StreamingSessionWindowDistributionSuite extends StreamTest
+ with StatefulOpClusteredDistributionTestHelper with Logging {
+
+ import testImplicits._
+
+ test("SPARK-38204: session window aggregation should require StatefulOpClusteredDistribution " +
+ "from children") {
+
+ withSQLConf(
+ // exclude partial merging session to simplify test
+ SQLConf.STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION.key -> "false") {
+
+ val inputData = MemoryStream[(String, String, Long)]
+
+ // Split the lines into words, treat words as sessionId of events
+ val events = inputData.toDF()
+ .select($"_1".as("value"), $"_2".as("userId"), $"_3".as("timestamp"))
+ .withColumn("eventTime", $"timestamp".cast("timestamp"))
+ .withWatermark("eventTime", "30 seconds")
+ .selectExpr("explode(split(value, ' ')) AS sessionId", "userId", "eventTime")
+
+ val sessionUpdates = events
+ .repartition($"userId")
+ .groupBy(session_window($"eventTime", "10 seconds") as 'session, 'sessionId, 'userId)
+ .agg(count("*").as("numEvents"))
+ .selectExpr("sessionId", "userId", "CAST(session.start AS LONG)",
+ "CAST(session.end AS LONG)",
+ "CAST(session.end AS LONG) - CAST(session.start AS LONG) AS durationMs",
+ "numEvents")
+
+ testStream(sessionUpdates, OutputMode.Append())(
+ AddData(inputData,
+ ("hello world spark streaming", "key1", 40L),
+ ("world hello structured streaming", "key2", 41L)
+ ),
+
+ // skip checking the result, since we focus to verify the physical plan
+ ProcessAllAvailable(),
+ Execute { query =>
+ val numPartitions = query.lastExecution.numStateStores
+
+ val operators = query.lastExecution.executedPlan.collect {
+ case s: SessionWindowStateStoreRestoreExec => s
+ case s: SessionWindowStateStoreSaveExec => s
+ }
+
+ assert(operators.nonEmpty)
+ operators.foreach { stateOp =>
+ assert(requireStatefulOpClusteredDistribution(stateOp, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ assert(hasDesiredHashPartitioningInChildren(stateOp, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ }
+
+ // Verify aggregations in between, except partial aggregation.
+ // This includes MergingSessionsExec.
+ val allAggregateExecs = query.lastExecution.executedPlan.collect {
+ case a: BaseAggregateExec => a
+ }
+
+ val aggregateExecsWithoutPartialAgg = allAggregateExecs.filter {
+ _.requiredChildDistribution.head != UnspecifiedDistribution
+ }
+
+ // We expect single partial aggregation since we disable partial merging sessions.
+ // Remaining agg execs should have child producing expected output partitioning.
+ assert(allAggregateExecs.length - 1 === aggregateExecsWithoutPartialAgg.length)
+
+ // For aggregate execs, we make sure output partitioning of the children is same as
+ // we expect, HashPartitioning with clustering keys & number of partitions.
+ aggregateExecsWithoutPartialAgg.foreach { aggr =>
+ assert(hasDesiredHashPartitioningInChildren(aggr, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ }
+ }
+ )
+ }
+ }
+
+ test("SPARK-38204: session window aggregation should require ClusteredDistribution " +
+ "from children if the query starts from checkpoint in 3.2") {
+
+ withSQLConf(
+ // exclude partial merging session to simplify test
+ SQLConf.STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION.key -> "false",
+ // explode(split(value, ' ')) with value as nullable field seems to give different
+ // nullability between Spark 3.2 and 3.3 (nullable in 3.2 and non-null in 3.3)
+ // This should be allowed, but unfortunately, nullability check in state store schema
+ // checker seems to have a bug (`from` and `to` are swapped). Applying temporary
+ // workaround for now.
+ // TODO: investigate the schema checker and remove this once it is resolved.
Review comment:
SPARK-38412 - https://github.com/apache/spark/pull/35731
--
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
[GitHub] [spark] HeartSaVioR edited a comment on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR edited a comment on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1057570601
cc. friendly reminder, @cloud-fan @viirya @sunchao @c21 @xuanyuanking
Worth noting that I marked this as a blocker for 3.3 - while this is not a regression, we should not leave this to break more queries silently once we identified a problem. Long-term fix can be done on demand. (e.g. let stateful operator have desired fixed output partitioning despite it's not full cluster keys)
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1067446257
@xuanyuanking Thanks for the review! I addressed your review comments. Please take a look again.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1071983312
FYI I submitted a PR for 3.2 #35908.
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r815951144
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala
##########
@@ -542,8 +543,8 @@ class StreamingAggregationSuite extends StateStoreMetricsTest with Assertions {
/**
* This method verifies certain properties in the SparkPlan of a streaming aggregation.
* First of all, it checks that the child of a `StateStoreRestoreExec` creates the desired
- * data distribution, where the child could be an Exchange, or a `HashAggregateExec` which already
- * provides the expected data distribution.
+ * data distribution, where the child is a `HashAggregateExec` which already provides
Review comment:
The change of this test is intended - the main change leads the streaming aggregation to always introduce shuffle just after partial aggregation, so that remaining parts execute in the same stage with same output partitioning. The test was relying on old behavior which triggers shuffle just before state store load node in some case.
--
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
[GitHub] [spark] xuanyuanking commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r825682652
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -153,7 +179,8 @@ object AggUtils {
// If we have session window expression in aggregation, we add UpdatingSessionsExec to
// calculate sessions for input rows and update rows' session column, so that further
// aggregations can aggregate input rows for the same session.
- val maySessionChild = mayAppendUpdatingSessionExec(groupingExpressions, child)
+ val maySessionChild = mayAppendUpdatingSessionExec(groupingExpressions, child,
Review comment:
ditto
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -447,8 +475,10 @@ object AggUtils {
val aggregateExpressions = functionsWithoutDistinct.map(_.copy(mode = PartialMerge))
val aggregateAttributes = aggregateExpressions.map(_.resultAttribute)
MergingSessionsExec(
- requiredChildDistributionExpressions = None,
- requiredChildDistributionOption = Some(restored.requiredChildDistribution),
+ requiredChildDistributionExpressions = Some(groupingWithoutSessionAttributes),
+ isStreaming = true,
+ // This will be replaced with actual value in state rule.
Review comment:
let's link the state rule class name here?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -491,10 +521,15 @@ object AggUtils {
private def mayAppendUpdatingSessionExec(
groupingExpressions: Seq[NamedExpression],
- maybeChildPlan: SparkPlan): SparkPlan = {
+ maybeChildPlan: SparkPlan,
+ isStreaming: Boolean): SparkPlan = {
groupingExpressions.find(_.metadata.contains(SessionWindow.marker)) match {
case Some(sessionExpression) =>
UpdatingSessionsExec(
+ isStreaming = isStreaming,
+ // numShufflePartitions will be set to None, and replaced to the actual value in the
+ // state rule if the query is streaming.
Review comment:
ditto
##########
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.
Review comment:
nit: [[StatefulOpClusteredDistribution]]
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -1773,6 +1773,23 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATEFUL_OPERATOR_USE_STRICT_DISTRIBUTION =
+ buildConf("spark.sql.streaming.statefulOperator.useStrictDistribution")
+ .internal()
+ .doc("When true, the stateful operator for streaming query will use " +
+ "StatefulOpClusteredDistribution which guarantees stable state partitioning as long as " +
+ "the operator provides consistent grouping keys across the lifetime of query. " +
+ "When false, the stateful operator for streaming query will use ClusteredDistribution " +
+ "which is not sufficient to guarantee stable state partitioning despite the operator " +
+ "provides consistent grouping keys across the lifetime of query. " +
+ "This config will be set to true for new streaming queries to guarantee stable state " +
+ "partitioning, and set to false for existing streaming queries to not break queries " +
+ "which are restored from existing checkpoints. Please refer SPARK-38204 for details. " +
+ "The purpose of this config is only compatibility; DO NOT MANUALLY CHANGE THIS!!!")
Review comment:
let's move the description of `This config is for compatility only; DO NOT MANUALLY CHANGE THIS!!!` as the first sentence in the doc.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -121,7 +147,7 @@ object AggUtils {
// If we have session window expression in aggregation, we add MergingSessionExec to
// merge sessions with calculating aggregation values.
val interExec: SparkPlan = mayAppendMergingSessionExec(groupingExpressions,
- aggregateExpressions, partialAggregate)
+ aggregateExpressions, partialAggregate, isStreaming = false)
Review comment:
nit: how about using default value in `mayAppendMergingSessionExec` and let here unchange?
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r826478105
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -491,10 +521,15 @@ object AggUtils {
private def mayAppendUpdatingSessionExec(
groupingExpressions: Seq[NamedExpression],
- maybeChildPlan: SparkPlan): SparkPlan = {
+ maybeChildPlan: SparkPlan,
+ isStreaming: Boolean): SparkPlan = {
groupingExpressions.find(_.metadata.contains(SessionWindow.marker)) match {
case Some(sessionExpression) =>
UpdatingSessionsExec(
+ isStreaming = isStreaming,
+ // numShufflePartitions will be set to None, and replaced to the actual value in the
+ // state rule if the query is streaming.
Review comment:
same here
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1063809995
Friendly reminder, @cloud-fan @viirya @sunchao @c21 @xuanyuanking
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r826667759
##########
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:
We create a node with numShufflePartitions = None and replace the value in state rule. That said, we can't check the condition before state rule has been performed.
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r826669518
##########
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:
`getCompatibleDistribution` sounds good to me. Not that long and produces all context. Thanks for the suggestion!
--
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
[GitHub] [spark] LuciferYang commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r826671421
##########
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 should use the new error framework to throw exception in newly added code.
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r815567106
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -696,11 +687,8 @@ case class SessionWindowStateStoreSaveExec(
override def outputPartitioning: Partitioning = child.outputPartitioning
override def requiredChildDistribution: Seq[Distribution] = {
- // NOTE: Please read through the NOTE on the classdoc of StatefulOpClusteredDistribution
- // before making any changes.
- // TODO(SPARK-38204)
- ClusteredDistribution(keyExpressions,
- requiredNumPartitions = stateInfo.map(_.numPartitions)) :: Nil
+ StatefulOperatorPartitioning.getClusteredDistributionWithBackwardCompatibility(
Review comment:
I just changed keyExpressions to keyWithoutSessionExpressions which is actually same, but clearer to see the behavior.
--
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
[GitHub] [spark] LuciferYang commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1054029846
4 UT failed, maybe related to the current pr:
- SPARK-19690: do not convert batch aggregation in streaming query to streaming - state format version 1
- SPARK-19690: do not convert batch aggregation in streaming query to streaming - state format version 2
- SPARK-21977: coalesce(1) with aggregation should still be repartitioned when it has non-empty grouping keys - state format version 1
- SPARK-21977: coalesce(1) with aggregation should still be repartitioned when it has non-empty grouping keys - state format version 2
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35673:
URL: https://github.com/apache/spark/pull/35673#discussion_r815567960
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSessionWindowDistributionSuite.scala
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.streaming
+
+import java.io.File
+
+import org.apache.commons.io.FileUtils
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.plans.physical.UnspecifiedDistribution
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.streaming.{MemoryStream, SessionWindowStateStoreRestoreExec, SessionWindowStateStoreSaveExec}
+import org.apache.spark.sql.functions.{count, session_window}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.streaming.util.StatefulOpClusteredDistributionTestHelper
+import org.apache.spark.util.Utils
+
+class StreamingSessionWindowDistributionSuite extends StreamTest
+ with StatefulOpClusteredDistributionTestHelper with Logging {
+
+ import testImplicits._
+
+ test("SPARK-38204: session window aggregation should require StatefulOpClusteredDistribution " +
+ "from children") {
+
+ withSQLConf(
+ // exclude partial merging session to simplify test
+ SQLConf.STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION.key -> "false") {
+
+ val inputData = MemoryStream[(String, String, Long)]
+
+ // Split the lines into words, treat words as sessionId of events
+ val events = inputData.toDF()
+ .select($"_1".as("value"), $"_2".as("userId"), $"_3".as("timestamp"))
+ .withColumn("eventTime", $"timestamp".cast("timestamp"))
+ .withWatermark("eventTime", "30 seconds")
+ .selectExpr("explode(split(value, ' ')) AS sessionId", "userId", "eventTime")
+
+ val sessionUpdates = events
+ .repartition($"userId")
+ .groupBy(session_window($"eventTime", "10 seconds") as 'session, 'sessionId, 'userId)
+ .agg(count("*").as("numEvents"))
+ .selectExpr("sessionId", "userId", "CAST(session.start AS LONG)",
+ "CAST(session.end AS LONG)",
+ "CAST(session.end AS LONG) - CAST(session.start AS LONG) AS durationMs",
+ "numEvents")
+
+ testStream(sessionUpdates, OutputMode.Append())(
+ AddData(inputData,
+ ("hello world spark streaming", "key1", 40L),
+ ("world hello structured streaming", "key2", 41L)
+ ),
+
+ // skip checking the result, since we focus to verify the physical plan
+ ProcessAllAvailable(),
+ Execute { query =>
+ val numPartitions = query.lastExecution.numStateStores
+
+ val operators = query.lastExecution.executedPlan.collect {
+ case s: SessionWindowStateStoreRestoreExec => s
+ case s: SessionWindowStateStoreSaveExec => s
+ }
+
+ assert(operators.nonEmpty)
+ operators.foreach { stateOp =>
+ assert(requireStatefulOpClusteredDistribution(stateOp, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ assert(hasDesiredHashPartitioningInChildren(stateOp, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ }
+
+ // Verify aggregations in between, except partial aggregation.
+ // This includes MergingSessionsExec.
+ val allAggregateExecs = query.lastExecution.executedPlan.collect {
+ case a: BaseAggregateExec => a
+ }
+
+ val aggregateExecsWithoutPartialAgg = allAggregateExecs.filter {
+ _.requiredChildDistribution.head != UnspecifiedDistribution
+ }
+
+ // We expect single partial aggregation since we disable partial merging sessions.
+ // Remaining agg execs should have child producing expected output partitioning.
+ assert(allAggregateExecs.length - 1 === aggregateExecsWithoutPartialAgg.length)
+
+ // For aggregate execs, we make sure output partitioning of the children is same as
+ // we expect, HashPartitioning with clustering keys & number of partitions.
+ aggregateExecsWithoutPartialAgg.foreach { aggr =>
+ assert(hasDesiredHashPartitioningInChildren(aggr, Seq(Seq("sessionId", "userId")),
+ numPartitions))
+ }
+ }
+ )
+ }
+ }
+
+ test("SPARK-38204: session window aggregation should require ClusteredDistribution " +
+ "from children if the query starts from checkpoint in 3.2") {
+
+ withSQLConf(
+ // exclude partial merging session to simplify test
+ SQLConf.STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION.key -> "false",
+ // explode(split(value, ' ')) with value as nullable field seems to give different
+ // nullability between Spark 3.2 and 3.3 (nullable in 3.2 and non-null in 3.3)
+ // This should be allowed, but unfortunately, nullability check in state store schema
+ // checker seems to have a bug (`from` and `to` are swapped). Applying temporary
+ // workaround for now.
+ // TODO: investigate the schema checker and remove this once it is resolved.
Review comment:
Will investigate and provide a fix.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1054104130
Thanks for letting me know. Probably the changed behavior breaks some UTs while it is intentional. I'll audit failed tests and see whether we should fix the tests or remove them.
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1071983312
FYI I submitted a PR for 3.2 #35908.
--
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
[GitHub] [spark] xuanyuanking commented on pull request #35673: [SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #35673:
URL: https://github.com/apache/spark/pull/35673#issuecomment-1068440781
Thank you @HeartSaVioR for the fix. Merged to master!
We can use a separate PR to do the 3.2 backport if we want.
--
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