Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-38809][SS] Implement option to skip null values in symmetric hash implementation of stream-stream joins #36090

Closed
wants to merge 6 commits into from
Closed

[SPARK-38809][SS] Implement option to skip null values in symmetric hash implementation of stream-stream joins #36090

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3baf99c
[SPARK-38809][SS] Implement option to skip null values in symmetric h…
anishshri-db Apr 6, 2022
066a03b
Change to fix comments in test
anishshri-db Apr 6, 2022
f6371e0
Change to address Jerry's comments
anishshri-db Apr 7, 2022
364790d
Change to address Jungtaek's comments
anishshri-db Apr 7, 2022
0d5ef66
Change to address Jungtaek's comments
anishshri-db Apr 7, 2022
45dee14
Change to add version to conf
anishshri-db Apr 8, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 17 additions & 20 deletions ...a/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManagerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@ import org.apache.spark.sql.catalyst.expressions.codegen.GeneratePredicate
import org.apache.spark.sql.catalyst.plans.logical.EventTimeWatermark
import org.apache.spark.sql.execution.streaming.StatefulOperatorStateInfo
import org.apache.spark.sql.execution.streaming.StreamingSymmetricHashJoinHelper.LeftSide
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.streaming.StreamTest
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
Expand Down Expand Up @@ -293,28 +292,26 @@ class SymmetricHashJoinStateManagerSuite extends StreamTest with BeforeAndAfter
manager.metrics.numKeys
}


def withJoinStateManager(
inputValueAttribs: Seq[Attribute],
joinKeyExprs: Seq[Expression],
stateFormatVersion: Int,
skipNullsForStreamStreamJoins: Boolean = false)
(f: SymmetricHashJoinStateManager => Unit): Unit = {
inputValueAttribs: Seq[Attribute],
joinKeyExprs: Seq[Expression],
stateFormatVersion: Int,
skipNullsForStreamStreamJoins: Boolean = false)
(f: SymmetricHashJoinStateManager => Unit): Unit = {

withTempDir { file =>
val sqlConf = new SQLConf
if (skipNullsForStreamStreamJoins) {
sqlConf.setConf(SQLConf.STATE_STORE_SKIP_NULLS_FOR_STREAM_STREAM_JOINS, true)
}
val storeConf = new StateStoreConf(sqlConf)
val stateInfo = StatefulOperatorStateInfo(file.getAbsolutePath, UUID.randomUUID, 0, 0, 5)
val manager = new SymmetricHashJoinStateManager(
LeftSide, inputValueAttribs, joinKeyExprs, Some(stateInfo), storeConf, new Configuration,
partitionId = 0, stateFormatVersion)
try {
f(manager)
} finally {
manager.abortIfNeeded()
withSQLConf("spark.sql.streaming.stateStore.skipNullsForStreamStreamJoins.enabled" ->
anishshri-db marked this conversation as resolved.
Show resolved Hide resolved
skipNullsForStreamStreamJoins.toString) {
val storeConf = new StateStoreConf(spark.sqlContext.conf)
val stateInfo = StatefulOperatorStateInfo(file.getAbsolutePath, UUID.randomUUID, 0, 0, 5)
val manager = new SymmetricHashJoinStateManager(
LeftSide, inputValueAttribs, joinKeyExprs, Some(stateInfo), storeConf, new Configuration,
partitionId = 0, stateFormatVersion)
try {
f(manager)
} finally {
manager.abortIfNeeded()
}
}
}
StateStore.stop()
Expand Down