Re: Write-through-cache in State logic

[Maximilian Michels]

Just to clarify, the repeated list of cache tokens in the process
bundle request is used to validate reading *and* stored when writing?
In that sense, should they just be called version identifiers or
something like that?

We could call them version identifiers, though cache tokens were always 
a means to identify versions of a state.

On 28.08.19 11:10, Maximilian Michels wrote:
cachetools sounds like a fine choice to me.

For the first version I've implemented a simple LRU cache. If you want 
to have a look: 
https://github.com/apache/beam/pull/9418/files#diff-ed2d70e99442b6e1668e30409d3383a6R60 


Open up a PR for the proto changes and we can work through any minor 
comments there.

Proto changes: https://github.com/apache/beam/pull/9440


Thanks,
Max

On 27.08.19 23:00, Robert Bradshaw wrote:
Just to clarify, the repeated list of cache tokens in the process
bundle request is used to validate reading *and* stored when writing?
In that sense, should they just be called version identifiers or
something like that?

On Tue, Aug 27, 2019 at 11:33 AM Maximilian Michels <m...@apache.org> 
wrote:

Thanks. Updated:

message ProcessBundleRequest {
   // (Required) A reference to the process bundle descriptor that 
must be
   // instantiated and executed by the SDK harness.
   string process_bundle_descriptor_reference = 1;

   // A cache token which can be used by an SDK to check for the 
validity
   // of cached elements which have a cache token associated.
   message CacheToken {

     // A flag to indicate a cache token is valid for user state.
     message UserState {}

     // A flag to indicate a cache token is valid for a side input.
     message SideInput {
       // The id of a side input.
       string side_input = 1;
     }

     // The scope of a cache token.
     oneof type {
       UserState user_state = 1;
       SideInput side_input = 2;
     }

     // The cache token identifier which should be globally unique.
     bytes token = 10;
   }

   // (Optional) A list of cache tokens that can be used by an SDK to 
reuse
   // cached data returned by the State API across multiple bundles.
   repeated CacheToken cache_tokens = 2;
}

On 27.08.19 19:22, Lukasz Cwik wrote:

SideInputState -> SideInput (side_input_state -> side_input)
+ more comments around the messages and the fields.


On Tue, Aug 27, 2019 at 10:18 AM Maximilian Michels <m...@apache.org> 
wrote:

We would have to differentiate cache tokens for user state and side 
inputs. How about something like this?

message ProcessBundleRequest {
   // (Required) A reference to the process bundle descriptor that 
must be
   // instantiated and executed by the SDK harness.
   string process_bundle_descriptor_reference = 1;

   message CacheToken {

     message UserState {
     }

     message SideInputState {
       string side_input_id = 1;
     }

     oneof type {
       UserState user_state = 1;
       SideInputState side_input_state = 2;
     }

     bytes token = 10;
   }

   // (Optional) A list of cache tokens that can be used by an SDK 
to reuse
   // cached data returned by the State API across multiple bundles.
   repeated CacheToken cache_tokens = 2;
}

-Max

On 27.08.19 18:43, Lukasz Cwik wrote:

The bundles view of side inputs should never change during 
processing and should have a point in time snapshot.

I was just trying to say that the cache token for side inputs being 
deferred till side input request time simplified the runners 
implementation since that is conclusively when the runner would need 
to take a look at the side input. Putting them as part of the 
ProcesBundleRequest complicates that but does make the SDK 
implementation significantly simpler which is a win.

On Tue, Aug 27, 2019 at 9:14 AM Maximilian Michels <m...@apache.org> 
wrote:

Thanks for the quick response.

Just to clarify, the issue with versioning side input is also present
when supplying the cache tokens on a request basis instead of per
bundle. The SDK never knows when the Runner receives a new version of
the side input. Like you pointed out, it needs to mark side inputs as
stale and generate new cache tokens for the stale side inputs.

