prodriguezdefino opened a new pull request, #40626:
URL: https://github.com/apache/beam/pull/40626
Problem:
In `TentativeData.ClearOrderedListState`
(sdks/go/pkg/beam/runners/prism/internal/engine/data.go), Prism indexed
directly into `d.state[stateID][w]` and wrote the cleared slice back to
`kMap[string(uKey)]` without checking whether `d.state`, `d.state[stateID]`, or
`winMap[w]` had been initialized:
```go
winMap := d.state[stateID]
w := d.toWindow(wKey)
kMap := winMap[w]
data := kMap[string(uKey)]
...
kMap[string(uKey)] = StateData{Bag: cleared} // panic: assignment to
entry in nil map
```
SDK harnesses (such as the Python SDK's
`SynchronousOrderedListRuntimeState.commit`) buffer ordered-list mutations in
memory during a bundle and flush pending `StateRequest_Clear` range requests
before `StateRequest_Append` requests at bundle commit. When a bundle clears an
ordered-list state cell (or range) in a window for which no prior bundle has
persisted state on the runner, `TentativeData.state` has no map for `(stateID,
window)`, `kMap` is nil, and Prism panics on line 336.
Fix:
Use the existing `d.clearState(stateID, wKey)` helper in
`ClearOrderedListState` and return early when `kMap` is nil, matching
`ClearBagState`, `ClearMultimapState`, and `ClearMultimapKeysState`.
Reproduction (Python SDK on PrismRunner):
```python
import apache_beam as beam
from apache_beam.coders import VarIntCoder
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.testing.util import assert_that, equal_to
from apache_beam.transforms.userstate import OrderedListStateSpec
from apache_beam.utils.timestamp import Timestamp
class ClearOrderedListDoFn(beam.DoFn):
EVENTS = OrderedListStateSpec("events", VarIntCoder())
def process(self, element, events=beam.DoFn.StateParam(EVENTS)):
key, (ts, op) = element
if op == "CLEAR":
events.clear()
yield f"cleared:{key}"
else:
events.add((Timestamp(micros=ts), op))
entries = [f"{int(t.micros)}={v}" for t, v in events.read()]
yield f"entry:{key}:{','.join(entries)}"
if __name__ == "__main__":
options = PipelineOptions(["--runner=PrismRunner",
"--environment_type=LOOPBACK"])
with beam.Pipeline(options=options) as p:
out = (
p
| beam.Create([("k", (30, 300)), ("k", (10, 100)), ("k", (0,
"CLEAR"))])
| beam.ParDo(ClearOrderedListDoFn())
)
assert_that(
out,
equal_to(["entry:k:30=300", "entry:k:10=100,30=300",
"cleared:k"]),
)
```
--
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]
