You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "HeartSaVioR (via GitHub)" <gi...@apache.org> on 2025/09/24 01:47:00 UTC
[PR] [SPARK-53687][SQL][SS] Introduce WATERMARK clause in SQL statement [spark]
HeartSaVioR opened a new pull request, #52428:
URL: https://github.com/apache/spark/pull/52428
### 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443534749
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
+ extends StateStoreMetricsTest
+ with BeforeAndAfter
+ with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 1") {
Review Comment:
I meant Data Manipulation Language - I can just specify it as "select statement".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #52428:
URL: https://github.com/apache/spark/pull/52428#issuecomment-3431012435
I've moved the check from AstBuilder to the analyzer rule since it's likely to be read more time than the AstBuilder impl.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415565543
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala:
##########
@@ -566,6 +566,17 @@ package object dsl extends SQLConfHelper {
}
def deduplicate(colNames: Attribute*): LogicalPlan = Deduplicate(colNames, logicalPlan)
+
+ def watermark(expr: Expression, delayThreshold: CalendarInterval): LogicalPlan = {
+ val namedExpression = expr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+ val proj = Project(Seq(namedExpression, UnresolvedStar(None)), logicalPlan)
Review Comment:
dsl API operates on unresolved plans, so using `UnresolvedStar` is right here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415569361
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
Review Comment:
shall we also add a parser test? there are so many places we can put the WATERMARK clause and we haven't tested all of them yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2382229164
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala:
##########
@@ -1228,3 +1229,21 @@ case class UnresolvedExecuteImmediate(
final override val nodePatterns: Seq[TreePattern] = Seq(EXECUTE_IMMEDIATE)
}
+
+case class UnresolvedEventTimeWatermark(
+ eventTimeColExpr: Expression,
+ delayInterval: Literal,
+ child: LogicalPlan)
+ extends UnresolvedUnaryNode {
+
+ final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_EVENT_TIME_WATERMARK)
+
+ private val delay = IntervalUtils.fromIntervalString(delayInterval.toString)
Review Comment:
can we do it in the parser? This logical plan should take `CalendarInterval` directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2382231085
##########
sql/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.
Review Comment:
do we still need this hack?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443673014
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -927,6 +929,10 @@ lateralView
: LATERAL VIEW (OUTER)? qualifiedName LEFT_PAREN (expression (COMMA expression)*)? RIGHT_PAREN tblName=identifier (AS? colName+=identifier (COMMA colName+=identifier)*)?
;
+watermarkClause
+ : WATERMARK colName=namedExpression DELAY OF delay=interval
Review Comment:
or we can enforce it in `AstBuilder`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2390547671
##########
sql/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.
Review Comment:
Let me try removing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447381452
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.child.outputSet.contains(u.eventTimeColExpr)) {
+ // We don't need to have projection since the attribute being referenced will be available.
+ EventTimeWatermark(uuid, u.eventTimeColExpr.toAttribute, u.delay, u.child)
+ } else {
+ // 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
Review Comment:
nit: let's move it before the `if-else`, as we need it in the `if` branch as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2373850247
##########
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" : {
Review Comment:
TODO: Need to assign one error class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443626036
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -927,6 +929,10 @@ lateralView
: LATERAL VIEW (OUTER)? qualifiedName LEFT_PAREN (expression (COMMA expression)*)? RIGHT_PAREN tblName=identifier (AS? colName+=identifier (COMMA colName+=identifier)*)?
;
+watermarkClause
+ : WATERMARK colName=namedExpression DELAY OF delay=interval
Review Comment:
How do we map this to the DF API `def withWatermark(eventTime: String, delayThreshold: String)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443657127
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -927,6 +929,10 @@ lateralView
: LATERAL VIEW (OUTER)? qualifiedName LEFT_PAREN (expression (COMMA expression)*)? RIGHT_PAREN tblName=identifier (AS? colName+=identifier (COMMA colName+=identifier)*)?
;
+watermarkClause
+ : WATERMARK colName=namedExpression DELAY OF delay=interval
Review Comment:
if we really want to enforce the name
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2448122301
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala:
##########
@@ -566,6 +566,23 @@ package object dsl extends SQLConfHelper {
}
def deduplicate(colNames: Attribute*): LogicalPlan = Deduplicate(colNames, logicalPlan)
+
+ def watermark(expr: Expression, delayThreshold: CalendarInterval): LogicalPlan = {
+ val namedExpression = expr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
Review Comment:
I'll just remove the method since it's unused at this point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2375142301
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
+ extends StateStoreMetricsTest
+ with BeforeAndAfter
+ with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 1") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ .withColumn("eventTime", timestamp_seconds(col("value")))
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src WATERMARK eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 2") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src
+ | WATERMARK timestamp_seconds(value) AS eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ private def testWindowedAggregation(
+ inputData: MemoryStream[Int],
+ dataFrame: DataFrame): Unit = {
+ testStream(dataFrame)(
+ AddData(inputData, 15),
+ CheckAnswer(),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputData, 10, 12, 14),
+ CheckAnswer(),
+ assertEventStats(min = 10, max = 14, avg = 12, wtrmark = 5),
+ AddData(inputData, 25),
+ CheckAnswer((10, 3)),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
+
+ test("stream-stream join with watermark in select DML - case 1") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+
+ val df1 = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time") as "leftTime",
+ ($"leftKey" * 2) as "leftValue")
+ val df2 = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time") as "rightTime",
+ ($"rightKey" * 3) as "rightValue")
+
+ df1.createOrReplaceTempView("stream_left")
+ df2.createOrReplaceTempView("stream_right")
+
+ val joined = spark.sql(
+ """
+ |SELECT
+ | leftKey, rightKey, CAST(leftTime AS INTEGER), CAST(rightTime AS INTEGER)
+ |FROM
+ | stream_left WATERMARK leftTime DELAY OF INTERVAL 0 second
+ |FULL OUTER JOIN
+ | stream_right WATERMARK rightTime DELAY OF INTERVAL 0 second
Review Comment:
No, that's just for simplification. You can have different definition of watermark including delay threshold. e.g. you can define watermark delay to 10 minutes in stream_left while defining delay to 5 minutes in stream_right.
The corresponding DataFrame query is following:
stream_left <- streaming DF
stream_right <- streaming DF
```
val leftDf = stream_left.withWatermark("leftTime, "0 second")
val rightDf = stream_right.withWatermark("rightTime", "0 second")
val query = stream.left.join(stream_right, expr("leftKey = rightKey AND leftTime BETWEEN rightTime - INTERVAL 5 SECONDS"), "full_outer")
query.writeStream.blabla.start()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2375379888
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
+ extends StateStoreMetricsTest
+ with BeforeAndAfter
+ with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 1") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ .withColumn("eventTime", timestamp_seconds(col("value")))
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src WATERMARK eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 2") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src
+ | WATERMARK timestamp_seconds(value) AS eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ private def testWindowedAggregation(
+ inputData: MemoryStream[Int],
+ dataFrame: DataFrame): Unit = {
+ testStream(dataFrame)(
+ AddData(inputData, 15),
+ CheckAnswer(),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputData, 10, 12, 14),
+ CheckAnswer(),
+ assertEventStats(min = 10, max = 14, avg = 12, wtrmark = 5),
+ AddData(inputData, 25),
+ CheckAnswer((10, 3)),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
+
+ test("stream-stream join with watermark in select DML - case 1") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+
+ val df1 = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time") as "leftTime",
+ ($"leftKey" * 2) as "leftValue")
+ val df2 = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time") as "rightTime",
+ ($"rightKey" * 3) as "rightValue")
+
+ df1.createOrReplaceTempView("stream_left")
+ df2.createOrReplaceTempView("stream_right")
+
+ val joined = spark.sql(
+ """
+ |SELECT
+ | leftKey, rightKey, CAST(leftTime AS INTEGER), CAST(rightTime AS INTEGER)
+ |FROM
+ | stream_left WATERMARK leftTime DELAY OF INTERVAL 0 second
+ |FULL OUTER JOIN
+ | stream_right WATERMARK rightTime DELAY OF INTERVAL 0 second
Review Comment:
I have the impression that, with DataFrame API, we can set watermark for all streaming operators, not only the streaming scans. Is this true?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2374878689
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
+ extends StateStoreMetricsTest
+ with BeforeAndAfter
+ with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 1") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ .withColumn("eventTime", timestamp_seconds(col("value")))
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src WATERMARK eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 2") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src
+ | WATERMARK timestamp_seconds(value) AS eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ private def testWindowedAggregation(
+ inputData: MemoryStream[Int],
+ dataFrame: DataFrame): Unit = {
+ testStream(dataFrame)(
+ AddData(inputData, 15),
+ CheckAnswer(),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputData, 10, 12, 14),
+ CheckAnswer(),
+ assertEventStats(min = 10, max = 14, avg = 12, wtrmark = 5),
+ AddData(inputData, 25),
+ CheckAnswer((10, 3)),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
+
+ test("stream-stream join with watermark in select DML - case 1") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+
+ val df1 = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time") as "leftTime",
+ ($"leftKey" * 2) as "leftValue")
+ val df2 = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time") as "rightTime",
+ ($"rightKey" * 3) as "rightValue")
+
+ df1.createOrReplaceTempView("stream_left")
+ df2.createOrReplaceTempView("stream_right")
+
+ val joined = spark.sql(
+ """
+ |SELECT
+ | leftKey, rightKey, CAST(leftTime AS INTEGER), CAST(rightTime AS INTEGER)
+ |FROM
+ | stream_left WATERMARK leftTime DELAY OF INTERVAL 0 second
+ |FULL OUTER JOIN
+ | stream_right WATERMARK rightTime DELAY OF INTERVAL 0 second
Review Comment:
do we require the watermark to be the same between two join sides? what's the corresponding dataframe query?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415563887
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ if (u.child.outputSet.contains(namedExpression)) {
+ // We don't need to have projection since the attribute being referenced will be available.
+ EventTimeWatermark(uuid, namedExpression.toAttribute, u.delay, u.child)
+ } else {
+ // We need to inject projection as we can't find the matching column directly in the
+ // child output.
+ val proj = Project(Seq(namedExpression, UnresolvedStar(None)), u.child)
+ val attrRef = proj.projectList.head.toAttribute
Review Comment:
We can fix it this way: https://github.com/apache/spark/pull/52428/files#r2415563482
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415562097
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ if (u.child.outputSet.contains(namedExpression)) {
+ // We don't need to have projection since the attribute being referenced will be available.
+ EventTimeWatermark(uuid, namedExpression.toAttribute, u.delay, u.child)
+ } else {
+ // We need to inject projection as we can't find the matching column directly in the
+ // child output.
+ val proj = Project(Seq(namedExpression, UnresolvedStar(None)), u.child)
+ val attrRef = proj.projectList.head.toAttribute
Review Comment:
```suggestion
val attrRef = namedExpression.toAttribute
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415564923
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala:
##########
@@ -566,6 +566,17 @@ package object dsl extends SQLConfHelper {
}
def deduplicate(colNames: Attribute*): LogicalPlan = Deduplicate(colNames, logicalPlan)
+
+ def watermark(expr: Expression, delayThreshold: CalendarInterval): LogicalPlan = {
+ val namedExpression = expr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
Review Comment:
ditto for here: https://github.com/apache/spark/pull/52428/files#r2415563887
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447463411
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala:
##########
@@ -2017,4 +2017,167 @@ class PlanParserSuite extends AnalysisTest {
assert(unresolvedRelation2.options == CaseInsensitiveStringMap.empty)
assert(unresolvedRelation2.isStreaming)
}
+
+ test("watermark clause - table & attribute reference") {
+ assertEqual(
+ """
+ |SELECT *
+ |FROM testData
+ |WATERMARK ts DELAY OF INTERVAL 10 seconds AS tbl
+ |WHERE a > 1
+ |""".stripMargin,
+ table("testData")
+ .as("tbl")
+ .watermarkUnresolved(
+ UnresolvedAttribute("ts"),
+ IntervalUtils.fromIntervalString("INTERVAL 10 seconds"))
+ .where($"a" > 1)
+ .select(UnresolvedStar(None))
+ )
+ }
+
+ test("watermark clause - table & expression with alias") {
+ assertEqual(
+ """
+ |SELECT *
+ |FROM testData
+ |WATERMARK timestamp_seconds(value) AS eventTime DELAY OF INTERVAL 10 seconds AS tbl
+ |WHERE a > 1
+ |""".stripMargin,
+ table("testData")
+ .as("tbl")
+ .watermarkUnresolved(
+ Alias(
+ UnresolvedFunction(
+ Seq("timestamp_seconds"), Seq(UnresolvedAttribute("value")), isDistinct = false),
+ "eventTime")(),
+ IntervalUtils.fromIntervalString("INTERVAL 10 seconds"))
+ .where($"a" > 1)
+ .select(UnresolvedStar(None))
+ )
+ }
+
+ test("watermark clause - table & expression without alias") {
+ val sql =
+ """
+ |SELECT *
+ |FROM testData
+ |WATERMARK timestamp_seconds(value) DELAY OF INTERVAL 10 seconds AS tbl
+ |WHERE a > 1
+ |""".stripMargin
+ checkError(
+ parseException(sql),
+ condition = "REQUIRES_EXPLICIT_NAME_IN_WATERMARK_CLAUSE",
+ sqlState = "42000",
+ parameters = Map("sqlExpr" -> "timestamp_seconds(value)")
+ )
+ }
+
+ test("watermark clause - aliased query") {
+ assertEqual(
+ """
+ |SELECT *
+ |FROM
+ |(
+ | SELECT *
+ | FROM testData
+ |)
+ |WATERMARK ts DELAY OF INTERVAL 10 seconds AS tbl
+ |WHERE a > 1
+ |""".stripMargin,
+ table("testData")
+ .select(UnresolvedStar(None))
+ .as("tbl")
+ .watermarkUnresolved(
+ UnresolvedAttribute("ts"),
+ IntervalUtils.fromIntervalString("INTERVAL 10 seconds"))
+ .where($"a" > 1)
+ .select(UnresolvedStar(None))
+ )
+ }
+
+ test("watermark clause - subquery") {
+ assertEqual(
+ """
+ |SELECT key, time
+ |FROM
+ |(
+ | SELECT key, time
+ | FROM
+ | testData
+ | WATERMARK timestamp_seconds(ts) AS time DELAY OF INTERVAL 10 seconds
+ |)
+ |AS tbl
+ |WHERE key = 'a'
+ |""".stripMargin,
+ table("testData")
+ .watermarkUnresolved(
+ Alias(
+ UnresolvedFunction(
+ Seq("timestamp_seconds"), Seq(UnresolvedAttribute("ts")), isDistinct = false),
+ "time")(),
+ IntervalUtils.fromIntervalString("INTERVAL 10 seconds"))
+ .select($"key", $"time")
+ .as("tbl")
+ .where($"key" === "a")
+ .select($"key", $"time")
+ )
+ }
+
+ test("watermark clause - aliasedRelation") {
+ val src1 = UnresolvedRelation(TableIdentifier("src1")).as("s1")
+ val src2 = UnresolvedRelation(TableIdentifier("src2")).as("s2")
+ assertEqual(
+ """
+ |SELECT *
+ |FROM
+ |(src1 s1 INNER JOIN src2 s2 ON s1.id = s2.id)
Review Comment:
how is this different from `watermark clause - aliased query`? The subquery after `FROM` clause can be any query and we don't need to prove it again here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #52428: [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement
URL: https://github.com/apache/spark/pull/52428
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443709924
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/WatermarkClauseSuite.scala:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.command
+
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedFunction, UnresolvedRelation, UnresolvedStar, UnresolvedTableValuedFunction}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Literal}
+import org.apache.spark.sql.catalyst.plans.Inner
+import org.apache.spark.sql.test.SharedSparkSession
+
+class WatermarkClauseSuite extends AnalysisTest with SharedSparkSession {
+ test("watermark") {
+ // attribute reference vs expression
+
+ // tableName, attribute reference, alias
+ assertEqual(
Review Comment:
It's probably proper to have these tests in PlanParserSuite.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447469786
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermarkSuite.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.analysis.TestRelations.streamingRelation
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.unsafe.types.CalendarInterval
+
+class ResolveEventTimeWatermarkSuite extends AnalysisTest {
+ override protected def extendedAnalysisRules: Seq[Rule[LogicalPlan]] = {
+ ResolveEventTimeWatermark +: super.extendedAnalysisRules
+ }
+
+ test("event time column expr refers to the column in child") {
+ val planBeforeRule = streamingRelation
+ .watermarkUnresolved($"ts", new CalendarInterval(0, 0, 1000))
+
+ val analyzed = getAnalyzer.execute(planBeforeRule)
+
+ // EventTimeWatermark node has UUID, hence we can't simply compare the plan
+ // with expected shape of plan as a whole.
+ val uuid = java.util.UUID.randomUUID()
+
+ val uuidInjectedAnalyzed = analyzed.transform {
+ case e: EventTimeWatermark => e.copy(nodeId = uuid)
+ }
+
+ comparePlans(
+ uuidInjectedAnalyzed,
+ EventTimeWatermark(
+ uuid,
+ streamingRelation.output.find(_.name == "ts").head,
+ new CalendarInterval(0, 0, 1000),
+ streamingRelation
+ )
Review Comment:
and shall we use `withWatermark` in the dsl API?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2448278192
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala:
##########
@@ -2017,4 +2017,167 @@ class PlanParserSuite extends AnalysisTest {
assert(unresolvedRelation2.options == CaseInsensitiveStringMap.empty)
assert(unresolvedRelation2.isStreaming)
}
+
+ test("watermark clause - table & attribute reference") {
+ assertEqual(
+ """
+ |SELECT *
+ |FROM testData
+ |WATERMARK ts DELAY OF INTERVAL 10 seconds AS tbl
+ |WHERE a > 1
+ |""".stripMargin,
+ table("testData")
+ .as("tbl")
+ .watermarkUnresolved(
+ UnresolvedAttribute("ts"),
+ IntervalUtils.fromIntervalString("INTERVAL 10 seconds"))
+ .where($"a" > 1)
+ .select(UnresolvedStar(None))
+ )
+ }
+
+ test("watermark clause - table & expression with alias") {
+ assertEqual(
+ """
+ |SELECT *
+ |FROM testData
+ |WATERMARK timestamp_seconds(value) AS eventTime DELAY OF INTERVAL 10 seconds AS tbl
+ |WHERE a > 1
+ |""".stripMargin,
+ table("testData")
+ .as("tbl")
+ .watermarkUnresolved(
+ Alias(
+ UnresolvedFunction(
+ Seq("timestamp_seconds"), Seq(UnresolvedAttribute("value")), isDistinct = false),
+ "eventTime")(),
+ IntervalUtils.fromIntervalString("INTERVAL 10 seconds"))
+ .where($"a" > 1)
+ .select(UnresolvedStar(None))
+ )
+ }
+
+ test("watermark clause - table & expression without alias") {
+ val sql =
+ """
+ |SELECT *
+ |FROM testData
+ |WATERMARK timestamp_seconds(value) DELAY OF INTERVAL 10 seconds AS tbl
+ |WHERE a > 1
+ |""".stripMargin
+ checkError(
+ parseException(sql),
+ condition = "REQUIRES_EXPLICIT_NAME_IN_WATERMARK_CLAUSE",
+ sqlState = "42000",
+ parameters = Map("sqlExpr" -> "timestamp_seconds(value)")
+ )
+ }
+
+ test("watermark clause - aliased query") {
+ assertEqual(
+ """
+ |SELECT *
+ |FROM
+ |(
+ | SELECT *
+ | FROM testData
+ |)
+ |WATERMARK ts DELAY OF INTERVAL 10 seconds AS tbl
+ |WHERE a > 1
+ |""".stripMargin,
+ table("testData")
+ .select(UnresolvedStar(None))
+ .as("tbl")
+ .watermarkUnresolved(
+ UnresolvedAttribute("ts"),
+ IntervalUtils.fromIntervalString("INTERVAL 10 seconds"))
+ .where($"a" > 1)
+ .select(UnresolvedStar(None))
+ )
+ }
+
+ test("watermark clause - subquery") {
+ assertEqual(
+ """
+ |SELECT key, time
+ |FROM
+ |(
+ | SELECT key, time
+ | FROM
+ | testData
+ | WATERMARK timestamp_seconds(ts) AS time DELAY OF INTERVAL 10 seconds
+ |)
+ |AS tbl
+ |WHERE key = 'a'
+ |""".stripMargin,
+ table("testData")
+ .watermarkUnresolved(
+ Alias(
+ UnresolvedFunction(
+ Seq("timestamp_seconds"), Seq(UnresolvedAttribute("ts")), isDistinct = false),
+ "time")(),
+ IntervalUtils.fromIntervalString("INTERVAL 10 seconds"))
+ .select($"key", $"time")
+ .as("tbl")
+ .where($"key" === "a")
+ .select($"key", $"time")
+ )
+ }
+
+ test("watermark clause - aliasedRelation") {
+ val src1 = UnresolvedRelation(TableIdentifier("src1")).as("s1")
+ val src2 = UnresolvedRelation(TableIdentifier("src2")).as("s2")
+ assertEqual(
+ """
+ |SELECT *
+ |FROM
+ |(src1 s1 INNER JOIN src2 s2 ON s1.id = s2.id)
Review Comment:
It's just to cover the case in the parser rule one by one - I can remove this if you think this is redundant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2375136170
##########
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" : {
Review Comment:
TODO: remove this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2376353965
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -995,9 +1001,11 @@ identifierComment
relationPrimary
: streamRelationPrimary #streamRelation
| identifierReference temporalClause?
- 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
Review Comment:
`watermarkClause` is already defined in `streamRelationPrimary`. Do we still need it here? Is it also applied for non-stream relation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2377576930
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -1042,10 +1052,13 @@ functionTableArgument
| functionArgument
;
+// This is only used in relationPrimary where having watermarkClause makes sense. If this becomes
+// referred by other clause, please check wheter watermarkClause makes sense to the clause.
+// If not, consider separate this rule.
functionTable
: funcName=functionName LEFT_PAREN
(functionTableArgument (COMMA functionTableArgument)*)?
- RIGHT_PAREN tableAlias
+ RIGHT_PAREN watermarkClause? tableAlias
Review Comment:
Ouch I think it's missed in that doc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415563482
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
Review Comment:
This alias is not user-facing and doesn't matter. We can simply do `Alias(e, "watermark_col")()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415566485
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala:
##########
@@ -595,9 +595,8 @@ class Dataset[T] private[sql](
val parsedDelay = IntervalUtils.fromIntervalString(delayThreshold)
require(!IntervalUtils.isNegative(parsedDelay),
s"delay threshold ($delayThreshold) should not be negative.")
- EliminateEventTimeWatermark(
Review Comment:
what did `EliminateEventTimeWatermark` do before?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2390544287
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ val isAttributeReference = namedExpression.isInstanceOf[AttributeReference]
+ val exprInChildOutput = u.child.output.exists(_.name == namedExpression.name)
Review Comment:
So when eventTimeColExpr is resolved, the expression can refer to the column by exprId? If then yes we can do that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443532862
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala:
##########
@@ -566,6 +566,17 @@ package object dsl extends SQLConfHelper {
}
def deduplicate(colNames: Attribute*): LogicalPlan = Deduplicate(colNames, logicalPlan)
+
+ def watermark(expr: Expression, delayThreshold: CalendarInterval): LogicalPlan = {
+ val namedExpression = expr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
Review Comment:
Ditto, we are not assigning the expression to "specific" column name. We expect the expression to be either an alias or column reference. It's more likely that this is not resolved by analyzer here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2448287449
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala:
##########
@@ -566,6 +566,23 @@ package object dsl extends SQLConfHelper {
}
def deduplicate(colNames: Attribute*): LogicalPlan = Deduplicate(colNames, logicalPlan)
+
+ def watermark(expr: Expression, delayThreshold: CalendarInterval): LogicalPlan = {
+ val namedExpression = expr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
Review Comment:
I re-added the method but changed the signature to receive NamedExpression to simplify the thing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #52428:
URL: https://github.com/apache/spark/pull/52428#issuecomment-3433360642
the docker test failure is unrelated, thanks, merging to master!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2378603067
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -2076,6 +2077,38 @@ class AstBuilder extends DataTypeAstBuilder
query)
}
+ /**
+ * Add an [[EventTimeWatermark]] to a logical plan.
+ */
+ private def withWatermark(
+ ctx: WatermarkClauseContext,
+ query: LogicalPlan): LogicalPlan = withOrigin(ctx) {
+ val expression = visitNamedExpression(ctx.namedExpression())
+
+ if (expression.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = expression match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ // 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?
Review Comment:
I think we need a new logical plan like `UnresolvedEventTimeWatermark`, which will be rewritten into `EventTimeWatermark` + `Project` during analysis.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #52428:
URL: https://github.com/apache/spark/pull/52428#issuecomment-3326162065
cc. @anishshri-db as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2377575979
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -995,9 +1001,11 @@ identifierComment
relationPrimary
: streamRelationPrimary #streamRelation
| identifierReference temporalClause?
- 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
Review Comment:
relation here can be a temp view which could be technically streaming without STREAM keyword
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2375142301
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
+ extends StateStoreMetricsTest
+ with BeforeAndAfter
+ with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 1") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ .withColumn("eventTime", timestamp_seconds(col("value")))
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src WATERMARK eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 2") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src
+ | WATERMARK timestamp_seconds(value) AS eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ private def testWindowedAggregation(
+ inputData: MemoryStream[Int],
+ dataFrame: DataFrame): Unit = {
+ testStream(dataFrame)(
+ AddData(inputData, 15),
+ CheckAnswer(),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputData, 10, 12, 14),
+ CheckAnswer(),
+ assertEventStats(min = 10, max = 14, avg = 12, wtrmark = 5),
+ AddData(inputData, 25),
+ CheckAnswer((10, 3)),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
+
+ test("stream-stream join with watermark in select DML - case 1") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+
+ val df1 = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time") as "leftTime",
+ ($"leftKey" * 2) as "leftValue")
+ val df2 = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time") as "rightTime",
+ ($"rightKey" * 3) as "rightValue")
+
+ df1.createOrReplaceTempView("stream_left")
+ df2.createOrReplaceTempView("stream_right")
+
+ val joined = spark.sql(
+ """
+ |SELECT
+ | leftKey, rightKey, CAST(leftTime AS INTEGER), CAST(rightTime AS INTEGER)
+ |FROM
+ | stream_left WATERMARK leftTime DELAY OF INTERVAL 0 second
+ |FULL OUTER JOIN
+ | stream_right WATERMARK rightTime DELAY OF INTERVAL 0 second
Review Comment:
No, that's just for simplification. You can have different definition of watermark including delay threshold. e.g. you can define watermark delay to 10 minutes in stream_left while defining delay to 5 minutes in stream_right.
The corresponding DataFrame query is following:
stream_left <- streaming DF
stream_right <- streaming DF
```
val leftDf = stream_left.withWatermark("leftTime, "0 second")
val rightDf = stream_right.withWatermark("rightTime", "0 second")
val query = leftDf.join(
rightDf,
expr("leftKey = rightKey AND leftTime BETWEEN rightTime - INTERVAL 5 SECONDS"),
"full_outer")
query.writeStream.blabla.start()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2375749544
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
+ extends StateStoreMetricsTest
+ with BeforeAndAfter
+ with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 1") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ .withColumn("eventTime", timestamp_seconds(col("value")))
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src WATERMARK eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 2") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src
+ | WATERMARK timestamp_seconds(value) AS eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ private def testWindowedAggregation(
+ inputData: MemoryStream[Int],
+ dataFrame: DataFrame): Unit = {
+ testStream(dataFrame)(
+ AddData(inputData, 15),
+ CheckAnswer(),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputData, 10, 12, 14),
+ CheckAnswer(),
+ assertEventStats(min = 10, max = 14, avg = 12, wtrmark = 5),
+ AddData(inputData, 25),
+ CheckAnswer((10, 3)),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
+
+ test("stream-stream join with watermark in select DML - case 1") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+
+ val df1 = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time") as "leftTime",
+ ($"leftKey" * 2) as "leftValue")
+ val df2 = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time") as "rightTime",
+ ($"rightKey" * 3) as "rightValue")
+
+ df1.createOrReplaceTempView("stream_left")
+ df2.createOrReplaceTempView("stream_right")
+
+ val joined = spark.sql(
+ """
+ |SELECT
+ | leftKey, rightKey, CAST(leftTime AS INTEGER), CAST(rightTime AS INTEGER)
+ |FROM
+ | stream_left WATERMARK leftTime DELAY OF INTERVAL 0 second
+ |FULL OUTER JOIN
+ | stream_right WATERMARK rightTime DELAY OF INTERVAL 0 second
Review Comment:
You can set watermark "before" the first stateful operator has appeared. But conceptually, it is advised to define watermark closer to stream scan. The concept of watermark is to mark that the engine has seen (processed) all the data before the timestamp ts, which is closely related to the input data from stream scan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2382224276
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ val isAttributeReference = namedExpression.isInstanceOf[AttributeReference]
+ val exprInChildOutput = u.child.output.exists(_.name == namedExpression.name)
Review Comment:
it's weird to search by name for resolved attributes, we can just do `u.child.outputSet.contains(namedExpression)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443628450
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/WatermarkClauseSuite.scala:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.command
+
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedFunction, UnresolvedRelation, UnresolvedStar, UnresolvedTableValuedFunction}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Literal}
+import org.apache.spark.sql.catalyst.plans.Inner
+import org.apache.spark.sql.test.SharedSparkSession
+
+class WatermarkClauseSuite extends AnalysisTest with SharedSparkSession {
+ test("watermark") {
+ // attribute reference vs expression
+
+ // tableName, attribute reference, alias
+ assertEqual(
Review Comment:
where is `assertEqual` defined?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2446835312
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -927,6 +929,10 @@ lateralView
: LATERAL VIEW (OUTER)? qualifiedName LEFT_PAREN (expression (COMMA expression)*)? RIGHT_PAREN tblName=identifier (AS? colName+=identifier (COMMA colName+=identifier)*)?
;
+watermarkClause
+ : WATERMARK colName=namedExpression DELAY OF delay=interval
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447389494
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala:
##########
@@ -566,6 +566,23 @@ package object dsl extends SQLConfHelper {
}
def deduplicate(colNames: Attribute*): LogicalPlan = Deduplicate(colNames, logicalPlan)
+
+ def watermark(expr: Expression, delayThreshold: CalendarInterval): LogicalPlan = {
+ val namedExpression = expr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
Review Comment:
shall we fail here? It's a test and we should require the test to name the event time expr always
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447412550
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermarkSuite.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.analysis.TestRelations.streamingRelation
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.unsafe.types.CalendarInterval
+
+class ResolveEventTimeWatermarkSuite extends AnalysisTest {
+ override protected def extendedAnalysisRules: Seq[Rule[LogicalPlan]] = {
+ ResolveEventTimeWatermark +: super.extendedAnalysisRules
+ }
+
+ test("event time column expr refers to the column in child") {
+ val planBeforeRule = streamingRelation
+ .watermarkUnresolved($"ts", new CalendarInterval(0, 0, 1000))
+
+ val analyzed = getAnalyzer.execute(planBeforeRule)
+
+ // EventTimeWatermark node has UUID, hence we can't simply compare the plan
+ // with expected shape of plan as a whole.
+ val uuid = java.util.UUID.randomUUID()
+
+ val uuidInjectedAnalyzed = analyzed.transform {
+ case e: EventTimeWatermark => e.copy(nodeId = uuid)
+ }
+
+ comparePlans(
+ uuidInjectedAnalyzed,
+ EventTimeWatermark(
+ uuid,
+ streamingRelation.output.find(_.name == "ts").head,
Review Comment:
```suggestion
streamingRelation.output(1),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #52428:
URL: https://github.com/apache/spark/pull/52428#issuecomment-3326158914
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2375142301
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
+ extends StateStoreMetricsTest
+ with BeforeAndAfter
+ with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 1") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ .withColumn("eventTime", timestamp_seconds(col("value")))
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src WATERMARK eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 2") {
+ // All event time metrics where watermarking is set
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF()
+ df.createOrReplaceTempView("stream_src")
+ val aggWithWatermark = spark.sql(
+ """
+ |SELECT
+ | CAST(window.start AS LONG), CAST(count(*) AS LONG) AS count
+ |FROM
+ | stream_src
+ | WATERMARK timestamp_seconds(value) AS eventTime DELAY OF INTERVAL 10 seconds
+ |GROUP BY window(eventTime, '5 seconds')
+ |""".stripMargin)
+
+ testWindowedAggregation(inputData, aggWithWatermark)
+ }
+
+ private def testWindowedAggregation(
+ inputData: MemoryStream[Int],
+ dataFrame: DataFrame): Unit = {
+ testStream(dataFrame)(
+ AddData(inputData, 15),
+ CheckAnswer(),
+ assertEventStats(min = 15, max = 15, avg = 15, wtrmark = 0),
+ AddData(inputData, 10, 12, 14),
+ CheckAnswer(),
+ assertEventStats(min = 10, max = 14, avg = 12, wtrmark = 5),
+ AddData(inputData, 25),
+ CheckAnswer((10, 3)),
+ assertEventStats(min = 25, max = 25, avg = 25, wtrmark = 5)
+ )
+ }
+
+ test("stream-stream join with watermark in select DML - case 1") {
+ val leftInput = MemoryStream[(Int, Int)]
+ val rightInput = MemoryStream[(Int, Int)]
+
+ val df1 = leftInput.toDF().toDF("leftKey", "time")
+ .select($"leftKey", timestamp_seconds($"time") as "leftTime",
+ ($"leftKey" * 2) as "leftValue")
+ val df2 = rightInput.toDF().toDF("rightKey", "time")
+ .select($"rightKey", timestamp_seconds($"time") as "rightTime",
+ ($"rightKey" * 3) as "rightValue")
+
+ df1.createOrReplaceTempView("stream_left")
+ df2.createOrReplaceTempView("stream_right")
+
+ val joined = spark.sql(
+ """
+ |SELECT
+ | leftKey, rightKey, CAST(leftTime AS INTEGER), CAST(rightTime AS INTEGER)
+ |FROM
+ | stream_left WATERMARK leftTime DELAY OF INTERVAL 0 second
+ |FULL OUTER JOIN
+ | stream_right WATERMARK rightTime DELAY OF INTERVAL 0 second
Review Comment:
No, that's just for simplification. You can have different definition of watermark including delay threshold. e.g. you can define watermark delay to 10 minutes in stream_left while defining delay to 5 minutes in stream_right.
The corresponding DataFrame query is following:
stream_left <- streaming DF
stream_right <- streaming DF
```
val leftDf = stream_left.withWatermark("leftTime, "0 second")
val rightDf = stream_right.withWatermark("rightTime", "0 second")
val query = stream.left.join(
stream_right,
expr("leftKey = rightKey AND leftTime BETWEEN rightTime - INTERVAL 5 SECONDS"),
"full_outer")
query.writeStream.blabla.start()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415569718
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
Review Comment:
example: `DropTableParserSuite`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2380631389
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -2076,6 +2077,38 @@ class AstBuilder extends DataTypeAstBuilder
query)
}
+ /**
+ * Add an [[EventTimeWatermark]] to a logical plan.
+ */
+ private def withWatermark(
+ ctx: WatermarkClauseContext,
+ query: LogicalPlan): LogicalPlan = withOrigin(ctx) {
+ val expression = visitNamedExpression(ctx.namedExpression())
+
+ if (expression.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = expression match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ // 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?
Review Comment:
Addressed. PTAL!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443671608
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -927,6 +929,10 @@ lateralView
: LATERAL VIEW (OUTER)? qualifiedName LEFT_PAREN (expression (COMMA expression)*)? RIGHT_PAREN tblName=identifier (AS? colName+=identifier (COMMA colName+=identifier)*)?
;
+watermarkClause
+ : WATERMARK colName=namedExpression DELAY OF delay=interval
Review Comment:
Will do.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447398442
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -2076,6 +2077,32 @@ class AstBuilder extends DataTypeAstBuilder
query)
}
+ /**
+ * Add an [[EventTimeWatermark]] to a logical plan.
+ */
+ private def withWatermark(
+ ctx: WatermarkClauseContext,
+ query: LogicalPlan): LogicalPlan = withOrigin(ctx) {
+ val expression = visitNamedExpression(ctx.namedExpression())
+
+ val namedExpression = expression match {
+ case e: NamedExpression => e
Review Comment:
have you tested with something like `a.b`? We should probably name it implicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443531113
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #52428:
URL: https://github.com/apache/spark/pull/52428#issuecomment-3434327364
Thanks for your thoughtful review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415568042
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/EventTimeWatermarkWithWatermarkDefInSelectSuite.scala:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 java.{util => ju}
+import java.text.SimpleDateFormat
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.UTC
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{col, timestamp_seconds}
+import org.apache.spark.sql.streaming.StateStoreMetricsTest
+
+class EventTimeWatermarkWithWatermarkDefInSelectSuite
+ extends StateStoreMetricsTest
+ with BeforeAndAfter
+ with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ test("event time and watermark metrics with watermark in select DML - case 1") {
Review Comment:
what does DML mean here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2376394555
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -1042,10 +1052,13 @@ functionTableArgument
| functionArgument
;
+// This is only used in relationPrimary where having watermarkClause makes sense. If this becomes
+// referred by other clause, please check wheter watermarkClause makes sense to the clause.
+// If not, consider separate this rule.
functionTable
: funcName=functionName LEFT_PAREN
(functionTableArgument (COMMA functionTableArgument)*)?
- RIGHT_PAREN tableAlias
+ RIGHT_PAREN watermarkClause? tableAlias
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2380688242
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ val isAttributeReference = namedExpression.isInstanceOf[AttributeReference]
Review Comment:
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2382222425
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.childrenResolved =>
Review Comment:
shall we wait for `u.eventTimeColExpr.resolved`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443525865
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
Review Comment:
The watermark column "name" is determined by the expression, not something we use the reserved name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443656539
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -927,6 +929,10 @@ lateralView
: LATERAL VIEW (OUTER)? qualifiedName LEFT_PAREN (expression (COMMA expression)*)? RIGHT_PAREN tblName=identifier (AS? colName+=identifier (COMMA colName+=identifier)*)?
;
+watermarkClause
+ : WATERMARK colName=namedExpression DELAY OF delay=interval
Review Comment:
BTW, `namedExpression` does not guarantee the expression to be named, we should explicit match `multipartIdentifier | expression AS errorCapturingIdentifier`
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -927,6 +929,10 @@ lateralView
: LATERAL VIEW (OUTER)? qualifiedName LEFT_PAREN (expression (COMMA expression)*)? RIGHT_PAREN tblName=identifier (AS? colName+=identifier (COMMA colName+=identifier)*)?
;
+watermarkClause
+ : WATERMARK colName=namedExpression DELAY OF delay=interval
Review Comment:
BTW, `namedExpression` does not guarantee the expression to be named, we should explicitly match `multipartIdentifier | expression AS errorCapturingIdentifier`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447444293
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermarkSuite.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.analysis.TestRelations.streamingRelation
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.unsafe.types.CalendarInterval
+
+class ResolveEventTimeWatermarkSuite extends AnalysisTest {
+ override protected def extendedAnalysisRules: Seq[Rule[LogicalPlan]] = {
+ ResolveEventTimeWatermark +: super.extendedAnalysisRules
+ }
+
+ test("event time column expr refers to the column in child") {
+ val planBeforeRule = streamingRelation
+ .watermarkUnresolved($"ts", new CalendarInterval(0, 0, 1000))
+
+ val analyzed = getAnalyzer.execute(planBeforeRule)
+
+ // EventTimeWatermark node has UUID, hence we can't simply compare the plan
+ // with expected shape of plan as a whole.
+ val uuid = java.util.UUID.randomUUID()
+
+ val uuidInjectedAnalyzed = analyzed.transform {
+ case e: EventTimeWatermark => e.copy(nodeId = uuid)
+ }
+
+ comparePlans(
+ uuidInjectedAnalyzed,
+ EventTimeWatermark(
+ uuid,
+ streamingRelation.output.find(_.name == "ts").head,
+ new CalendarInterval(0, 0, 1000),
+ streamingRelation
+ )
+ )
+ }
+
+ test("event time column expr deduces a new column from alias") {
+ val planBeforeRule = streamingRelation
+ .watermarkUnresolved(
+ Alias(
+ UnresolvedFunction(
+ Seq("timestamp_seconds"), Seq(UnresolvedAttribute("a")), isDistinct = false),
+ "event_time"
+ )(),
+ new CalendarInterval(0, 0, 1000))
+
+ val analyzed = getAnalyzer.execute(planBeforeRule)
+
+ // EventTimeWatermark node has UUID, hence we can't simply compare the plan
+ // with expected shape of plan as a whole.
+ val uuid = java.util.UUID.randomUUID()
+
+ val uuidInjectedAnalyzed = analyzed.transform {
+ case e: EventTimeWatermark => e.copy(nodeId = uuid)
+ }
+
+ val selectAsResolved = getAnalyzer.execute(
+ streamingRelation
+ .select(
+ Alias(
+ UnresolvedFunction(
+ Seq("timestamp_seconds"), Seq(UnresolvedAttribute("a")), isDistinct = false),
+ "event_time"
+ )(),
+ // `*` will be resolved to `a`, `ts`
+ streamingRelation.output.head,
+ streamingRelation.output(1)
+ )
+ )
+
+ val expectedLogicalPlan = EventTimeWatermark(
+ uuid,
+ selectAsResolved.output.find(_.name == "event_time").head,
+ new CalendarInterval(0, 0, 1000),
+ selectAsResolved
+ )
+ val expectedAnalyzed = getAnalyzer.execute(expectedLogicalPlan)
Review Comment:
ditto here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2448145209
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -2076,6 +2077,32 @@ class AstBuilder extends DataTypeAstBuilder
query)
}
+ /**
+ * Add an [[EventTimeWatermark]] to a logical plan.
+ */
+ private def withWatermark(
+ ctx: WatermarkClauseContext,
+ query: LogicalPlan): LogicalPlan = withOrigin(ctx) {
+ val expression = visitNamedExpression(ctx.namedExpression())
+
+ val namedExpression = expression match {
+ case e: NamedExpression => e
Review Comment:
I haven't checked, but we need to make sure the column we are marking as event time to be a top level column. For that, should we still force users to alias it explicitly if they refer to nested column?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2390539165
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ val isAttributeReference = namedExpression.isInstanceOf[AttributeReference]
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443534298
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala:
##########
@@ -595,9 +595,8 @@ class Dataset[T] private[sql](
val parsedDelay = IntervalUtils.fromIntervalString(delayThreshold)
require(!IntervalUtils.isNegative(parsedDelay),
s"delay threshold ($delayThreshold) should not be negative.")
- EliminateEventTimeWatermark(
Review Comment:
We eliminate the EventTimeWatermark node for batch query since it's not needed.
We already moved this to the rule of analyzer; we just forgot to remove it here.
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L528-L529
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2443675235
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/WatermarkClauseSuite.scala:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.command
+
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedFunction, UnresolvedRelation, UnresolvedStar, UnresolvedTableValuedFunction}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Literal}
+import org.apache.spark.sql.catalyst.plans.Inner
+import org.apache.spark.sql.test.SharedSparkSession
+
+class WatermarkClauseSuite extends AnalysisTest with SharedSparkSession {
+ test("watermark") {
+ // attribute reference vs expression
+
+ // tableName, attribute reference, alias
+ assertEqual(
Review Comment:
My bad. I'll push the fix sooner.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447377175
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
Review Comment:
We can fail it earlier in the parser
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447436192
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermarkSuite.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.analysis.TestRelations.streamingRelation
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.unsafe.types.CalendarInterval
+
+class ResolveEventTimeWatermarkSuite extends AnalysisTest {
+ override protected def extendedAnalysisRules: Seq[Rule[LogicalPlan]] = {
+ ResolveEventTimeWatermark +: super.extendedAnalysisRules
+ }
+
+ test("event time column expr refers to the column in child") {
+ val planBeforeRule = streamingRelation
+ .watermarkUnresolved($"ts", new CalendarInterval(0, 0, 1000))
+
+ val analyzed = getAnalyzer.execute(planBeforeRule)
+
+ // EventTimeWatermark node has UUID, hence we can't simply compare the plan
+ // with expected shape of plan as a whole.
+ val uuid = java.util.UUID.randomUUID()
+
+ val uuidInjectedAnalyzed = analyzed.transform {
+ case e: EventTimeWatermark => e.copy(nodeId = uuid)
+ }
+
+ comparePlans(
+ uuidInjectedAnalyzed,
+ EventTimeWatermark(
+ uuid,
+ streamingRelation.output.find(_.name == "ts").head,
+ new CalendarInterval(0, 0, 1000),
+ streamingRelation
+ )
Review Comment:
```suggestion
EventTimeWatermark(
uuid,
$"ts",
new CalendarInterval(0, 0, 1000),
streamingRelation
).analyze
```
`.analyze` invokes the default analyzer and won't resolve watermarks, which is a perfect match for this test to construct the expected plan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2447412550
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermarkSuite.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.analysis.TestRelations.streamingRelation
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.unsafe.types.CalendarInterval
+
+class ResolveEventTimeWatermarkSuite extends AnalysisTest {
+ override protected def extendedAnalysisRules: Seq[Rule[LogicalPlan]] = {
+ ResolveEventTimeWatermark +: super.extendedAnalysisRules
+ }
+
+ test("event time column expr refers to the column in child") {
+ val planBeforeRule = streamingRelation
+ .watermarkUnresolved($"ts", new CalendarInterval(0, 0, 1000))
+
+ val analyzed = getAnalyzer.execute(planBeforeRule)
+
+ // EventTimeWatermark node has UUID, hence we can't simply compare the plan
+ // with expected shape of plan as a whole.
+ val uuid = java.util.UUID.randomUUID()
+
+ val uuidInjectedAnalyzed = analyzed.transform {
+ case e: EventTimeWatermark => e.copy(nodeId = uuid)
+ }
+
+ comparePlans(
+ uuidInjectedAnalyzed,
+ EventTimeWatermark(
+ uuid,
+ streamingRelation.output.find(_.name == "ts").head,
Review Comment:
```suggestion
streamingRelation.output(1),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415560436
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
Review Comment:
nit: put this check at the beginning (before `val uuid = java.util.UUID.randomUUID()`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415562655
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ if (u.child.outputSet.contains(namedExpression)) {
+ // We don't need to have projection since the attribute being referenced will be available.
+ EventTimeWatermark(uuid, namedExpression.toAttribute, u.delay, u.child)
+ } else {
+ // We need to inject projection as we can't find the matching column directly in the
+ // child output.
+ val proj = Project(Seq(namedExpression, UnresolvedStar(None)), u.child)
+ val attrRef = proj.projectList.head.toAttribute
Review Comment:
wait, we can't do it because we haven't resolved the `UnresolvedAlias` yet
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2415561822
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.eventTimeColExpr.resolved && u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ if (u.child.outputSet.contains(namedExpression)) {
+ // We don't need to have projection since the attribute being referenced will be available.
+ EventTimeWatermark(uuid, namedExpression.toAttribute, u.delay, u.child)
+ } else {
+ // We need to inject projection as we can't find the matching column directly in the
+ // child output.
+ val proj = Project(Seq(namedExpression, UnresolvedStar(None)), u.child)
Review Comment:
```suggestion
val proj = Project(Seq(namedExpression, u.child.output), u.child)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53687][SQL][SS][SDP] Introduce WATERMARK clause in SQL statement [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #52428:
URL: https://github.com/apache/spark/pull/52428#discussion_r2382224978
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveEventTimeWatermark.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern
+
+/**
+ * Resolve [[UnresolvedEventTimeWatermark]] to [[EventTimeWatermark]].
+ */
+object ResolveEventTimeWatermark extends Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(TreePattern.UNRESOLVED_EVENT_TIME_WATERMARK), ruleId) {
+
+ case u: UnresolvedEventTimeWatermark if u.childrenResolved =>
+ val uuid = java.util.UUID.randomUUID()
+
+ if (u.eventTimeColExpr.isInstanceOf[MultiAlias]) {
+ throw new AnalysisException(
+ errorClass = "CANNOT_USE_MULTI_ALIASES_IN_WATERMARK_CLAUSE",
+ messageParameters = Map()
+ )
+ }
+
+ val namedExpression = u.eventTimeColExpr match {
+ case e: NamedExpression => e
+ case e: Expression => UnresolvedAlias(e)
+ }
+
+ val isAttributeReference = namedExpression.isInstanceOf[AttributeReference]
Review Comment:
it doesn't matter. We can also always add the Project, as optimizer can remove it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org