Re: [DISCUSS] KIP-923: Add A Grace Period to Stream Table Join

[Matthias J. Sax]

Walker,

thanks for the updates. The KIP itself reads fine (of course Victoria 
made good comments about some phrases), but there is a couple of things 
from your latest reply I don't understand, and that I still think need 
some more discussions.

Lukas, asked about in-memory option and `WindowStoreSupplier` and you 
mention "semantic concerns". There should not be any semantic difference 
from the underlying buffer implementation, so I am not sure what you 
mean here (also the relationship to suppress() is unclear to me)? -- I 
am ok to not make it configurable for now. We can always do it via a 
follow up KIP, and keep interface changes limited for now.

Does it really make sense to allow a grace period if the table is 
non-versioned? You also say: "If table is not materialized it will 
materialize it as versioned." -- What history retention time would we 
pick for this case (also asked by Victoria)? Or should we rather not 
support this and force the user to materialize the table explicitly, and 
thus explicitly picking a history retention time? It's tradeoff between 
usability and guiding uses that there will be a significant impact on 
disk usage. There is also compatibility concerns: If the table is not 
explicitly materialized in the old program, we would already need to 
materialize it also in the old program (of course, we would use a 
non-versioned store so far). Thus, if somebody adds a grace period, we 
cannot just switch the store type, as it would be a breaking change, 
potentially required an application re-set, or following the upgrade 
path for versioned state stores, and also changing the program to 
explicitly materialize using a versioned store. Also note, that we might 
not materialize the actual join table, but only an upstream table, and 
use `ValueGetter` to access the upstream data.

To this end, as you already mentioned, history retention of the table 
should be at least grace period. You proposed to include this in a 
follow up KIP, but I am wondering if it's a fundamental requirement and 
thus we should put a check in place right away and reject an invalid 
configuration? (It always easier to lift restriction than to introduce 
them later.) This would also imply that a non-versioned table cannot be 
supported, because it does not have a history retention that is larger 
than grace period, and maybe also answer the requirement about 
materialization: as we already always materialize something on the 
tablet side as non-versioned store right now, it seems difficult to 
migrate the store to a versioned store. Ie, it might be ok to push the 
burden onto the user and say: if you start using grace period, you also 
need to manually switch from non-versioned to versioned KTables. Doing 
stuff automatically under the hood if very complex for this case, we if 
we push the burden onto the user, it might be ok to not complicate this 
KIP significantly.

To summarize the last two paragraphs, I would propose to:
 - don't support non-versioned KTables
 - if grace period is added, users need to explicitly materialize the 
table as version (either directly, or upstream. Upstream only works if 
downstream tables "inherit" versioned semantics -- cf KIP-914)
 - the table's history retention time must be larger than the grace 
period (should be easy to check at runtime, when we build the topology)
 - because switching from non-versioned to version stores is not 
backward compatibly (cf KIP-914), users need to take care of this 
themselves, and this also implies that adding grace period is not a 
backward compatible change (even only if via indirect means)

About dropping late records: wondering if we should never drop a 
stream-side record for a left-join, even if it's late? In general, one 
thing I observed over the years is, that it's easier to keep stuff and 
let users filter explicitly downstream (or make it configurable), 
instead of dropping pro-actively, because users have no good way to 
resurrect record that got already dropped.

For ordering, sounds reasonable to me only start with one 
implementation, and maybe make it configurable as a follow up. However, 
I am wondering if starting with offset order might be the better option 
as it seems to align more with what we do so far? So instead of storing 
record ordered by timestamp, we can just store them ordered by offset, 
and still "poll" from the buffer based on the head records timestamp. Or 
would this complicate the implementation significantly?

I also think it's ok to not "sync" stream-time between the table and the 
stream in this KIP, but we should consider doing this as a follow up 
change (not sure if we would need a KIP or not for a change this this).