The difference between per-request tokens and per-bundle tokens 
would be
that the side input can only change after a bundle completes vs. 
during
the bundle. Side inputs are always fuzzy in that regard because 
there is
no precise instance where side inputs are atomically updated, other 
than
the assumption that they eventually will be updated. In that regard
per-bundle tokens for side input seem to be fine.

All of the above is not an issue for user state, as its cache can 
remain
valid for the lifetime of a Runner<=>SDK Harness connection. A simple
solution would be to not cache side input because there are many cases
where the caching just adds additional overhead. However, I can also
imagine cases where side input is valid forever and caching would be
very beneficial.

For the first version I want to focus on user state because that's 
where
I see the most benefit for caching. I don't see a problem though 
for the
Runner to detect new side input and reflect that in the cache tokens
supplied for a new bundle.

-Max

On 26.08.19 22:27, Lukasz Cwik wrote:
Your summary below makes sense to me. I can see that recovery from
rolling back doesn't need to be a priority and simplifies the 
solution
for user state caching down to one token.

Providing cache tokens upfront does require the Runner to know what
"version" of everything it may supply to the SDK upfront (instead 
of on
request) which would mean that the Runner may need to have a mapping
from cache token to internal version identifier for things like side
inputs which are typically broadcast. The Runner would also need 
to poll
to see if the side input has changed in the background to not block
processing bundles with "stale" side input data.

Ping me once you have the Runner PR updated and I'll take a look 
again.

On Mon, Aug 26, 2019 at 12:20 PM Maximilian Michels <m...@apache.org
<mailto:m...@apache.org>> wrote:

     Thank you for the summary Luke. I really appreciate the 
effort you put
     into this!

      > Based upon your discussion you seem to want option #1

     I'm actually for option #2. The option to cache/invalidate 
side inputs
     is important, and we should incorporate this in the design. 
That's why
     option #1 is not flexible enough. However, a first 
implementation could
     defer caching of side inputs.

     Option #3 was my initial thinking and the first version of 
the PR, but I
     think we agreed that there wouldn't be much gain from keeping 
a cache
     token per state id.

     Option #4 is what is specifically documented in the reference 
doc and
     already part of the Proto, where valid tokens are provided 
for each new
     bundle and also as part of the response of a get/put/clear. 
We mentioned
     that the reply does not have to be waited on synchronously (I 
mentioned
     it even), but it complicates the implementation. The idea 
Thomas and I
     expressed was that a response is not even necessary if we assume
     validity of the upfront provided cache tokens for the 
lifetime of a
     bundle and that cache tokens will be invalidated as soon as 
the Runner
     fails in any way. This is naturally the case for Flink 
because it will
     simply "forget" its current cache tokens.

     I currently envision the following schema:

     Runner
     ======

     - Runner generates a globally unique cache token, one for 
user state and
     one for each side input

     - The token is supplied to the SDK Harness for each bundle 
request

     - For the lifetime of a Runner<=>SDK Harness connection this 
cache token
     will not change
     - Runner will generate a new token if the connection/key 
space changes
     between Runner and SDK Harness


     SDK
     ===

     - For each bundle the SDK worker stores the list of valid 
cache tokens
     - The SDK Harness keep a global cache across all its (local) 
workers
     which is a LRU cache: state_key => (cache_token, value)
     - get: Lookup cache using the valid cache token for the 
state. If no
     match, then fetch from Runner and use the already available 
token for
     caching
     - put: Put value in cache with a valid cache token, put value 
to pending
     writes which will be flushed out latest when the bundle ends
     - clear: same as put but clear cache

     It does look like this is not too far off from what you were 
describing.
     The main difference is that we just work with a single cache 
token. In
     my opinion we do not need the second cache token for writes, 
as long as
     we ensure that we generate a new cache token if the
     bundle/checkpoint fails.

     I have a draft PR
        for the Runner: https://github.com/apache/beam/pull/9374
        for the SDK: https://github.com/apache/beam/pull/9418

     Note that the Runner PR needs to be updated to fully 
