[#36841] Spark 4 streaming: ValidatesRunner suite for the Structured Streaming runner

.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 6
+  "modification": 7
 }

runners/spark/4/build.gradle

@@ -51,6 +51,82 @@ tasks.validatesStructuredStreamingRunnerBatch {
   }
 }
 
+tasks.register("validatesStructuredStreamingRunnerStreaming", Test) {
+  applyStructuredStreamingValidatesRunnerSetup(it, ["--streaming": "true", "--streamingStopAfterIdleBatches": "5"], [
+    // Side inputs
+    'org.apache.beam.sdk.testing.UsesSideInputs',
+    // Unbounded
+    'org.apache.beam.sdk.testing.UsesTestStreamWithProcessingTime',
+    // State and Timers
+    'org.apache.beam.sdk.testing.UsesProcessingTimeTimers',
+    'org.apache.beam.sdk.testing.UsesLoopingTimer',
+    'org.apache.beam.sdk.testing.UsesMultimapState',
+    // Windowing
+    'org.apache.beam.sdk.testing.UsesCustomWindowMerging',
+    // Ordering
+    'org.apache.beam.sdk.testing.UsesRequiresTimeSortedInput',
+    // SDF
+    'org.apache.beam.sdk.testing.UsesBoundedSplittableParDo',
+    // Exceeds Java heap space
+    'org.apache.beam.sdk.testing.LargeKeys$Above100MB',
+  ])
+
+  filter {
+    // TODO(https://github.com/apache/beam/issues/36841): merging windows
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.CombineTest$WindowingTests.testSessionsCombine'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.GroupByKeyTest$WindowTests.testGroupByKeyMergingWindows'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterSessionsAndGroupByKey'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.ReshuffleTest.testReshuffleAfterSessionsAndGroupByKey'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.windowing.WindowingTest.testMergingWindowing'
+
+    // TODO(https://github.com/apache/beam/issues/36841): side inputs, PAssert.thatSingleton and Combine.globally with defaults
+    excludeTestsMatching 'org.apache.beam.sdk.io.CountingSourceTest.testBoundedSourceSplits'
+    excludeTestsMatching 'org.apache.beam.sdk.io.CountingSourceTest.testUnboundedSourceSplits'
+    excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testIsEqualTo'
+    excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testNotEqualTo'
+    excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testWindowedIsEqualTo'
+    excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testPAssertEqualsSingletonFalse'
+    excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testPAssertEqualsSingletonFalseDefaultReasonString'
+    excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testPAssertEqualsSingletonFailsForEmptyPCollection'
+    excludeTestsMatching 'org.apache.beam.sdk.testing.PAssertTest.testPAssertEqualsSingletonFailsForNonSingletonPCollection'
+
+    // TODO(https://github.com/apache/beam/issues/40949): ParDo with additional outputs
+    excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAllAttemptedMetrics'
+    excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedBoundedTrieMetrics'
+    excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedCounterMetrics'
+    excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedDistributionMetrics'
+    excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedGaugeMetrics'
+    excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testAttemptedStringSetMetrics'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.CombineTest$BasicTests.testHotKeyCombining'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.CombineTest$BasicTests.testHotKeyCombiningWithAccumulationMode'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.FlattenTest.testFlattenMultiplePCollectionsHavingMultipleConsumers'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoSchemaTest.testReadAndWriteMultiOutput'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$MultipleInputsAndOutputTests.testParDoEmptyWithTaggedOutput'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$MultipleInputsAndOutputTests.testParDoWithEmptyTaggedOutput'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$MultipleInputsAndOutputTests.testParDoWithOnlyTaggedOutput'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$MultipleInputsAndOutputTests.testParDoWithTaggedOutput'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$StateTests.testValueStateTaggedOutput'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$TimestampTests.testParDoTaggedOutputWithTimestamp'
+    excludeTestsMatching 'org.apache.beam.sdk.values.PCollectionTupleTest.testComposePCollectionTuple'
+
+    // TODO(https://github.com/apache/beam/issues/40950): Redistribute expands to GroupByKey with ReshuffleTrigger
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testJustRedistribute'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterFixedWindows'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterFixedWindowsAndGroupByKey'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterSlidingWindows'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributeAfterSlidingWindowsAndGroupByKey'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributePreservesMetadata'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.RedistributeTest.testRedistributePreservesTimestamps'
+
+    // TODO(https://github.com/apache/beam/issues/36841): PeriodicImpulse uses a splittable DoFn
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.PerKeyOrderingTest.testMultipleStatefulOrderingWithShuffle'
+    excludeTestsMatching 'org.apache.beam.sdk.transforms.PerKeyOrderingTest.testMultipleStatefulOrderingWithoutShuffle'
+
+    // TODO(https://github.com/apache/beam/issues/32021)
+    excludeTestsMatching 'org.apache.beam.sdk.metrics.MetricsTest$AttemptedMetricTests.testBoundedSourceMetricsInSplit'
+  }
+}
+
 tasks.validatesRunner.dependsOn(validatesStructuredStreamingRunnerBatch)
 
 // Exclude legacy DStream-based streaming tests (under runners/spark/translation/streaming)

