[BEAM-9935] Respect allowed split points in Python. #11653

robertwb · 2020-05-09T10:26:24Z

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.

robertwb · 2020-05-09T10:26:45Z

R: @lukecwik

lukecwik · 2020-05-11T16:41:49Z

sdks/python/apache_beam/runners/worker/bundle_processor.py

+      keep_of_element_remainder = keep / (1 - current_element_progress)
+      # If it's less than what's left of the current element,
+      # try splitting at the current element.
+      if (keep_of_element_remainder < 1 and is_valid_split_point(index) and


allowed_split_points definition is too vague in the case of multiple active elements and we need to scope it down to mean the set of allowed first_residual_element indices.

Agreed. I've added clarification to the proto.

lukecwik · 2020-05-11T16:52:01Z

sdks/python/apache_beam/runners/worker/bundle_processor.py

+    # pylint: disable=round-builtin
+    stop_index = index + max(1, int(round(current_element_progress + keep)))
+    if allowed_split_points and stop_index not in allowed_split_points:
+      allowed_split_points = sorted(allowed_split_points)


We should make it an error to have duplicate split points in allowed_split_points.

I agree that there's never a reason to have duplicates, but that wouldn't impact this code here (and I don't think should result in failure).

lukecwik · 2020-05-11T16:56:52Z

sdks/python/apache_beam/runners/worker/bundle_processor.py

+      else:
+        prev = allowed_split_points[closest - 1]
+        next = allowed_split_points[closest]
+        if index < prev and stop_index - prev < next - stop_index:


I think you should comment that your choosing the closer of the two points here.

lukecwik · 2020-05-11T16:57:55Z

sdks/python/apache_beam/runners/worker/bundle_processor_test.py

+    self.assertEqual(self.split(2, 0, 0.5, 16), simple_split(9))
+    self.assertEqual(self.split(6, 0, 0.5, 16), simple_split(11))
+
+  def test_split_with_element_progres(self):


Suggested change

def test_split_with_element_progres(self):

def test_split_with_element_progress(self):

lukecwik · 2020-05-11T17:01:54Z

sdks/python/apache_beam/runners/worker/bundle_processor_test.py

+    self.assertEqual(
+        self.split(0, 0, 0.25, 16, allowed=(2, 3, 6)), simple_split(3))
+
+    self.assertEqual(


It would be good to either add comments or break out the tests to separate methods to describe the different scenarios such as round to closest.

I kept them in the same methods, because it's easier to understand the values relative to the prior examples.

lukecwik · 2020-05-11T17:02:49Z

sdks/python/apache_beam/runners/worker/bundle_processor_test.py

+    self.assertEqual(
+        self.sdf_split(0, 0, 0.12, 4), (-1, 'Primary(0.5)', 'Residual(0.5)', 1))


Suggested change

self.assertEqual(

self.sdf_split(0, 0, 0.12, 4), (-1, 'Primary(0.5)', 'Residual(0.5)', 1))

self.assertEqual(

self.sdf_split(0, 0, 0.125, 4), (-1, 'Primary(0.5)', 'Residual(0.5)', 1))

I had this originally, but it bumped the formatting, and .12 was close enough. I can change this back.

I found it confusing that the rounding dropped the fractional part when I was running through the scenarios.

lukecwik · 2020-05-11T19:05:08Z

model/fn-execution/src/main/proto/beam_fn_api.proto

    // empty, there are no constraints on where to split.
+    // Specifically, the first_residual_element of a split result must be an
+    // allowed split point, and the last_primary_element must immediately
+    // preceded an allowed split point.


Suggested change

// preceded an allowed split point.

// precede an allowed split point.

lukecwik · 2020-05-11T19:07:55Z

CC: @lostluck @youngoli @boyuanzz We'll want to duplicate this logic within Java/Go for allowed split point selection.

robertwb · 2020-05-11T20:56:31Z

Run Java PreCommit

lostluck · 2020-05-11T21:51:39Z

Ack.

youngoli · 2020-05-11T22:16:22Z

Just wanna confirm, the allowed_split_points dictate splits at element boundaries only right?

lukecwik · 2020-05-11T22:59:07Z

The allowed split points tell you what is the valid set of return values for first_residual_index and last_primary_index+1 are so it applies to splittable DoFns as in if the runner tells you it can't split at position Y and your currently processing Y, there is no point of forwarding the split call to the SDF from the gRPC read port.

lukecwik · 2020-05-11T22:59:57Z

@ibzib, can you cherry pick this into the 2.21 branch. This would make the associated BEAM-9935 a non-blocking issue for 2.21 afterwards.

sdks/python/apache_beam/runners/worker/bundle_processor.py

[BEAM-9935] Respect allowed split points in Python.

3dad2e5

probot-autolabeler bot added the python label May 9, 2020

lukecwik reviewed May 11, 2020

View reviewed changes

More comments.

773854c

probot-autolabeler bot added the model label May 11, 2020

lukecwik approved these changes May 11, 2020

View reviewed changes

lukecwik merged commit cad0333 into apache:master May 11, 2020

ibzib mentioned this pull request May 11, 2020

[BEAM-9935] [release-2.21.0] Respect allowed split points in Python. #11671

Merged

4 tasks

lostluck reviewed May 11, 2020

View reviewed changes

sdks/python/apache_beam/runners/worker/bundle_processor.py Show resolved Hide resolved

boyuanzz mentioned this pull request May 12, 2020

[BEAM-9935, BEAM-3836] Respect allowed split points in Java #11688

Merged

4 tasks

	def test_split_with_element_progres(self):
	def test_split_with_element_progress(self):

		self.assertEqual(
		self.sdf_split(0, 0, 0.12, 4), (-1, 'Primary(0.5)', 'Residual(0.5)', 1))

	// preceded an allowed split point.
	// precede an allowed split point.

[BEAM-9935] Respect allowed split points in Python. #11653

[BEAM-9935] Respect allowed split points in Python. #11653

Uh oh!

Conversation

robertwb commented May 9, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

Uh oh!

robertwb commented May 9, 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!

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!

lukecwik commented May 11, 2020

Uh oh!

robertwb commented May 11, 2020

Uh oh!

lostluck commented May 11, 2020

Uh oh!

youngoli commented May 11, 2020

Uh oh!

lukecwik commented May 11, 2020

Uh oh!

lukecwik commented May 11, 2020

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants