[SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement #52428

HeartSaVioR · 2025-09-24T01:46:58Z

What changes were proposed in this pull request?

This PR proposes to introduce WATERMARK clause in SQL statement. WATERMARK clause is to define the watermark against the relation, especially streaming relation where STREAM keyword is added to the relation (or table valued function).

Please refer to the SQL reference doc for WATERMARK clause about its definition.
https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-qry-select-watermark

The PR also contains new tests which show how to use it.

Why are the changes needed?

This is needed to unblock the stateful workload in Streaming Table & Flow being described as SQL statement.

Does this PR introduce any user-facing change?

Yes, users can define the watermark for stateful workloads with SQL statement.

How was this patch tested?

New UTs.

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

No.

HeartSaVioR · 2025-09-24T01:50:43Z

cc. @sryza @cloud-fan @viirya
Could you please take a look? Thanks!

Also, do we have test purposed TVF definition for "streaming source"? I couldn't add E2E test for STREAM TVF + WATERMARK and it'd be ideal if we could add one.

HeartSaVioR · 2025-09-24T01:53:06Z

cc. @anishshri-db as well

HeartSaVioR · 2025-09-24T01:53:22Z

common/utils/src/main/resources/error/error-conditions.json

@@ -9880,6 +9880,11 @@
      "Doesn't support month or year interval: <interval>"
    ]
  },
+  "_LEGACY_ERROR_TEMP_3263" : {


TODO: Need to assign one error class

TODO: remove this

...g/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala

viirya · 2025-09-24T16:23:48Z

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

-      optionsClause? sample? tableAlias                     #tableName
-    | LEFT_PAREN query RIGHT_PAREN sample? tableAlias       #aliasedQuery
-    | LEFT_PAREN relation RIGHT_PAREN sample? tableAlias    #aliasedRelation
+      optionsClause? sample? watermarkClause? tableAlias    #tableName


watermarkClause is already defined in streamRelationPrimary. Do we still need it here? Is it also applied for non-stream relation?

relation here can be a temp view which could be technically streaming without STREAM keyword

viirya · 2025-09-24T16:41:02Z

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

 functionTable
    : funcName=functionName LEFT_PAREN
      (functionTableArgument (COMMA functionTableArgument)*)?
-      RIGHT_PAREN tableAlias
+      RIGHT_PAREN watermarkClause? tableAlias


From the doc https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-qry-select-watermark, it looks like table_valued_function has no watermark_clause support, but we want to have it here?

Ouch I think it's missed in that doc.

cloud-fan · 2025-09-25T10:29:57Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

+    }
+
+    // TODO: This seems to need to find the new expression by index. We can't rely on exprId
+    //  due to UnresolvedAlias. Are there better ways to do?


I think we need a new logical plan like UnresolvedEventTimeWatermark, which will be rewritten into EventTimeWatermark + Project during analysis.

Addressed. PTAL!

HeartSaVioR · 2025-09-26T02:12:35Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala

+        case e: Expression => UnresolvedAlias(e)
+      }
+
+      val isAttributeReference = namedExpression.isInstanceOf[AttributeReference]


@cloud-fan
Please help verify what I'm doing is correct.

What EventTimeWatermark node does is simply to attach metadata marker against the column (produced by child.output) referred by the expression. The node itself does not do any projection and just updates the accumulators and does the passthrough. If the eventTimeColExpr has to be evaluated, we have to inject Project to evaluate the expression, which we expect the result to be explicitly named.

Hence I skip injecting Project only if the expression is an AttributeReference and the column is available in the child.output, otherwise I simply add Project.

it doesn't matter. We can also always add the Project, as optimizer can remove it.

What I'm concerned is that we "change" the order of column when we add the project.

For expression with alias, that is not an existing column hence I think it's OK to consider it like withColumn, appending the column in the end. But if the expression is simply referring to the existing column, probably the ideal behavior is to not have a project which might change the column order.

cloud-fan · 2025-09-26T12:10:58Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala

+  override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+    _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+    case u: UnresolvedEventTimeWatermark if u.childrenResolved =>


shall we wait for u.eventTimeColExpr.resolved?

cloud-fan · 2025-09-26T12:11:52Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala

+      }
+
+      val isAttributeReference = namedExpression.isInstanceOf[AttributeReference]
+      val exprInChildOutput = u.child.output.exists(_.name == namedExpression.name)


it's weird to search by name for resolved attributes, we can just do u.child.outputSet.contains(namedExpression)

So when eventTimeColExpr is resolved, the expression can refer to the column by exprId? If then yes we can do that.

