[Python]Enable state cache to 100 MB

[AnandInguva]

Enable state_cache_size = 100 MB for python SDK.
Fixes: #28770

state_cache_size can be enabled using --state_cache_size=<X>MB. state_cache_size should be in terms of Megabytes.

EDIT:

From the doc - https://docs.google.com/document/u/1/d/1gllYsIFqKt4TWAxQmXU_-sw7SLnur2Q69d__N0XBMdE/edit?usp=drive_open&ouid=102749919556839394679, the consensus is to add a pipeline option named max_cache_memory_usage_mb and explain in Beam docs and runner docs on what this option is and how this options works.

---

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

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• 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.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[codecov]

Codecov Report

Merging #28781 (aa94309) into master (a93fa51) will decrease coverage by 0.05%.
Report is 24 commits behind head on master.
The diff coverage is 85.71%.

@@            Coverage Diff             @@
##           master   #28781      +/-   ##
==========================================
- Coverage   38.36%   38.32%   -0.05%     
==========================================
  Files         687      688       +1     
  Lines      101745   101833      +88     
==========================================
- Hits        39037    39027      -10     
- Misses      61129    61230     +101     
+ Partials     1579     1576       -3     

Flag	Coverage Δ
python	29.94% <85.71%> (-0.04%)	⬇️

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

Files	Coverage Δ
...dks/python/apache_beam/options/pipeline_options.py	64.91% <100.00%> (+0.07%)	⬆️
...apache_beam/runners/portability/portable_runner.py	28.27% <ø> (ø)
...thon/apache_beam/runners/worker/sdk_worker_main.py	66.66% <83.33%> (+0.77%)	⬆️

... and 8 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[AnandInguva]

R: @tvalentyn

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[AnandInguva]

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java

Line 247 in f30f6c5

Integer getWorkerCacheMb();

Java has default of 100 MB

Do we want to have a similar flag to workerCacheMB? state_cache_size to state_cache_size_mb or worker_cache_mb?

[tvalentyn]

state is a technical term that's not very user-friendly; it is good to have consistent naming. @lostluck does Go have state cache as well?

[lostluck]

state is a technical term that's not very user-friendly; it is good to have consistent naming. @lostluck does Go have state cache as well?

It does, but it's not memory aware, just element sized.

Technically, it's used for cross bundle applications across the State API which side inputs also use.

Unless the cache value also applies to the Combiner Lifting cache, it wouldn't be a true "worker" cache, vs a "state" cache.

[tvalentyn]

pretty sure there is already a helper that does this parsing.

[tvalentyn]

Any concerns to define the 100mb default here?

[AnandInguva]

Current flow: If it is None here, it gives us an opportunity to look in --experiements for state_cache_size.

If the value is defined here as 100 MB, and if the user passes --experiments=state_cache_size, we should override 100 MB for the --experiments=state_cache_size.

I don't see any concerns of setting default here. might need to change some code though

[damccorm]

Could you add a short description of what this cache is and link to https://beam.apache.org/releases/pydoc/2.50.0/apache_beam.options.pipeline_options.html#module-apache_beam.options.pipeline_options?

[damccorm]

Particularly, we should make it clear here that this impacts both user state and side inputs

[damccorm]

Run Python_PVR_Flink PreCommit

[tvalentyn]

nit: consider following wording

            'Size of the SDK Harness cache to store user state and side inputs '
            'in MB. Default is 100MB. If the cache is full, least recently '
            'used elements will be evicted.  This cache is per '
            'each SDK Harness instance. SDK Harness  is a component  responsible '
            'for executing the user code and communicating with the runner. ' 
            'Depending on the runner, '
            'there may be more than one SDK Harness process running on the same worker node. '
            'Increasing cache size  might improve performance of some pipelines, but can lead to an increase '
            'in memory consumption and OOM errors if workers are not appropriately provisioned.'

[AnandInguva]

Merging this since tests pass.