add tests

sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousAggregationSuite.scala

@@ -70,15 +70,39 @@ class ContinuousAggregationSuite extends ContinuousSuiteBase {
   test("multiple partitions with coalesce - multiple transformations") {
     val input = ContinuousMemoryStream[Int]
 
+    // We use a barrier to make sure predicates both before and after coalesce work
     val df = input.toDF()
-      .coalesce(1)
       .select('value as 'copy, 'value)
+      .where('copy =!= 1)
+      .planWithBarrier
+      .coalesce(1)
       .where('copy =!= 2)
       .agg(max('value))
 
     testStream(df, OutputMode.Complete)(
       AddData(input, 0, 1, 2),
-      CheckAnswer(1),
+      CheckAnswer(0),
+      StopStream,
+      AddData(input, 3, 4, 5),
+      StartStream(),
+      CheckAnswer(5),
+      AddData(input, -1, -2, -3),
+      CheckAnswer(5))
+  }
+
+  test("multiple partitions with multiple coalesce") {
+    val input = ContinuousMemoryStream[Int]
+
+    val df = input.toDF()
+      .coalesce(1)
+      .planWithBarrier
+      .coalesce(1)
+      .select('value as 'copy, 'value)
+      .agg(max('value))
+
+    testStream(df, OutputMode.Complete)(
+      AddData(input, 0, 1, 2),
+      CheckAnswer(2),
       StopStream,
       AddData(input, 3, 4, 5),
       StartStream(),