[SPARK-38124][SQL][SS] Introduce StatefulOpClusteredDistribution and apply to stream-stream join #35419

HeartSaVioR · 2022-02-07T08:28:55Z

What changes were proposed in this pull request?

This PR revives HashClusteredDistribution and renames to StatefulOpClusteredDistribution so that the rationalization of the distribution is clear from the name. Renaming is safe because this class no longer needs to be general one - in SPARK-35703 we moved out the usages of HashClusteredDistribution to ClusteredDistribution; stateful operators are exceptions.

Only HashPartitioning with same expressions and number of partitions can satisfy StatefulOpClusteredDistribution. That said, we cannot modify HashPartitioning unless we clone HashPartitioning and assign the clone to StatefulOpClusteredDistribution.

This PR documents the expectation of stateful operator on partitioning in the classdoc of StatefulOpClusteredDistribution.

This PR also changes stream-stream join to use StatefulOpClusteredDistribution instead of ClusteredDistribution. This effectively reverts a part of SPARK-35703 which hasn't been shipped to any releases. This PR doesn't deal with other stateful operators since it has been long standing issue (probably Spark 2.2.0+) and we need a plan for dealing with existing state.

Why are the changes needed?

Spark does not guarantee stable physical partitioning for stateful operators across query lifetime, and due to the relaxed distribution requirement it is hard to expect what would be the current physical partitioning of the state.
(We expect hash partitioning with grouping keys, but ClusteredDistribution does not "guarantee" the partitioning. It is much more relaxed.)

This PR will enforce the physical partitioning of stream-stream join operators to be hash partition with grouping keys, which is our general expectation of state store partitioning.

Does this PR introduce any user-facing change?

No, since SPARK-35703 hasn't been shipped to any release yet.

How was this patch tested?

Existing tests.

…apply to all stateful operators

HeartSaVioR · 2022-02-07T09:12:47Z

TODO:
I'll fix the build and make sure tests are passing.
I'll also try to see whether I can add tests for stateful operators to see whether the physical partitioning is consistent with several pre-partitioning.

HeartSaVioR · 2022-02-07T09:21:32Z

Implementation-wise, the PR is ready to review.
cc. @cloud-fan @sunchao @viirya @HyukjinKwon @c21 who are listed as reviewers of #32875.
cc. @tdas @zsxwing @brkyvz @jose-torres @xuanyuanking as reviewers on streaming area.

sunchao

Thanks @HeartSaVioR for the PR!

Sorry for my late response on the original PR: I was working on a PR but haven't figured out a way to test it.

sunchao · 2022-02-07T17:10:41Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

+ * Since this distribution relies on [[HashPartitioning]] on the physical partitioning of the
+ * stateful operator, only [[HashPartitioning]] can satisfy this distribution.
+ */
+case class StatefulOpClusteredDistribution(


I like the new name :) thanks for making it more specific.

Do we also need to update HashShuffleSpec so that two HashPartitionings can be compatible with each other when checking against StatefulOpClusteredDistributions? this is the previous behavior where Spark would avoid shuffle if both sides of the streaming join are co-partitioned.

Do we also need to update HashShuffleSpec so that two HashPartitionings can be compatible with each other when checking against StatefulOpClusteredDistributions? this is the previous behavior where Spark would avoid shuffle if both sides of the streaming join are co-partitioned.

Each input must follow the required distribution provided from stateful operator to respect the requirement of state partitioning. State partitioning is the first class, so even both sides of the streaming join are co-partitioned, Spark must perform shuffle if they don't match with state partitioning. (If that was the previous behavior, we broke something at some time point.)

Actually, I think this PR will skip shuffle if both sides of a streaming join are co-partitioned. In EnsureRequirements, we currently mainly do two things:

check if output partitioning can satisfy the required distribution

if there are two children, check if they are compatible with each other, and insert shuffle if not.

In the step 2) we'd only consider ClusteredDistribution at the moment, so in case of StatefulOpClusteredDistributions this step is simply skipped. Consequently, Spark will skip shuffle even if only step 1) is successful.

State partitioning is the first class, so even both sides of the streaming join are co-partitioned, Spark must perform shuffle if they don't match with state partitioning.

I'm not quite sure about this. Shouldn't we retain the behavior before #32875? Quoting the comment from @cloud-fan:

