You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "cloud-fan (via GitHub)" <gi...@apache.org> on 2026/03/25 14:39:47 UTC
[PR] [SPARK-XXX][SS] Refactor streaming join tests: split Base/Suite hierarchy and simplify mode dispatch [spark]
cloud-fan opened a new pull request, #55005:
URL: https://github.com/apache/spark/pull/55005
### What changes were proposed in this pull request?
Refactor streaming join test suites to separate test mode dispatch from the test hierarchy.
**Before:** `AlsoTestWithVirtualColumnFamilyJoins` runs every test in both VCF and non-VCF
modes within a single suite class. V4 suites extend these classes and use a string-matching
skip list to exclude incompatible tests.
**After:** Each join type has a two-layer hierarchy:
```
StreamingInnerJoinBase (universal tests — all state format versions)
├── StreamingInnerJoinSuite (adds V3-specific tests: window joins, metrics, schema)
│ ├── StreamingInnerJoinWithVCFSuite (useVirtualColumnFamilies = true)
│ └── StreamingInnerJoinWithoutVCFSuite (useVirtualColumnFamilies = false)
└── StreamingInnerJoinV4Suite (useVirtualColumnFamilies = true, stateFormatVersion = 4)
```
Mode dispatch uses a simple abstract boolean `useVirtualColumnFamilies` in `StreamingJoinSuite`,
replacing the `AlsoTestWithVirtualColumnFamilyJoins` trait. V4 suites extend `*Base` directly —
no skip list needed.
### Why are the changes needed?
1. **Failure isolation**: Suite name identifies the exact mode — no long disambiguating suffixes.
2. **Selective retrigger**: Rerun only the failing mode's suite without rerunning the passing mode.
3. **No skip lists**: V4 suites get only universal tests by inheriting from `*Base`, not `*Suite`. Adding V3-specific tests in the future won't accidentally break V4.
4. **Simpler dispatch**: A boolean replaces a trait with hard-coded dual-mode dispatch.
### Does this PR introduce _any_ user-facing change?
No. Test infrastructure only. Same tests, same coverage.
### How was this patch tested?
Existing tests — the `*WithVCFSuite` and `*WithoutVCFSuite` classes run the same tests as before, split by mode.
### Was this patch authored or co-authored using generative AI tooling?
Yes
--
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
Re: [PR] [SPARK-56202][SS] Refactor streaming join tests: split Base/Suite hierarchy and simplify mode dispatch [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #55005: [SPARK-56202][SS] Refactor streaming join tests: split Base/Suite hierarchy and simplify mode dispatch
URL: https://github.com/apache/spark/pull/55005
--
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
Re: [PR] [SPARK-56202][SS] Refactor streaming join tests: split Base/Suite hierarchy and simplify mode dispatch [spark]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #55005:
URL: https://github.com/apache/spark/pull/55005#discussion_r2988770301
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala:
##########
@@ -42,78 +42,78 @@ import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.execution.streaming.state._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.test.SQLTestUtils
import org.apache.spark.sql.types.{BooleanType, IntegerType, LongType, StructType}
import org.apache.spark.tags.SlowSQLTest
import org.apache.spark.util.Utils
-trait AlsoTestWithVirtualColumnFamilyJoins extends SQLTestUtils {
- /** Tests both with and without join ops using virtual column families */
+abstract class StreamingJoinSuite
+ extends StreamTest with StateStoreMetricsTest with BeforeAndAfter {
+
+ /** Whether this suite runs tests with virtual column families (VCF). */
+ protected def useVirtualColumnFamilies: Boolean
+
+ /** Dispatches each test to VCF or non-VCF mode based on [[useVirtualColumnFamilies]]. */
override protected def test(testName: String, testTags: Tag*)(testBody: => Any)(
implicit pos: Position): Unit = {
- // Test with virtual column family joins with changelog checkpointing enabled and disabled
- // Since virtual column family joins require RocksDB, we only test with RocksDB here.
- Seq("false", "true").foreach { enabled =>
- testWithVirtualColumnFamilyJoins(
- testName + s" with (with changelog checkpointing = $enabled)", testTags: _*) {
- withSQLConf(
- "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" -> enabled
- ) {
- testBody
+ if (useVirtualColumnFamilies) {
+ // Test with VCF with changelog checkpointing enabled and disabled.
+ // Since VCF requires RocksDB, we only test with RocksDB here.
+ Seq("false", "true").foreach { enabled =>
+ testWithVirtualColumnFamilyJoins(
+ testName + s" (with changelog checkpointing = $enabled)", testTags: _*) {
+ withSQLConf(
+ "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" -> enabled
+ ) {
+ testBody
+ }
+ }
+ }
+ } else {
+ // Test with both RocksDB and HDFS state store providers without VCF.
+ val providers = Seq(
+ classOf[RocksDBStateStoreProvider].getName,
+ classOf[HDFSBackedStateStoreProvider].getName
+ )
+ providers.foreach { provider =>
+ testWithoutVirtualColumnFamilyJoins(testName + s" (with $provider)", testTags: _*) {
+ withSQLConf(
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key -> provider
+ ) {
+ testBody
+ }
}
}
}
+ }
- // Test with both RocksDB and HDFS state store providers without virtual column family joins
- val providers = Seq(
- classOf[RocksDBStateStoreProvider].getName,
- classOf[HDFSBackedStateStoreProvider].getName
- )
-
- providers.foreach { provider =>
- testWithoutVirtualColumnFamilyJoins(testName + s" (with $provider)", testTags: _*) {
+ def testWithVirtualColumnFamilyJoins(testName: String, testTags: Tag*)(
+ testBody: => Any): Unit = {
+ if (useVirtualColumnFamilies) {
+ super.test(testName + " (with virtual column family joins)", testTags: _*) {
+ super.beforeEach()
withSQLConf(
- SQLConf.STATE_STORE_PROVIDER_CLASS.key -> provider
+ SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key -> "3",
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key -> classOf[RocksDBStateStoreProvider].getName
) {
testBody
}
+ super.afterEach()
}
}
}
- def testWithVirtualColumnFamilyJoins(testName: String, testTags: Tag*)(testBody: => Any): Unit = {
- super.test(testName + " (with virtual column family joins)", testTags: _*) {
- // in case tests have any code that needs to execute before every test
- super.beforeEach()
- // We must be using RocksDB for virtual column family joins
- withSQLConf(
- SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key -> "3",
- SQLConf.STATE_STORE_PROVIDER_CLASS.key -> classOf[RocksDBStateStoreProvider].getName
- ) {
- testBody
- }
- // in case tests have any code that needs to execute after every test
- super.afterEach()
- }
- }
-
def testWithoutVirtualColumnFamilyJoins(testName: String, testTags: Tag*)(
testBody: => Any): Unit = {
- super.test(testName + " (without virtual column family joins)", testTags: _*) {
- // in case tests have any code that needs to execute before every test
- super.beforeEach()
- withSQLConf(SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key -> "2") {
- testBody
+ if (!useVirtualColumnFamilies) {
+ super.test(testName + " (without virtual column family joins)", testTags: _*) {
+ super.beforeEach()
+ withSQLConf(SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key -> "2") {
+ testBody
+ }
Review Comment:
Same as above: `super.afterEach()` is not guaranteed to run when `testBody` fails. Use `try`/`finally` around the test body to ensure teardown always executes.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingJoinSuite.scala:
##########
@@ -42,78 +42,78 @@ import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.execution.streaming.state._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.test.SQLTestUtils
import org.apache.spark.sql.types.{BooleanType, IntegerType, LongType, StructType}
import org.apache.spark.tags.SlowSQLTest
import org.apache.spark.util.Utils
-trait AlsoTestWithVirtualColumnFamilyJoins extends SQLTestUtils {
- /** Tests both with and without join ops using virtual column families */
+abstract class StreamingJoinSuite
+ extends StreamTest with StateStoreMetricsTest with BeforeAndAfter {
+
+ /** Whether this suite runs tests with virtual column families (VCF). */
+ protected def useVirtualColumnFamilies: Boolean
+
+ /** Dispatches each test to VCF or non-VCF mode based on [[useVirtualColumnFamilies]]. */
override protected def test(testName: String, testTags: Tag*)(testBody: => Any)(
implicit pos: Position): Unit = {
- // Test with virtual column family joins with changelog checkpointing enabled and disabled
- // Since virtual column family joins require RocksDB, we only test with RocksDB here.
- Seq("false", "true").foreach { enabled =>
- testWithVirtualColumnFamilyJoins(
- testName + s" with (with changelog checkpointing = $enabled)", testTags: _*) {
- withSQLConf(
- "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" -> enabled
- ) {
- testBody
+ if (useVirtualColumnFamilies) {
+ // Test with VCF with changelog checkpointing enabled and disabled.
+ // Since VCF requires RocksDB, we only test with RocksDB here.
+ Seq("false", "true").foreach { enabled =>
+ testWithVirtualColumnFamilyJoins(
+ testName + s" (with changelog checkpointing = $enabled)", testTags: _*) {
+ withSQLConf(
+ "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled" -> enabled
+ ) {
+ testBody
+ }
+ }
+ }
+ } else {
+ // Test with both RocksDB and HDFS state store providers without VCF.
+ val providers = Seq(
+ classOf[RocksDBStateStoreProvider].getName,
+ classOf[HDFSBackedStateStoreProvider].getName
+ )
+ providers.foreach { provider =>
+ testWithoutVirtualColumnFamilyJoins(testName + s" (with $provider)", testTags: _*) {
+ withSQLConf(
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key -> provider
+ ) {
+ testBody
+ }
}
}
}
+ }
- // Test with both RocksDB and HDFS state store providers without virtual column family joins
- val providers = Seq(
- classOf[RocksDBStateStoreProvider].getName,
- classOf[HDFSBackedStateStoreProvider].getName
- )
-
- providers.foreach { provider =>
- testWithoutVirtualColumnFamilyJoins(testName + s" (with $provider)", testTags: _*) {
+ def testWithVirtualColumnFamilyJoins(testName: String, testTags: Tag*)(
+ testBody: => Any): Unit = {
+ if (useVirtualColumnFamilies) {
+ super.test(testName + " (with virtual column family joins)", testTags: _*) {
+ super.beforeEach()
withSQLConf(
- SQLConf.STATE_STORE_PROVIDER_CLASS.key -> provider
+ SQLConf.STREAMING_JOIN_STATE_FORMAT_VERSION.key -> "3",
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key -> classOf[RocksDBStateStoreProvider].getName
Review Comment:
`super.afterEach()` won’t run if `testBody` throws, which can leak per-test resources/state and make subsequent tests flaky. Wrap the body between `beforeEach`/`afterEach` in a `try`/`finally` (or move this setup/teardown into the ScalaTest fixture lifecycle) so teardown is guaranteed on failures.
--
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
Re: [PR] [SPARK-56202][SS] Refactor streaming join tests: split Base/Suite hierarchy and simplify mode dispatch [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #55005:
URL: https://github.com/apache/spark/pull/55005#issuecomment-4150027379
thanks for the review, merging to master!
--
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