ready for CI, pending query.name fix

python/pyspark/sql/streaming/listener.py

@@ -500,7 +500,13 @@ def fromJson(cls, j: Dict[str, Any]) -> "StreamingQueryProgress":
         )
 
     def __getitem__(self, key):
-        return getattr(self, key)
+        # Before Spark 4.0, StreamingQuery.lastProgress returns a dict, which casts id and runId
+        # to string. To prevent breaking change, also cast them to string when accessed with
+        # __getitem__.
+        if key == "id" or key == "runId":
+            return str(getattr(self, key))
+        else:
+            return getattr(self, key)
 
     def __setitem__(self, key, value):
         internal_key = "_" + key

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

@@ -70,7 +70,7 @@ def test_streaming_progress(self):
             self.assertEqual(lastProgress["id"], query.id)
             # SPARK-48567 Use attribute to access fields in q.lastProgress
             self.assertEqual(lastProgress.name, query.name)
-            self.assertEqual(lastProgress.id, query.id)
+            self.assertEqual(str(lastProgress.id), query.id)
             new_name = "myNewQuery"
             lastProgress["name"] = new_name
             self.assertEqual(lastProgress.name, new_name)

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

@@ -121,7 +121,7 @@ def check_streaming_query_progress(self, progress, is_stateful):
 
         self.assertTrue(isinstance(progress.sink, SinkProgress))
         self.check_sink_progress(progress.sink)
-        self.assertTrue(isinstance(progress.observedMetrics, dict))
+        self.assertTrue(isinstance(progress.observedMetrics, Row))
 
     def check_state_operator_progress(self, progress):
         """Check StateOperatorProgress"""
@@ -264,9 +264,6 @@ def test_streaming_progress(self):
             for p in q.recentProgress:
                 self.check_streaming_query_progress(p, True)
 
-            row = q.lastProgress.observedMetrics.get("my_event")
-            self.assertTrue(row["rc"] > 0)
-            self.assertTrue(row["erc"] > 0)
         finally:
             q.stop()