[SPARK-49630][SS] Add flatten option to process collection types with state data source reader

[anishshri-db]

What changes were proposed in this pull request?

Add flatten option to process collection types with state data source reader

Why are the changes needed?

Changes are needed to process entries row-by-row in case we don't have enough memory to fit these collections inside a single row

Does this PR introduce any user-facing change?

Yes

Users can provide the following query option:

        val stateReaderDf = spark.read
          .format("statestore")
          .option(StateSourceOptions.PATH, <state_checkpoint_loc>)
          .option(StateSourceOptions.STATE_VAR_NAME, <state_var_name>)
          .option(StateSourceOptions.FLATTEN_COLLECTION_TYPES, <true | false>)
          .load()

How was this patch tested?

Added unit tests

[info] Run completed in 1 minute, 10 seconds.
[info] Total number of tests run: 12
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 12, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.

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

No

[anishshri-db]

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

[pavel0fadeev]

If you are working on this file anyway, could you also fix the typo in the comment here?

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala

Line 152 in 931ab06

* Source. It reads the the state at a particular batchId.

the the state -> the state

[anishshri-db]

If you are working on this file anyway, could you also fix the typo in the comment here?

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala

Line 152 in 931ab06

* Source. It reads the the state at a particular batchId.

the the state -> the state

Sure done

[jingz-db]

Small nits otherwise LGTM! Thanks for making the change! Shall we also document in the PR API section that the default value for flatten option is True?

[HeartSaVioR]

First pass. Mostly minors.

[HeartSaVioR]

+1 pending CI

[HeartSaVioR]

The build only failed with test suite from SparkConnect and it seems to be flaky and not relevant to this change.
https://github.com/anishshri-db/spark/actions/runs/11008519536/job/30567579146

[HeartSaVioR]

Thanks! Merging to master.