tkaymak opened a new pull request, #40516: URL: https://github.com/apache/beam/pull/40516
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`, 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
