lostluck commented on a change in pull request #15594:
URL: https://github.com/apache/beam/pull/15594#discussion_r717119849
##########
File path: sdks/go/pkg/beam/core/runtime/exec/sideinput.go
##########
@@ -76,6 +78,19 @@ func (s *sideInputAdapter) NewIterable(ctx context.Context,
reader StateReader,
}, nil
}
+// QueryCache checks a reader's side input cache for an entry with a
PtransformID and sideInputID
+// and returns the entry.
+func (s *sideInputAdapter) QueryCache(reader StateReader) ReusableInput {
+ input := reader.GetSideInputCache().QueryCache(s.sid.PtransformID,
s.sideInputID)
+ return input
Review comment:
Aside: simply return in this situation rather than assigning to a new
variable, it's cleaner.
##########
File path: sdks/go/pkg/beam/core/runtime/exec/fn.go
##########
@@ -277,13 +268,24 @@ func makeSideInputs(ctx context.Context, w typex.Window,
side []SideInputAdapter
offset := len(param) - len(side)
var ret []ReusableInput
- for i := 0; i < len(streams); i++ {
- s, err := makeSideInput(in[i+1].Kind,
fn.Param[param[i+offset]].T, streams[i])
+ for i, adapter := range side {
+ // Cache hit
+ if c := adapter.QueryCache(reader); c != nil {
Review comment:
Why not put this logic into the NewIterable method instead? Is it
important that the makeSideInputs code is aware that a cache is being used?
Does this code *need* to know these details of how the adapter works?
Is there any reason the adapter can't make use of the cache itself, and
still produce ReStreams (and if still a good idea, cache ReusableInputs),
avoiding this code from needing to change or know the details of it happening?
Per the earlier failure, ReusableInputs aren't threadsafe, so we need to
cache something that is, that we can use to build what we use later on.
--
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]
