This is an automated email from the ASF dual-hosted git repository.
ashrigondekar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 0bc3f740c9ee [SPARK-56453][SS] Add ResolveEventTimeWatermark to
HiveSessionStateBuilder
0bc3f740c9ee is described below
commit 0bc3f740c9ee44602a1d9bfe06ceb10bce533e36
Author: ericm-db <[email protected]>
AuthorDate: Sun Apr 12 11:46:45 2026 -0700
[SPARK-56453][SS] Add ResolveEventTimeWatermark to HiveSessionStateBuilder
### What changes were proposed in this pull request?
Add `ResolveEventTimeWatermark` to the `extendedResolutionRules` in
`HiveSessionStateBuilder`.
`BaseSessionStateBuilder` already includes this rule, but
`HiveSessionStateBuilder` overrides `extendedResolutionRules` entirely without
including it. This causes `UnresolvedEventTimeWatermark` nodes (which have
`output = Nil`) to persist through analysis when using the Hive session state,
blocking column resolution in downstream operators such as JOIN conditions and
WHERE/ORDER BY clauses that reference aliased STREAM WATERMARK relations.
### Why are the changes needed?
This is a regression introduced by #52428 (SPARK-53477), which added
`ResolveEventTimeWatermark` to
`BaseSessionStateBuilder.extendedResolutionRules` but missed that
`HiveSessionStateBuilder` overrides that list entirely. When a query uses
`STREAM(...) WATERMARK ... <alias>` syntax with the Hive session state, the
`UnresolvedEventTimeWatermark` node is never resolved, causing
`UNRESOLVED_COLUMN` errors when referencing columns through the alias.
### Does this PR introduce _any_ user-facing change?
Yes. Streaming SQL queries using `WATERMARK` with table aliases now resolve
correctly when using the Hive session state, fixing `UNRESOLVED_COLUMN` errors.
### How was this patch tested?
Added `WatermarkColumnResolutionSuite` with three test cases:
- STREAM JOIN with WATERMARK and alias resolves columns correctly
- STREAM with WATERMARK and alias resolves columns in WHERE
- STREAM with WATERMARK and alias resolves columns in ORDER BY
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.6)
Closes #55312 from ericm-db/watermark-hive-resolution.
Authored-by: ericm-db <[email protected]>
Signed-off-by: Anish Shrigondekar <[email protected]>
---
.../spark/sql/hive/HiveSessionStateBuilder.scala | 3 +-
.../streaming/WatermarkColumnResolutionSuite.scala | 95 ++++++++++++++++++++++
2 files changed, 97 insertions(+), 1 deletion(-)
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala
index 9f5566407e38..0fbc41492e00 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala
@@ -25,7 +25,7 @@ import org.apache.hadoop.hive.ql.exec.{UDAF, UDF}
import org.apache.hadoop.hive.ql.udf.generic.{AbstractGenericUDAFResolver,
GenericUDF, GenericUDTF}
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.analysis.{Analyzer,
EvalSubqueriesForTimeTravel, InvokeProcedures, ReplaceCharWithVarchar,
ResolveDataSource, ResolveExecuteImmediate, ResolveMetricView,
ResolveSessionCatalog, ResolveTranspose}
+import org.apache.spark.sql.catalyst.analysis.{Analyzer,
EvalSubqueriesForTimeTravel, InvokeProcedures, ReplaceCharWithVarchar,
ResolveDataSource, ResolveEventTimeWatermark, ResolveExecuteImmediate,
ResolveMetricView, ResolveSessionCatalog, ResolveTranspose}
import org.apache.spark.sql.catalyst.analysis.resolver.ResolverExtension
import org.apache.spark.sql.catalyst.catalog.{ExternalCatalogWithListener,
InvalidUDFClassException}
import org.apache.spark.sql.catalyst.expressions.{Expression,
ExtractSemiStructuredFields}
@@ -136,6 +136,7 @@ class HiveSessionStateBuilder(
new InvokeProcedures(session) +:
ResolveExecuteImmediate(session, catalogManager) +:
ExtractSemiStructuredFields +:
+ ResolveEventTimeWatermark +:
customResolutionRules
override val postHocResolutionRules: Seq[Rule[LogicalPlan]] =
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/execution/streaming/WatermarkColumnResolutionSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/execution/streaming/WatermarkColumnResolutionSuite.scala
new file mode 100644
index 000000000000..577e0ca72f8d
--- /dev/null
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/execution/streaming/WatermarkColumnResolutionSuite.scala
@@ -0,0 +1,95 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.streaming
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.hive.test.TestHiveSingleton
+import org.apache.spark.sql.test.SQLTestUtils
+
+/**
+ * Regression tests for ResolveEventTimeWatermark in HiveSessionStateBuilder.
+ *
+ * SPARK-53687 introduced UnresolvedEventTimeWatermark /
ResolveEventTimeWatermark but only
+ * added the rule to BaseSessionStateBuilder, missing HiveSessionStateBuilder
(used in all
+ * Databricks production environments). Without the rule,
UnresolvedEventTimeWatermark
+ * (output=Nil) persists through analysis, breaking any SQL query that uses
the WATERMARK clause.
+ *
+ * Uses TestHiveSingleton to go through HiveSessionStateBuilder, matching
production behavior.
+ */
+class WatermarkColumnResolutionSuite extends QueryTest with TestHiveSingleton
with SQLTestUtils {
+
+ private def createTestTables(): Unit = {
+ sql("CREATE TABLE IF NOT EXISTS wm_signup_details " +
+ "(updated_at TIMESTAMP, consumer_profile_id INT) USING PARQUET")
+ sql("CREATE TABLE IF NOT EXISTS wm_profiles " +
+ "(updated_at TIMESTAMP, id INT) USING PARQUET")
+ sql("CREATE TABLE IF NOT EXISTS wm_events " +
+ "(ts TIMESTAMP, a INT) USING PARQUET")
+ }
+
+ private def analyzeStreamSQL(query: String): Unit = {
+ val plan = spark.sessionState.sqlParser.parsePlan(query)
+ val analyzed = spark.sessionState.analyzer.execute(plan)
+ spark.sessionState.analyzer.checkAnalysis(analyzed)
+ }
+
+ test("STREAM JOIN with WATERMARK and alias resolves columns correctly") {
+ createTestTables()
+ analyzeStreamSQL(
+ """
+ |SELECT *
+ |FROM
+ | STREAM(wm_signup_details) WATERMARK updated_at DELAY OF INTERVAL 60
SECONDS d
+ |JOIN
+ | STREAM(wm_profiles) WATERMARK updated_at DELAY OF INTERVAL 60
SECONDS p
+ |ON d.consumer_profile_id = p.id
+ | AND d.updated_at BETWEEN p.updated_at - INTERVAL '24 hours' AND
p.updated_at
+ |""".stripMargin)
+ }
+
+ test("STREAM with WATERMARK and alias resolves columns in WHERE") {
+ createTestTables()
+ analyzeStreamSQL(
+ """
+ |SELECT *
+ |FROM STREAM(wm_events) WATERMARK ts DELAY OF INTERVAL 10 SECONDS t
+ |WHERE t.a > 1
+ |""".stripMargin)
+ }
+
+ test("STREAM with WATERMARK and alias resolves columns in ORDER BY") {
+ createTestTables()
+ analyzeStreamSQL(
+ """
+ |SELECT *
+ |FROM STREAM(wm_events) WATERMARK ts DELAY OF INTERVAL 10 SECONDS d
+ |ORDER BY d.a
+ |""".stripMargin)
+ }
+
+ test("STREAM with computed WATERMARK expression resolves correctly") {
+ createTestTables()
+ analyzeStreamSQL(
+ """
+ |SELECT *
+ |FROM STREAM(wm_events) WATERMARK CAST(ts AS TIMESTAMP) AS wm_time
+ | DELAY OF INTERVAL 10 SECONDS d
+ |WHERE d.a > 1
+ |""".stripMargin)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
