[SPARK-48411][SS][PYTHON] Add E2E test for DropDuplicateWithinWatermark

### What changes were proposed in this pull request?
This PR adds a test for API DropDuplicateWithinWatermark in Python, which was previously missing.

### Why are the changes needed?
Check the correctness of API DropDuplicateWithinWatermark.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Passed:
```
python/run-tests --testnames pyspark.sql.tests.streaming.test_streaming
python/run-tests --testnames pyspark.sql.tests.connect.streaming.test_parity_streaming
```

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

Closes #46740 from eason-yuchen-liu/DropDuplicateWithinWatermark_test.

Authored-by: Yuchen Liu <yuchen.liu@databricks.com>
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

Author: eason-yuchen-liu

python/pyspark/sql/tests/streaming/test_streaming.py

@@ -22,7 +22,7 @@
 
 from pyspark.sql import Row
 from pyspark.sql.functions import lit
-from pyspark.sql.types import StructType, StructField, IntegerType, StringType
+from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType
 from pyspark.testing.sqlutils import ReusedSQLTestCase
 from pyspark.errors import PySparkValueError
 
@@ -413,6 +413,30 @@ def test_streaming_with_temporary_view(self):
                 set([Row(value="view_a"), Row(value="view_b"), Row(value="view_c")]), set(result)
             )
 
+    def test_streaming_drop_duplicate_within_watermark(self):
+        """
+        This verifies dropDuplicatesWithinWatermark works with a streaming dataframe.
+        """
+        user_schema = StructType().add("time", TimestampType()).add("id", "integer")
+        df = (
+            self.spark.readStream.option("sep", ";")
+            .schema(user_schema)
+            .csv("python/test_support/sql/streaming/time")
+        )
+        q1 = (
+            df.withWatermark("time", "2 seconds")
+            .dropDuplicatesWithinWatermark(["id"])
+            .writeStream.outputMode("update")
+            .format("memory")
+            .queryName("test_streaming_drop_duplicates_within_wm")
+            .start()
+        )
+        self.assertTrue(q1.isActive)
+        q1.processAllAvailable()
+        q1.stop()
+        result = self.spark.sql("SELECT * FROM test_streaming_drop_duplicates_within_wm").collect()
+        self.assertTrue(len(result) >= 6 and len(result) <= 9)
+
 
 class StreamingTests(StreamingTestsMixin, ReusedSQLTestCase):
     def _assert_exception_tree_contains_msg(self, exception, msg):

python/test_support/sql/streaming/time/text-with-time-test.txt

@@ -0,0 +1,10 @@
+2024-05-24 15:03:20;1
+2024-05-24 15:03:21;2
+2024-05-24 15:03:24;3
+2024-05-24 15:03:25;3
+2024-05-24 15:03:31;4
+2024-05-24 15:03:31;1
+2024-05-24 15:03:32;3
+2024-05-24 15:03:45;2
+2024-05-24 15:03:46;5
+2024-05-24 15:03:50;6