Skip to content

Commit

Permalink
[SPARK-50253][SS] Stream-Stream Join should not fetch checkpoint ID i…
Browse files Browse the repository at this point in the history
…f not supported

### What changes were proposed in this pull request?
In stream-stream join, only call getLatestCheckpointInfo() when the format version indicates that checkpoint version is supported.
The other place already have it applied: https://github.com/apache/spark/blob/07301ddb889bdf361499f65e1708b5fdcab7e539/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala#L350-L362

### Why are the changes needed?
These code is not needed and wasteful. Also inside it has assertion that we are not sure whether it is correct when the function is called when the state store is not created for the checkpointID feature supported.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Make sure existing CI passes.

### Was this patch authored or co-authored using generative AI tooling?
No.

Closes apache#48783 from siying/idjoinfix.

Authored-by: Siying Dong <siying.dong@databricks.com>
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
  • Loading branch information
siying authored and HeartSaVioR committed Nov 7, 2024
1 parent 07301dd commit 49b0811
Showing 1 changed file with 9 additions and 5 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -510,11 +510,15 @@ case class StreamingSymmetricHashJoinExec(
val rightSideMetrics = rightSideJoiner.commitStateAndGetMetrics()
val combinedMetrics = StateStoreMetrics.combine(Seq(leftSideMetrics, rightSideMetrics))

val checkpointInfo = SymmetricHashJoinStateManager.mergeStateStoreCheckpointInfo(
JoinStateStoreCkptInfo(
leftSideJoiner.getLatestCheckpointInfo(),
rightSideJoiner.getLatestCheckpointInfo()))
setStateStoreCheckpointInfo(checkpointInfo)
if (StatefulOperatorStateInfo.enableStateStoreCheckpointIds(conf)) {
val checkpointInfo = SymmetricHashJoinStateManager.mergeStateStoreCheckpointInfo(
JoinStateStoreCkptInfo(
leftSideJoiner.getLatestCheckpointInfo(),
rightSideJoiner.getLatestCheckpointInfo()
)
)
setStateStoreCheckpointInfo(checkpointInfo)
}

// Update SQL metrics
numUpdatedStateRows +=
Expand Down

0 comments on commit 49b0811

Please sign in to comment.