[#24789] Spot fix fullvalue wrapping for SDF.

apache · Dec 29, 2022 · 37955a9 · 37955a9
1 parent 80de3ef
commit 37955a9
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/sdks/go/pkg/beam/core/runtime/exec/sdf.go b/sdks/go/pkg/beam/core/runtime/exec/sdf.go
@@ -190,7 +190,8 @@ func (n *SplitAndSizeRestrictions) StartBundle(ctx context.Context, id string, d
 func (n *SplitAndSizeRestrictions) ProcessElement(ctx context.Context, elm *FullValue, values ...ReStream) error {
 	rest := elm.Elm2.(*FullValue).Elm
 	ws := elm.Elm2.(*FullValue).Elm2
-	mainElm := elm.Elm.(*FullValue)
+
+	mainElm := convertIfNeeded(elm.Elm, &FullValue{})
 
 	splitRests := n.splitInv.Invoke(mainElm, rest)