reflected the above
     scheme. The SDK implementation is WIP. I want to make sure 
that we
     clarify the design before this gets finalized.

     Thanks again for all your comments. Much appreciated!

     Cheers,
     Max

     On 26.08.19 19:58, Lukasz Cwik wrote:
      > There were originally a couple of ideas around how caching 
could
     work:
      > 1) One cache token for the entire bundle that is supplied up
     front. The
      > SDK caches everything using the given token. All
     reads/clear/append for
      > all types of state happen under this token. Anytime a side 
input
      > changes, key processing partition range changes or a 
bundle fails to
      > process, the runner chooses a new cache token effectively
     invalidating
      > everything in the past>
      > 2) One cache token per type of state that is supplied up 
front.
      > The SDK caches all requests for a given type using the 
given cache
      > token. The runner can selectively choose which type to 
keep and
     which to
      > invalidate. Bundle failure and key processing partition 
changes
      > invalidate all user state, side input change invalidates 
all side
     inputs.
      >
      > 3) One cache token per state id that is supplied up front.
      > The SDK caches all requests for the given state id using the
     given cache
      > token. The runner can selectively choose which to 
invalidate and
     which
      > to keep. Bundle failure and key processing partition changes
     invalidate
      > all user state, side input changes only invalidate the 
side input
     that
      > changed.
      >
      > 4) A cache token on each read/clear/append that is 
supplied on the
      > response of the call with an initial valid set that is 
supplied at
      > start. The runner can selectively choose which to keep on 
start.
     Bundle
      > failure allows runners to "roll back" to a known good 
state by
     selecting
      > the previous valid cache token as part of the initial set. 
Key
      > processing partition changes allow runners to keep cached 
state that
      > hasn't changed since it can be tied to a version number of 
the state
      > itself as part of the initial set. Side input changes only 
invalidate
      > the side input that changed.
      >
      > Based upon your discussion you seem to want option #1 which
     doesn't work
      > well with side inputs clearing cached state. If we want to 
have user
      > state survive a changing side input, we would want one of 
the other
      > options. I do agree that supplying the cache token upfront is
      > significantly simpler. Currently the protos are setup for 
#4 since it
      > was the most flexible and at the time the pros outweighed 
the cons.
      >
      > I don't understand why you think you need to wait for a 
response
     for the
      > append/clear to get its cache token since the only reason you
     need the
      > cache token is that you want to use that cached data when
     processing a
      > different bundle. I was thinking that the flow on the SDK 
side
     would be
      > something like (assuming there is a global cache of cache 
token
     -> (map
      > of state key -> data))
      > 1) Create a local cache of (map of state key -> data) 
using the
     initial
      > set of valid cache tokens
      > 2) Make all mutations in place on local cache without 
waiting for
     response.
      > 3) When response comes back, update global cache with new 
cache
     token ->
      > (map of state key -> data)) (this is when the data becomes 
visible to
      > other bundles that start processing)
      > 4) Before the bundle finishes processing, wait for all
     outstanding state
      > calls to finish.
      >
      > To implement caching on the runner side, you would keep 
track of
     at most
      > 2 cache tokens per state key, one cache token represents 
the initial
      > value when the bundle started while the second represents the
     modified
      > state. If the bundle succeeds the runner passes in the set 
of tokens
      > which represent the new state, if the bundle fails you 
process
     using the
      > original ones.
      >
      > After thinking through the implementation again, we could 
supply two
      > cache tokens for each state id, the first being the set of 
initial
      > tokens if no writes happen while the second represents the 
token
     to use
      > if the SDK changes the state. This gives us the 
simplification
     where we
      > don't need to wait for the response before we update the 
global cache
      > making a typical blocking cache much easier to do. We also 
get the
      > benefit that runners can supply either the same cache 
token for a
     state
      > id or different ones. If the runner supplies the same one 
then its
      > telling the SDK to make modifications in place without any 
rollback
      > (which is good on memory since we are reducing copies of 
stuff) or if
      > the runner supplies two different ones then its telling 
the SDK
     to keep
      > the old data around. If we went through with this new 
option the SDK
      > side logic would be (assuming there is a global cache of 
cache
     token ->
      > (map of state key -> data)):
      >
      > 1) Create an empty local set of state ids that are dirty when
     starting a
      > new bundle (dirty set)
      >
      > For reads/gets:
      > 2A) If the request is a read (get), use dirty set to 
choose which
     cache
      > token to lookup and use in the global cache. If the global 
cache is
      > missing data issue the appropriate request providing the 
result.
      >
      > For writes/appends/clear:
      > 2B) if the cache tokens are different for the state id, 
add the
     state id
      > to the dirty set if it isn't there and perform the 
appropriate
      > modification to convert the old cached state data to the new
     state data
      > 3B) modify the global caches data
      > 4B) issue the request to the runner
      > 5B*) add this request to the set of requests to block on 
before
      > completing the bundle.
      >
      > (* Note, there was another idea to update the process bundle
     response to
      > contain the id of the last state request which would allow 
the
     runner to
      > know when it has seen the last state request allowing the 
SDK to not
      > block at all when finishing the bundle)
      >
      > On Thu, Aug 22, 2019 at 10:11 AM Maximilian Michels
     <m...@apache.org <mailto:m...@apache.org>
      > <mailto:m...@apache.org <mailto:m...@apache.org>>> wrote:
      >
      >     Just to give a quick update here. Rakesh, Thomas, and 
I had a
     discussion
      >     about async writes from the Python SDK to the Runner. 
Robert
     was also
      >     present for some parts of the discussion.
      >
      >     We concluded that blocking writes with the need to 
refresh
     the cache
      >     token each time are not going to provide enough
     throughput/latency.
      >
      >     We figured that it will be enough to use a single 
cache token per
      >     Runner<=>SDK Harness connection. This cache token will be
     provided by
      >     the Runner in the ProcessBundleRequest. Writes will 
not yield
     a new
      >     cache token. The advantage is that we can use one 
cache token
     for the
      >     life time of the bundle and also across bundles, 
unless the
     Runner
      >     switches to a new Runner<=>SDK Harness connection; 
then the
     Runner would
      >     have to generate a new cache token.
      >
      >     We might require additional cache tokens for the side 
inputs.
     For now,
      >     I'm planning to only tackle user state which seems to 
be the
     area where
      >     users have expressed the most need for caching.
      >
      >     -Max
      >
      >     On 21.08.19 20:05, Maximilian Michels wrote:
      >     >> There is probably a misunderstanding here: I'm 
suggesting
     to use
      >     a worker ID instead of cache tokens, not additionally.
      >     >
      >     > Ah! Misread that. We need a changing token to 
indicate that the
      >     cache is
      >     > stale, e.g. checkpoint has failed / restoring from 
an old
      >     checkpoint. If
      >     > the _Runner_ generates a new unique token/id for 
workers
     which outlast
      >     > the Runner, then this should work fine. I don't 
think it is
     safe
      >     for the
      >     > worker to supply the id. The Runner should be in 
control of
     cache
      >     tokens
      >     > to avoid invalid tokens.
      >     >
      >     >> In the PR the token is modified as part of updating 
the state.
      >     Doesn't the SDK need the new token to update it's 
cache entry
     also?
      >     That's where it would help the SDK to know the new 
token upfront.
      >     >
      >     > If the state is updated in the Runner, a new token 
has to be
      >     generated.
      >     > The old one is not valid anymore. The SDK will use 
the updated
      >     token to
      >     > store the new value in the cache. I understand that 
it would be
      >     nice to
      >     > know the token upfront. That could be possible with 
