[BEAM-8872] Support split at fraction for OffsetRangeTracker #11418

boyuanzz · 2020-04-14T20:26:15Z

Computation logic is copied from python OffsetRestrictionTracker.try_split()

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

lukecwik

I looked at the Python implementation as well and I think we can improve it so that trySplit returns a split even if nothing has been claimed and so that we can get rid of checkpointed.

lukecwik · 2020-04-14T22:28:09Z

...ava/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTracker.java

We can split before any successfully claimed block by returning [from, to) and updating the current range to be [from, from)

This makes sense in some cases where we want to handoff all the work to someone else for the active element while this bundle finishes other processing.

Allowing split before first claiming makes sense to me. Python has already allowed that.

lukecwik · 2020-04-14T22:28:42Z

...ava/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTracker.java

Why do we need checkpointed?

Shouldn't the range restriction change so that to becomes lastClaimed (or from if nothing has been claimed)?

Just return early since we know there is no more split after checkpointing.

Changing the state makes the code more complicated though since the bounds checking varies based upon whether you got 0.0 or 0.00001 which is likely to produce the same final state with one having an additional boolean being set.

boyuanzz · 2020-04-16T17:12:47Z

Run Java PreCommit

boyuanzz · 2020-04-16T19:33:19Z

Run RAT PreCommit

lukecwik · 2020-04-16T19:44:25Z

...core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTrackerTest.java

-    OffsetRange checkpoint = tracker.trySplit(0).getResidual();
+    SplitResult res = tracker.trySplit(0);
+    assertEquals(new OffsetRange(100, 100), res.getPrimary());
+    assertEquals(new OffsetRange(100, 200), res.getResidual());


primary and residual shouldn't have the same value, primary should be an empty range like [100, 100)

In this test case, the expected primary is [100, 100) and the expected residual is [100, 200)

Thanks, that was my mistake, I read both lines as being [100, 200)

lukecwik · 2020-04-16T19:46:09Z

...n/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java

+        // residual is null when the entire restriction has been processed.
        if (processContext.numClaimedBlocks > 0) {
-          residual = checkNotNull(processContext.takeCheckpointNow());
+          residual = processContext.takeCheckpointNow();


takeCheckpointNow should work regardless whether numClaimedBlocks > 0 or not.

Even if tryClaim never happens, the watermark may advance.

I guess the original assumption is, checkpoint should happen after at least one tryClaim called. Since we change the assumption, the numClaimedBlocks can also be removed.

lukecwik · 2020-04-16T19:48:09Z

...n/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java

-          residual = checkNotNull(processContext.takeCheckpointNow());
+          residual = processContext.takeCheckpointNow();
          processContext.tracker.checkDone();
        } else {


The comments below will likely need updating

lukecwik · 2020-04-16T20:07:22Z

...core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTrackerTest.java

-    assertEquals(new OffsetRange(100, 161), tracker.currentRestriction());
-    assertEquals(new OffsetRange(161, 200), checkpoint);
+    assertNull(tracker.trySplit(0));
+  }


Suggested change

}

tracker.checkDone();

}

lukecwik · 2020-04-16T20:10:06Z

...core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTrackerTest.java

+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;

 import org.apache.beam.sdk.io.range.OffsetRange;


Can we add tests to verify tryClaim(0), tryClaim(0.1), tryClaim(1) on an empty range like [100, 100)

Can we also add tests to verify the behavior of tryClaim(1) on range [100, 200)

trySplit right?

Yes. Your right. Need more sleep.

lukecwik · 2020-04-16T20:11:08Z

...rs/core-java/src/main/java/org/apache/beam/runners/core/SplittableProcessElementInvoker.java

-      }
      this.residualRestriction = residualRestriction;
      this.futureOutputWatermark = futureOutputWatermark;
      this.futureWatermarkEstimatorState = futureWatermarkEstimatorState;


I believe the comment below could be incorrect. If we get stop(), we shouldn't have a residual restriction.

boyuanzz · 2020-04-17T20:22:18Z

Run Java PreCommit

iemejia · 2020-04-21T08:03:16Z

It seems this PR broke the ValidatesRunner tests for most runners. Can you guys PTAL. SplittableDoFnTest is failing with NullPointerException

For ref.
https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/7267/
https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/7297/

mxm · 2020-04-21T08:14:37Z

Running post commits here would have caught this (especially considering this is a new feature). The Flink ValidatesRunner test suite takes less than 10 minutes to complete. Perhaps we should move it to pre commit to be sure to catch these.

lukecwik · 2020-04-21T15:39:21Z

Fix in #11475

lukecwik · 2020-04-21T15:45:09Z

Fix in #11475

…11418) * [BEAM-8872] Support split at fraction for OffsetRangeTracker * Address comments * Update checkDone

probot-autolabeler bot added the java label Apr 14, 2020

lukecwik requested changes Apr 14, 2020

View reviewed changes

probot-autolabeler bot added core runners labels Apr 15, 2020

boyuanzz force-pushed the split branch from 49ccbc7 to e056d31 Compare April 15, 2020 23:13

[BEAM-8872] Support split at fraction for OffsetRangeTracker

800d20d

boyuanzz force-pushed the split branch from e056d31 to 800d20d Compare April 16, 2020 19:03

lukecwik requested changes Apr 16, 2020

View reviewed changes

Boyuan Zhang added 2 commits April 16, 2020 16:15

Address comments

5907ec3

Update checkDone

749174c

lukecwik approved these changes Apr 18, 2020

View reviewed changes

lukecwik merged commit fa4f418 into apache:master Apr 18, 2020

mxm mentioned this pull request Apr 21, 2020

[BEAM-9733] Always let ImpulseSourceFunction emit a final watermark #11362

Merged

yirutang pushed a commit to yirutang/beam that referenced this pull request Jul 23, 2020

[BEAM-8872] Support split at fraction for OffsetRangeTracker (apache#…

c5fad67

…11418) * [BEAM-8872] Support split at fraction for OffsetRangeTracker * Address comments * Update checkDone

[BEAM-8872] Support split at fraction for OffsetRangeTracker #11418

[BEAM-8872] Support split at fraction for OffsetRangeTracker #11418

Uh oh!

Conversation

boyuanzz commented Apr 14, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

Uh oh!

lukecwik left a comment

Choose a reason for hiding this comment

Uh oh!

lukecwik Apr 14, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

boyuanzz commented Apr 16, 2020

Uh oh!

boyuanzz commented Apr 16, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

boyuanzz commented Apr 17, 2020

Uh oh!

iemejia commented Apr 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mxm commented Apr 21, 2020

Uh oh!

lukecwik commented Apr 21, 2020

Uh oh!

lukecwik commented Apr 21, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

boyuanzz commented Apr 14, 2020 •

edited

Loading

lukecwik Apr 14, 2020 •

edited

Loading

iemejia commented Apr 21, 2020 •

edited

Loading