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
