[SPARK-53823][SS] Implement allow list for real time mode by jerrypeng · Pull Request #52891 · apache/spark

jerrypeng · 2025-11-05T06:41:29Z

What changes were proposed in this pull request?

Add an allowlist and some guardrails to help users understand what is supported and what is not supported in Real-time Mode. This should improve the user experience of real-time mode and minimize the chance of a new user use it in a way that is unexpected or untested which may produce undesirable outcomes.

Why are the changes needed?

To help guide users on what is currently supported and what is not.

Does this PR introduce any user-facing change?

Yes, running a query with currently unsupported features in RTM will throw an exception.

How was this patch tested?

Several tests are added

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

No

jerrypeng · 2025-11-05T06:49:28Z

...re/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala

@@ -436,7 +436,10 @@ class MicroBatchExecution(
      }
    }

-    if (containsStatefulOperator(analyzedPlan)) {
+    if (trigger.isInstanceOf[RealTimeTrigger]) {


@HeartSaVioR you recently added AQE support streaming queries. Can you take a look to see if this logic is appropriate to disable AQE for RTM quieries?

That should do the thing. Looks good to me.

viirya · 2025-11-06T16:43:25Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeAllowlistSuite.scala

+    }
+  }
+
+//  test("exactly once sink not supported") {


Do we want to keep commented code here?

viirya · 2025-11-06T17:39:29Z

...alyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala

+    "real-time without operators - append mode",
+    streamRelation, Append,
+    "STREAMING_REAL_TIME_MODE.OUTPUT_MODE_NOT_SUPPORTED"
+    //    Map("outputMode" -> "Append")


Remove this commented code?

forgot the push a commit that removes this :)

viirya · 2025-11-06T17:40:33Z

...alyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala

@@ -1017,6 +1030,30 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper {
    }
  }