some token
      >     > generation scheme. On the other hand, writes can be
     asynchronous and
      >     > thus not block the UDF.
      >     >
      >     >> But I believe there is no need to change the token 
in first
      >     place, unless bundles for the same key (ranges) can be
     processed by
      >     different workers.
      >     >
      >     > That's certainly possible, e.g. two workers A and B 
take turn
      >     processing
      >     > a certain key range, one bundle after another:
      >     >
      >     > You process a bundle with a token T with A, then 
worker B
     takes over.
      >     > Both have an entry with cache token T. So B goes on to
     modify the
      >     state
      >     > and uses the same cache token T. Then A takes over 
again. A
     would
      >     have a
      >     > stale cache entry but T would still be a valid cache 
token.
      >     >
      >     >> Indeed the fact that Dataflow can dynamically split 
and merge
      >     these ranges is what makes it trickier. If Flink does not
      >     repartition the ranges, then things are much easier.
      >     >
      >     > Flink does not dynamically repartition key ranges 
(yet). If
     it started
      >     > to support that, we would invalidate the cache 
tokens for
     the changed
      >     > partitions.
      >     >
      >     >
      >     > I'd suggest the following cache token generation 
scheme:
      >     >
      >     > One cache token per key range for user state and one 
cache
     token for
      >     > each side input. On writes to user state or changing 
side
     input, the
      >     > associated cache token will be renewed.
      >     >
      >     > On the SDK side, it should be sufficient to let the SDK
      >     re-associate all
      >     > its cached data belonging to a valid cache token 
with a new
     cache
      >     token
      >     > returned by a successful write. This has to happen 
in the
     active scope
      >     > (i.e. user state, or a particular side input).
      >     >
      >     > If the key range changes, new cache tokens have to
     generated. This
      >     > should happen automatically because the Runner does not
     checkpoint
      >     cache
      >     > tokens and will generate new ones when it restarts 
from an
     earlier
      >     > checkpoint.
      >     >
      >     > The current PR needs to be changed to (1) only keep a
     single cache
      >     token
      >     > per user state and key range (2) add support for cache
     tokens for each
      >     > side input.
      >     >
      >     > Hope that makes sense.
      >     >
      >     > -Max
      >     >
      >     > On 21.08.19 17:27, Reuven Lax wrote:
      >     >>
      >     >>
      >     >> On Wed, Aug 21, 2019 at 2:16 AM Maximilian Michels
      >     <m...@apache.org <mailto:m...@apache.org>
     <mailto:m...@apache.org <mailto:m...@apache.org>>
      >     >> <mailto:m...@apache.org <mailto:m...@apache.org>
     <mailto:m...@apache.org <mailto:m...@apache.org>>>> wrote:
      >     >>
      >     >>     Appreciate all your comments! Replying below.
      >     >>
      >     >>
      >     >>     @Luke:
      >     >>
      >     >>     > Having cache tokens per key would be very 
expensive
     indeed
      >     and I
      >     >>     believe we should go with a single cache token 
"per"
     bundle.
      >     >>
      >     >>     Thanks for your comments on the PR. I was 
thinking to
     propose
      >     something
      >     >>     along this lines of having cache tokens valid 
for a
     particular
      >     >>     checkpointing "epoch". That would require even 
less token
      >     renewal than
      >     >>     the per-bundle approach.
      >     >>
      >     >>
      >     >>     @Thomas, thanks for the input. Some remarks:
      >     >>
      >     >>     > Wouldn't it be simpler to have the runner 
just track a
      >     unique ID
      >     >>     for each worker and use that to communicate if the
     cache is
      >     valid or
      >     >>     not?
      >     >>
      >     >>     We do not need a unique id per worker. If a 
cache token is
      >     valid for a
      >     >>     particular worker, it is also valid for another
     worker. That
      >     is with the
      >     >>     assumption that key ranges are always disjoint 
