[SPARK-42941][SS][CONNECT][3.5] Python StreamingQueryListener

[bogao007]

What changes were proposed in this pull request?

Implement the python streaming query listener and the addListener method and removeListener method, follow up filed in: SPARK-44516 to actually terminate the query listener process when removeListener is called. SPARK-44516 depends on SPARK-44433.

Why are the changes needed?

SS Connect development

Does this PR introduce any user-facing change?

Yes now they can use connect listener

How was this patch tested?

Manual test and added unit test

addListener:

# Client side:
>>> from pyspark.sql.streaming.listener import StreamingQueryListener;from pyspark.sql.streaming.listener import (QueryStartedEvent, QueryProgressEvent, QueryTerminatedEvent, QueryIdleEvent)
>>> class MyListener(StreamingQueryListener):
...     def onQueryStarted(self, event: QueryStartedEvent) -> None: print("hi, event query id is: " +  str(event.id)); df=self.spark.createDataFrame(["10","11","13"], "string").toDF("age"); df.write.saveAsTable("tbllistener1")
...     def onQueryProgress(self, event: QueryProgressEvent) -> None: pass
...     def onQueryIdle(self, event: QueryIdleEvent) -> None: pass
...     def onQueryTerminated(self, event: QueryTerminatedEvent) -> None: pass
...
>>> spark.streams.addListener(MyListener())
>>> q = spark.readStream.format("rate").load().writeStream.format("console").start()
>>> q.stop()
>>> spark.read.table("tbllistener1").collect()
[Row(age='13'), Row(age='10'), Row(age='11’)]

# Server side:
##### event_type received from python process is 0
hi, event query id is: dd7ba1c4-6c8f-4369-9c3c-5dede22b8a2f

removeListener:

# Client side:
>>> listener = MyListener(); spark.streams.addListener(listener)
>>> spark.streams.removeListener(listener)

# Server side:
# nothing to print actually, the listener is removed from server side StreamingQueryManager and cache in sessionHolder, but the process still hangs there. Follow up SPARK-44516 filed to stop this process

[ueshin]

LGTM, pending tests.

[WweiL]

Thank you for resolving it!

[WweiL]

@HyukjinKwon @ueshin Hey guys, I think we could just merge this, the failed suites are flaky IMO. Last time the "other tests" failed but "slow test" succeeds. https://github.com/bogao007/spark/actions/runs/5722809496/job/15506495583

This time it's the opposite
https://github.com/bogao007/spark/actions/runs/5729966741/job/15527770333

[HyukjinKwon]

Merged to branch-3.5.