Add OrderedListState support for SparkRunner

[twosom]

Please add a meaningful description for your change here

fixes #33211
fixes #31724
fixes #31723

This PR contains these changes

• added OrderedList in SparkStateInternals
• implementation equals and hashcode in FlinkOrderedListState
• added OrderedListState test
• fixed FlinkRunner state test

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 59.01%. Comparing base (e5bd69d) to head (74033fa).

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #33212      +/-   ##
============================================
+ Coverage     57.42%   59.01%   +1.58%     
- Complexity     1475     3162    +1687     
============================================
  Files           970     1136     +166     
  Lines        154525   175078   +20553     
  Branches       1076     3354    +2278     
============================================
+ Hits          88743   103324   +14581     
- Misses        63578    68406    +4828     
- Partials       2204     3348    +1144     

Flag	Coverage Δ
java	70.27% <ø> (+1.69%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

---

🚨 Try these New Features:

• Flaky Tests Detection - Detect and resolve failed and flaky tests
• JS Bundle Analysis - Avoid shipping oversized bundles

[twosom]

Run Flink Container PreCommit

[liferoad]

R: @shunping

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[liferoad]

Did you rebase your branch?

[twosom]

Did you rebase your branch?

Thanks!

i rebased now!

[twosom]

Run Java_Spark3_Versions PreCommit

[liferoad]

Please check this:
:runners:spark:3:analyzeClassesDependencies FAILED

[twosom]

@liferoad
Sorry, my bad. Fixed the import statement to use Beam's vendored Guava Lists instead of clearspring's.
Thanks

[twosom]

Run Java PreCommit

[twosom]

Run Java PreCommit

[twosom]

Run Java_GCP_IO_Direct PreCommit

[twosom]

Run Java PreCommit

[kennknowles]

Can use Objects.hashCode

[twosom]

@kennknowles
The current implementation follows the same hashCode semantics used in FlinkStateInternals. I'm a bit unclear whether suggesting Objects.hashCode means we should replace the current implementation, or if you're suggesting that overriding hashCode is unnecessary altogether.

[kennknowles]

Overriding equality and hashcode should always occur together, so it is necessary to override hashCode.

I was just suggesting this pattern instead of doing your own math:

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SuccessOrFailure.java

Line 88 in 7c86bf3

return Objects.hashCode(isSuccess, site, throwable);

[kennknowles]

The reason for each additional kind of state is to efficiently offer a novel form of a state access. The state access here as the same performance characteristics as ValueState. It is actually better for the runner to reject a pipeline than to run it with performance characteristics that don't match the expected performance contract.

Is there some underlying mechanism in Spark that could implement OrderedListState efficiently and scalably?

[twosom]

I agree with your point. Let me share my thoughts on why I chose this implementation.

I've noticed that ListState/OrderedListState is mostly used in situations where writes happen much more frequently than reads. That's why I went with ArrayList instead of SortedMap - it's simply better at handling these frequent writes.

When it comes to reading data, it usually happens in just a couple of scenarios - either during OnTimer execution or when the list hits a certain size. So even if the read performance takes a small hit, it's not really going to affect the overall performance much.

It's also worth mentioning that FlinkOrderedListState uses the same approach, which gives me confidence in this design choice.

That's why I think the current implementation makes more sense for real-world usage patterns.

[kennknowles]

I see. If Flink is implemented then it is OK with me to follow that precedent. My point was that this does not actually add capability that is more than ValueState provides. It is just a minor API wrapper adjustment - still useful but not the main purpose.

So we can merge with this design. But if you think about following up, here is how we would really like this to behave:

• add should call some native Spark API that writes the element without reading the list
• readRange should only read the requested range, ideally seeking in near-constant time (aka without a scan or sort)
• clearRange should also seek in near-constant time
• isEmpty should not read the list

[twosom]

Run Java PreCommit