[Spark 4] Group the streaming stateful ParDo by encoded key bytes to support null keys

[tkaymak]

The Spark 4 Structured Streaming stateful ParDo grouped by the Beam key object, so a null key (VoidCoder, WithKeys.of((Void) null), Combine.globally) became a null Spark grouping key and transformWithState failed with INTERNAL_ERROR_TWS "Implicit key not found in state store" on the first state access.

The translator now groups by CoderHelpers.toByteArray(key, keyCoder) with Encoders.BINARY(), and BeamStatefulProcessor decodes the key once per handleInputRows and handleExpiredTimer call. Byte equality is Beam key equality, ParDo already rejects non deterministic key coders for stateful DoFns, and keys with custom coders were already grouped as bytes through EncoderHelpers.binaryEncoder.

The grouping key state schema changes from the key type to BinaryType, and the stateful path now uses the structured streaming CoderHelpers (outer context) for keys and timers, so a query restarted from a checkpoint written by #40281 or #40447 fails with STATE_STORE_KEY_SCHEMA_NOT_COMPATIBLE. That code is unreleased.

testNullKey fails on master with INTERNAL_ERROR_TWS and passes with this change.

Part of #36841.

[tkaymak]

R: @Abacn

[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

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 65.02%. Comparing base (76f6e6d) to head (b8b45ae).
⚠️ Report is 143 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #40516      +/-   ##
============================================
+ Coverage     56.16%   65.02%   +8.85%     
- Complexity     2288    13952   +11664     
============================================
  Files          1124     1728     +604     
  Lines        178336   108384   -69952     
  Branches       1489    11286    +9797     
============================================
- Hits         100170    70478   -29692     
+ Misses        75645    33703   -41942     
- Partials       2521     4203    +1682     

Flag	Coverage Δ
go	?
java	65.02% <ø> (-10.77%)	⬇️
python	?

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

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

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Abacn]

Thanks, just one minor comment

[Abacn]

Realize there is a org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers used by ReadUnboundedTranslator and etc. Consider deduplicate the class (merge structuredstreaming one into the common one)

[tkaymak]

Switched both stateful files to the structured streaming CoderHelpers in 5608646, so that package no longer imports the legacy one. I did not merge the classes here: the legacy class encodes in nested context (coder.encode(value, stream)) and the structured streaming one in outer context (CoderUtils), and the latter encodes the unbounded source checkpoint marks that shipped in 2.77.0, so folding one into the other changes persisted bytes. Happy to do the dedup in a follow up that keeps the outer encoding for the structured streaming callers.

[Abacn]

If the concern is compatibility it's fine. The code exists in Beam 2.77 but until validate runner running and we announce it in CHANGS.md, Spark4 DataSet API Streaming is considered incomplete and not usable in production yet