between the
      >     workers.
      >     >>
      >     >>     > * When the bundle is started, the runner 
tells the
     worker
      >     if the
      >     >>     cache has become invalid (since it knows if 
another
     worker has
      >     >>     mutated state)
      >     >>
      >     >>     This is simply done by not transferring the 
particular
     cache
      >     token. No
      >     >>     need to declare it invalid explicitly.
      >     >>
      >     >>     > * When the worker sends mutation requests to the
     runner, it
      >     >>     includes its own ID (or the runner already has 
it as
     contextual
      >     >>     information). No need to wait for a response.
      >     >>
      >     >>     Mutations of cached values can be freely done 
as long
     as the
      >     cache token
      >     >>     associated with the state is valid for a 
particular
     bundle.
      >     Only the
      >     >>     first time, the Runner needs to wait on the 
response
     to store
      >     the cache
      >     >>     token. This can also be done asynchronously.
      >     >>
      >     >>     > * When the bundle is finished, the runner 
records
     the last
      >     writer
      >     >>     (only if a change occurred)
      >     >>
      >     >>     I believe this is not necessary because there 
will only be
      >     one writer at
      >     >>     a time for a particular bundle and key range, 
hence
     only one
      >     writer
      >     >>     holds a valid cache token for a particular 
state and
     key range.
      >     >>
      >     >>
      >     >>     @Reuven:
      >     >>
      >     >>     >  Dataflow divides the keyspace up into 
lexicographic
      >     ranges, and
      >     >>     creates a cache token per range.
      >     >>
      >     >>     State is always processed partitioned by the 
Flink workers
      >     (hash-based,
      >     >>     not lexicopgrahical). I don't think that 
matters though
      >     because the key
      >     >>     ranges do not overlap between the workers. 
Flink does
     not support
      >     >>     dynamically repartitioning the key ranges. Even 
in case of
      >     fine-grained
      >     >>     recovery of workers and their key ranges, we 
would simply
      >     generate new
      >     >>     cache tokens for a particular worker.
      >     >>
      >     >>
      >     >> Dataflow's ranges are also hash based. When I said
     lexicographical, I
      >     >> meant lexicographical based on the hexadecimal hash 
value.
      >     >>
      >     >> Indeed the fact that Dataflow can dynamically split 
and
     merge these
      >     >> ranges is what makes it trickier. If Flink does not
     repartition the
      >     >> ranges, then things are much easier.
      >     >>
      >     >>
      >     >>
      >     >>     Thanks,
      >     >>     Max
      >     >>
      >     >>     On 21.08.19 09:33, Reuven Lax wrote:
      >     >>     > Dataflow does something like this, however 
since work is
      >     >>     > load balanced across workers a per-worker id 
doesn't
     work
      >     very well.
      >     >>     > Dataflow divides the keyspace up into 
lexicographic
     ranges, and
      >     >>     creates
      >     >>     > a cache token per range.
      >     >>     >
      >     >>     > On Tue, Aug 20, 2019 at 8:35 PM Thomas Weise
      >     <t...@apache.org <mailto:t...@apache.org>
     <mailto:t...@apache.org <mailto:t...@apache.org>>
      >     >>     <mailto:t...@apache.org <mailto:t...@apache.org>
     <mailto:t...@apache.org <mailto:t...@apache.org>>>
      >     >>     > <mailto:t...@apache.org <mailto:t...@apache.org>
     <mailto:t...@apache.org <mailto:t...@apache.org>>
      >     <mailto:t...@apache.org <mailto:t...@apache.org>
     <mailto:t...@apache.org <mailto:t...@apache.org>>>>> wrote:
      >     >>     >
      >     >>     >     Commenting here vs. on the PR since 
related to
     the overall
      >     >>     approach.
      >     >>     >
      >     >>     >     Wouldn't it be simpler to have the runner 
