[Spark][#36841] Add an opt-in Dataset-based backend for portable pipelines

.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.json

@@ -0,0 +1,3 @@
+{
+  "comment": "Modify this file in a trivial way to cause this test suite to run."
+}

.github/workflows/README.md

@@ -371,6 +371,7 @@ PostCommit Jobs run in a schedule against master branch and generally do not get
 | [ PostCommit Java PVR Spark Batch ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark_Batch.yml) | N/A |`beam_PostCommit_Java_PVR_Spark_Batch.json`| [![.github/workflows/beam_PostCommit_Java_PVR_Spark_Batch.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark_Batch.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark_Batch.yml?query=event%3Aschedule) |
 | [ PostCommit Java PVR Spark4 Batch ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_Batch.yml) | N/A |`beam_PostCommit_Java_PVR_Spark4_Batch.json`| [![.github/workflows/beam_PostCommit_Java_PVR_Spark4_Batch.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_Batch.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_Batch.yml?query=event%3Aschedule) |
 | [ PostCommit Java PVR Spark4 Streaming ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_Streaming.yml) | N/A |`beam_PostCommit_Java_PVR_Spark4_Streaming.json`| [![.github/workflows/beam_PostCommit_Java_PVR_Spark4_Streaming.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_Streaming.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_Streaming.yml?query=event%3Aschedule) |
+| [ PostCommit Java PVR Spark4 StructuredStreaming ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.yml) | N/A |`beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.json`| [![.github/workflows/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.yml?query=event%3Aschedule) |
 | [ PostCommit Java Tpcds Dataflow ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Dataflow.yml) | N/A |`beam_PostCommit_Java_Tpcds_Dataflow.json`| [![.github/workflows/beam_PostCommit_Java_Tpcds_Dataflow.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Dataflow.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Dataflow.yml?query=event%3Aschedule) |
 | [ PostCommit Java Tpcds Flink ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Flink.yml) | N/A |`beam_PostCommit_Java_Tpcds_Flink.json`| [![.github/workflows/beam_PostCommit_Java_Tpcds_Flink.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Flink.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Flink.yml?query=event%3Aschedule) |
 | [ PostCommit Java Tpcds Spark ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Spark.yml) | N/A |`beam_PostCommit_Java_Tpcds_Spark.json`| [![.github/workflows/beam_PostCommit_Java_Tpcds_Spark.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Spark.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Tpcds_Spark.yml?query=event%3Aschedule) |

.github/workflows/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.yml

@@ -0,0 +1,98 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: PostCommit Java PVR Spark4 StructuredStreaming
+
+on:
+  schedule:
+    - cron: '15 5/6 * * *'
+  pull_request_target:
+    paths: ['release/trigger_all_tests.json', '.github/trigger_files/beam_PostCommit_Java_PVR_Spark4_StructuredStreaming.json']
+  workflow_dispatch:
+
+# This allows a subsequently queued workflow run to interrupt previous runs
+concurrency:
+  group: '${{ github.workflow }} @ ${{ github.event.pull_request.number || github.sha || github.head_ref || github.ref }}-${{ github.event.schedule || github.event.comment.id || github.event.sender.login }}'
+  cancel-in-progress: true
+
+#Setting explicit permissions for the action to avoid the default permissions which are `write-all` in case of pull_request_target event
+permissions:
+  actions: write
+  pull-requests: write
+  checks: write
+  contents: read
+  deployments: read
+  id-token: none
+  issues: write
+  discussions: read
+  packages: read
+  pages: read
+  repository-projects: read
+  security-events: read
+  statuses: read
+
+env:
+  DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }}
+  GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GE_CACHE_USERNAME }}
+  GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }}
+
+jobs:
+  beam_PostCommit_Java_PVR_Spark4_StructuredStreaming:
+    name: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
+    runs-on: [self-hosted, ubuntu-24.04, main]
+    timeout-minutes: 180
+    strategy:
+      matrix:
+        job_name: [beam_PostCommit_Java_PVR_Spark4_StructuredStreaming]
+        job_phrase: [Run Java Spark v4 PortableValidatesRunner StructuredStreaming]
+    if: |
+      github.event_name == 'workflow_dispatch' ||
+      github.event_name == 'pull_request_target' ||
+      (github.event_name == 'schedule' && github.repository == 'apache/beam') ||
+      github.event.comment.body == 'Run Java Spark v4 PortableValidatesRunner StructuredStreaming'
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          persist-credentials: false
+      - name: Setup repository
+        uses: ./.github/actions/setup-action
+        with:
+          comment_phrase: ${{ matrix.job_phrase }}
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          github_job: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
+      - name: Setup environment
+        uses: ./.github/actions/setup-environment-action
+        with:
+          java-version: 17
+      - name: run PostCommit Java PortableValidatesRunner Spark4 StructuredStreaming script
+        uses: ./.github/actions/gradle-command-self-hosted-action
+        with:
+            gradle-command: :runners:spark:4:job-server:validatesPortableRunnerStructuredStreaming
+      - name: Archive JUnit Test Results
+        uses: actions/upload-artifact@v7
+        if: ${{ !success() }}
+        with:
+          name: JUnit Test Results
+          path: "**/build/reports/tests/"
+      - name: Publish JUnit Test Results
+        uses: EnricoMi/publish-unit-test-result-action@v2
+        if: always()
+        with:
+          commit: '${{ env.prsha || env.GITHUB_SHA }}'
+          comment_mode: ${{ github.event_name == 'issue_comment'  && 'always' || 'off' }}
+          files: '**/build/test-results/**/*.xml'
+          large_files: true

runners/spark/4/README.md

@@ -36,6 +36,12 @@ runner code.
 Batch only. Streaming is tracked in
 [#36841](https://github.com/apache/beam/issues/36841).
 
+The portable job server can run bounded pipelines on the Dataset-based backend
+with `--useStructuredStreaming`. That path is experimental and rejects unbounded
+input, state, and timers at translation. The
+`validatesPortableRunnerStructuredStreaming` task runs the streaming
+PortableValidatesRunner suite against it.
+
 ## Known issues
 
 ### `StackOverflowError` from `slf4j-jdk14` on the runtime classpath

runners/spark/job-server/spark_job_server.gradle

@@ -94,8 +94,11 @@ def sickbayTests = [
         'org.apache.beam.sdk.transforms.ReshuffleTest.testReshufflePreservesMetadata',
 ]
 
-def portableValidatesRunnerTask(String name, boolean streaming, boolean docker, ArrayList<String> sickbayTests) {
+def portableValidatesRunnerTask(String name, boolean streaming, boolean docker, ArrayList<String> sickbayTests, boolean structuredStreaming = false) {
   def pipelineOptions = []
+  if (structuredStreaming) {
+    pipelineOptions += "--useStructuredStreaming"
+  }
   def testCategories
   def testFilter
 
@@ -243,6 +246,9 @@ def portableValidatesRunnerTask(String name, boolean streaming, boolean docker,
 project.ext.validatesPortableRunnerDocker= portableValidatesRunnerTask("Docker", false, true, sickbayTests)
 project.ext.validatesPortableRunnerBatch = portableValidatesRunnerTask("Batch", false, false, sickbayTests)
 project.ext.validatesPortableRunnerStreaming = portableValidatesRunnerTask("Streaming", true, false, sickbayTests)
+// Structured Streaming variant of the streaming suite: same tests and exclusions, run on the
+// Dataset-based backend. It is the exit gate for making that backend the streaming default.
+project.ext.validatesPortableRunnerStructuredStreaming = portableValidatesRunnerTask("StructuredStreaming", true, false, sickbayTests, true)
 
 tasks.register("validatesPortableRunner") {
   dependsOn validatesPortableRunnerDocker
@@ -284,7 +290,7 @@ def sparkJobServerJvmArgs() {
 }
 
 // TestPortableRunner starts SparkJobServerDriver in-process in the test JVM.
-['validatesPortableRunnerDocker', 'validatesPortableRunnerBatch', 'validatesPortableRunnerStreaming'].each { taskName ->
+['validatesPortableRunnerDocker', 'validatesPortableRunnerBatch', 'validatesPortableRunnerStreaming', 'validatesPortableRunnerStructuredStreaming'].each { taskName ->
   tasks.named(taskName) {
     jvmArgs += sparkJobServerJvmArgs()
   }

runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java

@@ -86,4 +86,14 @@ public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
   boolean isCacheDisabled();
 
   void setCacheDisabled(boolean value);
+
+  @Description(
+      "Run portable pipelines on the Dataset-based backend. Fully bounded pipelines run as "
+          + "Spark batch jobs. Pipelines with unbounded input are "
+          + "rejected at translation until the Structured Streaming query is in place, see "
+          + "https://github.com/apache/beam/issues/36841.")
+  @Default.Boolean(false)
+  boolean getUseStructuredStreaming();
+
+  void setUseStructuredStreaming(boolean value);
 }

runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java

@@ -35,6 +35,7 @@
 import org.apache.beam.runners.spark.metrics.MetricsAccumulator;
 import org.apache.beam.runners.spark.translation.SparkBatchPortablePipelineTranslator;
 import org.apache.beam.runners.spark.translation.SparkContextFactory;
+import org.apache.beam.runners.spark.translation.SparkDatasetPortablePipelineTranslator;
 import org.apache.beam.runners.spark.translation.SparkPortablePipelineTranslator;
 import org.apache.beam.runners.spark.translation.SparkStreamingPortablePipelineTranslator;
 import org.apache.beam.runners.spark.translation.SparkStreamingTranslationContext;
@@ -82,7 +83,14 @@ public SparkPipelineRunner(SparkPipelineOptions pipelineOptions) {
   public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo) {
     SparkPortablePipelineTranslator translator;
     boolean isStreaming = pipelineOptions.isStreaming() || hasUnboundedPCollections(pipeline);
-    if (isStreaming) {
+    if (pipelineOptions.getUseStructuredStreaming()) {
+      // The Dataset backend evaluates its own leaves. It never starts a DStream context, and it
+      // rejects unbounded input at translation. Clear the streaming option so that everything
+      // reading it downstream, such as the metrics accumulator, agrees with how the job runs.
+      translator = new SparkDatasetPortablePipelineTranslator();
+      pipelineOptions.setStreaming(false);
+      isStreaming = false;
+    } else if (isStreaming) {
       translator = new SparkStreamingPortablePipelineTranslator();
     } else {
       translator = new SparkBatchPortablePipelineTranslator();