You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "lostluck (via GitHub)" <gi...@apache.org> on 2023/02/14 22:30:17 UTC
[GitHub] [beam] lostluck opened a new pull request, #25476: [#24789][prism] internal/coders.go and tests
lostluck opened a new pull request, #25476:
URL: https://github.com/apache/beam/pull/25476
internal/coders.go and tests
Initial version of the Go Direct Runner Replacement : Prism.
See https://github.com/apache/beam/issues/24789
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/get-started-contributing/#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #25476: [#24789][prism] internal/coders.go and tests
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on PR #25476:
URL: https://github.com/apache/beam/pull/25476#issuecomment-1430781450
R: @jrmccluskey
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] jrmccluskey commented on a diff in pull request #25476: [#24789][prism] internal/coders.go and tests
Posted by "jrmccluskey (via GitHub)" <gi...@apache.org>.
jrmccluskey commented on code in PR #25476:
URL: https://github.com/apache/beam/pull/25476#discussion_r1107690382
##########
sdks/go/pkg/beam/runners/prism/internal/coders.go:
##########
@@ -0,0 +1,243 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+ "bytes"
+ "fmt"
+ "io"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/ioutilx"
+ pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/urns"
+ "golang.org/x/exp/slog"
+ "google.golang.org/protobuf/encoding/prototext"
+)
+
+// leafCoders lists coder urns the runner knows how to manipulate.
+// In particular, ones that won't be a problem to parse, in general
+// because they have a known total size.
+var leafCoders = map[string]struct{}{
+ urns.CoderBytes: {},
+ urns.CoderStringUTF8: {},
+ urns.CoderLengthPrefix: {},
+ urns.CoderVarInt: {},
+ urns.CoderDouble: {},
+ urns.CoderBool: {},
+ urns.CoderGlobalWindow: {},
+ urns.CoderIntervalWindow: {},
+}
+
+func isLeafCoder(c *pipepb.Coder) bool {
+ _, ok := leafCoders[c.GetSpec().GetUrn()]
+ return ok
+}
+
+// makeWindowedValueCoder gets the coder for the PCollection, renders it safe, and adds it to the coders map.
+//
+// PCollection coders are not inherently WindowValueCoder wrapped, and they are added by the runner
+// for crossing the FnAPI boundary at data sources and data sinks.
+func makeWindowedValueCoder(pID string, comps *pipepb.Components, coders map[string]*pipepb.Coder) string {
+ col := comps.GetPcollections()[pID]
+ cID := lpUnknownCoders(col.GetCoderId(), coders, comps.GetCoders())
+ wcID := comps.GetWindowingStrategies()[col.GetWindowingStrategyId()].GetWindowCoderId()
+
+ // The runner needs to be defensive, and tell the SDK to Length Prefix
+ // any coders that it doesn't understand.
+ // So here, we look at the coder and its components, and produce
+ // new coders that we know how to deal with.
+
+ // Produce ID for the Windowed Value Coder
+ wvcID := "cwv_" + pID
+ wInC := &pipepb.Coder{
+ Spec: &pipepb.FunctionSpec{
+ Urn: urns.CoderWindowedValue,
+ },
+ ComponentCoderIds: []string{cID, wcID},
+ }
+ // Populate the coders to send with the new windowed value coder.
+ coders[wvcID] = wInC
+ return wvcID
+}
+
+// makeWindowCoders makes the coder pair but behavior is ultimately determined by the strategy's windowFn.
+func makeWindowCoders(wc *pipepb.Coder) (exec.WindowDecoder, exec.WindowEncoder) {
+ var cwc *coder.WindowCoder
+ switch wc.GetSpec().GetUrn() {
+ case urns.CoderGlobalWindow:
+ cwc = coder.NewGlobalWindow()
+ case urns.CoderIntervalWindow:
+ cwc = coder.NewIntervalWindow()
+ default:
+ slog.Log(slog.LevelError, "makeWindowCoders: unknown urn", slog.String("urn", wc.GetSpec().GetUrn()))
+ panic(fmt.Sprintf("makeWindowCoders, unknown urn: %v", prototext.Format(wc)))
+ }
+ return exec.MakeWindowDecoder(cwc), exec.MakeWindowEncoder(cwc)
+}
+
+// lpUnknownCoders takes a coder, and populates coders with any new coders
+// coders that the runner needs to be safe, and speedy.
+// It returns either the passed in coder id, or the new safe coder id.
+func lpUnknownCoders(cID string, bundle, base map[string]*pipepb.Coder) string {
+ // First check if we've already added the LP version of this coder to coders already.
+ lpcID := cID + "_lp"
+ // Check if we've done this one before.
+ if _, ok := bundle[lpcID]; ok {
+ return lpcID
+ }
+ // All coders in the coders map have been processed.
+ if _, ok := bundle[cID]; ok {
+ return cID
+ }
+ // Look up the canonical location.
+ c, ok := base[cID]
+ if !ok {
+ // We messed up somewhere.
+ panic(fmt.Sprint("unknown coder id:", cID))
+ }
+ // Add the original coder to the coders map.
+ bundle[cID] = c
+ // If we don't know this coder, and it has no sub components,
+ // we must LP it, and we return the LP'd version.
+ leaf := isLeafCoder(c)
+ if len(c.GetComponentCoderIds()) == 0 && !leaf {
+ lpc := &pipepb.Coder{
+ Spec: &pipepb.FunctionSpec{
+ Urn: urns.CoderLengthPrefix,
+ },
+ ComponentCoderIds: []string{cID},
+ }
+ bundle[lpcID] = lpc
+ return lpcID
+ }
+ // We know we have a composite, so if we count this as a leaf, move everything to
+ // the coders map.
+ if leaf {
+ // Copy the components from the base.
+ for _, cc := range c.GetComponentCoderIds() {
+ bundle[cc] = base[cc]
+ }
+ return cID
+ }
+ var needNewComposite bool
+ var comps []string
+ for _, cc := range c.GetComponentCoderIds() {
+ rcc := lpUnknownCoders(cc, bundle, base)
+ if cc != rcc {
+ needNewComposite = true
+ }
+ comps = append(comps, rcc)
+ }
+ if needNewComposite {
+ lpc := &pipepb.Coder{
+ Spec: c.GetSpec(),
+ ComponentCoderIds: comps,
+ }
+ bundle[lpcID] = lpc
+ return lpcID
+ }
+ return cID
+}
+
+// reconcileCoders ensures coders is primed with initial coders from
Review Comment:
```suggestion
// reconcileCoders ensures that bundle coders is primed with initial coders from
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #25476: [#24789][prism] internal/coders.go and tests
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #25476:
URL: https://github.com/apache/beam/pull/25476#discussion_r1107733619
##########
sdks/go/pkg/beam/runners/prism/internal/coders.go:
##########
@@ -0,0 +1,243 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+ "bytes"
+ "fmt"
+ "io"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/util/ioutilx"
+ pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/urns"
+ "golang.org/x/exp/slog"
+ "google.golang.org/protobuf/encoding/prototext"
+)
+
+// leafCoders lists coder urns the runner knows how to manipulate.
+// In particular, ones that won't be a problem to parse, in general
+// because they have a known total size.
+var leafCoders = map[string]struct{}{
+ urns.CoderBytes: {},
+ urns.CoderStringUTF8: {},
+ urns.CoderLengthPrefix: {},
+ urns.CoderVarInt: {},
+ urns.CoderDouble: {},
+ urns.CoderBool: {},
+ urns.CoderGlobalWindow: {},
+ urns.CoderIntervalWindow: {},
+}
+
+func isLeafCoder(c *pipepb.Coder) bool {
+ _, ok := leafCoders[c.GetSpec().GetUrn()]
+ return ok
+}
+
+// makeWindowedValueCoder gets the coder for the PCollection, renders it safe, and adds it to the coders map.
+//
+// PCollection coders are not inherently WindowValueCoder wrapped, and they are added by the runner
+// for crossing the FnAPI boundary at data sources and data sinks.
+func makeWindowedValueCoder(pID string, comps *pipepb.Components, coders map[string]*pipepb.Coder) string {
+ col := comps.GetPcollections()[pID]
+ cID := lpUnknownCoders(col.GetCoderId(), coders, comps.GetCoders())
+ wcID := comps.GetWindowingStrategies()[col.GetWindowingStrategyId()].GetWindowCoderId()
+
+ // The runner needs to be defensive, and tell the SDK to Length Prefix
+ // any coders that it doesn't understand.
+ // So here, we look at the coder and its components, and produce
+ // new coders that we know how to deal with.
+
+ // Produce ID for the Windowed Value Coder
+ wvcID := "cwv_" + pID
+ wInC := &pipepb.Coder{
+ Spec: &pipepb.FunctionSpec{
+ Urn: urns.CoderWindowedValue,
+ },
+ ComponentCoderIds: []string{cID, wcID},
+ }
+ // Populate the coders to send with the new windowed value coder.
+ coders[wvcID] = wInC
+ return wvcID
+}
+
+// makeWindowCoders makes the coder pair but behavior is ultimately determined by the strategy's windowFn.
+func makeWindowCoders(wc *pipepb.Coder) (exec.WindowDecoder, exec.WindowEncoder) {
+ var cwc *coder.WindowCoder
+ switch wc.GetSpec().GetUrn() {
+ case urns.CoderGlobalWindow:
+ cwc = coder.NewGlobalWindow()
+ case urns.CoderIntervalWindow:
+ cwc = coder.NewIntervalWindow()
+ default:
+ slog.Log(slog.LevelError, "makeWindowCoders: unknown urn", slog.String("urn", wc.GetSpec().GetUrn()))
+ panic(fmt.Sprintf("makeWindowCoders, unknown urn: %v", prototext.Format(wc)))
+ }
+ return exec.MakeWindowDecoder(cwc), exec.MakeWindowEncoder(cwc)
+}
+
+// lpUnknownCoders takes a coder, and populates coders with any new coders
+// coders that the runner needs to be safe, and speedy.
+// It returns either the passed in coder id, or the new safe coder id.
+func lpUnknownCoders(cID string, bundle, base map[string]*pipepb.Coder) string {
+ // First check if we've already added the LP version of this coder to coders already.
+ lpcID := cID + "_lp"
+ // Check if we've done this one before.
+ if _, ok := bundle[lpcID]; ok {
+ return lpcID
+ }
+ // All coders in the coders map have been processed.
+ if _, ok := bundle[cID]; ok {
+ return cID
+ }
+ // Look up the canonical location.
+ c, ok := base[cID]
+ if !ok {
+ // We messed up somewhere.
+ panic(fmt.Sprint("unknown coder id:", cID))
+ }
+ // Add the original coder to the coders map.
+ bundle[cID] = c
+ // If we don't know this coder, and it has no sub components,
+ // we must LP it, and we return the LP'd version.
+ leaf := isLeafCoder(c)
+ if len(c.GetComponentCoderIds()) == 0 && !leaf {
+ lpc := &pipepb.Coder{
+ Spec: &pipepb.FunctionSpec{
+ Urn: urns.CoderLengthPrefix,
+ },
+ ComponentCoderIds: []string{cID},
+ }
+ bundle[lpcID] = lpc
+ return lpcID
+ }
+ // We know we have a composite, so if we count this as a leaf, move everything to
+ // the coders map.
+ if leaf {
+ // Copy the components from the base.
+ for _, cc := range c.GetComponentCoderIds() {
+ bundle[cc] = base[cc]
+ }
+ return cID
+ }
+ var needNewComposite bool
+ var comps []string
+ for _, cc := range c.GetComponentCoderIds() {
+ rcc := lpUnknownCoders(cc, bundle, base)
+ if cc != rcc {
+ needNewComposite = true
+ }
+ comps = append(comps, rcc)
+ }
+ if needNewComposite {
+ lpc := &pipepb.Coder{
+ Spec: c.GetSpec(),
+ ComponentCoderIds: comps,
+ }
+ bundle[lpcID] = lpc
+ return lpcID
+ }
+ return cID
+}
+
+// reconcileCoders ensures coders is primed with initial coders from
Review Comment:
Nice thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck merged pull request #25476: [#24789][prism] internal/coders.go and tests
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck merged PR #25476:
URL: https://github.com/apache/beam/pull/25476
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org