runners/spark/spark_runner.gradle

@@ -486,33 +486,32 @@ def validatesRunnerStreaming = tasks.register("validatesRunnerStreaming", Test)
   }
 }
 
-tasks.register("validatesStructuredStreamingRunnerBatch", Test) {
-  group = "Verification"
+def applyStructuredStreamingValidatesRunnerSetup = { Test it, Map pipelineOptions, List<String> extraExcludedCategories ->
+  it.group = "Verification"
   // Disable gradle cache
-  outputs.upToDateWhen { false }
-  systemProperties sparkTestProperties(["--runner":"SparkStructuredStreamingRunner", "--testMode":"true"])
+  it.outputs.upToDateWhen { false }
+  it.systemProperties sparkTestProperties(["--runner":"SparkStructuredStreamingRunner", "--testMode":"true"] + pipelineOptions)
   // Register various other classes used in tests
-  systemProperty 'spark.kryo.classesToRegister',
+  it.systemProperty 'spark.kryo.classesToRegister',
     'org.apache.beam.sdk.transforms.MapViewTest$NonDeterministicStringCoder,' +
     'org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.RegularImmutableList'
-  jvmArgs += sparkTestJvmArgs()
-  jvmArgs '-Xmx7g' // Increase memory heap in order to avoid OOM errors
+  it.jvmArgs += sparkTestJvmArgs()
+  it.jvmArgs '-Xmx7g' // Increase memory heap in order to avoid OOM errors
 
-  classpath = configurations.validatesRunner
-  testClassesDirs = files(
+  it.classpath = project.configurations.validatesRunner
+  it.testClassesDirs = project.files(
     project(":sdks:java:core").sourceSets.test.output.classesDirs,
     project(":runners:core-java").sourceSets.test.output.classesDirs,
   )
-  testClassesDirs += files(project.sourceSets.test.output.classesDirs)
+  it.testClassesDirs += project.files(project.sourceSets.test.output.classesDirs)
 
-  maxParallelForks 4
+  it.maxParallelForks 4
 
-  useJUnit {
+  it.useJUnit {
     includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner'
     // Should be run only in a properly configured SDK harness environment
     excludeCategories 'org.apache.beam.sdk.testing.UsesSdkHarnessEnvironment'
     // Unbounded
-    excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedPCollections'
     excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream'
     // State and Timers
     excludeCategories 'org.apache.beam.sdk.testing.UsesOnWindowExpiration'
@@ -529,7 +528,13 @@ tasks.register("validatesStructuredStreamingRunnerBatch", Test) {
     excludeCategories 'org.apache.beam.sdk.testing.UsesExternalService'
     excludeCategories 'org.apache.beam.sdk.testing.UsesBundleFinalizer'
     excludeCategories 'org.apache.beam.sdk.testing.UsesTriggeredSideInputs'
+    extraExcludedCategories.each { excludeCategories it }
   }
+}
+ext.applyStructuredStreamingValidatesRunnerSetup = applyStructuredStreamingValidatesRunnerSetup
+
+tasks.register("validatesStructuredStreamingRunnerBatch", Test) {
+  applyStructuredStreamingValidatesRunnerSetup(it, [:], ['org.apache.beam.sdk.testing.UsesUnboundedPCollections'])
   filter {
     // These build on PeriodicImpulse, so the pipeline is unbounded and rejected by this batch only
     // runner, but they are not categorized as UsesUnboundedPCollections. Excluded by name rather
@@ -560,6 +565,9 @@ tasks.register("validatesRunner") {
   dependsOn validatesRunnerBatch
   dependsOn validatesRunnerBatchWithBoundedSDFExperiment
   dependsOn validatesRunnerStreaming
+  if (spark_major == '4') {
+    dependsOn "validatesStructuredStreamingRunnerStreaming"
+  }
   // It should be uncommented once all "validatesStructuredStreamingRunnerBatch" tests will pass.
   // Otherwise, it breaks Spark runner ValidatesRunner tests.
   //dependsOn validatesStructuredStreamingRunnerBatch

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/SparkTransformOverrides.java

@@ -32,6 +32,7 @@
 import org.apache.beam.sdk.transforms.Impulse;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.construction.EmptyFlattenAsCreateFactory;
 import org.apache.beam.sdk.util.construction.PTransformMatchers;
 import org.apache.beam.sdk.util.construction.PTransformTranslation;
 import org.apache.beam.sdk.util.construction.ReplacementOutputs;
@@ -70,7 +71,11 @@ public static List<PTransformOverride> getDefaultOverrides(boolean streaming) {
                   PTransformMatchers.urnEqualTo(PTransformTranslation.SPLITTABLE_PROCESS_KEYED_URN),
                   new SplittableParDoNaiveBounded.OverrideFactory()));
     } else {
+      // Before the Impulse override, the empty Flatten replacement expands to an Impulse.
       builder
+          .add(
+              PTransformOverride.of(
+                  PTransformMatchers.emptyFlatten(), EmptyFlattenAsCreateFactory.instance()))
           .add(
               PTransformOverride.of(
                   PTransformMatchers.classEqualTo(SplittableParDo.PrimitiveBoundedRead.class),