This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 3336b153784 [Prism] Guard ClearOrderedListState against uninitialized
state maps (#40626)
3336b153784 is described below
commit 3336b153784ce57ef226acfa6efafd8feccf2bd8
Author: pablo rodriguez defino <[email protected]>
AuthorDate: Fri Oct 16 07:11:09 2026 -0700
[Prism] Guard ClearOrderedListState against uninitialized state maps
(#40626)
* [Prism] Guard ClearOrderedListState against uninitialized state maps
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:
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 and
the Rust SDK's OrderedListCell) 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"]),
)
```
* Refactor comments in data_test.go
Updated comments for clarity in data_test.go.
---
.../pkg/beam/runners/prism/internal/engine/data.go | 7 ++--
.../runners/prism/internal/engine/data_test.go | 37 ++++++++++++++++++++++
2 files changed, 41 insertions(+), 3 deletions(-)
diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/data.go
b/sdks/go/pkg/beam/runners/prism/internal/engine/data.go
index 371e3f00b89..23553413f44 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/engine/data.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/engine/data.go
@@ -322,9 +322,10 @@ func findRange(bag [][]byte, start, end int64) (int, int) {
}
func (d *TentativeData) ClearOrderedListState(stateID LinkID, wKey, uKey
[]byte, start, end int64) {
- winMap := d.state[stateID]
- w := d.toWindow(wKey)
- kMap := winMap[w]
+ kMap := d.clearState(stateID, wKey)
+ if kMap == nil {
+ return
+ }
data := kMap[string(uKey)]
lo, hi := findRange(data.Bag, start, end)
diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/data_test.go
b/sdks/go/pkg/beam/runners/prism/internal/engine/data_test.go
index 1d049710418..8d800e3cd5f 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/engine/data_test.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/engine/data_test.go
@@ -219,4 +219,41 @@ func TestOrderedListState(t *testing.T) {
t.Errorf("OrderedList int32 \n%v", d)
}
})
+ t.Run("clear_uninitialized", func(t *testing.T) {
+ d := TentativeData{
+ stateTypeLen: map[LinkID]func([]byte) int{
+ linkID: func(_ []byte) int {
+ return 1
+ },
+ },
+ }
+
+ // d.state is nil.
+ d.ClearOrderedListState(linkID, wKey, uKey, math.MinInt64,
math.MaxInt64)
+ if d.state != nil {
+ t.Fatalf("ClearOrderedListState on empty TentativeData
allocated state map: %v", d.state)
+ }
+
+ // d.state is non-nil, but has no entry for linkID.
+ otherLink := LinkID{Transform: "dofn", Local: "otherState"}
+ d.AppendBagState(otherLink, wKey, uKey, []byte{1})
+ d.ClearOrderedListState(linkID, wKey, uKey, 0, 100)
+ if _, ok := d.state[linkID]; ok {
+ t.Fatalf("ClearOrderedListState created entry for
uninitialized stateID %v", linkID)
+ }
+
+ // d.state[linkID] exists for an interval window, but not for
the global window.
+ // An SDK that flushes range clears before appends on the first
bundle in a window
+ // must not panic and must preserve subsequent appends.
+ iwKey := make([]byte, 16)
+ binary.BigEndian.PutUint64(iwKey[0:8],
uint64(1000)^0x8000000000000000)
+ d.AppendOrderedListState(linkID, iwKey, uKey, cc(time1, 1))
+ d.ClearOrderedListState(linkID, wKey, uKey, 0, 30)
+ d.AppendOrderedListState(linkID, wKey, uKey, cc(time2, 1))
+ got := d.GetOrderedListState(linkID, wKey, uKey, 0, 60)
+ want := [][]byte{cc(time2, 1)}
+ if diff := cmp.Diff(want, got); diff != "" {
+ t.Errorf("OrderedList after clear_uninitialized +
append:\n%v", diff)
+ }
+ })
}
