You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lo...@apache.org on 2023/11/06 22:15:30 UTC
(beam) branch master updated: [#29210] Fix SDF progress fraction to not return NaN (#29307)
This is an automated email from the ASF dual-hosted git repository.
lostluck 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 ad287193c1e [#29210] Fix SDF progress fraction to not return NaN (#29307)
ad287193c1e is described below
commit ad287193c1ee3b47ea17a586a68381ad5065e8ce
Author: Johanna Öjeling <51...@users.noreply.github.com>
AuthorDate: Mon Nov 6 23:15:23 2023 +0100
[#29210] Fix SDF progress fraction to not return NaN (#29307)
---
sdks/go/pkg/beam/core/runtime/exec/sdf.go | 11 ++-
sdks/go/pkg/beam/core/runtime/exec/sdf_test.go | 117 ++++++++++++++++++-------
2 files changed, 96 insertions(+), 32 deletions(-)
diff --git a/sdks/go/pkg/beam/core/runtime/exec/sdf.go b/sdks/go/pkg/beam/core/runtime/exec/sdf.go
index b21b47b20ae..3977d31dda7 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/sdf.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/sdf.go
@@ -781,7 +781,7 @@ func (n *ProcessSizedElementsAndRestrictions) singleWindowSplit(ctx context.Cont
func (n *ProcessSizedElementsAndRestrictions) multiWindowSplit(ctx context.Context, f float64, pWeState any, rWeState any) ([]*FullValue, []*FullValue, error) {
// Get the split point in window range, to see what window it falls in.
done, rem := n.rt.GetProgress()
- cwp := done / (done + rem) // Progress in current window.
+ cwp := progressFraction(done, rem) // Progress in current window.
p := (float64(n.currW) + cwp) / float64(n.numW) // Progress of whole element.
sp := p + (f * (1.0 - p)) // Split point in range of entire element [0, 1].
wsp := sp * float64(n.numW) // Split point in window range [0, numW].
@@ -923,7 +923,7 @@ func (n *ProcessSizedElementsAndRestrictions) newSplitResult(ctx context.Context
// DoFns, so 1.0 is only returned once all windows have been processed.
func (n *ProcessSizedElementsAndRestrictions) GetProgress() float64 {
d, r := n.rt.GetProgress()
- frac := d / (d + r)
+ frac := progressFraction(d, r)
if n.numW == 1 {
return frac
@@ -959,6 +959,13 @@ func (n *ProcessSizedElementsAndRestrictions) GetOutputWatermark() map[string]*t
return nil
}
+func progressFraction(done float64, remaining float64) float64 {
+ if done == 0 {
+ return 0
+ }
+ return done / (done + remaining)
+}
+
// SdfFallback is an executor used when an SDF isn't expanded into steps by the
// runner, indicating that the runner doesn't support splitting. It executes all
// the SDF steps together in one unit.
diff --git a/sdks/go/pkg/beam/core/runtime/exec/sdf_test.go b/sdks/go/pkg/beam/core/runtime/exec/sdf_test.go
index a0380796e86..460a7111b11 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/sdf_test.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/sdf_test.go
@@ -715,6 +715,14 @@ func TestAsSplittableUnit(t *testing.T) {
currWindow: 0,
wantProgress: 0.5,
},
+ {
+ name: "SingleWindowZeroWork",
+ windows: testWindows,
+ doneWork: 0.0,
+ remainingWork: 0.0,
+ currWindow: 0,
+ wantProgress: 0.0,
+ },
{
name: "MultipleWindows",
windows: multiWindows,
@@ -724,6 +732,14 @@ func TestAsSplittableUnit(t *testing.T) {
// Progress should be halfway through second window.
wantProgress: 1.5 / 4.0,
},
+ {
+ name: "MultipleWindowsZeroWork",
+ windows: multiWindows,
+ doneWork: 0.0,
+ remainingWork: 0.0,
+ currWindow: 1,
+ wantProgress: 1.0 / 4.0,
+ },
}
for _, test := range tests {
test := test
@@ -776,15 +792,19 @@ func TestAsSplittableUnit(t *testing.T) {
name string
fn *graph.DoFn
frac float64
- doneRt bool // Result that RTracker will return for IsDone.
+ done float64
+ remaining float64
+ isDoneRt bool // Result that RTracker will return for IsDone.
in FullValue
wantPrimaries []*FullValue
wantResiduals []*FullValue
}{
{
- name: "SingleElem",
- fn: dfn,
- frac: 0.5,
+ name: "SingleElem",
+ fn: dfn,
+ frac: 0.5,
+ done: 0.0,
+ remaining: 1.0,
in: FullValue{
Elm: &FullValue{
Elm: 1,
@@ -823,9 +843,11 @@ func TestAsSplittableUnit(t *testing.T) {
}},
},
{
- name: "SingleElemStatefulWatermarkEstimating",
- fn: statefulWeFn,
- frac: 0.5,
+ name: "SingleElemStatefulWatermarkEstimating",
+ fn: statefulWeFn,
+ frac: 0.5,
+ done: 0.0,
+ remaining: 1.0,
in: FullValue{
Elm: &FullValue{
Elm: 1,
@@ -864,9 +886,11 @@ func TestAsSplittableUnit(t *testing.T) {
}},
},
{
- name: "KvElem",
- fn: kvdfn,
- frac: 0.5,
+ name: "KvElem",
+ fn: kvdfn,
+ frac: 0.5,
+ done: 0.0,
+ remaining: 1.0,
in: FullValue{
Elm: &FullValue{
Elm: &FullValue{
@@ -914,10 +938,12 @@ func TestAsSplittableUnit(t *testing.T) {
}},
},
{
- name: "DoneRTracker",
- fn: dfn,
- doneRt: true,
- frac: 0.5,
+ name: "DoneRTracker",
+ fn: dfn,
+ frac: 0.5,
+ done: 0.0,
+ remaining: 1.0,
+ isDoneRt: true,
in: FullValue{
Elm: &FullValue{
Elm: 1,
@@ -936,9 +962,11 @@ func TestAsSplittableUnit(t *testing.T) {
{
// MultiWindow split where split point lands inside currently
// processing restriction tracker.
- name: "MultiWindow/RestrictionSplit",
- fn: dfn,
- frac: 0.125, // Should be in the middle of the first (current) window.
+ name: "MultiWindow/RestrictionSplit",
+ fn: dfn,
+ frac: 0.125, // Should be in the middle of the first (current) window.
+ done: 0.0,
+ remaining: 1.0,
in: FullValue{
Elm: &FullValue{
Elm: 1,
@@ -990,9 +1018,11 @@ func TestAsSplittableUnit(t *testing.T) {
{
// MultiWindow split where the split lands outside the current
// window, and performs a window boundary split instead.
- name: "MultiWindow/WindowBoundarySplit",
- fn: dfn,
- frac: 0.55,
+ name: "MultiWindow/WindowBoundarySplit",
+ fn: dfn,
+ frac: 0.55,
+ done: 0.0,
+ remaining: 1.0,
in: FullValue{
Elm: &FullValue{
Elm: 1,
@@ -1033,10 +1063,12 @@ func TestAsSplittableUnit(t *testing.T) {
{
// Tests that a MultiWindow split with a Done RTracker will
// fallback to a window boundary split.
- name: "MultiWindow/DoneRTrackerSplit",
- fn: dfn,
- frac: 0.125,
- doneRt: true,
+ name: "MultiWindow/DoneRTrackerSplit",
+ fn: dfn,
+ frac: 0.125,
+ done: 0.0,
+ remaining: 1.0,
+ isDoneRt: true,
in: FullValue{
Elm: &FullValue{
Elm: 1,
@@ -1077,9 +1109,34 @@ func TestAsSplittableUnit(t *testing.T) {
{
// Test that if a window boundary split lands at the end of an
// element, it results in a no-op.
- name: "MultiWindow/NoResidual",
- fn: dfn,
- frac: 0.95, // Should round to end of element and cause a no-op.
+ name: "MultiWindow/NoResidual",
+ fn: dfn,
+ frac: 0.95, // Should round to end of element and cause a no-op.
+ done: 0.0,
+ remaining: 1.0,
+ in: FullValue{
+ Elm: &FullValue{
+ Elm: 1,
+ Elm2: &FullValue{
+ Elm: &VetRestriction{ID: "Sdf"},
+ Elm2: false,
+ },
+ },
+ Elm2: 1.0,
+ Timestamp: testTimestamp,
+ Windows: testMultiWindows,
+ },
+ wantPrimaries: []*FullValue{},
+ wantResiduals: []*FullValue{},
+ },
+ {
+ // Tests that an RTracker progress of 0.0 done and 0.0 remaining
+ // is treated as a current window progress of 0.0.
+ name: "MultiWindow/ZeroWork",
+ fn: dfn,
+ frac: 0.95,
+ done: 0.0,
+ remaining: 0.0,
in: FullValue{
Elm: &FullValue{
Elm: 1,
@@ -1104,9 +1161,9 @@ func TestAsSplittableUnit(t *testing.T) {
node := &ProcessSizedElementsAndRestrictions{PDo: n}
node.rt = &SplittableUnitRTracker{
VetRTracker: VetRTracker{Rest: test.in.Elm.(*FullValue).Elm2.(*FullValue).Elm.(*VetRestriction)},
- Done: 0,
- Remaining: 1.0,
- ThisIsDone: test.doneRt,
+ Done: test.done,
+ Remaining: test.remaining,
+ ThisIsDone: test.isDoneRt,
}
node.elm = &test.in
node.numW = len(test.in.Windows)