[BEAM-8739] Consistently use with Pipeline(...) syntax

[robertwb]

---

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.
• 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	Flink	Gearpump	Samza	Spark
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]

R: @ibzib lint and tests are now all happy.

[ibzib]

Before I actually go through all of this, a few general comments:

• I see you did a lot of these conversions automatically. I'd be curious what the scripts looked like.
• Ideally we would incorporate this into the linting process, but there's probably enough edge cases to make this infeasible.
• I noticed (as you pointed out on the Jira) that a few tests in here originally didn't call pipeline.run() at all (whether through with or directly). My biggest concern with this PR is that it might go the other direction and somewhere remove an existing call to pipeline.run(). (All other errors I could think of would have resulted in test failures, so you would have caught them already.)
• If one of the goals here is to encourage everyone to use this syntax, it'd be important to cover the code snippets on the website as well.

[robertwb]

Before I actually go through all of this, a few general comments:

• I see you did a lot of these conversions automatically. I'd be curious what the scripts looked like.

Yes, I'll attach the script. Nothing fancy.

• Ideally we would incorporate this into the linting process, but there's probably enough edge cases to make this infeasible.

Yeah. I'm hoping that mostly people will follow conventions of the surrounding code.

• I noticed (as you pointed out on the Jira) that a few tests in here originally didn't call pipeline.run() at all (whether through with or directly). My biggest concern with this PR is that it might go the other direction and somewhere remove an existing call to pipeline.run(). (All other errors I could think of would have resulted in test failures, so you would have caught them already.)

If something is converted to the with syntax, run() will always be called.

• If one of the goals here is to encourage everyone to use this syntax, it'd be important to cover the code snippets on the website as well.

Yes. That code is harder to auto (or manually) convert given the amount of mocking and ordering in which things are included/excluded. Taking a pass at the examples and snippets is certainly something that should be done and I'll keep this ticket open until then.

[robertwb]

Just a bunch of regular expression munging...

import re
import sys

for file in sys.argv[1:]:
  with open(file) as fin:
   contents = fin.read().split('\n')
  new_contents = []
  in_pipeline = False
  indent = None
  converted_pipelines = 0
  total_pipelines = 0
  for ix, line in enumerate(contents):
    if in_pipeline:
      if line.startswith('%s%s.run()' % (indent, pipeline_var)):
        converted_pipelines += 1
        new_contents.append('%swith %s as %s:' % (indent, pipeline_expr, pipeline_var))
        for b in body:
          new_contents.append('  ' + b if b.strip() and b[0] != '#' else b)
#         if body[-1]:
#           new_contents.append('\n')
        in_pipeline = False
      elif not line.startswith(indent) and line.strip() and not re.match('^ *#', line):
        print('END', file, ix, list(filter(lambda x: x, [b.strip() for b in body]))[-2:])
        new_contents.append(pipeline_init)
        new_contents.extend(body)
        new_contents.append(line)
        in_pipeline = False
      else:
        body.append(line)
    else:
      m = re.match(r'^( *)(\S+) = (\S*Pipeline[(].*)', line)
      if m:
        total_pipelines += 1
        in_pipeline = True
        indent = m.group(1)
        pipeline_init = line
        pipeline_var = m.group(2)
        pipeline_expr = m.group(3)
        body = []
      else:
        new_contents.append(line)

  if in_pipeline and body:
    new_contents.append(pipeline_init)
    new_contents.extend(body)

  if new_contents != contents:
    with open(file, 'w') as fout:
      fout.write('\n'.join(new_contents))
  if total_pipelines:
    print(file, converted_pipelines, total_pipelines)
    print()

[robertwb]

Post-holiday ping. Rebased on latest master.

[ibzib]

Please fix lint errors and precommit (causes commented inline).

[ibzib]

LGTM