Add @yields_batches and @yields_elements

[TheNeuralBit]

This PR adds two new decorators, @yields_batches and @yields_elements, which override the default interpretation for the output of DoFn.process and DoFn.process_batch. These decorators are handled via changes to OutputProcessor (now renamed to OutputHandler for clarity), which branches depending on whether batches or elements are expected in the results iterable.

Where possible, common logic in OutputHandler was extracted into private helper methods that are re-used. The .pxd definition for these methods indicates they should be inlined in cython-generated code so they don't affect performance.

Fixes #21656

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[TheNeuralBit]

Despite the inlining this PR seems to have a minor effect (0.1-0.2 0.01-0.02 microseconds/element) on map_fn_microbenchmark, presumably due to the new branch. Benchmarks on my desktop with Intel Xeon W-2135 CPU.

c779710:

Fixed cost   1.6063439090633391
Per-element  7.317467689514159e-07                                                                                  
R^2          0.9972847477152673

This PR:

Fixed cost   1.630842854329745
Per-element  7.487253376931855e-07
R^2          0.998137270424781

[codecov]

Codecov Report

Merging #19268 (e6c3735) into master (c779710) will decrease coverage by 0.06%.
The diff coverage is 85.81%.

@@            Coverage Diff             @@
##           master   #19268      +/-   ##
==========================================
- Coverage   74.09%   74.02%   -0.07%     
==========================================
  Files         697      698       +1     
  Lines       91986    92209     +223     
==========================================
+ Hits        68154    68258     +104     
- Misses      22583    22702     +119     
  Partials     1249     1249              

Flag	Coverage Δ
python	83.62% <85.81%> (-0.13%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...on/apache_beam/runners/direct/sdf_direct_runner.py	35.53% <75.00%> (ø)
sdks/python/apache_beam/runners/common.py	88.68% <83.33%> (+0.74%)	⬆️
sdks/python/apache_beam/transforms/core.py	92.38% <90.47%> (+0.08%)	⬆️
sdks/python/apache_beam/utils/windowed_value.py	92.52% <100.00%> (+0.17%)	⬆️
...ache_beam/runners/dataflow/dataflow_job_service.py	50.00% <0.00%> (-12.17%)	⬇️
sdks/python/apache_beam/dataframe/io.py	88.78% <0.00%> (-3.26%)	⬇️
...eam/runners/portability/fn_api_runner/fn_runner.py	87.51% <0.00%> (-2.50%)	⬇️
...nners/portability/fn_api_runner/worker_handlers.py	77.89% <0.00%> (-1.45%)	⬇️
sdks/python/apache_beam/internal/dill_pickler.py	84.89% <0.00%> (-1.44%)	⬇️
sdks/python/apache_beam/runners/direct/executor.py	96.46% <0.00%> (-0.55%)	⬇️
... and 15 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update c779710...e6c3735. Read the comment docs.

[robertwb]

The "per element" line seems to be missing from the benchmarks. What is the relative difference?

[TheNeuralBit]

Apologies, somehow the formatting was such that the per-element line was hidden after a bunch of spaces. It's there now.

[robertwb]

Thanks. The difference looks more like 0.01-0.02 microseconds, or a couple percent, which is more acceptable.

[TheNeuralBit]

retest this please

[TheNeuralBit]

Run Portable_Python PreCommit

[TheNeuralBit]

Run PythonDocker PreCommit

[TheNeuralBit]

Run PythonDocker PreCommit

[TheNeuralBit]

Run Portable_Python PreCommit