About increasing/decreasing grace period: what you describe make sense 
to me. If decreased, the next record would just trigger emitting a lot 
of records, and for increase, the buffer would just need to "fill up" 
again. For reprocessing getting a different result with a different 
grace period is expected, so that's ok IMHO. -- There seems to be one 
special corner case: grace period zero. For this case, we actually don't 
need any store, and the stream-side could be stateless. I think it can 
have the same behavior, but if we want to "add / remove" the store 
dynamically, we need to add specific code for it. For example, even if 
we start up with a grace period of zero, we would need to check if there 
is a local store, and still emit everything in it, before we can ditch 
the store (not sure if that's even easily done at all). Or: we would 
need to have a store for _all_ cases, even if grace period is zero (the 
store would be empty all the time though), to avoid super complex code?


-Matthias


On 5/25/23 10:53 AM, Lucas Brutschy wrote:
Hi Walker,

thanks for your responses. That makes sense. I guess there is always
the option to make the implementation more configurable later on, if
users request it. Also thanks for the clarifications. From my side,
the KIP is good to go.

Cheers,
Lucas

On Wed, May 24, 2023 at 11:54 PM Victoria Xia
<victoria....@confluent.io.invalid> wrote:

Thanks for the updates, Walker! Looks great, though I do have a couple
questions about the latest updates:

    1. The new example says that without stream-side buffering, "ex" and
    "fy" are possible join results. How could those join results happen? The
    example versioned table suggests that table record "x" has timestamp 2, and
    table record "y" has timestamp 3. If stream record "e" has timestamp 1,
    then it can never be joined against record "x", and similarly for stream
    record "f" with timestamp 2 being joined against "y".
    2. I see in your replies above that "If table is not materialized it
    will materialize it as versioned" but I don't see this called out in the
    KIP -- seems worth calling out. Also, what will the history retention for
    the versioned table be? Will it be the same as the join grace period, or
    will it be greater?


And some additional thoughts:

Sounds like there are a few things users should watch out for when enabling
the stream-side buffer:

    - Records will get "stuck" if there are no newer records to advance
    stream time.
    - If there are large gaps between the timestamps of stream-side records,
    then it's possible that versioned store history retention will have expired
    by the time a record is evicted from the join buffer, leading to a join
    "miss." For example, if the join grace period and table history retention
    are both 10, and records come in the order:

    table side t0 with ts=0
    stream side s1 with ts=1 <-- enters buffer
    table side t10 with ts=10
    table side t20 with ts=20
    stream side s21 with ts=21 <-- evicts record s1 from buffer, but
    versioned store no longer contains data for ts=1 due to history retention
    having elapsed

    This will result in the join result (s1, null) even though it should've
    been (s1, t0), due to t0 having been expired from the versioned store
    already.
    - Out-of-order records from the stream-side will be reordered, and late
    records will be dropped.

I don't think any of these are reasons to not go forward with this KIP, but
it'd be good to call them out in the eventual documentation to decrease the
chance users get tripped up.

We could maybe do an improvement later to advance stream time from table
side as well, but that might be debatable as we might get more late records.

Yes, the likelihood of late records increases but also the likelihood of
"join misses" due to versioned store history retention having elapsed
decreases, which feels important for certain use cases. Either way, agreed
that it can be a discussion for the future as incorporating this would
substantially complicate the implementation.

Also a couple nits:

    - The KIP currently says "We recently added versioned tables which allow
    the table side of the a join [...] but it is not taken advantage of in
    joins," but this doesn't seem true? If the table of a stream-table join is
    versioned, then the DSL's stream-table join processor will automatically
    perform timestamped lookups into the table, in order to take advantage of
    the new timestamp-aware store to provide better join semantics.
    - The KIP mentions "grace period" for versioned stores in a number of
    places but I think you actually mean "history retention"? The two happen to
    be the same today (it is not an option for users to configure the two
    separately) but this need not be true in the future. "History retention"
    governs how far back in time reads may occur, which is the relevant
    parameter for performing lookups as part of the stream-table join. "Grace
    period" in the context of versioned stores refers to how far back in time
    out-of-order writes may occur, which probably isn't directly relevant for
    introducing a stream-side buffer, though it's also possible I've overlooked
    something. (As a bonus, switching from "table grace period" in the KIP to
    "table history retention" also helps to clarify/distinguish that it's a
    different parameter from the "join grace period," which I could see being
    confusing to readers. :) )


Cheers,
Victoria

On Thu, May 18, 2023 at 1:43 PM Walker Carlson
<wcarl...@confluent.io.invalid> wrote:

Hey all,

Thanks for the comments, they gave me a lot to think about. I'll try to
address them all inorder. I have made some updates to the kip related to
them, but I mention where below.

Lucas

Good idea about the example. I added a simple one.

1) I have thought about including options for the underlying buffer
configuration. One of which might be adding an in memory option. My biggest
concern is about the semantic guarantees. This isn't like suppress or with
windows where producing incomplete results is repetitively harmless. Here
we would be possibly producing incorrect results. I also would like to keep
the interface changes as simple as I can. Making more than this change to
Joined I feel could make this more complicated than it needs to be. If we
really want to I could see adding a grace() option with a BufferConifg in
there or something, but I would rather not.

2) The buffer will be independent of if the table is versioned or not. If
table is not materialized it will materialize it as versioned. It might
make sense to do a follow up kip where we force the retention period  of
the versioned to be greater than whatever the max of the stream buffer is.

Victoria

1) Yes, records will exit in timestamp order not in offset order.
2) Late records will be dropped (Late as out of the grace period). From my
understanding that is the point of a grace period, no? Doesn't the same
thing happen with versioned stores?
3) The segment store already has an observed stream time, we advance based
on that. That should only advance based on records that enter the store. So
yes, only stream side records. We could maybe do an improvement later to
advance stream time from table side as well, but that might be debatable as
we might get more late records. Anyways I would rather have that as a
separate discussion.

in memory option? We can do that, for the buffer I plan to use the
TimeOrderedKeyValueBuffer interface which already has an in memory
implantation, so it would be simple.

I said more in my answer to Lucas's question. The concern I have with
buffer configs or in memory is complicating the interface. Also semantic
guarantees but in memory shouldn't effect that

Matthias

1) fixed out of order vs late terminology in the kip.

