[SPARK-42941][SS][CONNECT] 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

Closes #42116 from WweiL/listener-poc-newest.

Lead-authored-by: Wei Liu [email protected]

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?