Thanks Alexander for your reply. We can discuss the new interface when it comes out.
We are more inclined to deprecate the connector `async` option when discussing FLIP-234 [1]. We should use hint to let planner decide. Although the discussion has not yet produced a conclusion, can we remove this option from this FLIP? It doesn't seem to be related to this FLIP, but more to FLIP-234, and we can form a conclusion over there. [1] https://lists.apache.org/thread/9k1sl2519kh2n3yttwqc00p07xdfns3h Best, Jingsong On Wed, Jun 1, 2022 at 4:59 AM Jing Ge <j...@ververica.com> wrote: > Hi Jark, > > Thanks for clarifying it. It would be fine. as long as we could provide the > no-cache solution. I was just wondering if the client side cache could > really help when HBase is used, since the data to look up should be huge. > Depending how much data will be cached on the client side, the data that > should be lru in e.g. LruBlockCache will not be lru anymore. In the worst > case scenario, once the cached data at client side is expired, the request > will hit disk which will cause extra latency temporarily, if I am not > mistaken. > > Best regards, > Jing > > On Mon, May 30, 2022 at 9:59 AM Jark Wu <imj...@gmail.com> wrote: > > > Hi Jing Ge, > > > > What do you mean about the "impact on the block cache used by HBase"? > > In my understanding, the connector cache and HBase cache are totally two > > things. > > The connector cache is a local/client cache, and the HBase cache is a > > server cache. > > > > > does it make sense to have a no-cache solution as one of the > > default solutions so that customers will have no effort for the migration > > if they want to stick with Hbase cache > > > > The implementation migration should be transparent to users. Take the > HBase > > connector as > > an example, it already supports lookup cache but is disabled by default. > > After migration, the > > connector still disables cache by default (i.e. no-cache solution). No > > migration effort for users. > > > > HBase cache and connector cache are two different things. HBase cache > can't > > simply replace > > connector cache. Because one of the most important usages for connector > > cache is reducing > > the I/O request/response and improving the throughput, which can achieve > > by just using a server cache. > > > > Best, > > Jark > > > > > > > > > > On Fri, 27 May 2022 at 22:42, Jing Ge <j...@ververica.com> wrote: > > > > > Thanks all for the valuable discussion. The new feature looks very > > > interesting. > > > > > > According to the FLIP description: "*Currently we have JDBC, Hive and > > HBase > > > connector implemented lookup table source. All existing implementations > > > will be migrated to the current design and the migration will be > > > transparent to end users*." I was only wondering if we should pay > > attention > > > to HBase and similar DBs. Since, commonly, the lookup data will be huge > > > while using HBase, partial caching will be used in this case, if I am > not > > > mistaken, which might have an impact on the block cache used by HBase, > > e.g. > > > LruBlockCache. > > > Another question is that, since HBase provides a sophisticated cache > > > solution, does it make sense to have a no-cache solution as one of the > > > default solutions so that customers will have no effort for the > migration > > > if they want to stick with Hbase cache? > > > > > > Best regards, > > > Jing > > > > > > On Fri, May 27, 2022 at 11:19 AM Jingsong Li <jingsongl...@gmail.com> > > > wrote: > > > > > > > Hi all, > > > > > > > > I think the problem now is below: > > > > 1. AllCache and PartialCache interface on the non-uniform, one needs > to > > > > provide LookupProvider, the other needs to provide CacheBuilder. > > > > 2. AllCache definition is not flexible, for example, PartialCache can > > use > > > > any custom storage, while the AllCache can not, AllCache can also be > > > > considered to store memory or disk, also need a flexible strategy. > > > > 3. AllCache can not customize ReloadStrategy, currently only > > > > ScheduledReloadStrategy. > > > > > > > > In order to solve the above problems, the following are my ideas. > > > > > > > > ## Top level cache interfaces: > > > > > > > > ``` > > > > > > > > public interface CacheLookupProvider extends > > > > LookupTableSource.LookupRuntimeProvider { > > > > > > > > CacheBuilder createCacheBuilder(); > > > > } > > > > > > > > > > > > public interface CacheBuilder { > > > > Cache create(); > > > > } > > > > > > > > > > > > public interface Cache { > > > > > > > > /** > > > > * Returns the value associated with key in this cache, or null > if > > > > there is no cached value for > > > > * key. > > > > */ > > > > @Nullable > > > > Collection<RowData> getIfPresent(RowData key); > > > > > > > > /** Returns the number of key-value mappings in the cache. */ > > > > long size(); > > > > } > > > > > > > > ``` > > > > > > > > ## Partial cache > > > > > > > > ``` > > > > > > > > public interface PartialCacheLookupFunction extends > > CacheLookupProvider { > > > > > > > > @Override > > > > PartialCacheBuilder createCacheBuilder(); > > > > > > > > /** Creates an {@link LookupFunction} instance. */ > > > > LookupFunction createLookupFunction(); > > > > } > > > > > > > > > > > > public interface PartialCacheBuilder extends CacheBuilder { > > > > > > > > PartialCache create(); > > > > } > > > > > > > > > > > > public interface PartialCache extends Cache { > > > > > > > > /** > > > > * Associates the specified value rows with the specified key row > > > > in the cache. If the cache > > > > * previously contained value associated with the key, the old > > > > value is replaced by the > > > > * specified value. > > > > * > > > > * @return the previous value rows associated with key, or null > if > > > > there was no mapping for key. > > > > * @param key - key row with which the specified value is to be > > > > associated > > > > * @param value – value rows to be associated with the specified > > key > > > > */ > > > > Collection<RowData> put(RowData key, Collection<RowData> value); > > > > > > > > /** Discards any cached value for the specified key. */ > > > > void invalidate(RowData key); > > > > } > > > > > > > > ``` > > > > > > > > ## All cache > > > > ``` > > > > > > > > public interface AllCacheLookupProvider extends CacheLookupProvider { > > > > > > > > void registerReloadStrategy(ScheduledExecutorService > > > > executorService, Reloader reloader); > > > > > > > > ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider(); > > > > > > > > @Override > > > > AllCacheBuilder createCacheBuilder(); > > > > } > > > > > > > > > > > > public interface AllCacheBuilder extends CacheBuilder { > > > > > > > > AllCache create(); > > > > } > > > > > > > > > > > > public interface AllCache extends Cache { > > > > > > > > void putAll(Iterator<Map<RowData, RowData>> allEntries); > > > > > > > > void clearAll(); > > > > } > > > > > > > > > > > > public interface Reloader { > > > > > > > > void reload(); > > > > } > > > > > > > > ``` > > > > > > > > Best, > > > > Jingsong > > > > > > > > On Fri, May 27, 2022 at 11:10 AM Jingsong Li <jingsongl...@gmail.com > > > > > > wrote: > > > > > > > > > Thanks Qingsheng and all for your discussion. > > > > > > > > > > Very sorry to jump in so late. > > > > > > > > > > Maybe I missed something? > > > > > My first impression when I saw the cache interface was, why don't > we > > > > > provide an interface similar to guava cache [1], on top of guava > > cache, > > > > > caffeine also makes extensions for asynchronous calls.[2] > > > > > There is also the bulk load in caffeine too. > > > > > > > > > > I am also more confused why first from LookupCacheFactory.Builder > and > > > > then > > > > > to Factory to create Cache. > > > > > > > > > > [1] https://github.com/google/guava > > > > > [2] https://github.com/ben-manes/caffeine/wiki/Population > > > > > > > > > > Best, > > > > > Jingsong > > > > > > > > > > On Thu, May 26, 2022 at 11:17 PM Jark Wu <imj...@gmail.com> wrote: > > > > > > > > > >> After looking at the new introduced ReloadTime and Becket's > comment, > > > > >> I agree with Becket we should have a pluggable reloading strategy. > > > > >> We can provide some common implementations, e.g., periodic > > reloading, > > > > and > > > > >> daily reloading. > > > > >> But there definitely be some connector- or business-specific > > reloading > > > > >> strategies, e.g. > > > > >> notify by a zookeeper watcher, reload once a new Hive partition is > > > > >> complete. > > > > >> > > > > >> Best, > > > > >> Jark > > > > >> > > > > >> On Thu, 26 May 2022 at 11:52, Becket Qin <becket....@gmail.com> > > > wrote: > > > > >> > > > > >> > Hi Qingsheng, > > > > >> > > > > > >> > Thanks for updating the FLIP. A few comments / questions below: > > > > >> > > > > > >> > 1. Is there a reason that we have both "XXXFactory" and > > > "XXXProvider". > > > > >> > What is the difference between them? If they are the same, can > we > > > just > > > > >> use > > > > >> > XXXFactory everywhere? > > > > >> > > > > > >> > 2. Regarding the FullCachingLookupProvider, should the reloading > > > > policy > > > > >> > also be pluggable? Periodical reloading could be sometimes be > > tricky > > > > in > > > > >> > practice. For example, if user uses 24 hours as the cache > refresh > > > > >> interval > > > > >> > and some nightly batch job delayed, the cache update may still > see > > > the > > > > >> > stale data. > > > > >> > > > > > >> > 3. In DefaultLookupCacheFactory, it looks like InitialCapacity > > > should > > > > be > > > > >> > removed. > > > > >> > > > > > >> > 4. The purpose of LookupFunctionProvider#cacheMissingKey() > seems a > > > > >> little > > > > >> > confusing to me. If Optional<LookupCacheFactory> > getCacheFactory() > > > > >> returns > > > > >> > a non-empty factory, doesn't that already indicates the > framework > > to > > > > >> cache > > > > >> > the missing keys? Also, why is this method returning an > > > > >> Optional<Boolean> > > > > >> > instead of boolean? > > > > >> > > > > > >> > Thanks, > > > > >> > > > > > >> > Jiangjie (Becket) Qin > > > > >> > > > > > >> > > > > > >> > > > > > >> > On Wed, May 25, 2022 at 5:07 PM Qingsheng Ren < > renqs...@gmail.com > > > > > > > >> wrote: > > > > >> > > > > > >> >> Hi Lincoln and Jark, > > > > >> >> > > > > >> >> Thanks for the comments! If the community reaches a consensus > > that > > > we > > > > >> use > > > > >> >> SQL hint instead of table options to decide whether to use sync > > or > > > > >> async > > > > >> >> mode, it’s indeed not necessary to introduce the “lookup.async” > > > > option. > > > > >> >> > > > > >> >> I think it’s a good idea to let the decision of async made on > > query > > > > >> >> level, which could make better optimization with more > infomation > > > > >> gathered > > > > >> >> by planner. Is there any FLIP describing the issue in > > FLINK-27625? > > > I > > > > >> >> thought FLIP-234 is only proposing adding SQL hint for retry on > > > > missing > > > > >> >> instead of the entire async mode to be controlled by hint. > > > > >> >> > > > > >> >> Best regards, > > > > >> >> > > > > >> >> Qingsheng > > > > >> >> > > > > >> >> > On May 25, 2022, at 15:13, Lincoln Lee < > lincoln.8...@gmail.com > > > > > > > >> wrote: > > > > >> >> > > > > > >> >> > Hi Jark, > > > > >> >> > > > > > >> >> > Thanks for your reply! > > > > >> >> > > > > > >> >> > Currently 'lookup.async' just lies in HBase connector, I have > > no > > > > idea > > > > >> >> > whether or when to remove it (we can discuss it in another > > issue > > > > for > > > > >> the > > > > >> >> > HBase connector after FLINK-27625 is done), just not add it > > into > > > a > > > > >> >> common > > > > >> >> > option now. > > > > >> >> > > > > > >> >> > Best, > > > > >> >> > Lincoln Lee > > > > >> >> > > > > > >> >> > > > > > >> >> > Jark Wu <imj...@gmail.com> 于2022年5月24日周二 20:14写道： > > > > >> >> > > > > > >> >> >> Hi Lincoln, > > > > >> >> >> > > > > >> >> >> I have taken a look at FLIP-234, and I agree with you that > the > > > > >> >> connectors > > > > >> >> >> can > > > > >> >> >> provide both async and sync runtime providers simultaneously > > > > instead > > > > >> >> of one > > > > >> >> >> of them. > > > > >> >> >> At that point, "lookup.async" looks redundant. If this > option > > is > > > > >> >> planned to > > > > >> >> >> be removed > > > > >> >> >> in the long term, I think it makes sense not to introduce it > > in > > > > this > > > > >> >> FLIP. > > > > >> >> >> > > > > >> >> >> Best, > > > > >> >> >> Jark > > > > >> >> >> > > > > >> >> >> On Tue, 24 May 2022 at 11:08, Lincoln Lee < > > > lincoln.8...@gmail.com > > > > > > > > > >> >> wrote: > > > > >> >> >> > > > > >> >> >>> Hi Qingsheng, > > > > >> >> >>> > > > > >> >> >>> Sorry for jumping into the discussion so late. It's a good > > idea > > > > >> that > > > > >> >> we > > > > >> >> >> can > > > > >> >> >>> have a common table option. I have a minor comments on > > > > >> 'lookup.async' > > > > >> >> >> that > > > > >> >> >>> not make it a common option: > > > > >> >> >>> > > > > >> >> >>> The table layer abstracts both sync and async lookup > > > > capabilities, > > > > >> >> >>> connectors implementers can choose one or both, in the case > > of > > > > >> >> >> implementing > > > > >> >> >>> only one capability(status of the most of existing builtin > > > > >> connectors) > > > > >> >> >>> 'lookup.async' will not be used. And when a connector has > > both > > > > >> >> >>> capabilities, I think this choice is more suitable for > making > > > > >> >> decisions > > > > >> >> >> at > > > > >> >> >>> the query level, for example, table planner can choose the > > > > physical > > > > >> >> >>> implementation of async lookup or sync lookup based on its > > cost > > > > >> >> model, or > > > > >> >> >>> users can give query hint based on their own better > > > > >> understanding. If > > > > >> >> >>> there is another common table option 'lookup.async', it may > > > > confuse > > > > >> >> the > > > > >> >> >>> users in the long run. > > > > >> >> >>> > > > > >> >> >>> So, I prefer to leave the 'lookup.async' option in private > > > place > > > > >> (for > > > > >> >> the > > > > >> >> >>> current hbase connector) and not turn it into a common > > option. > > > > >> >> >>> > > > > >> >> >>> WDYT? > > > > >> >> >>> > > > > >> >> >>> Best, > > > > >> >> >>> Lincoln Lee > > > > >> >> >>> > > > > >> >> >>> > > > > >> >> >>> Qingsheng Ren <renqs...@gmail.com> 于2022年5月23日周一 14:54写道： > > > > >> >> >>> > > > > >> >> >>>> Hi Alexander, > > > > >> >> >>>> > > > > >> >> >>>> Thanks for the review! We recently updated the FLIP and > you > > > can > > > > >> find > > > > >> >> >>> those > > > > >> >> >>>> changes from my latest email. Since some terminologies has > > > > >> changed so > > > > >> >> >>> I’ll > > > > >> >> >>>> use the new concept for replying your comments. > > > > >> >> >>>> > > > > >> >> >>>> 1. Builder vs ‘of’ > > > > >> >> >>>> I’m OK to use builder pattern if we have additional > optional > > > > >> >> parameters > > > > >> >> >>>> for full caching mode (“rescan” previously). The > > > > >> schedule-with-delay > > > > >> >> >> idea > > > > >> >> >>>> looks reasonable to me, but I think we need to redesign > the > > > > >> builder > > > > >> >> API > > > > >> >> >>> of > > > > >> >> >>>> full caching to make it more descriptive for developers. > > Would > > > > you > > > > >> >> mind > > > > >> >> >>>> sharing your ideas about the API? For accessing the FLIP > > > > workspace > > > > >> >> you > > > > >> >> >>> can > > > > >> >> >>>> just provide your account ID and ping any PMC member > > including > > > > >> Jark. > > > > >> >> >>>> > > > > >> >> >>>> 2. Common table options > > > > >> >> >>>> We have some discussions these days and propose to > > introduce 8 > > > > >> common > > > > >> >> >>>> table options about caching. It has been updated on the > > FLIP. > > > > >> >> >>>> > > > > >> >> >>>> 3. Retries > > > > >> >> >>>> I think we are on the same page :-) > > > > >> >> >>>> > > > > >> >> >>>> For your additional concerns: > > > > >> >> >>>> 1) The table option has been updated. > > > > >> >> >>>> 2) We got “lookup.cache” back for configuring whether to > use > > > > >> partial > > > > >> >> or > > > > >> >> >>>> full caching mode. > > > > >> >> >>>> > > > > >> >> >>>> Best regards, > > > > >> >> >>>> > > > > >> >> >>>> Qingsheng > > > > >> >> >>>> > > > > >> >> >>>> > > > > >> >> >>>> > > > > >> >> >>>>> On May 19, 2022, at 17:25, Александр Смирнов < > > > > >> smirale...@gmail.com> > > > > >> >> >>>> wrote: > > > > >> >> >>>>> > > > > >> >> >>>>> Also I have a few additions: > > > > >> >> >>>>> 1) maybe rename 'lookup.cache.maximum-size' to > > > > >> >> >>>>> 'lookup.cache.max-rows'? I think it will be more clear > that > > > we > > > > >> talk > > > > >> >> >>>>> not about bytes, but about the number of rows. Plus it > fits > > > > more, > > > > >> >> >>>>> considering my optimization with filters. > > > > >> >> >>>>> 2) How will users enable rescanning? Are we going to > > separate > > > > >> >> caching > > > > >> >> >>>>> and rescanning from the options point of view? Like > > initially > > > > we > > > > >> had > > > > >> >> >>>>> one option 'lookup.cache' with values LRU / ALL. I think > > now > > > we > > > > >> can > > > > >> >> >>>>> make a boolean option 'lookup.rescan'. RescanInterval can > > be > > > > >> >> >>>>> 'lookup.rescan.interval', etc. > > > > >> >> >>>>> > > > > >> >> >>>>> Best regards, > > > > >> >> >>>>> Alexander > > > > >> >> >>>>> > > > > >> >> >>>>> чт, 19 мая 2022 г. в 14:50, Александр Смирнов < > > > > >> smirale...@gmail.com > > > > >> >> >>> : > > > > >> >> >>>>>> > > > > >> >> >>>>>> Hi Qingsheng and Jark, > > > > >> >> >>>>>> > > > > >> >> >>>>>> 1. Builders vs 'of' > > > > >> >> >>>>>> I understand that builders are used when we have > multiple > > > > >> >> >> parameters. > > > > >> >> >>>>>> I suggested them because we could add parameters later. > To > > > > >> prevent > > > > >> >> >>>>>> Builder for ScanRuntimeProvider from looking redundant I > > can > > > > >> >> suggest > > > > >> >> >>>>>> one more config now - "rescanStartTime". > > > > >> >> >>>>>> It's a time in UTC (LocalTime class) when the first > reload > > > of > > > > >> cache > > > > >> >> >>>>>> starts. This parameter can be thought of as > 'initialDelay' > > > > (diff > > > > >> >> >>>>>> between current time and rescanStartTime) in method > > > > >> >> >>>>>> ScheduleExecutorService#scheduleWithFixedDelay [1] . It > > can > > > be > > > > >> very > > > > >> >> >>>>>> useful when the dimension table is updated by some other > > > > >> scheduled > > > > >> >> >> job > > > > >> >> >>>>>> at a certain time. Or when the user simply wants a > second > > > scan > > > > >> >> >> (first > > > > >> >> >>>>>> cache reload) be delayed. This option can be used even > > > without > > > > >> >> >>>>>> 'rescanInterval' - in this case 'rescanInterval' will be > > one > > > > >> day. > > > > >> >> >>>>>> If you are fine with this option, I would be very glad > if > > > you > > > > >> would > > > > >> >> >>>>>> give me access to edit FLIP page, so I could add it > myself > > > > >> >> >>>>>> > > > > >> >> >>>>>> 2. Common table options > > > > >> >> >>>>>> I also think that FactoryUtil would be overloaded by all > > > cache > > > > >> >> >>>>>> options. But maybe unify all suggested options, not only > > for > > > > >> >> default > > > > >> >> >>>>>> cache? I.e. class 'LookupOptions', that unifies default > > > cache > > > > >> >> >> options, > > > > >> >> >>>>>> rescan options, 'async', 'maxRetries'. WDYT? > > > > >> >> >>>>>> > > > > >> >> >>>>>> 3. Retries > > > > >> >> >>>>>> I'm fine with suggestion close to > > RetryUtils#tryTimes(times, > > > > >> call) > > > > >> >> >>>>>> > > > > >> >> >>>>>> [1] > > > > >> >> >>>> > > > > >> >> >>> > > > > >> >> >> > > > > >> >> > > > > >> > > > > > > > > > > https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ScheduledExecutorService.html#scheduleWithFixedDelay-java.lang.Runnable-long-long-java.util.concurrent.TimeUnit- > > > > >> >> >>>>>> > > > > >> >> >>>>>> Best regards, > > > > >> >> >>>>>> Alexander > > > > >> >> >>>>>> > > > > >> >> >>>>>> ср, 18 мая 2022 г. в 16:04, Qingsheng Ren < > > > renqs...@gmail.com > > > > >: > > > > >> >> >>>>>>> > > > > >> >> >>>>>>> Hi Jark and Alexander, > > > > >> >> >>>>>>> > > > > >> >> >>>>>>> Thanks for your comments! I’m also OK to introduce > common > > > > table > > > > >> >> >>>> options. I prefer to introduce a new > > DefaultLookupCacheOptions > > > > >> class > > > > >> >> >> for > > > > >> >> >>>> holding these option definitions because putting all > options > > > > into > > > > >> >> >>>> FactoryUtil would make it a bit ”crowded” and not well > > > > >> categorized. > > > > >> >> >>>>>>> > > > > >> >> >>>>>>> FLIP has been updated according to suggestions above: > > > > >> >> >>>>>>> 1. Use static “of” method for constructing > > > > >> RescanRuntimeProvider > > > > >> >> >>>> considering both arguments are required. > > > > >> >> >>>>>>> 2. Introduce new table options matching > > > > >> DefaultLookupCacheFactory > > > > >> >> >>>>>>> > > > > >> >> >>>>>>> Best, > > > > >> >> >>>>>>> Qingsheng > > > > >> >> >>>>>>> > > > > >> >> >>>>>>> On Wed, May 18, 2022 at 2:57 PM Jark Wu < > > imj...@gmail.com> > > > > >> wrote: > > > > >> >> >>>>>>>> > > > > >> >> >>>>>>>> Hi Alex, > > > > >> >> >>>>>>>> > > > > >> >> >>>>>>>> 1) retry logic > > > > >> >> >>>>>>>> I think we can extract some common retry logic into > > > > utilities, > > > > >> >> >> e.g. > > > > >> >> >>>> RetryUtils#tryTimes(times, call). > > > > >> >> >>>>>>>> This seems independent of this FLIP and can be reused > by > > > > >> >> >> DataStream > > > > >> >> >>>> users. > > > > >> >> >>>>>>>> Maybe we can open an issue to discuss this and where > to > > > put > > > > >> it. > > > > >> >> >>>>>>>> > > > > >> >> >>>>>>>> 2) cache ConfigOptions > > > > >> >> >>>>>>>> I'm fine with defining cache config options in the > > > > framework. > > > > >> >> >>>>>>>> A candidate place to put is FactoryUtil which also > > > includes > > > > >> >> >>>> "sink.parallelism", "format" options. > > > > >> >> >>>>>>>> > > > > >> >> >>>>>>>> Best, > > > > >> >> >>>>>>>> Jark > > > > >> >> >>>>>>>> > > > > >> >> >>>>>>>> > > > > >> >> >>>>>>>> On Wed, 18 May 2022 at 13:52, Александр Смирнов < > > > > >> >> >>> smirale...@gmail.com> > > > > >> >> >>>> wrote: > > > > >> >> >>>>>>>>> > > > > >> >> >>>>>>>>> Hi Qingsheng, > > > > >> >> >>>>>>>>> > > > > >> >> >>>>>>>>> Thank you for considering my comments. > > > > >> >> >>>>>>>>> > > > > >> >> >>>>>>>>>> there might be custom logic before making retry, > such > > as > > > > >> >> >>>> re-establish the connection > > > > >> >> >>>>>>>>> > > > > >> >> >>>>>>>>> Yes, I understand that. I meant that such logic can > be > > > > >> placed in > > > > >> >> >> a > > > > >> >> >>>>>>>>> separate function, that can be implemented by > > connectors. > > > > >> Just > > > > >> >> >>> moving > > > > >> >> >>>>>>>>> the retry logic would make connector's LookupFunction > > > more > > > > >> >> >> concise > > > > >> >> >>> + > > > > >> >> >>>>>>>>> avoid duplicate code. However, it's a minor change. > The > > > > >> decision > > > > >> >> >> is > > > > >> >> >>>> up > > > > >> >> >>>>>>>>> to you. > > > > >> >> >>>>>>>>> > > > > >> >> >>>>>>>>>> We decide not to provide common DDL options and let > > > > >> developers > > > > >> >> >> to > > > > >> >> >>>> define their own options as we do now per connector. > > > > >> >> >>>>>>>>> > > > > >> >> >>>>>>>>> What is the reason for that? One of the main goals of > > > this > > > > >> FLIP > > > > >> >> >> was > > > > >> >> >>>> to > > > > >> >> >>>>>>>>> unify the configs, wasn't it? I understand that > current > > > > cache > > > > >> >> >>> design > > > > >> >> >>>>>>>>> doesn't depend on ConfigOptions, like was before. But > > > still > > > > >> we > > > > >> >> >> can > > > > >> >> >>>> put > > > > >> >> >>>>>>>>> these options into the framework, so connectors can > > reuse > > > > >> them > > > > >> >> >> and > > > > >> >> >>>>>>>>> avoid code duplication, and, what is more > significant, > > > > avoid > > > > >> >> >>> possible > > > > >> >> >>>>>>>>> different options naming. This moment can be pointed > > out > > > in > > > > >> >> >>>>>>>>> documentation for connector developers. > > > > >> >> >>>>>>>>> > > > > >> >> >>>>>>>>> Best regards, > > > > >> >> >>>>>>>>> Alexander > > > > >> >> >>>>>>>>> > > > > >> >> >>>>>>>>> вт, 17 мая 2022 г. в 17:11, Qingsheng Ren < > > > > >> renqs...@gmail.com>: > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>> Hi Alexander, > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>> Thanks for the review and glad to see we are on the > > same > > > > >> page! > > > > >> >> I > > > > >> >> >>>> think you forgot to cc the dev mailing list so I’m also > > > quoting > > > > >> your > > > > >> >> >>> reply > > > > >> >> >>>> under this email. > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>>> We can add 'maxRetryTimes' option into this class > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>> In my opinion the retry logic should be implemented > in > > > > >> lookup() > > > > >> >> >>>> instead of in LookupFunction#eval(). Retrying is only > > > meaningful > > > > >> >> under > > > > >> >> >>> some > > > > >> >> >>>> specific retriable failures, and there might be custom > logic > > > > >> before > > > > >> >> >>> making > > > > >> >> >>>> retry, such as re-establish the connection > > > > >> (JdbcRowDataLookupFunction > > > > >> >> >> is > > > > >> >> >>> an > > > > >> >> >>>> example), so it's more handy to leave it to the connector. > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>>> I don't see DDL options, that were in previous > > version > > > of > > > > >> >> FLIP. > > > > >> >> >>> Do > > > > >> >> >>>> you have any special plans for them? > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>> We decide not to provide common DDL options and let > > > > >> developers > > > > >> >> >> to > > > > >> >> >>>> define their own options as we do now per connector. > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>> The rest of comments sound great and I’ll update the > > > FLIP. > > > > >> Hope > > > > >> >> >> we > > > > >> >> >>>> can finalize our proposal soon! > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>> Best, > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>> Qingsheng > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>> > > > > >> >> >>>>>>>>>>> On May 17, 2022, at 13:46, Александр Смирнов < > > > > >> >> >>> smirale...@gmail.com> > > > > >> >> >>>> wrote: > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> Hi Qingsheng and devs! > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> I like the overall design of updated FLIP, however > I > > > have > > > > >> >> >> several > > > > >> >> >>>>>>>>>>> suggestions and questions. > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> 1) Introducing LookupFunction as a subclass of > > > > >> TableFunction > > > > >> >> >> is a > > > > >> >> >>>> good > > > > >> >> >>>>>>>>>>> idea. We can add 'maxRetryTimes' option into this > > > class. > > > > >> >> 'eval' > > > > >> >> >>>> method > > > > >> >> >>>>>>>>>>> of new LookupFunction is great for this purpose. > The > > > same > > > > >> is > > > > >> >> >> for > > > > >> >> >>>>>>>>>>> 'async' case. > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> 2) There might be other configs in future, such as > > > > >> >> >>>> 'cacheMissingKey' > > > > >> >> >>>>>>>>>>> in LookupFunctionProvider or 'rescanInterval' in > > > > >> >> >>>> ScanRuntimeProvider. > > > > >> >> >>>>>>>>>>> Maybe use Builder pattern in LookupFunctionProvider > > and > > > > >> >> >>>>>>>>>>> RescanRuntimeProvider for more flexibility (use one > > > > 'build' > > > > >> >> >>> method > > > > >> >> >>>>>>>>>>> instead of many 'of' methods in future)? > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> 3) What are the plans for existing > > > TableFunctionProvider > > > > >> and > > > > >> >> >>>>>>>>>>> AsyncTableFunctionProvider? I think they should be > > > > >> deprecated. > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> 4) Am I right that the current design does not > assume > > > > >> usage of > > > > >> >> >>>>>>>>>>> user-provided LookupCache in re-scanning? In this > > case, > > > > it > > > > >> is > > > > >> >> >> not > > > > >> >> >>>> very > > > > >> >> >>>>>>>>>>> clear why do we need methods such as 'invalidate' > or > > > > >> 'putAll' > > > > >> >> >> in > > > > >> >> >>>>>>>>>>> LookupCache. > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> 5) I don't see DDL options, that were in previous > > > version > > > > >> of > > > > >> >> >>> FLIP. > > > > >> >> >>>> Do > > > > >> >> >>>>>>>>>>> you have any special plans for them? > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> If you don't mind, I would be glad to be able to > make > > > > small > > > > >> >> >>>>>>>>>>> adjustments to the FLIP document too. I think it's > > > worth > > > > >> >> >>> mentioning > > > > >> >> >>>>>>>>>>> about what exactly optimizations are planning in > the > > > > >> future. > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>> Smirnov Alexander > > > > >> >> >>>>>>>>>>> > > > > >> >> >>>>>>>>>>> пт, 13 мая 2022 г. в 20:27, Qingsheng Ren < > > > > >> renqs...@gmail.com > > > > >> >> >>> : > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Hi Alexander and devs, > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Thank you very much for the in-depth discussion! > As > > > Jark > > > > >> >> >>>> mentioned we were inspired by Alexander's idea and made a > > > > >> refactor on > > > > >> >> >> our > > > > >> >> >>>> design. FLIP-221 [1] has been updated to reflect our > design > > > now > > > > >> and > > > > >> >> we > > > > >> >> >>> are > > > > >> >> >>>> happy to hear more suggestions from you! > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Compared to the previous design: > > > > >> >> >>>>>>>>>>>> 1. The lookup cache serves at table runtime level > > and > > > is > > > > >> >> >>>> integrated as a component of LookupJoinRunner as discussed > > > > >> >> previously. > > > > >> >> >>>>>>>>>>>> 2. Interfaces are renamed and re-designed to > reflect > > > the > > > > >> new > > > > >> >> >>>> design. > > > > >> >> >>>>>>>>>>>> 3. We separate the all-caching case individually > and > > > > >> >> >> introduce a > > > > >> >> >>>> new RescanRuntimeProvider to reuse the ability of > scanning. > > We > > > > are > > > > >> >> >>> planning > > > > >> >> >>>> to support SourceFunction / InputFormat for now > considering > > > the > > > > >> >> >>> complexity > > > > >> >> >>>> of FLIP-27 Source API. > > > > >> >> >>>>>>>>>>>> 4. A new interface LookupFunction is introduced to > > > make > > > > >> the > > > > >> >> >>>> semantic of lookup more straightforward for developers. > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> For replying to Alexander: > > > > >> >> >>>>>>>>>>>>> However I'm a little confused whether InputFormat > > is > > > > >> >> >> deprecated > > > > >> >> >>>> or not. Am I right that it will be so in the future, but > > > > currently > > > > >> >> it's > > > > >> >> >>> not? > > > > >> >> >>>>>>>>>>>> Yes you are right. InputFormat is not deprecated > for > > > > now. > > > > >> I > > > > >> >> >>> think > > > > >> >> >>>> it will be deprecated in the future but we don't have a > > clear > > > > plan > > > > >> >> for > > > > >> >> >>> that. > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Thanks again for the discussion on this FLIP and > > > looking > > > > >> >> >> forward > > > > >> >> >>>> to cooperating with you after we finalize the design and > > > > >> interfaces! > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> [1] > > > > >> >> >>>> > > > > >> >> >>> > > > > >> >> >> > > > > >> >> > > > > >> > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-221+Abstraction+for+lookup+source+cache+and+metric > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Qingsheng > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> On Fri, May 13, 2022 at 12:12 AM Александр > Смирнов < > > > > >> >> >>>> smirale...@gmail.com> wrote: > > > > >> >> >>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>> Hi Jark, Qingsheng and Leonard! > > > > >> >> >>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>> Glad to see that we came to a consensus on almost > > all > > > > >> >> points! > > > > >> >> >>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>> However I'm a little confused whether InputFormat > > is > > > > >> >> >> deprecated > > > > >> >> >>>> or > > > > >> >> >>>>>>>>>>>>> not. Am I right that it will be so in the future, > > but > > > > >> >> >> currently > > > > >> >> >>>> it's > > > > >> >> >>>>>>>>>>>>> not? Actually I also think that for the first > > version > > > > >> it's > > > > >> >> OK > > > > >> >> >>> to > > > > >> >> >>>> use > > > > >> >> >>>>>>>>>>>>> InputFormat in ALL cache realization, because > > > > supporting > > > > >> >> >> rescan > > > > >> >> >>>>>>>>>>>>> ability seems like a very distant prospect. But > for > > > > this > > > > >> >> >>>> decision we > > > > >> >> >>>>>>>>>>>>> need a consensus among all discussion > participants. > > > > >> >> >>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>> In general, I don't have something to argue with > > your > > > > >> >> >>>> statements. All > > > > >> >> >>>>>>>>>>>>> of them correspond my ideas. Looking ahead, it > > would > > > be > > > > >> nice > > > > >> >> >> to > > > > >> >> >>>> work > > > > >> >> >>>>>>>>>>>>> on this FLIP cooperatively. I've already done a > lot > > > of > > > > >> work > > > > >> >> >> on > > > > >> >> >>>> lookup > > > > >> >> >>>>>>>>>>>>> join caching with realization very close to the > one > > > we > > > > >> are > > > > >> >> >>>> discussing, > > > > >> >> >>>>>>>>>>>>> and want to share the results of this work. > Anyway > > > > >> looking > > > > >> >> >>>> forward for > > > > >> >> >>>>>>>>>>>>> the FLIP update! > > > > >> >> >>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>> Smirnov Alexander > > > > >> >> >>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>> чт, 12 мая 2022 г. в 17:38, Jark Wu < > > > imj...@gmail.com > > > > >: > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> Hi Alex, > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> Thanks for summarizing your points. > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> In the past week, Qingsheng, Leonard, and I have > > > > >> discussed > > > > >> >> >> it > > > > >> >> >>>> several times > > > > >> >> >>>>>>>>>>>>>> and we have totally refactored the design. > > > > >> >> >>>>>>>>>>>>>> I'm glad to say we have reached a consensus on > > many > > > of > > > > >> your > > > > >> >> >>>> points! > > > > >> >> >>>>>>>>>>>>>> Qingsheng is still working on updating the > design > > > docs > > > > >> and > > > > >> >> >>>> maybe can be > > > > >> >> >>>>>>>>>>>>>> available in the next few days. > > > > >> >> >>>>>>>>>>>>>> I will share some conclusions from our > > discussions: > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> 1) we have refactored the design towards to > "cache > > > in > > > > >> >> >>>> framework" way. > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> 2) a "LookupCache" interface for users to > > customize > > > > and > > > > >> a > > > > >> >> >>>> default > > > > >> >> >>>>>>>>>>>>>> implementation with builder for users to > easy-use. > > > > >> >> >>>>>>>>>>>>>> This can both make it possible to both have > > > > flexibility > > > > >> and > > > > >> >> >>>> conciseness. > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> 3) Filter pushdown is important for ALL and LRU > > > lookup > > > > >> >> >> cache, > > > > >> >> >>>> esp reducing > > > > >> >> >>>>>>>>>>>>>> IO. > > > > >> >> >>>>>>>>>>>>>> Filter pushdown should be the final state and > the > > > > >> unified > > > > >> >> >> way > > > > >> >> >>>> to both > > > > >> >> >>>>>>>>>>>>>> support pruning ALL cache and LRU cache, > > > > >> >> >>>>>>>>>>>>>> so I think we should make effort in this > > direction. > > > If > > > > >> we > > > > >> >> >> need > > > > >> >> >>>> to support > > > > >> >> >>>>>>>>>>>>>> filter pushdown for ALL cache anyway, why not > use > > > > >> >> >>>>>>>>>>>>>> it for LRU cache as well? Either way, as we > decide > > > to > > > > >> >> >>> implement > > > > >> >> >>>> the cache > > > > >> >> >>>>>>>>>>>>>> in the framework, we have the chance to support > > > > >> >> >>>>>>>>>>>>>> filter on cache anytime. This is an optimization > > and > > > > it > > > > >> >> >>> doesn't > > > > >> >> >>>> affect the > > > > >> >> >>>>>>>>>>>>>> public API. I think we can create a JIRA issue > to > > > > >> >> >>>>>>>>>>>>>> discuss it when the FLIP is accepted. > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> 4) The idea to support ALL cache is similar to > > your > > > > >> >> >> proposal. > > > > >> >> >>>>>>>>>>>>>> In the first version, we will only support > > > > InputFormat, > > > > >> >> >>>> SourceFunction for > > > > >> >> >>>>>>>>>>>>>> cache all (invoke InputFormat in join operator). > > > > >> >> >>>>>>>>>>>>>> For FLIP-27 source, we need to join a true > source > > > > >> operator > > > > >> >> >>>> instead of > > > > >> >> >>>>>>>>>>>>>> calling it embedded in the join operator. > > > > >> >> >>>>>>>>>>>>>> However, this needs another FLIP to support the > > > > re-scan > > > > >> >> >>> ability > > > > >> >> >>>> for FLIP-27 > > > > >> >> >>>>>>>>>>>>>> Source, and this can be a large work. > > > > >> >> >>>>>>>>>>>>>> In order to not block this issue, we can put the > > > > effort > > > > >> of > > > > >> >> >>>> FLIP-27 source > > > > >> >> >>>>>>>>>>>>>> integration into future work and integrate > > > > >> >> >>>>>>>>>>>>>> InputFormat&SourceFunction for now. > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> I think it's fine to use > > InputFormat&SourceFunction, > > > > as > > > > >> >> they > > > > >> >> >>>> are not > > > > >> >> >>>>>>>>>>>>>> deprecated, otherwise, we have to introduce > > another > > > > >> >> function > > > > >> >> >>>>>>>>>>>>>> similar to them which is meaningless. We need to > > > plan > > > > >> >> >> FLIP-27 > > > > >> >> >>>> source > > > > >> >> >>>>>>>>>>>>>> integration ASAP before InputFormat & > > SourceFunction > > > > are > > > > >> >> >>>> deprecated. > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> Best, > > > > >> >> >>>>>>>>>>>>>> Jark > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>> On Thu, 12 May 2022 at 15:46, Александр Смирнов > < > > > > >> >> >>>> smirale...@gmail.com> > > > > >> >> >>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> Hi Martijn! > > > > >> >> >>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> Got it. Therefore, the realization with > > InputFormat > > > > is > > > > >> not > > > > >> >> >>>> considered. > > > > >> >> >>>>>>>>>>>>>>> Thanks for clearing that up! > > > > >> >> >>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>> Smirnov Alexander > > > > >> >> >>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> чт, 12 мая 2022 г. в 14:23, Martijn Visser < > > > > >> >> >>>> mart...@ververica.com>: > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>> Hi, > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>> With regards to: > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> But if there are plans to refactor all > > connectors > > > > to > > > > >> >> >>> FLIP-27 > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>> Yes, FLIP-27 is the target for all connectors. > > The > > > > old > > > > >> >> >>>> interfaces will be > > > > >> >> >>>>>>>>>>>>>>>> deprecated and connectors will either be > > > refactored > > > > to > > > > >> >> use > > > > >> >> >>>> the new ones > > > > >> >> >>>>>>>>>>>>>>> or > > > > >> >> >>>>>>>>>>>>>>>> dropped. > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>> The caching should work for connectors that > are > > > > using > > > > >> >> >>> FLIP-27 > > > > >> >> >>>> interfaces, > > > > >> >> >>>>>>>>>>>>>>>> we should not introduce new features for old > > > > >> interfaces. > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>> Martijn > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>> On Thu, 12 May 2022 at 06:19, Александр > Смирнов > > < > > > > >> >> >>>> smirale...@gmail.com> > > > > >> >> >>>>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> Hi Jark! > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> Sorry for the late response. I would like to > > make > > > > >> some > > > > >> >> >>>> comments and > > > > >> >> >>>>>>>>>>>>>>>>> clarify my points. > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> 1) I agree with your first statement. I think > > we > > > > can > > > > >> >> >>> achieve > > > > >> >> >>>> both > > > > >> >> >>>>>>>>>>>>>>>>> advantages this way: put the Cache interface > in > > > > >> >> >>>> flink-table-common, > > > > >> >> >>>>>>>>>>>>>>>>> but have implementations of it in > > > > >> flink-table-runtime. > > > > >> >> >>>> Therefore if a > > > > >> >> >>>>>>>>>>>>>>>>> connector developer wants to use existing > cache > > > > >> >> >> strategies > > > > >> >> >>>> and their > > > > >> >> >>>>>>>>>>>>>>>>> implementations, he can just pass > lookupConfig > > to > > > > the > > > > >> >> >>>> planner, but if > > > > >> >> >>>>>>>>>>>>>>>>> he wants to have its own cache implementation > > in > > > > his > > > > >> >> >>>> TableFunction, it > > > > >> >> >>>>>>>>>>>>>>>>> will be possible for him to use the existing > > > > >> interface > > > > >> >> >> for > > > > >> >> >>>> this > > > > >> >> >>>>>>>>>>>>>>>>> purpose (we can explicitly point this out in > > the > > > > >> >> >>>> documentation). In > > > > >> >> >>>>>>>>>>>>>>>>> this way all configs and metrics will be > > unified. > > > > >> WDYT? > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> If a filter can prune 90% of data in the > > cache, > > > we > > > > >> will > > > > >> >> >>>> have 90% of > > > > >> >> >>>>>>>>>>>>>>>>> lookup requests that can never be cached > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> 2) Let me clarify the logic filters > > optimization > > > in > > > > >> case > > > > >> >> >> of > > > > >> >> >>>> LRU cache. > > > > >> >> >>>>>>>>>>>>>>>>> It looks like Cache<RowData, > > > Collection<RowData>>. > > > > >> Here > > > > >> >> >> we > > > > >> >> >>>> always > > > > >> >> >>>>>>>>>>>>>>>>> store the response of the dimension table in > > > cache, > > > > >> even > > > > >> >> >>>> after > > > > >> >> >>>>>>>>>>>>>>>>> applying calc function. I.e. if there are no > > rows > > > > >> after > > > > >> >> >>>> applying > > > > >> >> >>>>>>>>>>>>>>>>> filters to the result of the 'eval' method of > > > > >> >> >>> TableFunction, > > > > >> >> >>>> we store > > > > >> >> >>>>>>>>>>>>>>>>> the empty list by lookup keys. Therefore the > > > cache > > > > >> line > > > > >> >> >>> will > > > > >> >> >>>> be > > > > >> >> >>>>>>>>>>>>>>>>> filled, but will require much less memory (in > > > > bytes). > > > > >> >> >> I.e. > > > > >> >> >>>> we don't > > > > >> >> >>>>>>>>>>>>>>>>> completely filter keys, by which result was > > > pruned, > > > > >> but > > > > >> >> >>>> significantly > > > > >> >> >>>>>>>>>>>>>>>>> reduce required memory to store this result. > If > > > the > > > > >> user > > > > >> >> >>>> knows about > > > > >> >> >>>>>>>>>>>>>>>>> this behavior, he can increase the 'max-rows' > > > > option > > > > >> >> >> before > > > > >> >> >>>> the start > > > > >> >> >>>>>>>>>>>>>>>>> of the job. But actually I came up with the > > idea > > > > >> that we > > > > >> >> >>> can > > > > >> >> >>>> do this > > > > >> >> >>>>>>>>>>>>>>>>> automatically by using the 'maximumWeight' > and > > > > >> 'weigher' > > > > >> >> >>>> methods of > > > > >> >> >>>>>>>>>>>>>>>>> GuavaCache [1]. Weight can be the size of the > > > > >> collection > > > > >> >> >> of > > > > >> >> >>>> rows > > > > >> >> >>>>>>>>>>>>>>>>> (value of cache). Therefore cache can > > > automatically > > > > >> fit > > > > >> >> >>> much > > > > >> >> >>>> more > > > > >> >> >>>>>>>>>>>>>>>>> records than before. > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> Flink SQL has provided a standard way to do > > > > filters > > > > >> and > > > > >> >> >>>> projects > > > > >> >> >>>>>>>>>>>>>>>>> pushdown, i.e., SupportsFilterPushDown and > > > > >> >> >>>> SupportsProjectionPushDown. > > > > >> >> >>>>>>>>>>>>>>>>>> Jdbc/hive/HBase haven't implemented the > > > > interfaces, > > > > >> >> >> don't > > > > >> >> >>>> mean it's > > > > >> >> >>>>>>>>>>>>>>> hard > > > > >> >> >>>>>>>>>>>>>>>>> to implement. > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> It's debatable how difficult it will be to > > > > implement > > > > >> >> >> filter > > > > >> >> >>>> pushdown. > > > > >> >> >>>>>>>>>>>>>>>>> But I think the fact that currently there is > no > > > > >> database > > > > >> >> >>>> connector > > > > >> >> >>>>>>>>>>>>>>>>> with filter pushdown at least means that this > > > > feature > > > > >> >> >> won't > > > > >> >> >>>> be > > > > >> >> >>>>>>>>>>>>>>>>> supported soon in connectors. Moreover, if we > > > talk > > > > >> about > > > > >> >> >>>> other > > > > >> >> >>>>>>>>>>>>>>>>> connectors (not in Flink repo), their > databases > > > > might > > > > >> >> not > > > > >> >> >>>> support all > > > > >> >> >>>>>>>>>>>>>>>>> Flink filters (or not support filters at > all). > > I > > > > >> think > > > > >> >> >>> users > > > > >> >> >>>> are > > > > >> >> >>>>>>>>>>>>>>>>> interested in supporting cache filters > > > optimization > > > > >> >> >>>> independently of > > > > >> >> >>>>>>>>>>>>>>>>> supporting other features and solving more > > > complex > > > > >> >> >> problems > > > > >> >> >>>> (or > > > > >> >> >>>>>>>>>>>>>>>>> unsolvable at all). > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> 3) I agree with your third statement. > Actually > > in > > > > our > > > > >> >> >>>> internal version > > > > >> >> >>>>>>>>>>>>>>>>> I also tried to unify the logic of scanning > and > > > > >> >> reloading > > > > >> >> >>>> data from > > > > >> >> >>>>>>>>>>>>>>>>> connectors. But unfortunately, I didn't find > a > > > way > > > > to > > > > >> >> >> unify > > > > >> >> >>>> the logic > > > > >> >> >>>>>>>>>>>>>>>>> of all ScanRuntimeProviders (InputFormat, > > > > >> >> SourceFunction, > > > > >> >> >>>> Source,...) > > > > >> >> >>>>>>>>>>>>>>>>> and reuse it in reloading ALL cache. As a > > result > > > I > > > > >> >> >> settled > > > > >> >> >>>> on using > > > > >> >> >>>>>>>>>>>>>>>>> InputFormat, because it was used for scanning > > in > > > > all > > > > >> >> >> lookup > > > > >> >> >>>>>>>>>>>>>>>>> connectors. (I didn't know that there are > plans > > > to > > > > >> >> >>> deprecate > > > > >> >> >>>>>>>>>>>>>>>>> InputFormat in favor of FLIP-27 Source). IMO > > > usage > > > > of > > > > >> >> >>>> FLIP-27 source > > > > >> >> >>>>>>>>>>>>>>>>> in ALL caching is not good idea, because this > > > > source > > > > >> was > > > > >> >> >>>> designed to > > > > >> >> >>>>>>>>>>>>>>>>> work in distributed environment > > (SplitEnumerator > > > on > > > > >> >> >>>> JobManager and > > > > >> >> >>>>>>>>>>>>>>>>> SourceReaders on TaskManagers), not in one > > > operator > > > > >> >> >> (lookup > > > > >> >> >>>> join > > > > >> >> >>>>>>>>>>>>>>>>> operator in our case). There is even no > direct > > > way > > > > to > > > > >> >> >> pass > > > > >> >> >>>> splits from > > > > >> >> >>>>>>>>>>>>>>>>> SplitEnumerator to SourceReader (this logic > > works > > > > >> >> through > > > > >> >> >>>>>>>>>>>>>>>>> SplitEnumeratorContext, which requires > > > > >> >> >>>>>>>>>>>>>>>>> OperatorCoordinator.SubtaskGateway to send > > > > >> >> >> AddSplitEvents). > > > > >> >> >>>> Usage of > > > > >> >> >>>>>>>>>>>>>>>>> InputFormat for ALL cache seems much more > > clearer > > > > and > > > > >> >> >>>> easier. But if > > > > >> >> >>>>>>>>>>>>>>>>> there are plans to refactor all connectors to > > > > >> FLIP-27, I > > > > >> >> >>>> have the > > > > >> >> >>>>>>>>>>>>>>>>> following ideas: maybe we can refuse from > > lookup > > > > join > > > > >> >> ALL > > > > >> >> >>>> cache in > > > > >> >> >>>>>>>>>>>>>>>>> favor of simple join with multiple scanning > of > > > > batch > > > > >> >> >>> source? > > > > >> >> >>>> The point > > > > >> >> >>>>>>>>>>>>>>>>> is that the only difference between lookup > join > > > ALL > > > > >> >> cache > > > > >> >> >>>> and simple > > > > >> >> >>>>>>>>>>>>>>>>> join with batch source is that in the first > > case > > > > >> >> scanning > > > > >> >> >>> is > > > > >> >> >>>> performed > > > > >> >> >>>>>>>>>>>>>>>>> multiple times, in between which state > (cache) > > is > > > > >> >> cleared > > > > >> >> >>>> (correct me > > > > >> >> >>>>>>>>>>>>>>>>> if I'm wrong). So what if we extend the > > > > >> functionality of > > > > >> >> >>>> simple join > > > > >> >> >>>>>>>>>>>>>>>>> to support state reloading + extend the > > > > >> functionality of > > > > >> >> >>>> scanning > > > > >> >> >>>>>>>>>>>>>>>>> batch source multiple times (this one should > be > > > > easy > > > > >> >> with > > > > >> >> >>>> new FLIP-27 > > > > >> >> >>>>>>>>>>>>>>>>> source, that unifies streaming/batch reading > - > > we > > > > >> will > > > > >> >> >> need > > > > >> >> >>>> to change > > > > >> >> >>>>>>>>>>>>>>>>> only SplitEnumerator, which will pass splits > > > again > > > > >> after > > > > >> >> >>>> some TTL). > > > > >> >> >>>>>>>>>>>>>>>>> WDYT? I must say that this looks like a > > long-term > > > > >> goal > > > > >> >> >> and > > > > >> >> >>>> will make > > > > >> >> >>>>>>>>>>>>>>>>> the scope of this FLIP even larger than you > > said. > > > > >> Maybe > > > > >> >> >> we > > > > >> >> >>>> can limit > > > > >> >> >>>>>>>>>>>>>>>>> ourselves to a simpler solution now > > > (InputFormats). > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> So to sum up, my points is like this: > > > > >> >> >>>>>>>>>>>>>>>>> 1) There is a way to make both concise and > > > flexible > > > > >> >> >>>> interfaces for > > > > >> >> >>>>>>>>>>>>>>>>> caching in lookup join. > > > > >> >> >>>>>>>>>>>>>>>>> 2) Cache filters optimization is important > both > > > in > > > > >> LRU > > > > >> >> >> and > > > > >> >> >>>> ALL caches. > > > > >> >> >>>>>>>>>>>>>>>>> 3) It is unclear when filter pushdown will be > > > > >> supported > > > > >> >> >> in > > > > >> >> >>>> Flink > > > > >> >> >>>>>>>>>>>>>>>>> connectors, some of the connectors might not > > have > > > > the > > > > >> >> >>>> opportunity to > > > > >> >> >>>>>>>>>>>>>>>>> support filter pushdown + as I know, > currently > > > > filter > > > > >> >> >>>> pushdown works > > > > >> >> >>>>>>>>>>>>>>>>> only for scanning (not lookup). So cache > > filters > > > + > > > > >> >> >>>> projections > > > > >> >> >>>>>>>>>>>>>>>>> optimization should be independent from other > > > > >> features. > > > > >> >> >>>>>>>>>>>>>>>>> 4) ALL cache realization is a complex topic > > that > > > > >> >> involves > > > > >> >> >>>> multiple > > > > >> >> >>>>>>>>>>>>>>>>> aspects of how Flink is developing. Refusing > > from > > > > >> >> >>>> InputFormat in favor > > > > >> >> >>>>>>>>>>>>>>>>> of FLIP-27 Source will make ALL cache > > realization > > > > >> really > > > > >> >> >>>> complex and > > > > >> >> >>>>>>>>>>>>>>>>> not clear, so maybe instead of that we can > > extend > > > > the > > > > >> >> >>>> functionality of > > > > >> >> >>>>>>>>>>>>>>>>> simple join or not refuse from InputFormat in > > > case > > > > of > > > > >> >> >>> lookup > > > > >> >> >>>> join ALL > > > > >> >> >>>>>>>>>>>>>>>>> cache? > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>>> Smirnov Alexander > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> [1] > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> > > > > >> >> >>>> > > > > >> >> >>> > > > > >> >> >> > > > > >> >> > > > > >> > > > > > > > > > > https://guava.dev/releases/18.0/api/docs/com/google/common/cache/CacheBuilder.html#weigher(com.google.common.cache.Weigher) > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> чт, 5 мая 2022 г. в 20:34, Jark Wu < > > > > imj...@gmail.com > > > > >> >: > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> It's great to see the active discussion! I > > want > > > to > > > > >> >> share > > > > >> >> >>> my > > > > >> >> >>>> ideas: > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> 1) implement the cache in framework vs. > > > connectors > > > > >> base > > > > >> >> >>>>>>>>>>>>>>>>>> I don't have a strong opinion on this. Both > > ways > > > > >> should > > > > >> >> >>>> work (e.g., > > > > >> >> >>>>>>>>>>>>>>> cache > > > > >> >> >>>>>>>>>>>>>>>>>> pruning, compatibility). > > > > >> >> >>>>>>>>>>>>>>>>>> The framework way can provide more concise > > > > >> interfaces. > > > > >> >> >>>>>>>>>>>>>>>>>> The connector base way can define more > > flexible > > > > >> cache > > > > >> >> >>>>>>>>>>>>>>>>>> strategies/implementations. > > > > >> >> >>>>>>>>>>>>>>>>>> We are still investigating a way to see if > we > > > can > > > > >> have > > > > >> >> >>> both > > > > >> >> >>>>>>>>>>>>>>> advantages. > > > > >> >> >>>>>>>>>>>>>>>>>> We should reach a consensus that the way > > should > > > > be a > > > > >> >> >> final > > > > >> >> >>>> state, > > > > >> >> >>>>>>>>>>>>>>> and we > > > > >> >> >>>>>>>>>>>>>>>>>> are on the path to it. > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> 2) filters and projections pushdown: > > > > >> >> >>>>>>>>>>>>>>>>>> I agree with Alex that the filter pushdown > > into > > > > >> cache > > > > >> >> >> can > > > > >> >> >>>> benefit a > > > > >> >> >>>>>>>>>>>>>>> lot > > > > >> >> >>>>>>>>>>>>>>>>> for > > > > >> >> >>>>>>>>>>>>>>>>>> ALL cache. > > > > >> >> >>>>>>>>>>>>>>>>>> However, this is not true for LRU cache. > > > > Connectors > > > > >> use > > > > >> >> >>>> cache to > > > > >> >> >>>>>>>>>>>>>>> reduce > > > > >> >> >>>>>>>>>>>>>>>>> IO > > > > >> >> >>>>>>>>>>>>>>>>>> requests to databases for better throughput. > > > > >> >> >>>>>>>>>>>>>>>>>> If a filter can prune 90% of data in the > > cache, > > > we > > > > >> will > > > > >> >> >>>> have 90% of > > > > >> >> >>>>>>>>>>>>>>>>> lookup > > > > >> >> >>>>>>>>>>>>>>>>>> requests that can never be cached > > > > >> >> >>>>>>>>>>>>>>>>>> and hit directly to the databases. That > means > > > the > > > > >> cache > > > > >> >> >> is > > > > >> >> >>>>>>>>>>>>>>> meaningless in > > > > >> >> >>>>>>>>>>>>>>>>>> this case. > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> IMO, Flink SQL has provided a standard way > to > > do > > > > >> >> filters > > > > >> >> >>>> and projects > > > > >> >> >>>>>>>>>>>>>>>>>> pushdown, i.e., SupportsFilterPushDown and > > > > >> >> >>>>>>>>>>>>>>> SupportsProjectionPushDown. > > > > >> >> >>>>>>>>>>>>>>>>>> Jdbc/hive/HBase haven't implemented the > > > > interfaces, > > > > >> >> >> don't > > > > >> >> >>>> mean it's > > > > >> >> >>>>>>>>>>>>>>> hard > > > > >> >> >>>>>>>>>>>>>>>>> to > > > > >> >> >>>>>>>>>>>>>>>>>> implement. > > > > >> >> >>>>>>>>>>>>>>>>>> They should implement the pushdown > interfaces > > to > > > > >> reduce > > > > >> >> >> IO > > > > >> >> >>>> and the > > > > >> >> >>>>>>>>>>>>>>> cache > > > > >> >> >>>>>>>>>>>>>>>>>> size. > > > > >> >> >>>>>>>>>>>>>>>>>> That should be a final state that the scan > > > source > > > > >> and > > > > >> >> >>>> lookup source > > > > >> >> >>>>>>>>>>>>>>> share > > > > >> >> >>>>>>>>>>>>>>>>>> the exact pushdown implementation. > > > > >> >> >>>>>>>>>>>>>>>>>> I don't see why we need to duplicate the > > > pushdown > > > > >> logic > > > > >> >> >> in > > > > >> >> >>>> caches, > > > > >> >> >>>>>>>>>>>>>>> which > > > > >> >> >>>>>>>>>>>>>>>>>> will complex the lookup join design. > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> 3) ALL cache abstraction > > > > >> >> >>>>>>>>>>>>>>>>>> All cache might be the most challenging part > > of > > > > this > > > > >> >> >> FLIP. > > > > >> >> >>>> We have > > > > >> >> >>>>>>>>>>>>>>> never > > > > >> >> >>>>>>>>>>>>>>>>>> provided a reload-lookup public interface. > > > > >> >> >>>>>>>>>>>>>>>>>> Currently, we put the reload logic in the > > "eval" > > > > >> method > > > > >> >> >> of > > > > >> >> >>>>>>>>>>>>>>> TableFunction. > > > > >> >> >>>>>>>>>>>>>>>>>> That's hard for some sources (e.g., Hive). > > > > >> >> >>>>>>>>>>>>>>>>>> Ideally, connector implementation should > share > > > the > > > > >> >> logic > > > > >> >> >>> of > > > > >> >> >>>> reload > > > > >> >> >>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>>> scan, i.e. ScanTableSource with > > > > >> >> >>>> InputFormat/SourceFunction/FLIP-27 > > > > >> >> >>>>>>>>>>>>>>>>> Source. > > > > >> >> >>>>>>>>>>>>>>>>>> However, InputFormat/SourceFunction are > > > > deprecated, > > > > >> and > > > > >> >> >>> the > > > > >> >> >>>> FLIP-27 > > > > >> >> >>>>>>>>>>>>>>>>> source > > > > >> >> >>>>>>>>>>>>>>>>>> is deeply coupled with SourceOperator. > > > > >> >> >>>>>>>>>>>>>>>>>> If we want to invoke the FLIP-27 source in > > > > >> LookupJoin, > > > > >> >> >>> this > > > > >> >> >>>> may make > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>> scope of this FLIP much larger. > > > > >> >> >>>>>>>>>>>>>>>>>> We are still investigating how to abstract > the > > > ALL > > > > >> >> cache > > > > >> >> >>>> logic and > > > > >> >> >>>>>>>>>>>>>>> reuse > > > > >> >> >>>>>>>>>>>>>>>>>> the existing source interfaces. > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> Best, > > > > >> >> >>>>>>>>>>>>>>>>>> Jark > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>> On Thu, 5 May 2022 at 20:22, Roman Boyko < > > > > >> >> >>>> ro.v.bo...@gmail.com> > > > > >> >> >>>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>> It's a much more complicated activity and > > lies > > > > out > > > > >> of > > > > >> >> >> the > > > > >> >> >>>> scope of > > > > >> >> >>>>>>>>>>>>>>> this > > > > >> >> >>>>>>>>>>>>>>>>>>> improvement. Because such pushdowns should > be > > > > done > > > > >> for > > > > >> >> >>> all > > > > >> >> >>>>>>>>>>>>>>>>> ScanTableSource > > > > >> >> >>>>>>>>>>>>>>>>>>> implementations (not only for Lookup ones). > > > > >> >> >>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>> On Thu, 5 May 2022 at 19:02, Martijn > Visser < > > > > >> >> >>>>>>>>>>>>>>> martijnvis...@apache.org> > > > > >> >> >>>>>>>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> Hi everyone, > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> One question regarding "And Alexander > > > correctly > > > > >> >> >>> mentioned > > > > >> >> >>>> that > > > > >> >> >>>>>>>>>>>>>>> filter > > > > >> >> >>>>>>>>>>>>>>>>>>>> pushdown still is not implemented for > > > > >> >> >> jdbc/hive/hbase." > > > > >> >> >>>> -> Would > > > > >> >> >>>>>>>>>>>>>>> an > > > > >> >> >>>>>>>>>>>>>>>>>>>> alternative solution be to actually > > implement > > > > >> these > > > > >> >> >>> filter > > > > >> >> >>>>>>>>>>>>>>> pushdowns? > > > > >> >> >>>>>>>>>>>>>>>>> I > > > > >> >> >>>>>>>>>>>>>>>>>>>> can > > > > >> >> >>>>>>>>>>>>>>>>>>>> imagine that there are many more benefits > to > > > > doing > > > > >> >> >> that, > > > > >> >> >>>> outside > > > > >> >> >>>>>>>>>>>>>>> of > > > > >> >> >>>>>>>>>>>>>>>>> lookup > > > > >> >> >>>>>>>>>>>>>>>>>>>> caching and metrics. > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> Martijn Visser > > > > >> >> >>>>>>>>>>>>>>>>>>>> https://twitter.com/MartijnVisser82 > > > > >> >> >>>>>>>>>>>>>>>>>>>> https://github.com/MartijnVisser > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> On Thu, 5 May 2022 at 13:58, Roman Boyko < > > > > >> >> >>>> ro.v.bo...@gmail.com> > > > > >> >> >>>>>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> Hi everyone! > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> Thanks for driving such a valuable > > > improvement! > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> I do think that single cache > implementation > > > > >> would be > > > > >> >> >> a > > > > >> >> >>>> nice > > > > >> >> >>>>>>>>>>>>>>>>> opportunity > > > > >> >> >>>>>>>>>>>>>>>>>>>> for > > > > >> >> >>>>>>>>>>>>>>>>>>>>> users. And it will break the "FOR > > SYSTEM_TIME > > > > AS > > > > >> OF > > > > >> >> >>>> proc_time" > > > > >> >> >>>>>>>>>>>>>>>>> semantics > > > > >> >> >>>>>>>>>>>>>>>>>>>>> anyway - doesn't matter how it will be > > > > >> implemented. > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> Putting myself in the user's shoes, I can > > say > > > > >> that: > > > > >> >> >>>>>>>>>>>>>>>>>>>>> 1) I would prefer to have the opportunity > > to > > > > cut > > > > >> off > > > > >> >> >>> the > > > > >> >> >>>> cache > > > > >> >> >>>>>>>>>>>>>>> size > > > > >> >> >>>>>>>>>>>>>>>>> by > > > > >> >> >>>>>>>>>>>>>>>>>>>>> simply filtering unnecessary data. And > the > > > most > > > > >> >> handy > > > > >> >> >>>> way to do > > > > >> >> >>>>>>>>>>>>>>> it > > > > >> >> >>>>>>>>>>>>>>>>> is > > > > >> >> >>>>>>>>>>>>>>>>>>>> apply > > > > >> >> >>>>>>>>>>>>>>>>>>>>> it inside LookupRunners. It would be a > bit > > > > >> harder to > > > > >> >> >>>> pass it > > > > >> >> >>>>>>>>>>>>>>>>> through the > > > > >> >> >>>>>>>>>>>>>>>>>>>>> LookupJoin node to TableFunction. And > > > Alexander > > > > >> >> >>> correctly > > > > >> >> >>>>>>>>>>>>>>> mentioned > > > > >> >> >>>>>>>>>>>>>>>>> that > > > > >> >> >>>>>>>>>>>>>>>>>>>>> filter pushdown still is not implemented > > for > > > > >> >> >>>> jdbc/hive/hbase. > > > > >> >> >>>>>>>>>>>>>>>>>>>>> 2) The ability to set the different > caching > > > > >> >> >> parameters > > > > >> >> >>>> for > > > > >> >> >>>>>>>>>>>>>>> different > > > > >> >> >>>>>>>>>>>>>>>>>>>> tables > > > > >> >> >>>>>>>>>>>>>>>>>>>>> is quite important. So I would prefer to > > set > > > it > > > > >> >> >> through > > > > >> >> >>>> DDL > > > > >> >> >>>>>>>>>>>>>>> rather > > > > >> >> >>>>>>>>>>>>>>>>> than > > > > >> >> >>>>>>>>>>>>>>>>>>>>> have similar ttla, strategy and other > > options > > > > for > > > > >> >> all > > > > >> >> >>>> lookup > > > > >> >> >>>>>>>>>>>>>>> tables. > > > > >> >> >>>>>>>>>>>>>>>>>>>>> 3) Providing the cache into the framework > > > > really > > > > >> >> >>>> deprives us of > > > > >> >> >>>>>>>>>>>>>>>>>>>>> extensibility (users won't be able to > > > implement > > > > >> >> their > > > > >> >> >>> own > > > > >> >> >>>>>>>>>>>>>>> cache). > > > > >> >> >>>>>>>>>>>>>>>>> But > > > > >> >> >>>>>>>>>>>>>>>>>>>> most > > > > >> >> >>>>>>>>>>>>>>>>>>>>> probably it might be solved by creating > > more > > > > >> >> >> different > > > > >> >> >>>> cache > > > > >> >> >>>>>>>>>>>>>>>>> strategies > > > > >> >> >>>>>>>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>>>>>> a wider set of configurations. > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> All these points are much closer to the > > > schema > > > > >> >> >> proposed > > > > >> >> >>>> by > > > > >> >> >>>>>>>>>>>>>>>>> Alexander. > > > > >> >> >>>>>>>>>>>>>>>>>>>>> Qingshen Ren, please correct me if I'm > not > > > > right > > > > >> and > > > > >> >> >>> all > > > > >> >> >>>> these > > > > >> >> >>>>>>>>>>>>>>>>>>>> facilities > > > > >> >> >>>>>>>>>>>>>>>>>>>>> might be simply implemented in your > > > > architecture? > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>>>>>>> Roman Boyko > > > > >> >> >>>>>>>>>>>>>>>>>>>>> e.: ro.v.bo...@gmail.com > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> On Wed, 4 May 2022 at 21:01, Martijn > > Visser < > > > > >> >> >>>>>>>>>>>>>>>>> martijnvis...@apache.org> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> Hi everyone, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> I don't have much to chip in, but just > > > wanted > > > > to > > > > >> >> >>>> express that > > > > >> >> >>>>>>>>>>>>>>> I > > > > >> >> >>>>>>>>>>>>>>>>> really > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> appreciate the in-depth discussion on > this > > > > topic > > > > >> >> >> and I > > > > >> >> >>>> hope > > > > >> >> >>>>>>>>>>>>>>> that > > > > >> >> >>>>>>>>>>>>>>>>>>>> others > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> will join the conversation. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> Martijn > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> On Tue, 3 May 2022 at 10:15, Александр > > > > Смирнов < > > > > >> >> >>>>>>>>>>>>>>>>> smirale...@gmail.com> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> Hi Qingsheng, Leonard and Jark, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> Thanks for your detailed feedback! > > > However, I > > > > >> have > > > > >> >> >>>> questions > > > > >> >> >>>>>>>>>>>>>>>>> about > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> some of your statements (maybe I didn't > > get > > > > >> >> >>>> something?). > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Caching actually breaks the semantic > of > > > "FOR > > > > >> >> >>>> SYSTEM_TIME > > > > >> >> >>>>>>>>>>>>>>> AS OF > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> proc_time” > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> I agree that the semantics of "FOR > > > > SYSTEM_TIME > > > > >> AS > > > > >> >> >> OF > > > > >> >> >>>>>>>>>>>>>>> proc_time" > > > > >> >> >>>>>>>>>>>>>>>>> is > > > > >> >> >>>>>>>>>>>>>>>>>>>> not > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> fully implemented with caching, but as > > you > > > > >> said, > > > > >> >> >>> users > > > > >> >> >>>> go > > > > >> >> >>>>>>>>>>>>>>> on it > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> consciously to achieve better > performance > > > (no > > > > >> one > > > > >> >> >>>> proposed > > > > >> >> >>>>>>>>>>>>>>> to > > > > >> >> >>>>>>>>>>>>>>>>> enable > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> caching by default, etc.). Or by users > do > > > you > > > > >> mean > > > > >> >> >>>> other > > > > >> >> >>>>>>>>>>>>>>>>> developers > > > > >> >> >>>>>>>>>>>>>>>>>>>> of > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> connectors? In this case developers > > > > explicitly > > > > >> >> >>> specify > > > > >> >> >>>>>>>>>>>>>>> whether > > > > >> >> >>>>>>>>>>>>>>>>> their > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> connector supports caching or not (in > the > > > > list > > > > >> of > > > > >> >> >>>> supported > > > > >> >> >>>>>>>>>>>>>>>>>>>> options), > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> no one makes them do that if they don't > > > want > > > > >> to. > > > > >> >> So > > > > >> >> >>>> what > > > > >> >> >>>>>>>>>>>>>>>>> exactly is > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> the difference between implementing > > caching > > > > in > > > > >> >> >>> modules > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> flink-table-runtime and in > > > flink-table-common > > > > >> from > > > > >> >> >>> the > > > > >> >> >>>>>>>>>>>>>>>>> considered > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> point of view? How does it affect on > > > > >> >> >>>> breaking/non-breaking > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> semantics of "FOR SYSTEM_TIME AS OF > > > > proc_time"? > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> confront a situation that allows table > > > > >> options in > > > > >> >> >>> DDL > > > > >> >> >>>> to > > > > >> >> >>>>>>>>>>>>>>>>> control > > > > >> >> >>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> behavior of the framework, which has > > never > > > > >> >> happened > > > > >> >> >>>>>>>>>>>>>>> previously > > > > >> >> >>>>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>>>>>> should > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> be cautious > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> If we talk about main differences of > > > > semantics > > > > >> of > > > > >> >> >> DDL > > > > >> >> >>>>>>>>>>>>>>> options > > > > >> >> >>>>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> config options("table.exec.xxx"), isn't > > it > > > > >> about > > > > >> >> >>>> limiting > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>> scope > > > > >> >> >>>>>>>>>>>>>>>>>>>> of > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> the options + importance for the user > > > > business > > > > >> >> >> logic > > > > >> >> >>>> rather > > > > >> >> >>>>>>>>>>>>>>> than > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> specific location of corresponding > logic > > in > > > > the > > > > >> >> >>>> framework? I > > > > >> >> >>>>>>>>>>>>>>>>> mean > > > > >> >> >>>>>>>>>>>>>>>>>>>> that > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> in my design, for example, putting an > > > option > > > > >> with > > > > >> >> >>>> lookup > > > > >> >> >>>>>>>>>>>>>>> cache > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> strategy in configurations would be > the > > > > wrong > > > > >> >> >>>> decision, > > > > >> >> >>>>>>>>>>>>>>>>> because it > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> directly affects the user's business > > logic > > > > (not > > > > >> >> >> just > > > > >> >> >>>>>>>>>>>>>>> performance > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> optimization) + touches just several > > > > functions > > > > >> of > > > > >> >> >> ONE > > > > >> >> >>>> table > > > > >> >> >>>>>>>>>>>>>>>>> (there > > > > >> >> >>>>>>>>>>>>>>>>>>>> can > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> be multiple tables with different > > caches). > > > > >> Does it > > > > >> >> >>>> really > > > > >> >> >>>>>>>>>>>>>>>>> matter for > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> the user (or someone else) where the > > logic > > > is > > > > >> >> >>> located, > > > > >> >> >>>>>>>>>>>>>>> which is > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> affected by the applied option? > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> Also I can remember DDL option > > > > >> 'sink.parallelism', > > > > >> >> >>>> which in > > > > >> >> >>>>>>>>>>>>>>>>> some way > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> "controls the behavior of the > framework" > > > and > > > > I > > > > >> >> >> don't > > > > >> >> >>>> see any > > > > >> >> >>>>>>>>>>>>>>>>> problem > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> here. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> introduce a new interface for this > > > > all-caching > > > > >> >> >>>> scenario > > > > >> >> >>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>> design > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> would become more complex > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> This is a subject for a separate > > > discussion, > > > > >> but > > > > >> >> >>>> actually > > > > >> >> >>>>>>>>>>>>>>> in our > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> internal version we solved this problem > > > quite > > > > >> >> >> easily > > > > >> >> >>> - > > > > >> >> >>>> we > > > > >> >> >>>>>>>>>>>>>>> reused > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> InputFormat class (so there is no need > > for > > > a > > > > >> new > > > > >> >> >>> API). > > > > >> >> >>>> The > > > > >> >> >>>>>>>>>>>>>>>>> point is > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> that currently all lookup connectors > use > > > > >> >> >> InputFormat > > > > >> >> >>>> for > > > > >> >> >>>>>>>>>>>>>>>>> scanning > > > > >> >> >>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> data in batch mode: HBase, JDBC and > even > > > Hive > > > > >> - it > > > > >> >> >>> uses > > > > >> >> >>>>>>>>>>>>>>> class > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> PartitionReader, that is actually just > a > > > > >> wrapper > > > > >> >> >>> around > > > > >> >> >>>>>>>>>>>>>>>>> InputFormat. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> The advantage of this solution is the > > > ability > > > > >> to > > > > >> >> >>> reload > > > > >> >> >>>>>>>>>>>>>>> cache > > > > >> >> >>>>>>>>>>>>>>>>> data > > > > >> >> >>>>>>>>>>>>>>>>>>>> in > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> parallel (number of threads depends on > > > number > > > > >> of > > > > >> >> >>>>>>>>>>>>>>> InputSplits, > > > > >> >> >>>>>>>>>>>>>>>>> but > > > > >> >> >>>>>>>>>>>>>>>>>>>> has > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> an upper limit). As a result cache > reload > > > > time > > > > >> >> >>>> significantly > > > > >> >> >>>>>>>>>>>>>>>>> reduces > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> (as well as time of input stream > > > blocking). I > > > > >> know > > > > >> >> >>> that > > > > >> >> >>>>>>>>>>>>>>> usually > > > > >> >> >>>>>>>>>>>>>>>>> we > > > > >> >> >>>>>>>>>>>>>>>>>>>> try > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> to avoid usage of concurrency in Flink > > > code, > > > > >> but > > > > >> >> >>> maybe > > > > >> >> >>>> this > > > > >> >> >>>>>>>>>>>>>>> one > > > > >> >> >>>>>>>>>>>>>>>>> can > > > > >> >> >>>>>>>>>>>>>>>>>>>> be > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> an exception. BTW I don't say that it's > > an > > > > >> ideal > > > > >> >> >>>> solution, > > > > >> >> >>>>>>>>>>>>>>> maybe > > > > >> >> >>>>>>>>>>>>>>>>>>>> there > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> are better ones. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Providing the cache in the framework > > might > > > > >> >> >> introduce > > > > >> >> >>>>>>>>>>>>>>>>> compatibility > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> issues > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> It's possible only in cases when the > > > > developer > > > > >> of > > > > >> >> >> the > > > > >> >> >>>>>>>>>>>>>>> connector > > > > >> >> >>>>>>>>>>>>>>>>>>>> won't > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> properly refactor his code and will use > > new > > > > >> cache > > > > >> >> >>>> options > > > > >> >> >>>>>>>>>>>>>>>>>>>> incorrectly > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> (i.e. explicitly provide the same > options > > > > into > > > > >> 2 > > > > >> >> >>>> different > > > > >> >> >>>>>>>>>>>>>>> code > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> places). For correct behavior all he > will > > > > need > > > > >> to > > > > >> >> >> do > > > > >> >> >>>> is to > > > > >> >> >>>>>>>>>>>>>>>>> redirect > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> existing options to the framework's > > > > >> LookupConfig > > > > >> >> (+ > > > > >> >> >>>> maybe > > > > >> >> >>>>>>>>>>>>>>> add an > > > > >> >> >>>>>>>>>>>>>>>>>>>> alias > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> for options, if there was different > > > naming), > > > > >> >> >>> everything > > > > >> >> >>>>>>>>>>>>>>> will be > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> transparent for users. If the developer > > > won't > > > > >> do > > > > >> >> >>>>>>>>>>>>>>> refactoring at > > > > >> >> >>>>>>>>>>>>>>>>> all, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> nothing will be changed for the > connector > > > > >> because > > > > >> >> >> of > > > > >> >> >>>>>>>>>>>>>>> backward > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> compatibility. Also if a developer > wants > > to > > > > use > > > > >> >> his > > > > >> >> >>> own > > > > >> >> >>>>>>>>>>>>>>> cache > > > > >> >> >>>>>>>>>>>>>>>>> logic, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> he just can refuse to pass some of the > > > > configs > > > > >> >> into > > > > >> >> >>> the > > > > >> >> >>>>>>>>>>>>>>>>> framework, > > > > >> >> >>>>>>>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> instead make his own implementation > with > > > > >> already > > > > >> >> >>>> existing > > > > >> >> >>>>>>>>>>>>>>>>> configs > > > > >> >> >>>>>>>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> metrics (but actually I think that > it's a > > > > rare > > > > >> >> >> case). > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> filters and projections should be > pushed > > > all > > > > >> the > > > > >> >> >> way > > > > >> >> >>>> down > > > > >> >> >>>>>>>>>>>>>>> to > > > > >> >> >>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>> table > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> function, like what we do in the scan > > > source > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> It's the great purpose. But the truth > is > > > that > > > > >> the > > > > >> >> >>> ONLY > > > > >> >> >>>>>>>>>>>>>>> connector > > > > >> >> >>>>>>>>>>>>>>>>>>>> that > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> supports filter pushdown is > > > > >> FileSystemTableSource > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> (no database connector supports it > > > > currently). > > > > >> >> Also > > > > >> >> >>>> for some > > > > >> >> >>>>>>>>>>>>>>>>>>>> databases > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> it's simply impossible to pushdown such > > > > complex > > > > >> >> >>> filters > > > > >> >> >>>>>>>>>>>>>>> that we > > > > >> >> >>>>>>>>>>>>>>>>> have > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> in Flink. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> only applying these optimizations to > the > > > > cache > > > > >> >> >> seems > > > > >> >> >>>> not > > > > >> >> >>>>>>>>>>>>>>>>> quite > > > > >> >> >>>>>>>>>>>>>>>>>>>>> useful > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> Filters can cut off an arbitrarily > large > > > > >> amount of > > > > >> >> >>> data > > > > >> >> >>>>>>>>>>>>>>> from the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> dimension table. For a simple example, > > > > suppose > > > > >> in > > > > >> >> >>>> dimension > > > > >> >> >>>>>>>>>>>>>>>>> table > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> 'users' > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> we have column 'age' with values from > 20 > > to > > > > 40, > > > > >> >> and > > > > >> >> >>>> input > > > > >> >> >>>>>>>>>>>>>>> stream > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> 'clicks' that is ~uniformly distributed > > by > > > > age > > > > >> of > > > > >> >> >>>> users. If > > > > >> >> >>>>>>>>>>>>>>> we > > > > >> >> >>>>>>>>>>>>>>>>> have > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> filter 'age > 30', > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> there will be twice less data in cache. > > > This > > > > >> means > > > > >> >> >>> the > > > > >> >> >>>> user > > > > >> >> >>>>>>>>>>>>>>> can > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> increase 'lookup.cache.max-rows' by > > almost > > > 2 > > > > >> >> times. > > > > >> >> >>> It > > > > >> >> >>>> will > > > > >> >> >>>>>>>>>>>>>>>>> gain a > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> huge > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> performance boost. Moreover, this > > > > optimization > > > > >> >> >> starts > > > > >> >> >>>> to > > > > >> >> >>>>>>>>>>>>>>> really > > > > >> >> >>>>>>>>>>>>>>>>>>>> shine > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> in 'ALL' cache, where tables without > > > filters > > > > >> and > > > > >> >> >>>> projections > > > > >> >> >>>>>>>>>>>>>>>>> can't > > > > >> >> >>>>>>>>>>>>>>>>>>>> fit > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> in memory, but with them - can. This > > opens > > > up > > > > >> >> >>>> additional > > > > >> >> >>>>>>>>>>>>>>>>>>>> possibilities > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> for users. And this doesn't sound as > 'not > > > > quite > > > > >> >> >>>> useful'. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> It would be great to hear other voices > > > > >> regarding > > > > >> >> >> this > > > > >> >> >>>> topic! > > > > >> >> >>>>>>>>>>>>>>>>> Because > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> we have quite a lot of controversial > > > points, > > > > >> and I > > > > >> >> >>>> think > > > > >> >> >>>>>>>>>>>>>>> with > > > > >> >> >>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>> help > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> of others it will be easier for us to > > come > > > > to a > > > > >> >> >>>> consensus. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> Smirnov Alexander > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> пт, 29 апр. 2022 г. в 22:33, Qingsheng > > Ren > > > < > > > > >> >> >>>>>>>>>>>>>>> renqs...@gmail.com > > > > >> >> >>>>>>>>>>>>>>>>>> : > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Hi Alexander and Arvid, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Thanks for the discussion and sorry > for > > my > > > > >> late > > > > >> >> >>>> response! > > > > >> >> >>>>>>>>>>>>>>> We > > > > >> >> >>>>>>>>>>>>>>>>> had > > > > >> >> >>>>>>>>>>>>>>>>>>>> an > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> internal discussion together with Jark > > and > > > > >> Leonard > > > > >> >> >>> and > > > > >> >> >>>> I’d > > > > >> >> >>>>>>>>>>>>>>> like > > > > >> >> >>>>>>>>>>>>>>>>> to > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> summarize our ideas. Instead of > > > implementing > > > > >> the > > > > >> >> >>> cache > > > > >> >> >>>>>>>>>>>>>>> logic in > > > > >> >> >>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>> table > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> runtime layer or wrapping around the > > > > >> user-provided > > > > >> >> >>>> table > > > > >> >> >>>>>>>>>>>>>>>>> function, > > > > >> >> >>>>>>>>>>>>>>>>>>>> we > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> prefer to introduce some new APIs > > extending > > > > >> >> >>>> TableFunction > > > > >> >> >>>>>>>>>>>>>>> with > > > > >> >> >>>>>>>>>>>>>>>>> these > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> concerns: > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> 1. Caching actually breaks the > semantic > > of > > > > >> "FOR > > > > >> >> >>>>>>>>>>>>>>> SYSTEM_TIME > > > > >> >> >>>>>>>>>>>>>>>>> AS OF > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> proc_time”, because it couldn’t truly > > > reflect > > > > >> the > > > > >> >> >>>> content > > > > >> >> >>>>>>>>>>>>>>> of the > > > > >> >> >>>>>>>>>>>>>>>>>>>> lookup > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> table at the moment of querying. If > users > > > > >> choose > > > > >> >> to > > > > >> >> >>>> enable > > > > >> >> >>>>>>>>>>>>>>>>> caching > > > > >> >> >>>>>>>>>>>>>>>>>>>> on > > > > >> >> >>>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> lookup table, they implicitly indicate > > that > > > > >> this > > > > >> >> >>>> breakage is > > > > >> >> >>>>>>>>>>>>>>>>>>>> acceptable > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> in > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> exchange for the performance. So we > > prefer > > > > not > > > > >> to > > > > >> >> >>>> provide > > > > >> >> >>>>>>>>>>>>>>>>> caching on > > > > >> >> >>>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> table runtime level. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> 2. If we make the cache implementation > > in > > > > the > > > > >> >> >>>> framework > > > > >> >> >>>>>>>>>>>>>>>>> (whether > > > > >> >> >>>>>>>>>>>>>>>>>>>> in a > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> runner or a wrapper around > > TableFunction), > > > we > > > > >> have > > > > >> >> >> to > > > > >> >> >>>>>>>>>>>>>>> confront a > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> situation > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> that allows table options in DDL to > > control > > > > the > > > > >> >> >>>> behavior of > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> framework, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> which has never happened previously and > > > > should > > > > >> be > > > > >> >> >>>> cautious. > > > > >> >> >>>>>>>>>>>>>>>>> Under > > > > >> >> >>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> current design the behavior of the > > > framework > > > > >> >> should > > > > >> >> >>>> only be > > > > >> >> >>>>>>>>>>>>>>>>>>>> specified > > > > >> >> >>>>>>>>>>>>>>>>>>>>> by > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> configurations (“table.exec.xxx”), and > > it’s > > > > >> hard > > > > >> >> to > > > > >> >> >>>> apply > > > > >> >> >>>>>>>>>>>>>>> these > > > > >> >> >>>>>>>>>>>>>>>>>>>> general > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> configs to a specific table. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> 3. We have use cases that lookup > source > > > > loads > > > > >> and > > > > >> >> >>>> refresh > > > > >> >> >>>>>>>>>>>>>>> all > > > > >> >> >>>>>>>>>>>>>>>>>>>> records > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> periodically into the memory to achieve > > > high > > > > >> >> lookup > > > > >> >> >>>>>>>>>>>>>>> performance > > > > >> >> >>>>>>>>>>>>>>>>>>>> (like > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> Hive > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> connector in the community, and also > > widely > > > > >> used > > > > >> >> by > > > > >> >> >>> our > > > > >> >> >>>>>>>>>>>>>>> internal > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> connectors). Wrapping the cache around > > the > > > > >> user’s > > > > >> >> >>>>>>>>>>>>>>> TableFunction > > > > >> >> >>>>>>>>>>>>>>>>>>>> works > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> fine > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> for LRU caches, but I think we have to > > > > >> introduce a > > > > >> >> >>> new > > > > >> >> >>>>>>>>>>>>>>>>> interface for > > > > >> >> >>>>>>>>>>>>>>>>>>>>> this > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> all-caching scenario and the design > would > > > > >> become > > > > >> >> >> more > > > > >> >> >>>>>>>>>>>>>>> complex. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> 4. Providing the cache in the > framework > > > > might > > > > >> >> >>>> introduce > > > > >> >> >>>>>>>>>>>>>>>>>>>> compatibility > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> issues to existing lookup sources like > > > there > > > > >> might > > > > >> >> >>>> exist two > > > > >> >> >>>>>>>>>>>>>>>>> caches > > > > >> >> >>>>>>>>>>>>>>>>>>>>> with > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> totally different strategies if the > user > > > > >> >> >> incorrectly > > > > >> >> >>>>>>>>>>>>>>> configures > > > > >> >> >>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>> table > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> (one in the framework and another > > > implemented > > > > >> by > > > > >> >> >> the > > > > >> >> >>>> lookup > > > > >> >> >>>>>>>>>>>>>>>>> source). > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> As for the optimization mentioned by > > > > >> Alexander, I > > > > >> >> >>>> think > > > > >> >> >>>>>>>>>>>>>>>>> filters > > > > >> >> >>>>>>>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> projections should be pushed all the > way > > > down > > > > >> to > > > > >> >> >> the > > > > >> >> >>>> table > > > > >> >> >>>>>>>>>>>>>>>>> function, > > > > >> >> >>>>>>>>>>>>>>>>>>>>> like > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> what we do in the scan source, instead > of > > > the > > > > >> >> >> runner > > > > >> >> >>>> with > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>> cache. > > > > >> >> >>>>>>>>>>>>>>>>>>>>> The > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> goal of using cache is to reduce the > > > network > > > > >> I/O > > > > >> >> >> and > > > > >> >> >>>>>>>>>>>>>>> pressure > > > > >> >> >>>>>>>>>>>>>>>>> on the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> external system, and only applying > these > > > > >> >> >>> optimizations > > > > >> >> >>>> to > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>> cache > > > > >> >> >>>>>>>>>>>>>>>>>>>>> seems > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> not quite useful. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> I made some updates to the FLIP[1] to > > > > reflect > > > > >> our > > > > >> >> >>>> ideas. > > > > >> >> >>>>>>>>>>>>>>> We > > > > >> >> >>>>>>>>>>>>>>>>>>>> prefer to > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> keep the cache implementation as a part > > of > > > > >> >> >>>> TableFunction, > > > > >> >> >>>>>>>>>>>>>>> and we > > > > >> >> >>>>>>>>>>>>>>>>>>>> could > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> provide some helper classes > > > > >> (CachingTableFunction, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> AllCachingTableFunction, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> CachingAsyncTableFunction) to > developers > > > and > > > > >> >> >> regulate > > > > >> >> >>>>>>>>>>>>>>> metrics > > > > >> >> >>>>>>>>>>>>>>>>> of the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> cache. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> Also, I made a POC[2] for your > reference. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Looking forward to your ideas! > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> [1] > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> > > > > >> >> >>>> > > > > >> >> >>> > > > > >> >> >> > > > > >> >> > > > > >> > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-221+Abstraction+for+lookup+source+cache+and+metric > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> [2] > > > > >> >> >>> https://github.com/PatrickRen/flink/tree/FLIP-221 > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Qingsheng > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> On Tue, Apr 26, 2022 at 4:45 PM > > Александр > > > > >> Смирнов > > > > >> >> >> < > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> smirale...@gmail.com> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> Thanks for the response, Arvid! > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> I have few comments on your message. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> but could also live with an easier > > > > solution > > > > >> as > > > > >> >> >> the > > > > >> >> >>>>>>>>>>>>>>> first > > > > >> >> >>>>>>>>>>>>>>>>> step: > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> I think that these 2 ways are > mutually > > > > >> exclusive > > > > >> >> >>>>>>>>>>>>>>> (originally > > > > >> >> >>>>>>>>>>>>>>>>>>>>> proposed > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> by Qingsheng and mine), because > > > > conceptually > > > > >> >> they > > > > >> >> >>>> follow > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>> same > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> goal, but implementation details are > > > > >> different. > > > > >> >> >> If > > > > >> >> >>> we > > > > >> >> >>>>>>>>>>>>>>> will > > > > >> >> >>>>>>>>>>>>>>>>> go one > > > > >> >> >>>>>>>>>>>>>>>>>>>>> way, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> moving to another way in the future > > will > > > > mean > > > > >> >> >>>> deleting > > > > >> >> >>>>>>>>>>>>>>>>> existing > > > > >> >> >>>>>>>>>>>>>>>>>>>> code > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> and once again changing the API for > > > > >> connectors. > > > > >> >> >> So > > > > >> >> >>> I > > > > >> >> >>>>>>>>>>>>>>> think we > > > > >> >> >>>>>>>>>>>>>>>>>>>> should > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> reach a consensus with the community > > > about > > > > >> that > > > > >> >> >> and > > > > >> >> >>>> then > > > > >> >> >>>>>>>>>>>>>>> work > > > > >> >> >>>>>>>>>>>>>>>>>>>>> together > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> on this FLIP, i.e. divide the work on > > > tasks > > > > >> for > > > > >> >> >>>> different > > > > >> >> >>>>>>>>>>>>>>>>> parts > > > > >> >> >>>>>>>>>>>>>>>>>>>> of > > > > >> >> >>>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> flip (for example, LRU cache > > unification > > > / > > > > >> >> >>>> introducing > > > > >> >> >>>>>>>>>>>>>>>>> proposed > > > > >> >> >>>>>>>>>>>>>>>>>>>> set > > > > >> >> >>>>>>>>>>>>>>>>>>>>> of > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> metrics / further work…). WDYT, > > > Qingsheng? > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> as the source will only receive the > > > > requests > > > > >> >> >> after > > > > >> >> >>>>>>>>>>>>>>> filter > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> Actually if filters are applied to > > fields > > > > of > > > > >> the > > > > >> >> >>>> lookup > > > > >> >> >>>>>>>>>>>>>>>>> table, we > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> firstly must do requests, and only > > after > > > > >> that we > > > > >> >> >>> can > > > > >> >> >>>>>>>>>>>>>>> filter > > > > >> >> >>>>>>>>>>>>>>>>>>>>> responses, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> because lookup connectors don't have > > > filter > > > > >> >> >>>> pushdown. So > > > > >> >> >>>>>>>>>>>>>>> if > > > > >> >> >>>>>>>>>>>>>>>>>>>>> filtering > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> is done before caching, there will be > > > much > > > > >> less > > > > >> >> >>> rows > > > > >> >> >>>> in > > > > >> >> >>>>>>>>>>>>>>>>> cache. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> @Alexander unfortunately, your > > > > architecture > > > > >> is > > > > >> >> >> not > > > > >> >> >>>>>>>>>>>>>>> shared. > > > > >> >> >>>>>>>>>>>>>>>>> I > > > > >> >> >>>>>>>>>>>>>>>>>>>> don't > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> know the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> solution to share images to be > honest. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> Sorry for that, I’m a bit new to such > > > kinds > > > > >> of > > > > >> >> >>>>>>>>>>>>>>> conversations > > > > >> >> >>>>>>>>>>>>>>>>> :) > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> I have no write access to the > > confluence, > > > > so > > > > >> I > > > > >> >> >>> made a > > > > >> >> >>>>>>>>>>>>>>> Jira > > > > >> >> >>>>>>>>>>>>>>>>> issue, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> where described the proposed changes > in > > > > more > > > > >> >> >>> details > > > > >> >> >>>> - > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >> https://issues.apache.org/jira/browse/FLINK-27411. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> Will happy to get more feedback! > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> Best, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> Smirnov Alexander > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>> пн, 25 апр. 2022 г. в 19:49, Arvid > > Heise > > > < > > > > >> >> >>>>>>>>>>>>>>> ar...@apache.org>: > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> Hi Qingsheng, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> Thanks for driving this; the > > > inconsistency > > > > >> was > > > > >> >> >> not > > > > >> >> >>>>>>>>>>>>>>>>> satisfying > > > > >> >> >>>>>>>>>>>>>>>>>>>> for > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> me. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> I second Alexander's idea though but > > > could > > > > >> also > > > > >> >> >>> live > > > > >> >> >>>>>>>>>>>>>>> with > > > > >> >> >>>>>>>>>>>>>>>>> an > > > > >> >> >>>>>>>>>>>>>>>>>>>>> easier > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> solution as the first step: Instead > of > > > > >> making > > > > >> >> >>>> caching > > > > >> >> >>>>>>>>>>>>>>> an > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> implementation > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> detail of TableFunction X, rather > > > devise a > > > > >> >> >> caching > > > > >> >> >>>>>>>>>>>>>>> layer > > > > >> >> >>>>>>>>>>>>>>>>>>>> around X. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> So > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> proposal would be a > > CachingTableFunction > > > > >> that > > > > >> >> >>>>>>>>>>>>>>> delegates to > > > > >> >> >>>>>>>>>>>>>>>>> X in > > > > >> >> >>>>>>>>>>>>>>>>>>>>> case > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> of > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> misses and else manages the cache. > > > Lifting > > > > >> it > > > > >> >> >> into > > > > >> >> >>>> the > > > > >> >> >>>>>>>>>>>>>>>>> operator > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> model > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> as > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> proposed would be even better but is > > > > >> probably > > > > >> >> >>>>>>>>>>>>>>> unnecessary > > > > >> >> >>>>>>>>>>>>>>>>> in > > > > >> >> >>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> first step > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> for a lookup source (as the source > > will > > > > only > > > > >> >> >>> receive > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>> requests > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> after > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> filter; applying projection may be > > more > > > > >> >> >>> interesting > > > > >> >> >>>> to > > > > >> >> >>>>>>>>>>>>>>> save > > > > >> >> >>>>>>>>>>>>>>>>>>>>> memory). > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> Another advantage is that all the > > > changes > > > > of > > > > >> >> >> this > > > > >> >> >>>> FLIP > > > > >> >> >>>>>>>>>>>>>>>>> would be > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> limited to > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> options, no need for new public > > > > interfaces. > > > > >> >> >>>> Everything > > > > >> >> >>>>>>>>>>>>>>> else > > > > >> >> >>>>>>>>>>>>>>>>>>>>> remains > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> an > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> implementation of Table runtime. > That > > > > means > > > > >> we > > > > >> >> >> can > > > > >> >> >>>>>>>>>>>>>>> easily > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> incorporate > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> optimization potential that > Alexander > > > > >> pointed > > > > >> >> >> out > > > > >> >> >>>>>>>>>>>>>>> later. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> @Alexander unfortunately, your > > > > architecture > > > > >> is > > > > >> >> >> not > > > > >> >> >>>>>>>>>>>>>>> shared. > > > > >> >> >>>>>>>>>>>>>>>>> I > > > > >> >> >>>>>>>>>>>>>>>>>>>> don't > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> know the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> solution to share images to be > honest. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> On Fri, Apr 22, 2022 at 5:04 PM > > > Александр > > > > >> >> >> Смирнов > > > > >> >> >>> < > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> smirale...@gmail.com> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> wrote: > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> Hi Qingsheng! My name is Alexander, > > I'm > > > > >> not a > > > > >> >> >>>>>>>>>>>>>>> committer > > > > >> >> >>>>>>>>>>>>>>>>> yet, > > > > >> >> >>>>>>>>>>>>>>>>>>>> but > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> I'd > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> really like to become one. And this > > > FLIP > > > > >> >> really > > > > >> >> >>>>>>>>>>>>>>>>> interested > > > > >> >> >>>>>>>>>>>>>>>>>>>> me. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> Actually I have worked on a similar > > > > >> feature in > > > > >> >> >> my > > > > >> >> >>>>>>>>>>>>>>>>> company’s > > > > >> >> >>>>>>>>>>>>>>>>>>>>> Flink > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> fork, and we would like to share > our > > > > >> thoughts > > > > >> >> >> on > > > > >> >> >>>>>>>>>>>>>>> this and > > > > >> >> >>>>>>>>>>>>>>>>>>>> make > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> code > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> open source. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> I think there is a better > alternative > > > > than > > > > >> >> >>>>>>>>>>>>>>> introducing an > > > > >> >> >>>>>>>>>>>>>>>>>>>>> abstract > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> class for TableFunction > > > > >> >> (CachingTableFunction). > > > > >> >> >>> As > > > > >> >> >>>>>>>>>>>>>>> you > > > > >> >> >>>>>>>>>>>>>>>>> know, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> TableFunction exists in the > > > > >> flink-table-common > > > > >> >> >>>>>>>>>>>>>>> module, > > > > >> >> >>>>>>>>>>>>>>>>> which > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> provides > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> only an API for working with > tables – > > > > it’s > > > > >> >> very > > > > >> >> >>>>>>>>>>>>>>>>> convenient > > > > >> >> >>>>>>>>>>>>>>>>>>>> for > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> importing > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> in connectors. In turn, > > > > >> CachingTableFunction > > > > >> >> >>>> contains > > > > >> >> >>>>>>>>>>>>>>>>> logic > > > > >> >> >>>>>>>>>>>>>>>>>>>> for > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> runtime execution, so this class > and > > > > >> >> >> everything > > > > >> >> >>>>>>>>>>>>>>>>> connected > > > > >> >> >>>>>>>>>>>>>>>>>>>> with > > > > >> >> >>>>>>>>>>>>>>>>>>>>> it > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> should be located in another > module, > > > > >> probably > > > > >> >> >> in > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> flink-table-runtime. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> But this will require connectors to > > > > depend > > > > >> on > > > > >> >> >>>> another > > > > >> >> >>>>>>>>>>>>>>>>> module, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> which > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> contains a lot of runtime logic, > > which > > > > >> doesn’t > > > > >> >> >>>> sound > > > > >> >> >>>>>>>>>>>>>>>>> good. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> I suggest adding a new method > > > > >> >> ‘getLookupConfig’ > > > > >> >> >>> to > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> LookupTableSource > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> or LookupRuntimeProvider to allow > > > > >> connectors > > > > >> >> to > > > > >> >> >>>> only > > > > >> >> >>>>>>>>>>>>>>> pass > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> configurations to the planner, > > > therefore > > > > >> they > > > > >> >> >>> won’t > > > > >> >> >>>>>>>>>>>>>>>>> depend on > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> runtime > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> realization. Based on these configs > > > > planner > > > > >> >> >> will > > > > >> >> >>>>>>>>>>>>>>>>> construct a > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> lookup > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> join operator with corresponding > > > runtime > > > > >> logic > > > > >> >> >>>>>>>>>>>>>>>>>>>> (ProcessFunctions > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> in > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> module flink-table-runtime). > > > Architecture > > > > >> >> looks > > > > >> >> >>>> like > > > > >> >> >>>>>>>>>>>>>>> in > > > > >> >> >>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>> pinned > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> image (LookupConfig class there is > > > > actually > > > > >> >> >> yours > > > > >> >> >>>>>>>>>>>>>>>>>>>> CacheConfig). > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> Classes in flink-table-planner, > that > > > will > > > > >> be > > > > >> >> >>>>>>>>>>>>>>> responsible > > > > >> >> >>>>>>>>>>>>>>>>> for > > > > >> >> >>>>>>>>>>>>>>>>>>>>> this > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> – > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> CommonPhysicalLookupJoin and his > > > > >> inheritors. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> Current classes for lookup join in > > > > >> >> >>>>>>>>>>>>>>> flink-table-runtime > > > > >> >> >>>>>>>>>>>>>>>>> - > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> LookupJoinRunner, > > > AsyncLookupJoinRunner, > > > > >> >> >>>>>>>>>>>>>>>>>>>>> LookupJoinRunnerWithCalc, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> AsyncLookupJoinRunnerWithCalc. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> I suggest adding classes > > > > >> >> >> LookupJoinCachingRunner, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> LookupJoinCachingRunnerWithCalc, > etc. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> And here comes another more > powerful > > > > >> advantage > > > > >> >> >> of > > > > >> >> >>>>>>>>>>>>>>> such a > > > > >> >> >>>>>>>>>>>>>>>>>>>>> solution. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> If > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> we have caching logic on a lower > > level, > > > > we > > > > >> can > > > > >> >> >>>> apply > > > > >> >> >>>>>>>>>>>>>>> some > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> optimizations to it. > > > > >> LookupJoinRunnerWithCalc > > > > >> >> >> was > > > > >> >> >>>>>>>>>>>>>>> named > > > > >> >> >>>>>>>>>>>>>>>>> like > > > > >> >> >>>>>>>>>>>>>>>>>>>>> this > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> because it uses the ‘calc’ > function, > > > > which > > > > >> >> >>> actually > > > > >> >> >>>>>>>>>>>>>>>>> mostly > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> consists > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> of > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> filters and projections. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> For example, in join table A with > > > lookup > > > > >> table > > > > >> >> >> B > > > > >> >> >>>>>>>>>>>>>>>>> condition > > > > >> >> >>>>>>>>>>>>>>>>>>>>> ‘JOIN … > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> ON > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> A.id = B.id AND A.age = B.age + 10 > > > WHERE > > > > >> >> >>> B.salary > > > > > >> >> >>>>>>>>>>>>>>> 1000’ > > > > >> >> >>>>>>>>>>>>>>>>>>>>> ‘calc’ > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> function will contain filters > A.age = > > > > >> B.age + > > > > >> >> >> 10 > > > > >> >> >>>> and > > > > >> >> >>>>>>>>>>>>>>>>>>>> B.salary > > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> 1000. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> If we apply this function before > > > storing > > > > >> >> >> records > > > > >> >> >>> in > > > > >> >> >>>>>>>>>>>>>>>>> cache, > > > > >> >> >>>>>>>>>>>>>>>>>>>> size > > > > >> >> >>>>>>>>>>>>>>>>>>>>> of > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> cache will be significantly > reduced: > > > > >> filters = > > > > >> >> >>>> avoid > > > > >> >> >>>>>>>>>>>>>>>>> storing > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> useless > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> records in cache, projections = > > reduce > > > > >> >> records’ > > > > >> >> >>>>>>>>>>>>>>> size. So > > > > >> >> >>>>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> initial > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> max number of records in cache can > be > > > > >> >> increased > > > > >> >> >>> by > > > > >> >> >>>>>>>>>>>>>>> the > > > > >> >> >>>>>>>>>>>>>>>>> user. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> What do you think about it? > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> On 2022/04/19 02:47:11 Qingsheng > Ren > > > > wrote: > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Hi devs, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Yuan and I would like to start a > > > > >> discussion > > > > >> >> >>> about > > > > >> >> >>>>>>>>>>>>>>>>>>>> FLIP-221[1], > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> which > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> introduces an abstraction of lookup > > > table > > > > >> >> cache > > > > >> >> >>> and > > > > >> >> >>>>>>>>>>>>>>> its > > > > >> >> >>>>>>>>>>>>>>>>>>>> standard > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> metrics. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Currently each lookup table source > > > > should > > > > >> >> >>>> implement > > > > >> >> >>>>>>>>>>>>>>>>> their > > > > >> >> >>>>>>>>>>>>>>>>>>>> own > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> cache to > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> store lookup results, and there > > isn’t a > > > > >> >> >> standard > > > > >> >> >>> of > > > > >> >> >>>>>>>>>>>>>>>>> metrics > > > > >> >> >>>>>>>>>>>>>>>>>>>> for > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> users and > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> developers to tuning their jobs > with > > > > lookup > > > > >> >> >>> joins, > > > > >> >> >>>>>>>>>>>>>>> which > > > > >> >> >>>>>>>>>>>>>>>>> is a > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> quite > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> common > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> use case in Flink table / SQL. > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Therefore we propose some new APIs > > > > >> including > > > > >> >> >>>> cache, > > > > >> >> >>>>>>>>>>>>>>>>>>>> metrics, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> wrapper > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> classes of TableFunction and new > > table > > > > >> >> options. > > > > >> >> >>>>>>>>>>>>>>> Please > > > > >> >> >>>>>>>>>>>>>>>>> take a > > > > >> >> >>>>>>>>>>>>>>>>>>>>> look > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> at the > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> FLIP page [1] to get more details. > > Any > > > > >> >> >>> suggestions > > > > >> >> >>>>>>>>>>>>>>> and > > > > >> >> >>>>>>>>>>>>>>>>>>>> comments > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> would be > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> appreciated! > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> [1] > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> > > > > >> >> >>>> > > > > >> >> >>> > > > > >> >> >> > > > > >> >> > > > > >> > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-221+Abstraction+for+lookup+source+cache+and+metric > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Qingsheng > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> -- > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Best Regards, > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Qingsheng Ren > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Real-time Computing Team > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Alibaba Cloud > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>>> Email: renqs...@gmail.com > > > > >> >> >>>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>>>> -- > > > > >> >> >>>>>>>>>>>>>>>>>>> Best regards, > > > > >> >> >>>>>>>>>>>>>>>>>>> Roman Boyko > > > > >> >> >>>>>>>>>>>>>>>>>>> e.: ro.v.bo...@gmail.com > > > > >> >> >>>>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> -- > > > > >> >> >>>>>>>>>>>> Best Regards, > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Qingsheng Ren > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Real-time Computing Team > > > > >> >> >>>>>>>>>>>> Alibaba Cloud > > > > >> >> >>>>>>>>>>>> > > > > >> >> >>>>>>>>>>>> Email: renqs...@gmail.com > > > > >> >> >>>>>>>>>> > > > > >> >> >>>> > > > > >> >> >>>> > > > > >> >> >>> > > > > >> >> >> > > > > >> >> > > > > >> >> > > > > >> > > > > > > > > > > > > > > >
