HeartSaVioR commented on a change in pull request #35526:
URL: https://github.com/apache/spark/pull/35526#discussion_r807907460
##########
File path:
sql/core/src/test/scala/org/apache/spark/sql/DataFrameTimeWindowingSuite.scala
##########
@@ -490,4 +490,81 @@ class DataFrameTimeWindowingSuite extends QueryTest with
SharedSparkSession {
assert(attributeReference.dataType == tuple._2)
}
}
+
+ test("No need to filter data when the sliding window length is not
redundant") {
+ // check the value column
+ val df1 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df2 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ val df3 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "-2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df4 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ Seq(df1, df2).foreach { df =>
+ val filter = df.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ val exist = filter.get.constraints.iterator.toStream.filter(e =>
Review comment:
1. Please add `assert(filter.isDefined)` in above line.
2. You may be able to directly call `constraints.filter { e => ... ` instead
of adding `.iterator.toStream`.
##########
File path:
sql/core/src/test/scala/org/apache/spark/sql/DataFrameTimeWindowingSuite.scala
##########
@@ -490,4 +490,81 @@ class DataFrameTimeWindowingSuite extends QueryTest with
SharedSparkSession {
assert(attributeReference.dataType == tuple._2)
}
}
+
+ test("No need to filter data when the sliding window length is not
redundant") {
+ // check the value column
+ val df1 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df2 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ val df3 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "-2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df4 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ Seq(df1, df2).foreach { df =>
Review comment:
I guess we can test df1, df2, df3, df4 here instead of having two
separate code blocks.
##########
File path:
sql/core/src/test/scala/org/apache/spark/sql/DataFrameTimeWindowingSuite.scala
##########
@@ -490,4 +490,81 @@ class DataFrameTimeWindowingSuite extends QueryTest with
SharedSparkSession {
assert(attributeReference.dataType == tuple._2)
}
}
+
+ test("No need to filter data when the sliding window length is not
redundant") {
+ // check the value column
+ val df1 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df2 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ val df3 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "-2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df4 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ Seq(df1, df2).foreach { df =>
+ val filter = df.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ val exist = filter.get.constraints.iterator.toStream.filter(e =>
+ e.toString.contains(">=") || e.toString.contains("<"))
+ assert(exist.isEmpty, "No need to filter data between " +
+ "window.start and window.end when the sliding window length is not
redundant")
+
+ checkAnswer(
Review comment:
We don't need to validate the results here since other tests also do
this.
##########
File path:
sql/core/src/test/scala/org/apache/spark/sql/DataFrameTimeWindowingSuite.scala
##########
@@ -490,4 +490,81 @@ class DataFrameTimeWindowingSuite extends QueryTest with
SharedSparkSession {
assert(attributeReference.dataType == tuple._2)
}
}
+
+ test("No need to filter data when the sliding window length is not
redundant") {
+ // check the value column
+ val df1 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df2 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ val df3 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "-2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df4 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ Seq(df1, df2).foreach { df =>
+ val filter = df.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ val exist = filter.get.constraints.iterator.toStream.filter(e =>
+ e.toString.contains(">=") || e.toString.contains("<"))
+ assert(exist.isEmpty, "No need to filter data between " +
+ "window.start and window.end when the sliding window length is not
redundant")
+
+ checkAnswer(
+ df,
+ Seq(Row(4), Row(4), Row(4), Row(1), Row(1), Row(1), Row(2), Row(2),
Row(2))
+ )
+ }
+
+ Seq(df3, df4).foreach { df =>
Review comment:
This can be removed if we test against df1 to df4 in above.
##########
File path:
sql/core/src/test/scala/org/apache/spark/sql/DataFrameTimeWindowingSuite.scala
##########
@@ -490,4 +490,81 @@ class DataFrameTimeWindowingSuite extends QueryTest with
SharedSparkSession {
assert(attributeReference.dataType == tuple._2)
}
}
+
+ test("No need to filter data when the sliding window length is not
redundant") {
+ // check the value column
+ val df1 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df2 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ val df3 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "-2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df4 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ Seq(df1, df2).foreach { df =>
+ val filter = df.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ val exist = filter.get.constraints.iterator.toStream.filter(e =>
+ e.toString.contains(">=") || e.toString.contains("<"))
+ assert(exist.isEmpty, "No need to filter data between " +
+ "window.start and window.end when the sliding window length is not
redundant")
+
+ checkAnswer(
+ df,
+ Seq(Row(4), Row(4), Row(4), Row(1), Row(1), Row(1), Row(2), Row(2),
Row(2))
+ )
+ }
+
+ Seq(df3, df4).foreach { df =>
+ val filter = df.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ val exist = filter.get.constraints.iterator.toStream.filter(e =>
+ e.toString.contains(">=") || e.toString.contains("<"))
+ assert(exist.isEmpty, "No need to filter data between " +
+ "window.start and window.end when the sliding window length is not
redundant")
+
+ checkAnswer(
+ df,
+ Seq(Row(4), Row(4), Row(4), Row(1), Row(1), Row(1), Row(2), Row(2),
Row(2))
+ )
+ }
+
+ // check produces right windows
Review comment:
We don't need to validate the results here since other tests also do
this. If you feel we don't have matching tests, please add separated general
test.
##########
File path:
sql/core/src/test/scala/org/apache/spark/sql/DataFrameTimeWindowingSuite.scala
##########
@@ -490,4 +490,81 @@ class DataFrameTimeWindowingSuite extends QueryTest with
SharedSparkSession {
assert(attributeReference.dataType == tuple._2)
}
}
+
+ test("No need to filter data when the sliding window length is not
redundant") {
+ // check the value column
+ val df1 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df2 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "0 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ val df3 = Seq(
+ ("2022-02-15 19:39:34", 1, "a"),
+ ("2022-02-15 19:39:56", 2, "a"),
+ ("2022-02-15 19:39:27", 4, "b")).toDF("time", "value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "-2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+ val df4 = Seq(
+ (LocalDateTime.parse("2022-02-15T19:39:34"), 1, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:56"), 2, "a"),
+ (LocalDateTime.parse("2022-02-15T19:39:27"), 4, "b")).toDF("time",
"value", "id")
+ .select(window($"time", "9 seconds", "3 seconds", "2 second"), $"value")
+ .orderBy($"window.start".asc, $"value".desc).select("value")
+
+ Seq(df1, df2).foreach { df =>
+ val filter = df.queryExecution.optimizedPlan.find(_.isInstanceOf[Filter])
+ val exist = filter.get.constraints.iterator.toStream.filter(e =>
+ e.toString.contains(">=") || e.toString.contains("<"))
+ assert(exist.isEmpty, "No need to filter data between " +
+ "window.start and window.end when the sliding window length is not
redundant")
Review comment:
`when the sliding window length is not redundant` Honestly I don't get
what this means. Could you please elaborate about the the meaning of
`redundant` here, or could we use mathematical expression like `multiple of` or
`a factor of`?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