cloud-fan · 2025-09-26T12:13:31Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala

+
+  final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_EVENT_TIME_WATERMARK)
+
+  private val delay = IntervalUtils.fromIntervalString(delayInterval.toString)


can we do it in the parser? This logical plan should take CalendarInterval directly.

cloud-fan · 2025-09-26T12:14:03Z

...catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/EventTimeWatermark.scala

@@ -81,8 +81,9 @@ case class EventTimeWatermark(
  // This is not allowed by default - WatermarkPropagator will throw an exception. We keep the
  // logic here because we also maintain the compatibility flag. (See
  // SQLConf.STATEFUL_OPERATOR_ALLOW_MULTIPLE for details.)
+  // Should be defined as lazy so that attributes can be resolved before calling this.


do we still need this hack?

Let me try removing it.

...talyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala

cloud-fan · 2025-10-09T04:51:59Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala

+
+      val namedExpression = u.eventTimeColExpr match {
+        case e: NamedExpression => e
+        case e: Expression => UnresolvedAlias(e)


This alias is not user-facing and doesn't matter. We can simply do Alias(e, "watermark_col")()

The watermark column "name" is determined by the expression, not something we use the reserved name.

But wait, now this pattern is only executed "after" the eventTimeColExpr is resolved. Maybe we can just throw an exception if it's still not a named expression.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala

sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala

cloud-fan · 2025-10-09T04:55:54Z

...g/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala

+    sqlContext.streams.active.foreach(_.stop())
+  }
+
+  test("event time and watermark metrics with watermark in select DML - case 1") {


what does DML mean here?

I meant Data Manipulation Language - I can just specify it as "select statement".

...g/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

cloud-fan · 2025-10-20T01:36:44Z

sql/core/src/test/scala/org/apache/spark/sql/execution/command/WatermarkClauseSuite.scala

+    // attribute reference vs expression
+
+    // tableName, attribute reference, alias
+    assertEqual(


where is assertEqual defined?

My bad. I'll push the fix sooner.

It's probably proper to have these tests in PlanParserSuite.

cloud-fan · 2025-10-21T09:05:51Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala

+    _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+    case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+      if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {


We can fail it earlier in the parser

cloud-fan · 2025-10-21T09:06:57Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala

+        // We need to inject projection as we can't find the matching column directly in the
+        // child output.
+        val proj = Project(Seq(u.eventTimeColExpr) ++ u.child.output, u.child)
+        val attrRef = u.eventTimeColExpr.toAttribute


nit: let's move it before the if-else, as we need it in the if branch as well

Co-authored-by: Wenchen Fan <[email protected]>

HeartSaVioR · 2025-10-22T08:16:22Z

I've moved the check from AstBuilder to the analyzer rule since it's likely to be read more time than the AstBuilder impl.

cloud-fan · 2025-10-22T17:06:14Z

the docker test failure is unrelated, thanks, merging to master!

HeartSaVioR · 2025-10-22T21:57:53Z

Thanks for your thoughtful review!

### What changes were proposed in this pull request? This PR proposes to introduce WATERMARK clause in SQL statement. WATERMARK clause is to define the watermark against the relation, especially streaming relation where STREAM keyword is added to the relation (or table valued function). Please refer to the SQL reference doc for WATERMARK clause about its definition. https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-qry-select-watermark The PR also contains new tests which show how to use it. ### Why are the changes needed? This is needed to unblock the stateful workload in Streaming Table & Flow being described as SQL statement. ### Does this PR introduce _any_ user-facing change? Yes, users can define the watermark for stateful workloads with SQL statement. ### How was this patch tested? New UTs. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#52428 from HeartSaVioR/SPARK-53687. Authored-by: Jungtaek Lim <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

github-actions bot added SQL STRUCTURED STREAMING labels Sep 24, 2025

HeartSaVioR changed the title ~~[SPARK-53687][SQL][SS] Introduce WATERMARK clause in SQL statement~~ [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement Sep 24, 2025

HeartSaVioR commented Sep 24, 2025

View reviewed changes

github-actions bot added the DOCS label Sep 24, 2025

cloud-fan reviewed Sep 24, 2025

View reviewed changes

...g/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala Show resolved Hide resolved

viirya reviewed Sep 24, 2025

View reviewed changes

cloud-fan reviewed Sep 25, 2025

View reviewed changes

HeartSaVioR commented Sep 26, 2025

View reviewed changes

cloud-fan reviewed Sep 26, 2025

View reviewed changes