+  /** Assert that the logical plan is supported for real-time mode */
+  def assertSupportedForRealTime(name: String, plan: LogicalPlan, outputMode: OutputMode): Unit = {


Hmm, where is it used? I don't find any.

ah forgot to push a commit. The test is added

viirya · 2025-11-07T01:13:31Z

Note that this will be merged into master and branch-4.1.

dongjoon-hyun · 2025-11-07T03:08:39Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -3094,6 +3032,12 @@ object SQLConf {
    .timeConf(TimeUnit.MILLISECONDS)
    .createWithDefault(5000)

+  val STREAMING_REAL_TIME_MODE_ALLOWLIST_CHECK = buildConf(
+    "spark.sql.streaming.realTimeMode.allowlistCheck")


nit. Indentation looks a little wrong to me for all realTimeMode configurations, @jerrypeng and @viirya .

spark.sql.streaming.realTimeMode.minBatchDuration

spark.sql.streaming.realTimeMode.allowlistCheck

It would be great if you can match with the other code around here.

dongjoon-hyun · 2025-11-07T03:09:29Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -3094,6 +3032,12 @@ object SQLConf {
    .timeConf(TimeUnit.MILLISECONDS)
    .createWithDefault(5000)

+  val STREAMING_REAL_TIME_MODE_ALLOWLIST_CHECK = buildConf(
+    "spark.sql.streaming.realTimeMode.allowlistCheck")
+    .doc("Whether to check all operators, sinks used in real-time mode are in the allowlist.")


We need a version field.

.version("4.1.0")

dongjoon-hyun · 2025-11-07T03:09:43Z

...alyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala

+  =======================================================================================
+                                 REAL-TIME STREAMING
+  =======================================================================================
+ */


indentation?

dongjoon-hyun · 2025-11-07T03:11:38Z

...alyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala

+  /*
+  =======================================================================================
+                                 REAL-TIME STREAMING
+  =======================================================================================


Also, this is a little mismatched.

dongjoon-hyun · 2025-11-07T03:12:26Z

...alyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala

@@ -853,6 +853,24 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper {
        Deduplicate(Seq(attribute), streamRelation)), outputMode = Append)
  }

+  /*
+  =======================================================================================
+                                 REAL-TIME STREAMING


BTW, for the actual test methods, it seems that the following {...} grouping is used in this file, doesn't it?

// REAL-TIME STREAMING { ... }

dongjoon-hyun · 2025-11-07T03:13:09Z

...alyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala

+    }
+  }
+
+  /** Assert that the logical plan is not supported inside a streaming plan with the


nit. Maybe, the following?

/** * Assert that the logical plan is not supported inside a streaming plan with the ...

dongjoon-hyun · 2025-11-07T03:20:39Z

.../src/main/scala/org/apache/spark/sql/execution/streaming/runtime/RealTimeModeAllowlist.scala

+    "org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec",
+    "org.apache.spark.sql.execution.exchange.BroadcastExchangeExec",
+    "org.apache.spark.sql.execution.joins.BroadcastHashJoinExec",
+    "org.apache.spark.sql.execution.exchange.ReusedExchangeExec",


I guess this was initially designed to be a sorted list alphabetically in the code side. Could you change like the following because it looks like the only exceptions?

- "org.apache.spark.sql.execution.joins.BroadcastHashJoinExec", "org.apache.spark.sql.execution.exchange.ReusedExchangeExec", + "org.apache.spark.sql.execution.joins.BroadcastHashJoinExec",

dongjoon-hyun · 2025-11-07T03:24:00Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeAllowlistSuite.scala

+  }
+
+  test("rtm operator allowlist") {
+    withSQLConf("spark.sql.autoBroadcastJoinThreshold" -> "0") {


nit.

- withSQLConf("spark.sql.autoBroadcastJoinThreshold" -> "0") { + withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "0") {

dongjoon-hyun · 2025-11-07T03:24:29Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeAllowlistSuite.scala

+    }
+  }
+
+  // TODO : Remove this test after RTM can shuffle to multiple stages


Please file a JIRA issue and make this IDed TODO like TODO(SPARK-XXX). It helps the community move forward from your contributions.

dongjoon-hyun · 2025-11-07T03:25:22Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeAllowlistSuite.scala

+      }
+  }
+
+  // TODO : Remove this test after RTM supports stateful queries


ditto. We need IDed TODO.

dongjoon-hyun

+1, LGTM (except a few minor style comments)

jerrypeng · 2025-11-07T07:46:28Z

@dongjoon-hyun thank you for your review! I have addressed your comments.

dongjoon-hyun · 2025-11-07T21:08:13Z

Thank you, @jerrypeng and @viirya .
Merged to master/4.1 for Apache Spark 4.1.0.

### What changes were proposed in this pull request? Add an allowlist and some guardrails to help users understand what is supported and what is not supported in Real-time Mode. This should improve the user experience of real-time mode and minimize the chance of a new user use it in a way that is unexpected or untested which may produce undesirable outcomes. ### Why are the changes needed? To help guide users on what is currently supported and what is not. ### Does this PR introduce _any_ user-facing change? Yes, running a query with currently unsupported features in RTM will throw an exception. ### How was this patch tested? Several tests are added ### Was this patch authored or co-authored using generative AI tooling? No Closes #52891 from jerrypeng/SPARK-53823. Authored-by: Jerry Peng <jerry.peng@databricks.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org> (cherry picked from commit 178f0f4) Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

viirya · 2025-11-07T22:16:25Z

Thank you @jerrypeng @dongjoon-hyun

### What changes were proposed in this pull request? Add an allowlist and some guardrails to help users understand what is supported and what is not supported in Real-time Mode. This should improve the user experience of real-time mode and minimize the chance of a new user use it in a way that is unexpected or untested which may produce undesirable outcomes. ### Why are the changes needed? To help guide users on what is currently supported and what is not. ### Does this PR introduce _any_ user-facing change? Yes, running a query with currently unsupported features in RTM will throw an exception. ### How was this patch tested? Several tests are added ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#52891 from jerrypeng/SPARK-53823. Authored-by: Jerry Peng <jerry.peng@databricks.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

see - apache/spark#52473 - apache/spark#52870 - apache/spark#52891

* Move GlutenStreamingQuerySuite to correct package * Add Spark 4.1 new test suites for Gluten * Enable new and existing Gluten test suites for Spark 4.1 UT * Update workflow trigger paths to exclude Spark 4.0 and 4.1 shims directories for clickhouse backend * Add support for Spark 4.1 in build script * Merge Spark 4.1.0 sql-tests into Gluten Spark 4.1 (three-way merge) Three-way merge performed using Git: - Base: Spark 4.0.1 (29434ea766b) - Left: Spark 4.1.0 (e221b56be7b) - Right: Gluten Spark 4.1 backends-velox Summary: - Auto-merged: 165 files - New tests added: 31 files (collations, edge cases, recursion, spatial, etc.) - Modified tests: 134 files - Deleted tests: 2 files (collations.sql -> split into 4 files, timestamp-ntz.sql) Conflicts resolved: - inputs/timestamp-ntz.sql: Right deleted + Left modified -> DELETED (per resolution rule) New test suites from Spark 4.1.0: - Collations (4 files): aliases, basic, padding-trim, string-functions - Edge cases (6 files): alias-resolution, extract-value, join-resolution, etc. - Advanced features: cte-recursion, generators, kllquantiles, thetasketch, time - Name resolution: order-by-alias, session-variable-precedence, runtime-replaceable - Spatial functions: st-functions (ANSI and non-ANSI variants) - Various resolution edge cases Total files after merge: 671 (up from 613) * Enable additional Spark 4.1 SQL tests by resolving TODOs * Add new Spark 4.1 test files to VeloxSQLQueryTestSettings * [Fix] Replace `RuntimeReplaceable` with its `replacement` to fix UT. see apache/spark#50287 * [4.1.0] Exclude "infer shredding with mixed scale" see apache/spark#52406 * [Fix] Implement Kryo serialization for CachedColumnarBatch see apache/spark#50599 * [4.1.0] Exclude GlutenMapStatusEndToEndSuite and configure parallelism see apache/spark#50230 * [4.1.0] Exclude Spark Structure Steaming tests in Gluten see - apache/spark#52473 - apache/spark#52870 - apache/spark#52891 * [4.1.0] Exclude failing SQL tests on Spark 4.1 * Replace SparkException.require with standard require in ColumnarCachedBatchSerializer to work across different spark versions * [Fix] Replace `RuntimeReplaceable` with its `replacement` to fix UT. see apache/spark#50287 * Exclude Spark 4.0 and 4.1 paths in clickhouse_be_trigger using `!` prefix * [Fix] Update GlutenShowNamespacesParserSuite to use GlutenSQLTestsBaseTrait

github-actions bot added SQL STRUCTURED STREAMING labels Nov 5, 2025

jerrypeng force-pushed the SPARK-53823 branch 2 times, most recently from e2047f0 to 1dab9ec Compare November 5, 2025 06:48

jerrypeng commented Nov 5, 2025

View reviewed changes

jerrypeng changed the title ~~[WIP][SPARK-53823] Implement allow list for real time mode~~ [SPARK-53823] Implement allow list for real time mode Nov 5, 2025

HyukjinKwon changed the title ~~[SPARK-53823] Implement allow list for real time mode~~ [SPARK-53823][SS] Implement allow list for real time mode Nov 5, 2025

viirya reviewed Nov 6, 2025

View reviewed changes

viirya approved these changes Nov 6, 2025

View reviewed changes

viirya reviewed Nov 6, 2025

View reviewed changes

dongjoon-hyun reviewed Nov 7, 2025

View reviewed changes

dongjoon-hyun approved these changes Nov 7, 2025

View reviewed changes

jerrypeng added 3 commits November 6, 2025 23:44

[SPARK-53823] Implement allow list for real time mode

7162aab

adding additional tests and cleaning up

caf3a50

addressing comments

9703b2e

jerrypeng force-pushed the SPARK-53823 branch from ab913ef to 9703b2e Compare November 7, 2025 07:44

dongjoon-hyun closed this in 178f0f4 Nov 7, 2025

baibaichen added a commit to baibaichen/gluten that referenced this pull request Jan 12, 2026

[4.1.0] Exclude Spark Structure Steaming tests in Gluten

13d6588

see - apache/spark#52473 - apache/spark#52870 - apache/spark#52891

baibaichen added a commit to baibaichen/gluten that referenced this pull request Jan 13, 2026

[4.1.0] Exclude Spark Structure Steaming tests in Gluten

7c4f944

see - apache/spark#52473 - apache/spark#52870 - apache/spark#52891

This was referenced Jan 13, 2026

[GLUTEN-11355][UT] Add new Spark 4.1 tests apache/incubator-gluten#11380

Merged

[VL] Track on Spark-4.1.x failed unit tests apache/incubator-gluten#11400

Open

baibaichen added a commit to baibaichen/gluten that referenced this pull request Jan 13, 2026

[4.1.0] Exclude Spark Structure Steaming tests in Gluten

31bc8f4

see - apache/spark#52473 - apache/spark#52870 - apache/spark#52891

Conversation

jerrypeng commented Nov 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya commented Nov 7, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Nov 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

jerrypeng commented Nov 7, 2025

Uh oh!

dongjoon-hyun commented Nov 7, 2025

Uh oh!

viirya commented Nov 7, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

jerrypeng commented Nov 5, 2025 •

edited

Loading

dongjoon-hyun Nov 7, 2025 •

edited

Loading