I think this is kind of a potential bug. Let's say that we have 2 tables that can report hash partitioning optionally (e.g. controlled by a flag). Assume a streaming query is first run with the flag off, which means the tables do not report hash partitioning, then Spark will add shuffles before the stream-stream join, and the join state (steaming checkpoint) is partitioned by Spark's murmur3 hash function. Then we restart the streaming query with the flag on, and the 2 tables report hash partitioning (not the same as Spark's murmur3). Spark will not add shuffles before stream-stream join this time, and leads to wrong result, because the left/right join child is not co-partitioned with the join state in the previous run.

If we respect co-partitioning and avoid shuffle before #32875 but start shuffle after this PR, I think similar issue like described in the comment can happen?

check if output partitioning can satisfy the required distribution

For stream-stream join, once each input satisfy the required "hash" distribution of each, they will be co-partitioned. stream-stream join must guarantee this.

The problem brought up because ClusteredDistribution has much more relaxed requirement; what we really need to require for "any" stateful operator including stream-stream join is that for all children a specific tuple having specific grouping key must be bound to the deterministic partition "ID", which only HashClusteredDistribution could guarantee.

I think one behavior difference between this PR and the state before #32875 is that, previously, we'd also check spark.sql.shuffle.partitions and insert shuffle if there's not enough parallelism from the input. However, this PR doesn't do that since it skips the step 2) above.

HashClusteredDistribution also has a requirement of the number of partitions, so step 1) should fulfill it.

Yes, StatefulOpClusteredDistributions is very strict and requires numPartitions as well. I don't think we need extra co-partition check for it.

I see, it should be good then!

c21 · 2022-02-08T00:07:20Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala

@@ -337,7 +337,7 @@ case class StateStoreRestoreExec(
    if (keyExpressions.isEmpty) {
      AllTuples :: Nil
    } else {
-      ClusteredDistribution(keyExpressions, stateInfo.map(_.numPartitions)) :: Nil
+      StatefulOpClusteredDistribution(keyExpressions, stateInfo.map(_.numPartitions)) :: Nil


I am wondering if this change could introduce extra shuffle for streaming aggregate. Previously the operator requires ClusteredDistribution, and right now it requires StatefulOpClusteredDistribution/HashClusteredDistribution.

ClusteredDistribution is more relaxed than HashClusteredDistribution in the sense that a HashPartitioning(c1) can satisfy ClusteredDistribution(c1, c2), but cannot satisfy HashClusteredDistribution(c1, c2). In short, ClusteredDistribution allows child to be hash-partitioned on subset of required keys. So for aggregate, if the plan is already shuffled on subset of group-by columns, Spark will not add a shuffle again before group-by.

For example:

MemoryStream[(Int, Int)].toDF() .repartition($"_1") .groupBy($"_1", $"_2") .agg(count("*")) .as[(Int, Int, Long)]

and the query plan:

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5940f7c2, org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy$$Lambda$1940/1200613952@4861dac3 +- *(4) HashAggregate(keys=[_1#588, _2#589], functions=[count(1)], output=[_1#588, _2#589, count(1)#596L]) +- StateStoreSave [_1#588, _2#589], state info [ checkpoint = file:/private/var/folders/y5/hnsw8mz93vs57ngcd30y6y9c0000gn/T/streaming.metadata-0d7cb004-92dd-4b0d-9d90-5a65c0d2934c/state, runId = 68598bd1-cf35-4bf7-a167-5f73dc9f4d84, opId = 0, ver = 0, numPartitions = 5], Complete, 0, 1 +- *(3) HashAggregate(keys=[_1#588, _2#589], functions=[merge_count(1)], output=[_1#588, _2#589, count#663L]) +- StateStoreRestore [_1#588, _2#589], state info [ checkpoint = file:/private/var/folders/y5/hnsw8mz93vs57ngcd30y6y9c0000gn/T/streaming.metadata-0d7cb004-92dd-4b0d-9d90-5a65c0d2934c/state, runId = 68598bd1-cf35-4bf7-a167-5f73dc9f4d84, opId = 0, ver = 0, numPartitions = 5], 1 +- *(2) HashAggregate(keys=[_1#588, _2#589], functions=[merge_count(1)], output=[_1#588, _2#589, count#663L]) +- *(2) HashAggregate(keys=[_1#588, _2#589], functions=[partial_count(1)], output=[_1#588, _2#589, count#663L]) +- Exchange hashpartitioning(_1#588, 5), REPARTITION_BY_COL, [id=#2008] +- *(1) Project [_1#588, _2#589] +- MicroBatchScan[_1#588, _2#589] MemoryStreamDataSource

One can argue the previous behavior for streaming aggregate is not wrong. As long as all rows for same keys are colocated in same partition, StateStoreRestore/Store should output correct answer for streaming aggregate. If we make the change here, I assume one extra shuffle on ($"_1", $"_2") would be introduced, and it might yield incorrect result when running the new query plan against the existing state store?

But we don't define "repartition just before stateful operator" as "unsupported operation across query lifetime", no?

The thing is that once the query is run, the partitioning of stateful operator must not be changed during lifetime. Since we don't store the information of partitioning against stateful operator in the checkpoint, we have no way around other than enforcing the partitioning of stateful operator as the "one" what we basically expect.

As I said in #32875, there is a room for improvement, but the effort on improvement must be performed after we fix this issue.

But we don't define "repartition just before stateful operator" as "unsupported operation across query lifetime", no?

I ran the query in StreamingAggregationSuite.scala and it seems fine. I briefly checked UnsupportedOperationChecker.scala and didn't find we disallow "repartition just before stateful operator".

The thing is that once the query is run, the partitioning of stateful operator must not be changed during lifetime. Since we don't store the information of partitioning against stateful operator in the checkpoint, we have no way around other than enforcing the partitioning of stateful operator as the "one" what we basically expect.

I agree with you @HeartSaVioR. I want to raise a concern here it might change the query plan for certain streaming aggregate query (as above synthetic query), and it could break existing state store when running with next Spark 3.3 code, based on my limited understanding.

If we make the change here, I assume one extra shuffle on ($"_1", $"_2") would be introduced, and it might yield incorrect result when running the new query plan against the existing state store?

Unfortunately yes. We may need to craft some tools to analyze the state and repartition if the partitioning is already messed up. But leaving this as it is would bring more chances to let users' state be indeterministic.

But we don't define "repartition just before stateful operator" as "unsupported operation across query lifetime", no?

I ran the query in StreamingAggregationSuite.scala and it seems fine. I briefly checked UnsupportedOperationChecker.scala and didn't find we disallow "repartition just before stateful operator".

That is the problem we have. We didn't disallow the case where it brings silent correctness issue.

That is the problem we have. We didn't disallow the case where it brings silent correctness issue.

This is just a synthetic example I composed to verify my theory. But I think it might break for more cases, such as GROUP BY c1, c2 after JOIN ON c1. I am trying to say it would break for the queries which are partitioned on subset of group-by keys.

Even nowadays multiple stateful operators don't work properly due to the global watermark, so we don't need to worry about the partitioning between stateful operators. We just need to worry about the partitioning between upstream (in most cases non stateful) and the stateful operator.

HeartSaVioR · 2022-02-08T05:23:05Z

I see the concern of fixing stateful operators due to the state on existing queries. stream-stream join has been using HashClusteredDistribution and it shouldn't suffer this long standing problem.

So if we want to split down the problem with stream-stream join and others, I'll move out the changes on other stateful operators, and in there we have to also come up with the plan how to deal with state on existing queries.

Would it work for all?

HeartSaVioR · 2022-02-08T06:42:03Z

(I'm reading through #19080, and just realized there has been NO such strict requirement before we introduced HashClusteredDistribution in #19080 if I understand correctly... If then, it doesn't even start from 2.3+. It starts from 2.2+ where we introduced stateful operators in SS.)

c21 · 2022-02-08T07:27:21Z

So if we want to split down the problem with stream-stream join and others, I'll move out the changes on other stateful operators, and in there we have to also come up with the plan how to deal with state on existing queries.

+1, SGTM.

HeartSaVioR · 2022-02-08T07:29:59Z

Just updated JIRA ticket, PR description, PR code diff to only contain the change of stream-stream join.

Now this PR is effectively a partial revert of SPARK-35703, which hasn't been released. That said, this PR doesn't bring any breaking change.

I'll file another JIRA ticket for dealing with other stateful operators.

c21 · 2022-02-08T07:44:49Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

+ * (including restart), the result of evaluation on `partitionIdExpression` must be unchanged
+ * across Spark versions. Violation of this requirement may bring silent correctness issue.


shall we enforce this assumption in unit test as well? e.g. in StreamingJoinSuite. It's great to highlight in comment here, but people always forget and the unit test will fail loudly when we introduce any invalid change.

We have a test for verifying this, although it is not exhaustive.

spark/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala

Lines 574 to 605 in 2e703ae

test("streaming join should require HashClusteredDistribution from children") {

val input1 = MemoryStream[Int]

val input2 = MemoryStream[Int]

val df1 = input1.toDF.select('value as 'a, 'value * 2 as 'b)

val df2 = input2.toDF.select('value as 'a, 'value * 2 as 'b).repartition('b)

val joined = df1.join(df2, Seq("a", "b")).select('a)

testStream(joined)(

AddData(input1, 1.to(1000): _*),

AddData(input2, 1.to(1000): _*),

CheckAnswer(1.to(1000): _*),

Execute { query =>

// Verify the query plan

def partitionExpressionsColumns(expressions: Seq[Expression]): Seq[String] = {

expressions.flatMap {

case ref: AttributeReference => Some(ref.name)

}

}

val numPartitions = spark.sqlContext.conf.getConf(SQLConf.SHUFFLE_PARTITIONS)

assert(query.lastExecution.executedPlan.collect {

case j @ StreamingSymmetricHashJoinExec(_, _, _, _, _, _, _, _,

ShuffleExchangeExec(opA: HashPartitioning, _, _),

ShuffleExchangeExec(opB: HashPartitioning, _, _))

if partitionExpressionsColumns(opA.expressions) === Seq("a", "b")

&& partitionExpressionsColumns(opB.expressions) === Seq("a", "b")

&& opA.numPartitions == numPartitions && opB.numPartitions == numPartitions => j

}.size == 1)

})

}

If we want to be exhaustive, I can make a combination of repartitions which could have not triggered shuffle with hash partitioning against joining keys if stream-stream join uses ClusteredDistribution. It may not be exhaustive for future-proof indeed.

Instead, if we are pretty sure StateOpClusteredDistribution would work as expected, we can simply check the required child distribution of the physical plan of stream-stream join, and additionally check the output partitioning of each child to be HashPartitioning with joining keys (this effectively verifies StateOpClusteredDistribution indeed).

oh actually I was referring to the assumption:

HashPartitioning.partitionIdExpression has to be exactly Pmod(new Murmur3Hash(expressions), Literal(numPartitions)).

It would be just to add some logic to check opA/opB.partitionIdExpression for the opA/opB at Line 598/599. I can also do it later if it's not clear to you.

We checked HashPartitioning and partitionExpression here - the remaining is partitionIdExpression, which is the implementation of HashPartitioning.

That said, it would be nice if we have a separate test against HashPartitioning if we don't have one. Could you please check and craft one if we don't have it?

Sure I can add one later this week.

c21 · 2022-02-08T07:47:03Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

+ * Spark versions. Violation of this requirement may bring silent correctness issue.
+ *
+ * Since this distribution relies on [[HashPartitioning]] on the physical partitioning of the
+ * stateful operator, only [[HashPartitioning]] can satisfy this distribution.


do we want to also explain briefly this only applies to StreamingSymmetricHashJoinExec now, and the challenge to apply it to other stateful operators? Maybe we can also file a JIRA for other stateful operators and leave a TODO here.

Where to leave a comment is the issue. It is unlikely that we often look at StatefulOpClusteredDistribution - probably giving more chance to get attention if we put a comment on every stateful operators wherever using ClusteredDistribution. Totally redundant, but gives a sign of warning whenever they try to change it.

Yeah either works for me. The comment is also non-blocking for this PR, as this is an improvement for documentation.

c21 · 2022-02-08T07:58:01Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

+ * Spark versions. Violation of this requirement may bring silent correctness issue.
+ *
+ * Since this distribution relies on [[HashPartitioning]] on the physical partitioning of the
+ * stateful operator, only [[HashPartitioning]] can satisfy this distribution.


only [[HashPartitioning]] can satisfy this distribution. -> only [[HashPartitioning]], and [[PartitioningCollection]] of [[HashPartitioning]] can satisfy this distribution. ?

c21

LGTM, thanks @HeartSaVioR for the fix!

HeartSaVioR · 2022-02-08T08:35:11Z

In the meanwhile I'll think of how to deal with other stateful operators with existing state. One rough idea is having the functionality of validation on state partitioning against child output partitioning, although it is not "future-proof" validation.

sunchao

LGTM, thanks @HeartSaVioR !

viirya · 2022-02-08T17:42:22Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

+      "a single partition.")
+
+  override def createPartitioning(numPartitions: Int): Partitioning = {
+    assert(requiredNumPartitions.isEmpty || requiredNumPartitions.get == numPartitions,


Hmm, is there any chance we specify empty requiredNumPartitions?

In sense of "defensive programming", it shouldn't. I just didn't change the implementation of HashClusteredPartition, but now I think it worths to do.

viirya

Looks good. Only a few question.

viirya · 2022-02-08T17:44:41Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

@@ -90,6 +90,35 @@ case class ClusteredDistribution(
  }
 }

+/**
+ * Represents the requirement of distribution on the stateful operator.


nit: do we need to put "structured streaming" before "stateful operator"?

I think "stateful" is already representing the streaming context, but no big deal if we repeat here.

viirya · 2022-02-08T17:48:15Z

...src/main/scala/org/apache/spark/sql/execution/streaming/StreamingSymmetricHashJoinExec.scala

+    StatefulOpClusteredDistribution(leftKeys, stateInfo.map(_.numPartitions)) ::
+      StatefulOpClusteredDistribution(rightKeys, stateInfo.map(_.numPartitions)) :: Nil


There is other ClusteredDistribution usages in statefulOperators, e.g. ClusteredDistribution, do we need to update them too? As they are also stateful operators, they also need strict partition requirement?

Please refer the long comment thread #35419 (comment)

We have to fix them, but we should have a plan to avoid introducing "silently broken" on existing queries. We need more time to think through how to address the "already broken" thing. They seem to be broken from their introduction (Spark 2.2+), so it could be possible someone is even leveraging the relaxed requirement as a "feature", despite it would be very risky if they tried to adjust partitioning by theirselves. Even for this case we can't simply break their query.

I'll create a new JIRA ticket, and/or initiate discussion thread in dev@ regarding this. I need some time to build a plan (with options) to address this safely.

it could be possible someone is even leveraging the relaxed requirement as a "feature"

Suppose they have a stream of event logs having (userId, time, blabla), and do time-window aggregation like this:

df .withWatermark("time", "1 hour") .groupBy("userId", window("time", "10 minutes")) .agg(count("*"))

groupBy won't trigger shuffle for various output partitionings of df, since streaming aggregation requires ClusteredDistribution. The thing is, it could be from the intention to 1) reduce shuffle in any way, or 2) try to control the partitioning to deal with skew. (I can't easily think of skew from applying hash function against "grouping keys + time window", but once they see it, they will try to fix it. ...Technically saying, they must not try to fix it as state partitioning will be no longer the same with operator's partitioning...)

Both are very risky (as of now, changing the partitioning during query lifetime would lead to correctness issue), but it's still from users' intention and they already did it anyway so we can't simply enforce the partitioning and silently break this again.

Furthermore, we seem to allow data source to produce output partitioning by itself, which can satisfy ClusteredDistribution. This is still very risky for stateful operator's perspective, but once the output partitioning is guaranteed to be not changed, it's still a great change to reduce (unnecessary) shuffle.
(Just saying hypothetically; stateful operator has to require specific output partitioning once the state is built, so it's unlikely that we can leverage the partitioning of data source. We may find a way later but not now.)

a few questions need to address

HeartSaVioR · 2022-02-08T21:15:29Z

@viirya Could you please take a look again? I've reflected your review comments via 753a5b6

viirya · 2022-02-08T21:48:15Z

Thanks @HeartSaVioR . LGTM

xuanyuanking

LGTM

HeartSaVioR · 2022-02-08T23:31:16Z

Thanks! Merging to master.

…distribution of stateful operator ### What changes were proposed in this pull request? This PR proposes to add the context of current challenge on fixing distribution of stateful operator, even the distribution is a sort of "broken" now. This PR addresses the review comment #35419 (comment) ### Why are the changes needed? In SPARK-38124 we figured out the existing long-standing problem in stateful operator, but it is not easy to fix since the fix may break the existing query if the fix is not carefully designed. Anyone should also be pretty much careful when touching the required distribution. We want to document this explicitly to help others to be careful whenever someone is around the codebase. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Code comment only changes. Closes #35512 from HeartSaVioR/SPARK-38124-followup. Authored-by: Jungtaek Lim <[email protected]> Signed-off-by: Jungtaek Lim <[email protected]>

…tioning requirement ### What changes were proposed in this pull request? This is a followup of #35419 (comment), to add unit test to harden the assumption of SS partitioning and distribution requirement: * Check the `HashPartitioning.partitionIdExpression` to be exactly expected format * Check all different kinds of `Partitioning` against `StatefulOpClusteredDistribution`. Also add a minor comment for `StatefulOpClusteredDistribution`, as `SinglePartition` can also satisfy the distribution. ### Why are the changes needed? Document our assumption of SS in code as unit test. So next time when we introduce intrusive code change, the unit test can save us by failing loudly. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? The added unit test itself. Closes #35529 from c21/partition-test. Authored-by: Cheng Su <[email protected]> Signed-off-by: Jungtaek Lim <[email protected]>

HeartSaVioR added 2 commits February 7, 2022 16:05

[SPARK-38124][SQL][SS] Introduce StatefulOpClusteredDistribution and …

90595dd

…apply to all stateful operators

Remove StatefulOpPartitioning and merge to HashPartitioning

4639b7e

github-actions bot added SQL STRUCTURED STREAMING labels Feb 7, 2022

HeartSaVioR marked this pull request as draft February 7, 2022 09:10

HeartSaVioR mentioned this pull request Feb 7, 2022

[SPARK-35703][SQL] Relax constraint for bucket join and remove HashClusteredDistribution #32875

Closed

fix compilation

adfe796

sunchao reviewed Feb 7, 2022

View reviewed changes

c21 reviewed Feb 8, 2022

View reviewed changes

HeartSaVioR changed the title ~~[SPARK-38124][SQL][SS] Introduce StatefulOpClusteredDistribution and apply to all stateful operators~~ [SPARK-38124][SQL][SS] Introduce StatefulOpClusteredDistribution and apply to stream-stream join Feb 8, 2022

HeartSaVioR marked this pull request as ready for review February 8, 2022 07:26

HeartSaVioR force-pushed the SPARK-38124 branch from 11eabf0 to d0aa192 Compare February 8, 2022 07:27

c21 reviewed Feb 8, 2022

View reviewed changes

Revert the changes in non stream-stream join operators

d722b2c

c21 reviewed Feb 8, 2022

View reviewed changes

HeartSaVioR force-pushed the SPARK-38124 branch from d0aa192 to d722b2c Compare February 8, 2022 07:47

cloud-fan approved these changes Feb 8, 2022

View reviewed changes

c21 reviewed Feb 8, 2022

View reviewed changes

c21 approved these changes Feb 8, 2022

View reviewed changes

update comment

4b68d28

HeartSaVioR requested a review from sunchao February 8, 2022 12:16

sunchao approved these changes Feb 8, 2022

View reviewed changes

viirya reviewed Feb 8, 2022

View reviewed changes

viirya previously approved these changes Feb 8, 2022

View reviewed changes

viirya reviewed Feb 8, 2022

View reviewed changes

reflect review comments

753a5b6

HeartSaVioR requested a review from viirya February 8, 2022 21:15

viirya approved these changes Feb 8, 2022

View reviewed changes

xuanyuanking approved these changes Feb 8, 2022

View reviewed changes

HeartSaVioR closed this in 43cce92 Feb 8, 2022

HeartSaVioR mentioned this pull request Feb 14, 2022

[SPARK-38124][SS][FOLLOWUP] Document the current challenge on fixing distribution of stateful operator #35512

Closed

c21 mentioned this pull request Feb 15, 2022

[SPARK-38124][SS][FOLLOWUP] Add test to harden assumption of SS partitioning requirement #35529

Closed

HeartSaVioR mentioned this pull request Feb 17, 2022

[SPARK-38237][SQL][SS] Introduce a new config to require all cluster keys on Aggregate #35552

Closed

HeartSaVioR mentioned this pull request Feb 28, 2022

[SPARK-38204][SS] Use StatefulOpClusteredDistribution for stateful operators with respecting backward compatibility #35673

Closed

		* (including restart), the result of evaluation on `partitionIdExpression` must be unchanged
		* across Spark versions. Violation of this requirement may bring silent correctness issue.

	test("streaming join should require HashClusteredDistribution from children") {
	val input1 = MemoryStream[Int]
	val input2 = MemoryStream[Int]

	val df1 = input1.toDF.select('value as 'a, 'value * 2 as 'b)
	val df2 = input2.toDF.select('value as 'a, 'value * 2 as 'b).repartition('b)
	val joined = df1.join(df2, Seq("a", "b")).select('a)

	testStream(joined)(
	AddData(input1, 1.to(1000): _*),
	AddData(input2, 1.to(1000): _*),
	CheckAnswer(1.to(1000): _*),
	Execute { query =>
	// Verify the query plan
	def partitionExpressionsColumns(expressions: Seq[Expression]): Seq[String] = {
	expressions.flatMap {
	case ref: AttributeReference => Some(ref.name)
	}
	}

	val numPartitions = spark.sqlContext.conf.getConf(SQLConf.SHUFFLE_PARTITIONS)

	assert(query.lastExecution.executedPlan.collect {
	case j @ StreamingSymmetricHashJoinExec(_, _, _, _, _, _, _, _,
	ShuffleExchangeExec(opA: HashPartitioning, _, _),
	ShuffleExchangeExec(opB: HashPartitioning, _, _))
	if partitionExpressionsColumns(opA.expressions) === Seq("a", "b")
	&& partitionExpressionsColumns(opB.expressions) === Seq("a", "b")
	&& opA.numPartitions == numPartitions && opB.numPartitions == numPartitions => j
	}.size == 1)
	})
	}

		StatefulOpClusteredDistribution(leftKeys, stateInfo.map(_.numPartitions)) ::
		StatefulOpClusteredDistribution(rightKeys, stateInfo.map(_.numPartitions)) :: Nil

[SPARK-38124][SQL][SS] Introduce StatefulOpClusteredDistribution and apply to stream-stream join #35419

[SPARK-38124][SQL][SS] Introduce StatefulOpClusteredDistribution and apply to stream-stream join #35419

Conversation

HeartSaVioR commented Feb 7, 2022 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

HeartSaVioR commented Feb 7, 2022

HeartSaVioR commented Feb 7, 2022

sunchao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HeartSaVioR Feb 7, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HeartSaVioR Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HeartSaVioR Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HeartSaVioR Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HeartSaVioR Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

HeartSaVioR commented Feb 8, 2022

HeartSaVioR commented Feb 8, 2022 • edited Loading

c21 commented Feb 8, 2022

HeartSaVioR commented Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

HeartSaVioR Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

c21 left a comment

Choose a reason for hiding this comment

HeartSaVioR commented Feb 8, 2022

sunchao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

HeartSaVioR Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

HeartSaVioR Feb 8, 2022 • edited Loading

Choose a reason for hiding this comment

HeartSaVioR commented Feb 8, 2022

viirya commented Feb 8, 2022

xuanyuanking left a comment

Choose a reason for hiding this comment

HeartSaVioR commented Feb 8, 2022

HeartSaVioR commented Feb 7, 2022 •

edited

Loading

HeartSaVioR Feb 7, 2022 •

edited

Loading

HeartSaVioR Feb 8, 2022 •

edited

Loading

HeartSaVioR Feb 8, 2022 •

edited

Loading

HeartSaVioR Feb 8, 2022 •

edited

Loading

HeartSaVioR Feb 8, 2022 •

edited

Loading

HeartSaVioR commented Feb 8, 2022 •

edited

Loading

HeartSaVioR commented Feb 8, 2022 •

edited

Loading

HeartSaVioR Feb 8, 2022 •

edited

Loading

viirya left a comment •

edited

Loading

viirya Feb 8, 2022 •

edited

Loading

HeartSaVioR Feb 8, 2022 •

edited

Loading

HeartSaVioR Feb 8, 2022 •

edited

Loading