MapState with List Type for values

[Aaron Langford]

Hello Flink Community,

I have a question about using MapState with lists as values. Here is a
description of the use case:

I have an operator over a keyed stream where for each record that comes, it
needs to look into some state to determine if a value has arrived or not.
If the value has not arrived yet, the record needs to be en-queued for when
the value will eventually arrive. When that value does arrive, the queued
records need to be flushed, and associated list state cleared.

Here's an emoji representation of the stream:
---------------------------------------------------------------------
🐿🦞🐿🦄🐿🦄🦄🐉 ...
---------------------------------------------------------------------
Where 🐿 must wait for at least 1 🦞 to be output (otherwise be queued) and
🦄 must wait for 🐉to be output (otherwise be queued). If you can't tell,
this is basically/sort of a join without a window.

The output should be something abstractly like this:

---------------------------------------------------------------------
{🐿,🦞}{🐿,🦞}{🐿,🦞}{🦄,🐉}{🦄,🐉}{🦄,🐉}...
---------------------------------------------------------------------

Many records might be en-queued while waiting for a given value. Many
records may be waiting for many different values, but any record that is
en-queued will only be waiting for a record of one type to show up.

Records on this stream are keyed by some shared parent key, and I have
reasons to avoid keying the stream on the "join key" as it were. Namely I'm
getting a CDC stream with a lot of different tables, and I want to avoid a
topology with N operators for N different tables if I can.

If I lean on MapState<KEY, Seq<VALUE>> to get this done for me, then my job
suffers considerably in terms of performance. I believe one of the biggest
bottlenecks is that for each time I need to interact with a Seq<VALUE>
(like for appends), I must deserialize the entire list.

Is there a way to set up a MapState whose value is a ListState? Or is there
any guidance for how I might serialize/deserialize a list type in the
MapState in such a way that appends aren't so expensive? Open to other
suggestions/approaches as well.

Aaron