Good Point Victoria. I just removed the compacted topic mention from the KIP. I agree with Burno about using a normal topic and deleting records that have been processed.
On Tue, Jun 6, 2023 at 2:28 AM Bruno Cadonna <cado...@apache.org> wrote: > Hi, > > another idea that came to my mind. Instead of using a compacted topic, > the buffer could use a non-compacted topic and regularly delete records > before a given offset as Streams does for repartition topics. > > Best, > Bruno > > On 05.06.23 21:48, Bruno Cadonna wrote: > > Hi Victoria, > > > > that is a good point! > > > > I think, the topic needs to be a compacted topic to be able to get rid > > of records that are evicted from the buffer. So the key might be > > something with the key, the timestamp, and a sequence number to > > distinguish between records with the same key and same timestamp. > > > > Just an idea! Maybe Walker comes up with something better. > > > > Best, > > Bruno > > > > On 05.06.23 20:38, Victoria Xia wrote: > >> Hi Walker, > >> > >> Thanks for the latest updates! The KIP looks great. Just one question > >> about > >> the changelog topic for the join buffer: The KIP says "When a failure > >> occurs the buffer will try to recover from an OffsetCheckpoint if > >> possible. > >> If not it will reload the buffer from a compacted change-log topic." > This > >> is a new changelog topic that will be introduced specifically for the > >> join > >> buffer, right? Why is the changelog topic compacted? What are the keys? > I > >> am confused because the buffer contains records from the stream-side > >> of the > >> join, for which multiple records with the same key should be treated as > >> separate updates will all must be tracked in the buffer, rather than > >> updates which replace each other. > >> > >> Thanks, > >> Victoria > >> > >> On Mon, Jun 5, 2023 at 1:47 AM Bruno Cadonna <cado...@apache.org> > wrote: > >> > >>> Hi Walker, > >>> > >>> Thanks once more for the updates to the KIP! > >>> > >>> Do you also plan to expose metrics for the buffer? > >>> > >>> Best, > >>> Bruno > >>> > >>> On 02.06.23 17:16, Walker Carlson wrote: > >>>> Hello Bruno, > >>>> > >>>> I think this covers your questions. Let me know what you think > >>>> > >>>> 2. > >>>> We can use a changelog topic. I think we can treat it like any other > >>> store > >>>> and recover in the usual manner. Also implementation is on disk > >>>> > >>>> 3. > >>>> The description is in the public interfaces description. I will copy > it > >>>> into the proposed changes as well. > >>>> > >>>> This is a bit of an implementation detail that I didn't want to add > >>>> into > >>>> the kip, but the record will be added to the buffer to keep the stream > >>> time > >>>> consistent, it will just be ejected immediately. If of course if this > >>>> causes performance issues we will skip this step and track stream time > >>>> separately. I will update the kip to say that stream time advances > >>>> when a > >>>> stream record enters the node. > >>>> > >>>> Also, yes, updated. > >>>> > >>>> 5. > >>>> No there is no difference right now, everything gets processed as it > >>> comes > >>>> in and tries to find a record for its time stamp. > >>>> > >>>> Walker > >>>> > >>>> On Fri, Jun 2, 2023 at 6:41 AM Bruno Cadonna <cado...@apache.org> > >>>> wrote: > >>>> > >>>>> Hi Walker, > >>>>> > >>>>> Thanks for the updates! > >>>>> > >>>>> 2. > >>>>> It is still not clear to me how a failure is handled. I do not > >>>>> understand what you mean by "recover from an OffsetCheckpoint". > >>>>> > >>>>> My understanding is that the buffer needs to be replicated into its > >>>>> own > >>>>> Kafka topic. The input topic is not enough. The offset of a record is > >>>>> added to the offsets to commit once the record is streamed through > the > >>>>> subtopology. That means once the record is added to the buffer its > >>>>> offset is added to the offsets to commit -- independently of > >>>>> whether the > >>>>> record was evicted from the buffer and sent to the join node or not. > >>>>> Now, let's assume the following scenario > >>>>> 1. a record is read from the input topic and added to the buffer, but > >>>>> not evicted to be processed by the join node. > >>>>> 2. When the processing of the subtopology finishes the offset of the > >>>>> record is added to the offsets to commit. > >>>>> 3. A commit happens. > >>>>> 4. A failure happens > >>>>> > >>>>> After the failure the buffer is empty but the record will not be read > >>>>> anymore from the input topic since its offset has been already > >>>>> committed. The record is lost. > >>>>> One solution to avoid the loss is to recreate the buffer from a > >>>>> compacted Kafka topic as we do for suppression buffers. I do not > >>>>> think, > >>>>> we need any offset checkpoint here since, we keep the buffer in > >>>>> memory, > >>>>> right? Or do you plan to back the buffer with a persistent store? > Even > >>>>> in that case, a compacted Kafka topic would be needed. > >>>>> > >>>>> > >>>>> 3. > >>>>> From the KIP it is still not clear to me what happens if a > >>>>> record is > >>>>> outside of the grace period. I guess the record that falls outside of > >>>>> the grace period will not be added to the buffer, but will be send to > >>>>> the join node. Since it is outside of the grace period it will also > >>>>> not > >>>>> increase stream time and it will not trigger an eviction. Also the > >>>>> head > >>>>> of the buffer will not contain a record that needs to be evicted > since > >>>>> the the timestamp of the head record will be within the interval > >>>>> stream > >>>>> time minus grace period. Is this correct? Please add such a > >>>>> description > >>>>> to the KIP. > >>>>> Furthermore, I think there is a mistake in the text: > >>>>> "... will dequeue when the record timestamp is greater than stream > >>>>> time > >>>>> plus the grace period". I guess that should be "... will dequeue when > >>>>> the record timestamp is less than (or equal?) stream time minus the > >>>>> grace period" > >>>>> > >>>>> > >>>>> 5. > >>>>> What is the difference between not setting the grace period and > >>>>> setting > >>>>> it to zero? If there is a difference, why is there a difference? > >>>>> > >>>>> > >>>>> Best, > >>>>> Bruno > >>>>> > >>>>> > >>>>> On 01.06.23 23:58, Walker Carlson wrote: > >>>>>> Hey Bruno thanks for the feedback. > >>>>>> > >>>>>> 1) > >>>>>> I will add this to the kip, but stream time only advances as the > when > >>> the > >>>>>> buffer receives a new record. > >>>>>> > >>>>>> 2) > >>>>>> You are correct, I will add a failure section on to the kip. Since > >>>>>> the > >>>>>> records wont change in the buffer from when they are read from the > >>> topic > >>>>>> they are replicated already. > >>>>>> > >>>>>> 3) > >>>>>> I see that I'm out voted on the dropping of records thing. We will > >>>>>> pass > >>>>>> them on and try to join them if possible. This might cause some null > >>>>>> results, but increasing the table history retention should help > that. > >>>>>> > >>>>>> 4) > >>>>>> I can add some on the kip. But its pretty directly adding whatever > >>>>>> the > >>>>>> grace period is to the latency. I don't see a way around it. > >>>>>> > >>>>>> Walker > >>>>>> > >>>>>> On Thu, Jun 1, 2023 at 5:23 AM Bruno Cadonna <cado...@apache.org> > >>> wrote: > >>>>>> > >>>>>>> Hi Walker, > >>>>>>> > >>>>>>> thanks for the KIP! > >>>>>>> > >>>>>>> Here my feedback: > >>>>>>> > >>>>>>> 1. > >>>>>>> It is still not clear to me when stream time for the buffer > >>>>>>> advances. > >>>>>>> What is the event that let the stream time advance? In the > >>> discussion, I > >>>>>>> do not understand what you mean by "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." Where does this segment > >>>>>>> store > >>>>>>> come from? Anyways, I think it would be great to also state how > >>>>>>> stream > >>>>>>> time advances in the KIP. > >>>>>>> > >>>>>>> 2. > >>>>>>> How does the buffer behave in case of a failure? I think I > >>>>>>> understand > >>>>>>> that the buffer will use an implementation of > >>> TimeOrderedKeyValueBuffer > >>>>>>> and therefore the records in the buffer will be replicated to a > >>>>>>> topic > >>> in > >>>>>>> Kafka, but I am not completely sure. Could you elaborate on this in > >>> the > >>>>>>> KIP? > >>>>>>> > >>>>>>> 3. > >>>>>>> I agree with Matthias about dropping late records. We use grace > >>> periods > >>>>>>> in scenarios where we records are grouped like in windowed > >>> aggregations > >>>>>>> and windowed joins. The stream buffer you propose does not really > >>> group > >>>>>>> any records. It rather delays records and reorders them. I am not > >>>>>>> sure > >>>>>>> if grace period is the right naming/concept to apply here. > >>>>>>> Instead of > >>>>>>> dropping records that fall outside of the buffer's time interval > the > >>>>>>> join should skip the buffer and try to join the record > >>>>>>> immediately. In > >>>>>>> the end, a stream-table join is a unwindowed join, i.e., no > grouping > >>> is > >>>>>>> applied to the records. > >>>>>>> What do you and other folks think about this proposal? > >>>>>>> > >>>>>>> 4. > >>>>>>> How does the proposed buffer, affects processing latency? Could you > >>>>>>> please add some words about this to the KIP? > >>>>>>> > >>>>>>> > >>>>>>> Best, > >>>>>>> Bruno > >>>>>>> > >>>>>>> > >>>>>>> > >>>>>>> > >>>>>>> On 31.05.23 01:49, Walker Carlson wrote: > >>>>>>>> Thanks for all the additional comments. I will either address them > >>> here > >>>>>>> or > >>>>>>>> update the kip accordingly. > >>>>>>>> > >>>>>>>> > >>>>>>>> I mentioned a follow kip to add extra features before and in the > >>>>>>> responses. > >>>>>>>> I will try to briefly summarize what options and optimizations I > >>>>>>>> plan > >>>>> to > >>>>>>>> include. If a concern is not covered in this list I for sure talk > >>> about > >>>>>>> it > >>>>>>>> below. > >>>>>>>> > >>>>>>>> * Allowing non versioned tables to still use the stream buffer > >>>>>>>> * Automatically materializing tables instead of forcing the user > to > >>> do > >>>>> it > >>>>>>>> * Configurable for in memory buffer > >>>>>>>> * Order the records in offset order or in time order > >>>>>>>> * Non memory use buffer (offset order, delayed pull from stream.) > >>>>>>>> * Time synced between stream and table side (maybe) > >>>>>>>> * Do not drop late records and process them as they come in > >>>>>>>> instead. > >>>>>>>> > >>>>>>>> > >>>>>>>> First, Victoria. > >>>>>>>> > >>>>>>>> 1) (One of your nits covers this, but you are correct it doesn't > >>>>>>>> make > >>>>>>>> sense. so I removed that part of the example.) > >>>>>>>> For those examples with the "bad" join results I said without > >>> buffering > >>>>>>> the > >>>>>>>> stream it would look like that, but that was incomplete. If the > >>>>>>>> look > >>> up > >>>>>>> was > >>>>>>>> simply looking at the latest version of the table when the stream > >>>>> records > >>>>>>>> came in then the results were possible. If we are using the > >>>>>>>> point in > >>>>> time > >>>>>>>> lookup that versioned tables let us then you are correct the > future > >>>>>>> results > >>>>>>>> are not possible. > >>>>>>>> > >>>>>>>> 2) I'll get to this later as Matthias brought up something > related. > >>>>>>>> > >>>>>>>> To your additional thoughts, I agree that we need to call those > >>> things > >>>>>>> out > >>>>>>>> in the documentation. I'm writing up a follow up kip with a lot of > >>> the > >>>>>>>> ideas we have discussed so that we can improve this feature beyond > >>> the > >>>>>>> base > >>>>>>>> implementation if it's needed. > >>>>>>>> > >>>>>>>> I addressed the nits in the kip. I somehow missed the table stream > >>>>> table > >>>>>>>> join processor improvement, it makes your first question make a > lot > >>>>> more > >>>>>>>> sense. Table history retention is a much cleaner way to > >>>>>>>> describe it. > >>>>>>>> > >>>>>>>> As to your mention of the syncing the time for the table and > >>>>>>>> stream. > >>>>>>>> Matthias mentioned that as well. I will address both here. I > >>>>>>>> plan to > >>>>>>> bring > >>>>>>>> that up in the future, but for now we will leave it out. I > >>>>>>>> suppose it > >>>>>>> will > >>>>>>>> be more useful after the table history retention is separable from > >>> the > >>>>>>>> table grace period. > >>>>>>>> > >>>>>>>> > >>>>>>>> To address Matthias comments. > >>>>>>>> > >>>>>>>> You are correct by saying the in memory store shouldn't cause any > >>>>>>> semantic > >>>>>>>> concerns. My concern would be more with if we limited the number > of > >>>>>>> records > >>>>>>>> on the buffer and what we would do if we hit said limits, > (emitting > >>>>> those > >>>>>>>> records might be an issue, throwing an error and halting would > >>>>>>>> not). > >>> I > >>>>>>>> think we can leave this discussion to the follow up kip along > >>>>>>>> with a > >>>>> few > >>>>>>>> other options. > >>>>>>>> > >>>>>>>> I will go through your proposals now. > >>>>>>>> > >>>>>>>> - don't support non-versioned KTables > >>>>>>>> > >>>>>>>> Sure, we can always expand this later on. Will include as part > >>>>>>>> of the > >>>>> of > >>>>>>>> the improvement kip > >>>>>>>> > >>>>>>>> - 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) > >>>>>>>> > >>>>>>>> again, that works for me for now, if we find a use we can always > >>>>>>>> add > >>>>>>> later. > >>>>>>>> > >>>>>>>> - 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) > >>>>>>>> > >>>>>>>> agreed > >>>>>>>> > >>>>>>>> - 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) > >>>>>>>> > >>>>>>>> sure, this works > >>>>>>>> > >>>>>>>> As to the dropping of late records, I'm not sure. One one hand I > >>>>>>>> like > >>>>> not > >>>>>>>> dropping things. But on the other I struggle to see how a user can > >>>>> filter > >>>>>>>> out late records that might have incomplete join results. The > point > >>> in > >>>>>>> time > >>>>>>>> look up will aggressively expire old data and if new data has been > >>>>>>> replaced > >>>>>>>> it will return null if outside of the retention. This seems like > it > >>>>> could > >>>>>>>> corrupt the integrity of the join output. Seeing that we drop late > >>>>>>> records > >>>>>>>> on the table side as well I would think it makes sense to drop > late > >>>>>>> records > >>>>>>>> on the stream buffer. I could be convinced otherwise I suppose, I > >>> could > >>>>>>> see > >>>>>>>> adding this as an option in a follow up kip. It would be very > >>>>>>>> easy to > >>>>>>>> implement either way. For now unless no one else objects I'm > >>>>>>>> going to > >>>>>>> stick > >>>>>>>> with dropping the records for the sake of getting this kip > >>>>>>>> passed. It > >>>>> is > >>>>>>>> functionally a small change to make and we can update later if you > >>> feel > >>>>>>>> strongly about it. > >>>>>>>> > >>>>>>>> For the ordering. I have to say that it would be more > >>>>>>>> complicated to > >>>>>>>> implement it to be in offset order, if the goal it to get as > >>>>>>>> many of > >>>>> the > >>>>>>>> records validly joined as possible. Because we would process as > >>> things > >>>>>>> left > >>>>>>>> the buffer a sufficiency early enough record could hold up records > >>> that > >>>>>>>> would otherwise be valid past the table history retention. To fix > >>> this > >>>>> we > >>>>>>>> could process by timestamp then store in a second queue and emit > by > >>>>>>> offset, > >>>>>>>> but that would be a lot more complicated. If we didn't care > >>>>>>>> about not > >>>>>>>> missing some valid joins we could just have no store and pull from > >>> the > >>>>>>>> topic at a delay only caring about the timestamp of the next > >>>>>>>> offset. > >>>>> For > >>>>>>>> now I want to stick with the timestamp ordering as it makes much > >>>>>>>> more > >>>>>>> sense > >>>>>>>> to me, but would propose we add both of the other options I have > >>>>>>>> laid > >>>>> out > >>>>>>>> here in the follow up kip. > >>>>>>>> > >>>>>>>> Lastly, I think having an empty store with zero grace period > >>>>>>>> would be > >>>>>>> super > >>>>>>>> simple and not costly, so we might as well make it even if nothing > >>> gets > >>>>>>>> entered. > >>>>>>>> > >>>>>>>> I hope that address all your concerns, > >>>>>>>> > >>>>>>>> Walker > >>>>>>>> > >>>>>>>> On Thu, May 25, 2023 at 9:50 AM Matthias J. Sax <mj...@apache.org > > > >>>>>>> wrote: > >>>>>>>> > >>>>>>>>> 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 > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>>> > >>>> > >>> > >> >
