[SPARK-50128][SS] Add stateful processor handle APIs using implicit encoders in Scala

[anishshri-db]

What changes were proposed in this pull request?

Add stateful processor handle APIs using implicit encoders in Scala

Why are the changes needed?

Without the changes, users have to pass explicit SQL encoders for state types while acquiring an instance of the underlying state variable

Does this PR introduce any user-facing change?

Yes

Users can now implicits available in Scala through import spark.implicits._ and only provide the type while getting the state objects. For eg -

      override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
         _myValueState = getHandle.getValueState[Long]("myValueState", TTLConfig.NONE)
      }

How was this patch tested?

Existing unit tests

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

No

[anishshri-db]

cc - @jingz-db @HeartSaVioR - PTAL, thx !

[HeartSaVioR]

The change itself looks good to me.

I feel like requiring users to provide TTLConfig(Duration.ZERO) isn't great though. Shall we check whether we could provide a dedicated instance for this? Like TTLConfig.NONE or TTLConfig.none() (if the former does not work with Java).

[anishshri-db]

I feel like requiring users to provide TTLConfig(Duration.ZERO) isn't great though. Shall we check whether we could provide a dedicated instance for this? Like TTLConfig.NONE or TTLConfig.none()

Done

[HeartSaVioR]

+1 pending CI

[HeartSaVioR]

https://github.com/anishshri-db/spark/actions/runs/11675450690/job/32510196280

• Run / Build modules: pyspark-mllib, pyspark-ml, pyspark-ml-connect
• Run / Run Docker integration tests

These failures are unrelated to this PR.

[HeartSaVioR]

Thanks! Merging to master.