2) I was referring to having a stream. So after this kip we can have a
buffered stream or a normal one. For the table we can use a versioned table
or a normal table.

3 Good call out. I clarified this as "If the table side uses a materialized
version store, it can store multiple versions of each record within its
defined grace period." and modified the rest of the paragraph a bit.

4) I get the preserving off offset ordering, but if the stream is buffered
to join on timestamp instead of offset doesn't it already seem like we care
more about time in this case?

If we end up adding more options it might make sense to do this. Maybe
offset order processing can be a follow up?

I'll add a section for this in Rejected Alternatives. I think it makes
sense to do something like this but maybe in a follow up.

5) I hadn't thought about this. I suppose if they changed this in an
upgrade the next record would either evict a lot of records (if the grace
period decreased) or there would be a pause until the new grace period
reached. Increasing is a bit more problematic, especially if the table
grace period and retention time stays the same. If the data is reprocessed
after a change like that then there would be different results, but I feel
like that would be expected after such a change.

What do you think should happen?

Hopefully this answers your questions!

Walker

On Mon, May 8, 2023 at 11:32 AM Matthias J. Sax <mj...@apache.org> wrote:

Thanks for the KIP! Also some question/comments from my side:

10) Notation: you use the term "late data" but I think you mean
out-of-order. We reserve the term "late" to records that arrive after
grace period passed, and thus, "late == out-of-order data that is
dropped".


20) "There is only one option from the stream side and only recently is
there a second option on the table side."

What are those options? Victoria already asked about the table side, but
I am also not sure what option you mean for the stream side?


30) "If the table side uses a materialized version store the value is
the latest by stream time rather than by offset within its defined grace
period."

The phrase "the value is the latest by stream time" is confusing -- in
the end, a versioned stores multiple versions, not just one.


40) I am also wondering about ordering. In general, KS tries to preserve
offset-order during processing (with some exception, when offset order
preservation is not clearly defined). Given that the stream-side buffer
is really just a "linear buffer", we could easily preserve offset-order.
But I also see a benefit of re-ordering and emitting out-of-order data
right away when read (instead of blocking them behind in-order records
that are not ready yet). -- It might even be a possibility, to let users
pick a emit strategy eg "EmitStrategy.preserveOffsets" (name just a
placeholder).

The KIP should explain this in more detail and also discuss different
options and mention them in "Rejected alternatives" in case we don't
want to include them.


50) What happens when users change the grace period? Especially, when
they turn it on/off (but also increasing/decreasing is an interesting
point)? I think we should try to support this if possible; the
"Compatibility" section needs to cover switching on/off in more detail.


-Matthias




On 5/2/23 2:06 PM, Victoria Xia wrote:
Cool KIP, Walker! Thanks for sharing this proposal.

A few clarifications:

1. Is the order that records exit the buffer in necessarily the same as
the
order that records enter the buffer in, or no? Based on the description
in
the KIP, it sounds like the answer is no, i.e., records will exit the
buffer in increasing timestamp order, which means that they may be
ordered
(even for the same key) compared to the input order.

2. What happens if the join grace period is nonzero, and a stream-side
record arrives with a timestamp that is older than the current stream
time
minus the grace period? Will this record trigger a join result, or will
it
be dropped? Based on the description for what happens when the join
grace
period is set to zero, it sounds like the late record will be dropped,
even
if the join grace period is nonzero. Is that true?

3. What could cause stream time to advance, for purposes of removing
records from the join buffer? For example, will new records arriving on
the
table side of the join cause stream time to advance? From the KIP it
sounds
like only stream-side records will advance stream time -- does that
mean
that the join processor itself will have to track this stream time?

Also +1 to Lucas's question about what options will be available for
configuring the join buffer. Will users have the option to choose
whether
they want the buffer to be in-memory vs persistent?

- Victoria

On Fri, Apr 28, 2023 at 11:54 AM Lucas Brutschy
<lbruts...@confluent.io.invalid> wrote:

HI Walker,

thanks for the KIP! We definitely need this. I have two questions:

   - Have you considered allowing the customization of the underlying
buffer implementation? As I can see, `StreamJoined` lets you customize
the underlying store via a `WindowStoreSupplier`. Would it make sense
for `Joined` to have this as well? I can imagine one may want to limit
the number of records in the buffer, for example. If we hit the
maximum, the only option would be to drop semantic guarantees, but
users may still want to do this.
   - With "second option on the table side" you are referring to
versioned tables, right? Will the buffer on the stream side behave any
different whether the table side is versioned or not?

Finally, I think a simple example in the motivation section could help
non-experts understand the KIP.

Best,
Lucas

On Tue, Apr 25, 2023 at 9:13 PM Walker Carlson
<wcarl...@confluent.io.invalid> wrote:

Hello everybody,

I have a stream proposal to improve the stream table join by adding a
grace
period and buffer to the stream side of the join to allow processing
in
timestamp order matching the recent improvements of the versioned
tables.

Please take a look here <
https://cwiki.apache.org/confluence/x/lAs0Dw>
and
share your thoughts.

best,
Walker