just
     track a
      >     unique
      >     >>     ID for
      >     >>     >     each worker and use that to communicate 
if the
     cache is
      >     valid
      >     >>     or not?
      >     >>     >
      >     >>     >     * When the bundle is started, the runner 
tells the
      >     worker if the
      >     >>     >     cache has become invalid (since it knows 
if another
      >     worker has
      >     >>     >     mutated state)
      >     >>     >     * When the worker sends mutation requests 
to the
     runner, it
      >     >>     includes
      >     >>     >     its own ID (or the runner already has it as
     contextual
      >     >>     information).
      >     >>     >     No need to wait for a response.
      >     >>     >     * When the bundle is finished, the runner
     records the
      >     last writer
      >     >>     >     (only if a change occurred)
      >     >>     >
      >     >>     >     Whenever current worker ID and last 
writer ID
     doesn't
      >     match, cache
      >     >>     >     is invalid.
      >     >>     >
      >     >>     >     Thomas
      >     >>     >
      >     >>     >
      >     >>     >     On Tue, Aug 20, 2019 at 11:42 AM Lukasz Cwik
      >     <lc...@google.com <mailto:lc...@google.com>
     <mailto:lc...@google.com <mailto:lc...@google.com>>
      >     >>     <mailto:lc...@google.com <mailto:lc...@google.com>
     <mailto:lc...@google.com <mailto:lc...@google.com>>>
      >     >>     >     <mailto:lc...@google.com
     <mailto:lc...@google.com> <mailto:lc...@google.com
     <mailto:lc...@google.com>>
      >     <mailto:lc...@google.com <mailto:lc...@google.com>
     <mailto:lc...@google.com <mailto:lc...@google.com>>>>> wrote:
      >     >>     >
      >     >>     >         Having cache tokens per key would be 
very
     expensive
      >     indeed
      >     >>     and I
      >     >>     >         believe we should go with a single 
cache token
      >     "per" bundle.
      >     >>     >
      >     >>     >         On Mon, Aug 19, 2019 at 11:36 AM 
Maximilian
     Michels
      >     >>     >         <m...@apache.org <mailto:m...@apache.org>
     <mailto:m...@apache.org <mailto:m...@apache.org>>
      >     <mailto:m...@apache.org <mailto:m...@apache.org>
     <mailto:m...@apache.org <mailto:m...@apache.org>>>
      >     >>     <mailto:m...@apache.org <mailto:m...@apache.org>
     <mailto:m...@apache.org <mailto:m...@apache.org>>
      >     <mailto:m...@apache.org <mailto:m...@apache.org>
     <mailto:m...@apache.org <mailto:m...@apache.org>>>>> wrote:
      >     >>     >
      >     >>     >             Maybe a Beam Python expert can 
chime in for
      >     Rakesh's
      >     >>     question?
      >     >>     >
      >     >>     >             Luke, I was assuming cache tokens 
to be
     per key
      >     and state
      >     >>     >             id. During
      >     >>     >             implementing an initial support 
on the
     Runner
      >     side, I
      >     >>     >             realized that we
      >     >>     >             probably want cache tokens to 
only be
     per state
      >     id. Note
      >     >>     >             that if we had
      >     >>     >             per-key cache tokens, the number 
of cache
      >     tokens would
      >     >>     >             approach the
      >     >>     >             total number of keys in an 
application.
      >     >>     >
      >     >>     >             If anyone wants to have a look, 
here is
     a first
      >     version of
      >     >>     >             the Runner
      >     >>     >             side for cache tokens. Note that 
I only
      >     implemented cache
      >     >>     >             tokens for
      >     >>     >             BagUserState for now, but it can 
be easily
      >     added for side
      >     >>     >             inputs as well.
      >     >>     >
      >     >>     > https://github.com/apache/beam/pull/9374
      >     >>     >
      >     >>     >             -Max
      >     >>     >
      >     >>     >
      >     >>
      >