Hi Ron, Thanks for sharing your thoughts! It makes sense. It would be helpful if these references of Hive, Polardb, etc. could be added into the FLIP.
Best regards, Jing On Tue, Jun 20, 2023 at 5:41 PM liu ron <ron9....@gmail.com> wrote: > Hi, Jing > > The default value for this ratio is a reference to other systems, such as > Hive. As long as Runtime Filter can filter out more than half of the data, > we can benefit from it. Of course, normally, as long as we can get the > statistics, ndv are present, the use of rowCount should be less, so I think > the formula is valid in most cases. This formula we are also borrowed from > some systems, such as the polardb of AliCloud. your concern is valuable for > this FLIP, but currently, we do not know how to adjust is reasonably, too > complex may lead to the user also can not understand, so I think we should > first follow the simple way, the subsequent gradual optimization. The first > step may be that we can verify the reasonableness of current formula by > TPC-DS case. > > Best, > Ron > > Jing Ge <j...@ververica.com.invalid> 于2023年6月20日周二 19:46写道： > > > Hi Ron, > > > > Thanks for the clarification. That answered my questions. > > > > Regarding the ratio, since my gut feeling is that any value less than 0.8 > > or 0.9 won't help too much(I might be wrong). I was thinking of adapting > > the formula to somehow map the current 0.9-1 to 0-1, i.e. if user config > > 0.5, it will be mapped to e.g. 0.95 (or e.g. 0.85, the real number > > needs more calculation) for the current formula described in the FLIP. > But > > I am not sure it is a feasible solution. It deserves more discussion. > Maybe > > some real performance tests could give us some hints. > > > > Best regards, > > Jing > > > > On Tue, Jun 20, 2023 at 5:19 AM liu ron <ron9....@gmail.com> wrote: > > > > > Hi, Jing > > > > > > Thanks for your feedback. > > > > > > > Afaiu, the runtime Filter will only be Injected when the gap between > > the > > > build data size and prob data size is big enough. Let's make an extreme > > > example. If the small table(build side) has one row and the large > > > table(probe side) contains tens of billions of rows. This will be the > > ideal > > > use case for the runtime filter and the improvement will be > significant. > > Is > > > this correct? > > > > > > Yes, you are right. > > > > > > > Speaking of the "Conditions of injecting Runtime Filter" in the FLIP, > > > will > > > the value of max-build-data-size and min-prob-data-size depend on the > > > parallelism config? I.e. with the same data-size setting, is it > possible > > to > > > inject or don't inject runtime filters by adjusting the parallelism? > > > > > > First, let me clarify two points. The first is that RuntimeFilter > decides > > > whether to inject or not in the optimization phase, but we do not > > consider > > > operator parallelism in the SQL optimization phase currently, which is > > set > > > at the ExecNode level. The second is that in batch mode, the default > > > AdaptiveBatchScheduler[1] is now used, which will derive the > parallelism > > of > > > the downstream operator based on the amount of data produced by the > > > upstream operator, that is, the parallelism is determined by runtime > > > adaptation. In the above case, we cannot decide whether to inject > > > BloomFilter in the optimization stage based on parallelism. > > > A more important point is that the purpose of Runtime Filter is to > reduce > > > the amount of data for shuffle, and thus the amount of data processed > by > > > the downstream join operator. Therefore, I understand that regardless > of > > > the parallelism of the probe, the amount of data in the shuffle must be > > > reduced after inserting the Runtime Filter, which is beneficial to the > > join > > > operator, so whether to insert the RuntimeFilter or not is not > dependent > > on > > > the parallelism. > > > > > > > Does it make sense to reconsider the formula of ratio > > > calculation to help users easily control the filter injection? > > > > > > Only when ndv does not exist will row count be considered. when size > uses > > > the default value and ndv cannot be taken, it is true that this > condition > > > may always hold, but this does not seem to affect anything, and the > user > > is > > > also likely to change the value of the size. One question, how do you > > think > > > we should make it easier for users to control the filter injection？ > > > > > > > > > [1]: > > > > > > > > > https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/elastic_scaling/#adaptive-batch-scheduler > > > > > > Best, > > > Ron > > > > > > Jing Ge <j...@ververica.com.invalid> 于2023年6月20日周二 07:11写道： > > > > > > > Hi Lijie, > > > > > > > > Thanks for your proposal. It is a really nice feature. I'd like to > ask > > a > > > > few questions to understand your thoughts. > > > > > > > > Afaiu, the runtime Filter will only be Injected when the gap between > > the > > > > build data size and prob data size is big enough. Let's make an > extreme > > > > example. If the small table(build side) has one row and the large > > > > table(probe side) contains tens of billions of rows. This will be the > > > ideal > > > > use case for the runtime filter and the improvement will be > > significant. > > > Is > > > > this correct? > > > > > > > > Speaking of the "Conditions of injecting Runtime Filter" in the FLIP, > > > will > > > > the value of max-build-data-size and min-prob-data-size depend on the > > > > parallelism config? I.e. with the same data-size setting, is it > > possible > > > to > > > > inject or don't inject runtime filters by adjusting the parallelism? > > > > > > > > In the FLIP, there are default values for the new configuration > > > parameters > > > > that will be used to check the injection condition. If ndv cannot be > > > > estimated, row count will be used. Given the max-build-data-size is > > 10MB > > > > and the min-prob-data-size is 10GB, in the worst case, the > > > min-filter-ratio > > > > will be 0.999, i.e. the probeNdv is 1000 times buildNdv . If we > > consider > > > > the duplication and the fact that the large table might have more > > columns > > > > than the small table, the probeNdv should still be 100 or 10 times > > > > buildNdv, which ends up with a min-filter-ratio equals to 0.99 or > 0.9. > > > Both > > > > are bigger than the default value 0.5 in the FLIP. If I am not > > mistaken, > > > > commonly, a min-filter-ratio less than 0.99 will always allow > injecting > > > the > > > > runtime filter. Does it make sense to reconsider the formula of ratio > > > > calculation to help users easily control the filter injection? > > > > > > > > Best regards, > > > > Jing > > > > > > > > On Mon, Jun 19, 2023 at 4:42 PM Lijie Wang <wangdachui9...@gmail.com > > > > > > wrote: > > > > > > > > > Hi Stefan, > > > > > > > > > > >> bypassing the dataflow > > > > > I believe it's a possible solution, but it may require more > > > coordination > > > > > and extra conditions (such as DFS), I do think it should be > excluded > > > from > > > > > the first version. I'll put it in Future+Improvements as a > potential > > > > > improvement. > > > > > > > > > > Thanks again for your quick reply :) > > > > > > > > > > Best, > > > > > Lijie > > > > > > > > > > Stefan Richter <srich...@confluent.io.invalid> 于2023年6月19日周一 > > 20:51写道： > > > > > > > > > > > > > > > > > Hi Lijie, > > > > > > > > > > > > I think you understood me correctly. But I would not consider > this > > a > > > > true > > > > > > cyclic dependency in the dataflow because I would not suggest to > > send > > > > the > > > > > > filter through an edge in the job graph from join to scan. I’d > > rather > > > > > > bypass the stream graph for exchanging bringing the filter to the > > > scan. > > > > > For > > > > > > example, the join could report the filter after the build phase, > > e.g. > > > > to > > > > > > the JM or a predefined DFS folder. And when the probe scan is > > > > scheduled, > > > > > > the JM provides the filter information to the scan when it gets > > > > scheduled > > > > > > for execution or the scan looks in DFS if it can find any filter > > that > > > > it > > > > > > can use as part of initialization. I’m not suggesting to do it > > > exactly > > > > in > > > > > > those ways, but just to show what I mean by "bypassing the > > dataflow". > > > > > > > > > > > > Anyways, I’m fine with excluding this optimization from the > current > > > > FLIP > > > > > > if you believe it would be hard to implement in Flink. > > > > > > > > > > > > Best, > > > > > > Stefan > > > > > > > > > > > > > > > > > > > On 19. Jun 2023, at 14:07, Lijie Wang < > wangdachui9...@gmail.com> > > > > > wrote: > > > > > > > > > > > > > > Hi Stefan, > > > > > > > > > > > > > > If I understand correctly(I hope so), the hash join operator > > needs > > > to > > > > > > send > > > > > > > the bloom filter to probe scan, and probe scan also needs to > send > > > the > > > > > > > filtered data to the hash join operator. This means there will > > be a > > > > > cycle > > > > > > > in the data flow, it will be hard for current Flink to schedule > > > this > > > > > kind > > > > > > > of graph. I admit we can find a way to do this, but that's > > > probably a > > > > > > > bit outside the scope of this FLIP. So let's do these complex > > > > > > > optimizations later, WDYT？ > > > > > > > > > > > > > > Best, > > > > > > > Lijie > > > > > > > > > > > > > > Stefan Richter <srich...@confluent.io.invalid <mailto: > > > > > > srich...@confluent.io.invalid>> 于2023年6月19日周一 18:15写道： > > > > > > > > > > > > > >> Hi Lijie, > > > > > > >> > > > > > > >> Exactly, my proposal was to build the bloom filter in the hash > > > > > > operator. I > > > > > > >> don’t know about all the details about the implementation of > > > Flink’s > > > > > > join > > > > > > >> operator, but I’d assume that even if the join is a two input > > > > operator > > > > > > it > > > > > > >> gets scheduled for 2 different pipelines. First the build > phase > > > with > > > > > the > > > > > > >> scan from the dimension table and after that’s completed the > > probe > > > > > phase > > > > > > >> with the scan of the fact table. I’m not proposing the use the > > > bloom > > > > > > filter > > > > > > >> only in the join operator, but rather send the bloom filter to > > the > > > > > probe > > > > > > >> scan before starting the probe. I assume this would require > some > > > > form > > > > > of > > > > > > >> side channel to transport the filter and coordination to tell > > the > > > > > > sources > > > > > > >> that such a filter is available. I cannot answer how hard > those > > > > would > > > > > > be to > > > > > > >> implement, but the idea doesn’t seem impossible to me. > > > > > > >> > > > > > > >> Best, > > > > > > >> Stefan > > > > > > >> > > > > > > >> > > > > > > >>> On 19. Jun 2023, at 11:56, Lijie Wang < > > wangdachui9...@gmail.com> > > > > > > wrote: > > > > > > >>> > > > > > > >>> Hi Stefan, > > > > > > >>> > > > > > > >>> Now I know what you mean about point 1. But currently it is > > > > > unfeasible > > > > > > >> for > > > > > > >>> Flink, because the building of the hash table is inside the > > hash > > > > join > > > > > > >>> operator. The hash join operator has two inputs, it will > first > > > > > process > > > > > > >> the > > > > > > >>> data of the build-input to build a hash table, and then use > the > > > > hash > > > > > > >> table > > > > > > >>> to process the data of the probe-input. If we want to use the > > > built > > > > > > hash > > > > > > >>> table to deduplicate data for bloom filter, we must put the > > bloom > > > > > > filter > > > > > > >>> inside the hash join operator. However, in this way, the > data > > > > > reaching > > > > > > >> the > > > > > > >>> join operator cannot be reduced (the shuffle/network overhead > > > > cannot > > > > > be > > > > > > >>> reduced), which is not what we expected. > > > > > > >>> > > > > > > >>> Regarding the filter type, I agree with you, more types of > > > filters > > > > > can > > > > > > >>> get further > > > > > > >>> optimization, and it is in our future plan (We described it > in > > > the > > > > > > >> section > > > > > > >>> Future+Improvements#More+underlying+implementations). > > > > > > >>> > > > > > > >>> Best, > > > > > > >>> Lijie > > > > > > >>> > > > > > > >>> Stefan Richter <srich...@confluent.io.invalid <mailto: > > > > > > srich...@confluent.io.invalid> <mailto: > > > > > > >> srich...@confluent.io.invalid <mailto: > > > srich...@confluent.io.invalid > > > > > >>> > > > > > > 于2023年6月19日周一 15:58写道： > > > > > > >>> > > > > > > >>>> > > > > > > >>>> Hi Lijie, > > > > > > >>>> > > > > > > >>>> thanks for your response, I agree with what you said about > > > points > > > > 2 > > > > > > and > > > > > > >> 3. > > > > > > >>>> Let me explain a bit more about point 1. This would not > apply > > to > > > > all > > > > > > >> types > > > > > > >>>> of joins and my suggestion is also *not* to build a hash > table > > > > only > > > > > > for > > > > > > >> the > > > > > > >>>> purpose to build the bloom filter. > > > > > > >>>> I was thinking about the scenario of a hash join, where you > > > would > > > > > > build > > > > > > >>>> the hash table as part of the join algorithm anyways and > then > > > use > > > > > the > > > > > > >>>> keyset of that hash table to 1) have better insights on > about > > > NDV > > > > > and > > > > > > >> 2) be > > > > > > >>>> able to construct the bloom filter without duplicates and > > > > therefore > > > > > > >> faster. > > > > > > >>>> So the preconditions where I would use this is if you are > > > > building a > > > > > > >> hash > > > > > > >>>> table as part of the join and you know you are not building > > for > > > a > > > > > key > > > > > > >>>> column (because there would be no duplicates to eliminate). > > Then > > > > > your > > > > > > >> bloom > > > > > > >>>> filter construction could benefit already from the > > deduplication > > > > > work > > > > > > >> that > > > > > > >>>> was done for building the hash table. > > > > > > >>>> > > > > > > >>>> I also wanted to point out that besides bloom filter and IN > > > filter > > > > > you > > > > > > >>>> could also think of other types of filter that can become > > > > > interesting > > > > > > >> for > > > > > > >>>> certain distributions and meta data. For example, if you > have > > > > > min/max > > > > > > >>>> information about columns and partitions you could have a > bit > > > > vector > > > > > > >>>> represent equilibrium-sized ranges of the key space between > > min > > > > and > > > > > > max > > > > > > >> and > > > > > > >>>> have the bits represent what part of the range is present > and > > > push > > > > > > that > > > > > > >>>> information down to the scan. > > > > > > >>>> > > > > > > >>>> Best, > > > > > > >>>> Stefan > > > > > > >>>> > > > > > > >>>> > > > > > > >>>>> On 19. Jun 2023, at 08:26, Lijie Wang < > > > wangdachui9...@gmail.com > > > > > > <mailto:wangdachui9...@gmail.com>> > > > > > > >> wrote: > > > > > > >>>>> > > > > > > >>>>> Hi Stefan, > > > > > > >>>>> > > > > > > >>>>> Thanks for your feedback. Let me briefly summarize the > > > > optimization > > > > > > >>>> points > > > > > > >>>>> you mentioned above (Please correct me if I'm wrong): > > > > > > >>>>> > > > > > > >>>>> 1. Build an extra hash table for deduplication before > > building > > > > the > > > > > > >> bloom > > > > > > >>>>> filter. > > > > > > >>>>> 2. Use the two-phase approach to build the bloom > filter(first > > > > > local, > > > > > > >> then > > > > > > >>>>> OR-combine). > > > > > > >>>>> 3. Use blocked bloom filters to improve the cache > efficiency. > > > > > > >>>>> > > > > > > >>>>> For the above 3 points, I have the following questions or > > > > opinions: > > > > > > >>>>> > > > > > > >>>>> For point 1, it seems that building a hash table also > > requires > > > > > > >> traversing > > > > > > >>>>> all build side data, and the overhead seems to be the same > as > > > > > > building > > > > > > >> a > > > > > > >>>>> bloom filter directly? In addition, the hash table will > take > > up > > > > > more > > > > > > >>>> space > > > > > > >>>>> when the amount of data is large, which is why we choose to > > use > > > > > bloom > > > > > > >>>>> filter instead of hash table. > > > > > > >>>>> > > > > > > >>>>> For point 2, I think it's a good idea to use the two-phase > > > > approach > > > > > > to > > > > > > >>>>> build the bloom filter. But rather than directly > broadcasting > > > the > > > > > > local > > > > > > >>>>> bloom filter to the probe side, I prefer to introduce a > > global > > > > node > > > > > > for > > > > > > >>>> the > > > > > > >>>>> OR-combine(like two-phase-agg[1]), then broadcast the > > combined > > > > > bloom > > > > > > >>>> filter > > > > > > >>>>> to the probe side. The latter can reduce the amount of data > > > > > > transferred > > > > > > >>>> by > > > > > > >>>>> the network. I will change the FLIP like this. > > > > > > >>>>> > > > > > > >>>>> For point 3, I think it's a nice optimization, but I prefer > > to > > > > put > > > > > it > > > > > > >> to > > > > > > >>>>> the future improvements. There is already an implementation > > of > > > > > bloom > > > > > > >>>> filter > > > > > > >>>>> in flink, we can simply reuse it. Introducing a new bloom > > > filter > > > > > > >>>>> implementation introduces some complexity (we need to > > > implement > > > > > it, > > > > > > >> test > > > > > > >>>>> it, etc), and is not the focus of this FLIP. > > > > > > >>>>> > > > > > > >>>>> [1] > > > > > > >>>>> > > > > > > >>>> > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://www.google.com/url?q=https://www.google.com/url?q%3Dhttps://www.google.com/url?q%253Dhttps://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/tuning/%252523local-global-aggregation%2526source%253Dgmail-imap%2526ust%253D1687760804000000%2526usg%253DAOvVaw2eoXknGWmG4TSiznxtHFWG%26source%3Dgmail-imap%26ust%3D1687773407000000%26usg%3DAOvVaw3V4Sv1o119cpU4xfP0ifkj&source=gmail-imap&ust=1687781326000000&usg=AOvVaw033xxrkenJpx27XzCVKsda > > > > > > >>>>> > > > > > > >>>>> Best, > > > > > > >>>>> Lijie > > > > > > >>>>> > > > > > > >>>>> Stefan Richter <srich...@confluent.io.invalid <mailto: > > > > > > srich...@confluent.io.invalid> <mailto: > > > > > > >> srich...@confluent.io.invalid <mailto: > > > srich...@confluent.io.invalid > > > > >> > > > > > > <mailto: > > > > > > >>>> srich...@confluent.io.invalid <mailto: > > > > srich...@confluent.io.invalid > > > > > > > > > > > > <mailto:srich...@confluent.io.invalid>>> > > > > > > >> 于2023年6月16日周五 16:45写道： > > > > > > >>>>> > > > > > > >>>>>> Hi, > > > > > > >>>>>> > > > > > > >>>>>> Thanks for the proposal of this feature! I have a question > > > about > > > > > the > > > > > > >>>>>> filter build and a some suggestions for potential > > > improvements. > > > > > > >> First, I > > > > > > >>>>>> wonder why you suggest to run the filter builder as > separate > > > > > > operator > > > > > > >>>> with > > > > > > >>>>>> parallelism 1. I’d suggest to integrate the filter > > distributed > > > > > build > > > > > > >>>> with > > > > > > >>>>>> the hash table build phase as follows: > > > > > > >>>>>> > > > > > > >>>>>> 1. Build the hash table completely in each subtask. > > > > > > >>>>>> 2. The keyset of the hash table is giving us a precise NDV > > > count > > > > > for > > > > > > >>>> every > > > > > > >>>>>> subtask. > > > > > > >>>>>> 3. Build a filter from the subtask hash table. For low > > > > cardinality > > > > > > >>>> tables, > > > > > > >>>>>> I’d go with the suggested optimization of IN-filter. > > > > > > >>>>>> 4. Each build subtask transfers the local bloom filter to > > all > > > > > probe > > > > > > >>>>>> operators. > > > > > > >>>>>> 5. On the probe operator we can either probe against the > > > > > individual > > > > > > >>>>>> filters, or we OR-combine all subtask filters into > > aggregated > > > > > bloom > > > > > > >>>> filter. > > > > > > >>>>>> > > > > > > >>>>>> I’m suggesting this because building inserting into a > > (larger) > > > > > bloom > > > > > > >>>>>> filter can be costly, especially once the filter exceeds > > cache > > > > > sizes > > > > > > >>>> and is > > > > > > >>>>>> therefor better parallelized. First inserting into the > hash > > > > table > > > > > > also > > > > > > >>>>>> deduplicates the keys and we avoid inserting records twice > > > into > > > > > the > > > > > > >>>> bloom > > > > > > >>>>>> filter. If we want to improve cache efficiency for the > build > > > of > > > > > > larger > > > > > > >>>>>> filters, we could structure them as blocked bloom filters, > > > where > > > > > the > > > > > > >>>> filter > > > > > > >>>>>> is separated into blocks and all bits of one key go only > > into > > > > one > > > > > > >> block. > > > > > > >>>>>> That allows us to apply software managed buffering to > first > > > > group > > > > > > keys > > > > > > >>>> that > > > > > > >>>>>> go into the same partition (ideally fitting into cache) > and > > > then > > > > > > bulk > > > > > > >>>> load > > > > > > >>>>>> partitions once we collected enough keys for one round of > > > > loading. > > > > > > >>>>>> > > > > > > >>>>>> Best, > > > > > > >>>>>> Stefan > > > > > > >>>>>> > > > > > > >>>>>> > > > > > > >>>>>> < > > > > > > >>>> > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://www.google.com/url?q=https://www.google.com/url?q%3Dhttps://www.google.com/url?q%253Dhttps://www.confluent.io/%2526source%253Dgmail-imap%2526ust%253D1687760804000000%2526usg%253DAOvVaw3p0tBjuVsWz3SLYyPQukfL%26source%3Dgmail-imap%26ust%3D1687773407000000%26usg%3DAOvVaw1THgA9fFMrOd7QpGpwiRx6&source=gmail-imap&ust=1687781326000000&usg=AOvVaw1f-3D9-2lZDGsvFBjeFlvn > > > > > > >>>>> > > > > > > >>>>>> Stefan Richter > > > > > > >>>>>> Principal Engineer II > > > > > > >>>>>> > > > > > > >>>>>> Follow us: < > > > > > > >>>>>> > > > > > > >>>> > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://www.google.com/url?q=https://www.google.com/url?q%3Dhttps://www.google.com/url?q%253Dhttps://www.confluent.io/blog?utm_source%25253Dfooter%252526utm_medium%25253Demail%252526utm_campaign%25253Dch.email-signature_type.community_content.blog%2526source%253Dgmail-imap%2526ust%253D1687760804000000%2526usg%253DAOvVaw2VU_JTYB24Wp4bF2JshdU7%26source%3Dgmail-imap%26ust%3D1687773407000000%26usg%3DAOvVaw37ghBlQPqP0tTXCfNJCqKv&source=gmail-imap&ust=1687781326000000&usg=AOvVaw20v4QTnSyAz_HAHbMyVY7J > > > > > > >>>>> > > > > > > >>>>>> < > > > > > > >>>> > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://www.google.com/url?q=https://www.google.com/url?q%3Dhttps://www.google.com/url?q%253Dhttps://twitter.com/ConfluentInc%2526source%253Dgmail-imap%2526ust%253D1687760804000000%2526usg%253DAOvVaw2irnDxUAhXR0N8FUk2orze%26source%3Dgmail-imap%26ust%3D1687773407000000%26usg%3DAOvVaw0ItT553mEuA5KaeJWSH36D&source=gmail-imap&ust=1687781326000000&usg=AOvVaw1mNvHaIwjIKU_gqOuDYLDK > > > > > > >>>>> > > > > > > >>>>>> > > > > > > >>>>>> > > > > > > >>>>>> > > > > > > >>>>>>> On 15. Jun 2023, at 13:35, Lijie Wang < > > > > wangdachui9...@gmail.com > > > > > > <mailto:wangdachui9...@gmail.com> > > > > > > >> <mailto:wangdachui9...@gmail.com> > > > > > > >>>> <mailto:wangdachui9...@gmail.com>> wrote: > > > > > > >>>>>>> > > > > > > >>>>>>> Hi, Benchao and Aitozi, > > > > > > >>>>>>> > > > > > > >>>>>>> Thanks for your feedback about this FLIP. > > > > > > >>>>>>> > > > > > > >>>>>>> @Benchao > > > > > > >>>>>>> > > > > > > >>>>>>>>> I think it would be reasonable to also support > "pipeline > > > > > shuffle" > > > > > > >> if > > > > > > >>>>>>> possible. > > > > > > >>>>>>> As I said above, runtime filter can work well with all > > > shuffle > > > > > > mode, > > > > > > >>>>>>> including pipeline shuffle. > > > > > > >>>>>>> > > > > > > >>>>>>>>> if the RuntimeFIlterBuilder could be done quickly than > > > > > > >> RuntimeFilter > > > > > > >>>>>>> operator, it can still filter out additional data > > afterwards. > > > > > > >>>>>>> I think the main purpose of runtime filter is to reduce > the > > > > > shuffle > > > > > > >>>> data > > > > > > >>>>>>> and the data arriving at join. Although eagerly running > the > > > > large > > > > > > >>>>>>> table side can process datas in advance, most of the data > > may > > > > be > > > > > > >>>>>>> irrelevant, causing huge shuffle overhead and slowing the > > > join. > > > > > In > > > > > > >>>>>>> addition, if the join is a hash-join, the probe side of > the > > > > > > hash-join > > > > > > >>>>>> also > > > > > > >>>>>>> needs to wait for its build side to complete, so the > large > > > > table > > > > > > side > > > > > > >>>> is > > > > > > >>>>>>> likely to be back-pressed. > > > > > > >>>>>>> In addition, I don't tend to add too many configuration > > > options > > > > > in > > > > > > >> the > > > > > > >>>>>>> first version, which may make it more difficult to use > > (users > > > > > need > > > > > > to > > > > > > >>>>>>> understand a lot of internal implementation details). > Maybe > > > it > > > > > > could > > > > > > >>>> be a > > > > > > >>>>>>> future improvement (if it's worthwhile)? > > > > > > >>>>>>> > > > > > > >>>>>>> > > > > > > >>>>>>> @Aitozi > > > > > > >>>>>>> > > > > > > >>>>>>>>> IMO, In the current implementation two source table > > > operators > > > > > > will > > > > > > >> be > > > > > > >>>>>>> executed simultaneously. > > > > > > >>>>>>> The example in FLIP uses blocking shuffle(I will add this > > > point > > > > > to > > > > > > >>>> FLIP). > > > > > > >>>>>>> The runtime filter is generally chained with the large > > table > > > > side > > > > > > to > > > > > > >>>>>> reduce > > > > > > >>>>>>> the shuffle data (as shown in Figure 2 of FLIP). The job > > > > vertices > > > > > > >>>> should > > > > > > >>>>>> be > > > > > > >>>>>>> scheduled in topological order, so the large table side > can > > > > only > > > > > be > > > > > > >>>>>>> scheduled after the RuntimeFilterBuilder finishes. > > > > > > >>>>>>> > > > > > > >>>>>>>>> Are there some tests to show the default value of > > > > > > >>>>>>> table.optimizer.runtime-filter.min-probe-data-size 10G > is a > > > > good > > > > > > >>>> default > > > > > > >>>>>>> value. > > > > > > >>>>>>> It's not tested yet, but it will be done before merge the > > > code. > > > > > The > > > > > > >>>>>> current > > > > > > >>>>>>> value refers to systems such as spark and hive. Before > code > > > > > > merging, > > > > > > >> we > > > > > > >>>>>>> will test on TPC-DS 10 T to find an optimal set of > values. > > If > > > > you > > > > > > >> have > > > > > > >>>>>>> relevant experience on it, welcome to give some > > suggestions. > > > > > > >>>>>>> > > > > > > >>>>>>>>> What's the representation of the runtime filter node in > > > > > planner ? > > > > > > >>>>>>> As shown in Figure 1 of FLIP, we intend to add two new > > > physical > > > > > > >> nodes, > > > > > > >>>>>>> RuntimeFilterBuilder and RuntimeFilter. > > > > > > >>>>>>> > > > > > > >>>>>>> Best, > > > > > > >>>>>>> Lijie > > > > > > >>>>>>> > > > > > > >>>>>>> Aitozi <gjying1...@gmail.com <mailto: > gjying1...@gmail.com> > > > > > > <mailto:gjying1...@gmail.com> <mailto: > > > > > > >> gjying1...@gmail.com <mailto:gjying1...@gmail.com>> <mailto: > > > > > > >>>> gjying1...@gmail.com <mailto:gjying1...@gmail.com> <mailto: > > > > > > gjying1...@gmail.com>>> > > > > > > >>>>>> 于2023年6月15日周四 15:52写道： > > > > > > >>>>>>> > > > > > > >>>>>>>> Hi Lijie, > > > > > > >>>>>>>> > > > > > > >>>>>>>> Nice to see this valuable feature. After reading the > FLIP > > I > > > > have > > > > > > >>>> some > > > > > > >>>>>>>> questions below: > > > > > > >>>>>>>> > > > > > > >>>>>>>>> Schedule the TableSource(dim) first. > > > > > > >>>>>>>> > > > > > > >>>>>>>> How does it know to schedule the TableSource(dim) first > ? > > > IMO, > > > > > In > > > > > > >> the > > > > > > >>>>>>>> current implementation two source table operators will > be > > > > > executed > > > > > > >>>>>>>> simultaneously. > > > > > > >>>>>>>> > > > > > > >>>>>>>>> If the data volume on the probe side is too small, the > > > > overhead > > > > > > of > > > > > > >>>>>>>> building runtime filter is not worth it. > > > > > > >>>>>>>> > > > > > > >>>>>>>> Are there some tests to show the default value of > > > > > > >>>>>>>> table.optimizer.runtime-filter.min-probe-data-size 10G > is > > a > > > > good > > > > > > >>>> default > > > > > > >>>>>>>> value. The same to > > > > > > >> table.optimizer.runtime-filter.max-build-data-size > > > > > > >>>>>>>> > > > > > > >>>>>>>>> the runtime filter can be pushed down along the probe > > side, > > > > as > > > > > > >> close > > > > > > >>>> to > > > > > > >>>>>>>> data sources as possible > > > > > > >>>>>>>> > > > > > > >>>>>>>> What's the representation of the runtime filter node in > > > > planner > > > > > ? > > > > > > Is > > > > > > >>>> it > > > > > > >>>>>> a > > > > > > >>>>>>>> Filternode > > > > > > >>>>>>>> > > > > > > >>>>>>>> Best, > > > > > > >>>>>>>> > > > > > > >>>>>>>> Aitozi. > > > > > > >>>>>>>> > > > > > > >>>>>>>> Benchao Li <libenc...@apache.org <mailto: > > > libenc...@apache.org > > > > > > > > > > > <mailto:libenc...@apache.org> > > > > > > >> <mailto:libenc...@apache.org>> > > > > > > >>>> 于2023年6月15日周四 14:30写道： > > > > > > >>>>>>>> > > > > > > >>>>>>>>> Hi Lijie, > > > > > > >>>>>>>>> > > > > > > >>>>>>>>> Regarding the shuffle mode, I think it would be > > reasonable > > > to > > > > > > also > > > > > > >>>>>>>> support > > > > > > >>>>>>>>> "pipeline shuffle" if possible. > > > > > > >>>>>>>>> > > > > > > >>>>>>>>> "pipeline shuffle" is a essential for OLAP/MPP > computing, > > > > > > although > > > > > > >>>> this > > > > > > >>>>>>>> has > > > > > > >>>>>>>>> not been much exposed to users for now, I know a few > > > > companies > > > > > > that > > > > > > >>>>>> uses > > > > > > >>>>>>>>> Flink as a MPP computing engine, and there is an > ongoing > > > > > > effort[1] > > > > > > >> to > > > > > > >>>>>>>> make > > > > > > >>>>>>>>> this usage more powerful. > > > > > > >>>>>>>>> > > > > > > >>>>>>>>> Back to your concern that "Even if the RuntimeFilter > > > becomes > > > > > > >> running > > > > > > >>>>>>>> before > > > > > > >>>>>>>>> the RuntimeFilterBuilder finished, it will not process > > any > > > > data > > > > > > and > > > > > > >>>>>> will > > > > > > >>>>>>>>> occupy resources", whether it benefits us depends on > the > > > > scale > > > > > of > > > > > > >>>> data, > > > > > > >>>>>>>> if > > > > > > >>>>>>>>> the RuntimeFIlterBuilder could be done quickly than > > > > > RuntimeFilter > > > > > > >>>>>>>> operator, > > > > > > >>>>>>>>> it can still filter out additional data afterwards. > Hence > > > in > > > > my > > > > > > >>>>>> opinion, > > > > > > >>>>>>>> we > > > > > > >>>>>>>>> do not need to make the edge between > RuntimeFilterBuilder > > > and > > > > > > >>>>>>>> RuntimeFilter > > > > > > >>>>>>>>> BLOCKING only, at least it can be configured. > > > > > > >>>>>>>>> > > > > > > >>>>>>>>> [1] > > > > > > >>>>>> > > > > > > >>>> > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://www.google.com/url?q=https://www.google.com/url?q%3Dhttps://www.google.com/url?q%253Dhttps://www.google.com/url?q%25253Dhttps://issues.apache.org/jira/browse/FLINK-25318%252526source%25253Dgmail-imap%252526ust%25253D1687433776000000%252526usg%25253DAOvVaw3GqdpuiCqegqRLDv1PjMiL%2526source%253Dgmail-imap%2526ust%253D1687760804000000%2526usg%253DAOvVaw1oNzOlNn0UCDtz1M9jAw1x%26source%3Dgmail-imap%26ust%3D1687773407000000%26usg%3DAOvVaw3Zt14Wvxs_b8ghD0dIgPfH&source=gmail-imap&ust=1687781326000000&usg=AOvVaw0HsmkkqPeZGZOBvFiA8NOA > > > > > > >>>>>>>>> > > > > > > >>>>>>>>> Lijie Wang <wangdachui9...@gmail.com <mailto: > > > > > > wangdachui9...@gmail.com> <mailto: > > > > > > >> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com>> > > > > <mailto: > > > > > > >>>> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com> > > > > <mailto: > > > > > > wangdachui9...@gmail.com>> <mailto: > > > > > > >> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com>>> > > > > > > >>>>>> 于2023年6月15日周四 14:18写道： > > > > > > >>>>>>>>> > > > > > > >>>>>>>>>> Hi Yuxia, > > > > > > >>>>>>>>>> > > > > > > >>>>>>>>>> I made a mistake in the above response. > > > > > > >>>>>>>>>> > > > > > > >>>>>>>>>> The runtime filter can work well with all shuffle > mode. > > > > > However, > > > > > > >>>>>> hybrid > > > > > > >>>>>>>>>> shuffle and blocking shuffle are currently recommended > > for > > > > > batch > > > > > > >>>> jobs > > > > > > >>>>>>>>>> (piepline shuffle is not recommended). > > > > > > >>>>>>>>>> > > > > > > >>>>>>>>>> One more thing to mention here is that we will force > the > > > > edge > > > > > > >>>> between > > > > > > >>>>>>>>>> RuntimeFilterBuilder and RuntimeFilter to be > > > > > BLOCKING(regardless > > > > > > >> of > > > > > > >>>>>>>> which > > > > > > >>>>>>>>>> BatchShuffleMode is set). Because the RuntimeFilter > > really > > > > > > doesn’t > > > > > > >>>>>> need > > > > > > >>>>>>>>> to > > > > > > >>>>>>>>>> run before the RuntimeFilterBuilder finished. Even if > > the > > > > > > >>>>>> RuntimeFilter > > > > > > >>>>>>>>>> becomes running before the RuntimeFilterBuilder > > finished, > > > it > > > > > > will > > > > > > >>>> not > > > > > > >>>>>>>>>> process any data and will occupy resources. > > > > > > >>>>>>>>>> > > > > > > >>>>>>>>>> Best, > > > > > > >>>>>>>>>> Lijie > > > > > > >>>>>>>>>> > > > > > > >>>>>>>>>> Lijie Wang <wangdachui9...@gmail.com <mailto: > > > > > > wangdachui9...@gmail.com> <mailto: > > > > > > >> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com>> > > > > <mailto: > > > > > > >>>> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com> > > > > <mailto: > > > > > > wangdachui9...@gmail.com>> <mailto: > > > > > > >> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com>>> > > > > > > >>>>>> 于2023年6月15日周四 09:48写道： > > > > > > >>>>>>>>>> > > > > > > >>>>>>>>>>> Hi Yuxia, > > > > > > >>>>>>>>>>> > > > > > > >>>>>>>>>>> Thanks for your feedback. The answers of your > questions > > > are > > > > > as > > > > > > >>>>>>>> follows: > > > > > > >>>>>>>>>>> > > > > > > >>>>>>>>>>> 1. Yes, the row count comes from statistic of > > underlying > > > > > > table(Or > > > > > > >>>>>>>>>>> estimated based on the statistic of underlying table, > > if > > > > the > > > > > > >> build > > > > > > >>>>>>>> side > > > > > > >>>>>>>>>> or > > > > > > >>>>>>>>>>> probe side is not TableScan). If the statistic > > > > unavailable, > > > > > we > > > > > > >>>> will > > > > > > >>>>>>>>> not > > > > > > >>>>>>>>>>> inject a runtime filter(As you said, we can hardly > > > evaluate > > > > > the > > > > > > >>>>>>>>>> benefits). > > > > > > >>>>>>>>>>> Besides, AFAIK, the estimated data size of build side > > is > > > > also > > > > > > >> based > > > > > > >>>>>>>> on > > > > > > >>>>>>>>>> the > > > > > > >>>>>>>>>>> row count statistics, that is, if the statistics is > > > > > > unavailable, > > > > > > >>>> the > > > > > > >>>>>>>>>>> requirement > > > > > > "table.optimizer.runtime-filter.max-build-data-size" > > > > > > >>>>>>>> cannot > > > > > > >>>>>>>>>> be > > > > > > >>>>>>>>>>> evaluated either. I'll add this point into FLIP. > > > > > > >>>>>>>>>>> > > > > > > >>>>>>>>>>> 2. > > > > > > >>>>>>>>>>> Estimated data size does not meet requirement (in > > planner > > > > > > >>>>>>>> optimization > > > > > > >>>>>>>>>>> phase) -> No filter > > > > > > >>>>>>>>>>> Estimated data size meets the requirement (in planner > > > > > > >> optimization > > > > > > >>>>>>>>>> phase), > > > > > > >>>>>>>>>>> but the real data size does not meet the > requirement(in > > > > > > execution > > > > > > >>>>>>>>> phase) > > > > > > >>>>>>>>>> -> > > > > > > >>>>>>>>>>> Fake filter > > > > > > >>>>>>>>>>> > > > > > > >>>>>>>>>>> 3. Yes, the runtime filter is only for batch > > > jobs/blocking > > > > > > >> shuffle. > > > > > > >>>>>>>>>>> > > > > > > >>>>>>>>>>> Best, > > > > > > >>>>>>>>>>> Lijie > > > > > > >>>>>>>>>>> > > > > > > >>>>>>>>>>> yuxia <luoyu...@alumni.sjtu.edu.cn <mailto: > > > > > > luoyu...@alumni.sjtu.edu.cn> <mailto: > > > > > > >> luoyu...@alumni.sjtu.edu.cn <mailto: > luoyu...@alumni.sjtu.edu.cn > > >> > > > > > > <mailto: > > > > > > >>>> luoyu...@alumni.sjtu.edu.cn <mailto: > > luoyu...@alumni.sjtu.edu.cn > > > > > > > > > > <mailto:luoyu...@alumni.sjtu.edu.cn>> > > > > > > >> <mailto: > > > > > > >>>>>> luoyu...@alumni.sjtu.edu.cn <mailto: > > > luoyu...@alumni.sjtu.edu.cn > > > > > > > > > > > <mailto:luoyu...@alumni.sjtu.edu.cn> > > > > > > >> <mailto:luoyu...@alumni.sjtu.edu.cn>>> > > > > > > >>>> 于2023年6月14日周三 20:37写道： > > > > > > >>>>>>>>>>> > > > > > > >>>>>>>>>>>> Thanks Lijie for starting this discussion. Excited > to > > > see > > > > > > >> runtime > > > > > > >>>>>>>>> filter > > > > > > >>>>>>>>>>>> is to be implemented in Flink. > > > > > > >>>>>>>>>>>> I have few questions about it: > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> 1: As the FLIP said, `if the ndv cannot be > estimated, > > > use > > > > > row > > > > > > >>>> count > > > > > > >>>>>>>>>>>> instead`. So, does row count comes from the > statistic > > > from > > > > > > >>>>>>>> underlying > > > > > > >>>>>>>>>>>> table? What if the the statistic is also unavailable > > > > > > considering > > > > > > >>>>>>>> users > > > > > > >>>>>>>>>>>> maynot always remember to generate statistic in > > > > production. > > > > > > >>>>>>>>>>>> I'm wondering whether it make senese that just > disable > > > > > runtime > > > > > > >>>>>>>> filter > > > > > > >>>>>>>>> if > > > > > > >>>>>>>>>>>> statistic is unavailable since in that case, we can > > > hardly > > > > > > >>>> evaluate > > > > > > >>>>>>>>> the > > > > > > >>>>>>>>>>>> benefits of runtime-filter. > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> 2: The FLIP said: "We will inject the runtime > filters > > > only > > > > > if > > > > > > >> the > > > > > > >>>>>>>>>>>> following requirements are met:xxx", but it also > said, > > > > "Once > > > > > > >> this > > > > > > >>>>>>>>> limit > > > > > > >>>>>>>>>> is > > > > > > >>>>>>>>>>>> exceeded, it will output a fake filter(which always > > > > returns > > > > > > >> true)" > > > > > > >>>>>>>> in > > > > > > >>>>>>>>>>>> `RuntimeFilterBuilderOperator` part; Seems they are > > > > > > >> contradictory, > > > > > > >>>>>>>> so > > > > > > >>>>>>>>>> i'm > > > > > > >>>>>>>>>>>> wondering what's the real behavior, no filter will > be > > > > > injected > > > > > > >> or > > > > > > >>>>>>>> fake > > > > > > >>>>>>>>>>>> filter? > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> 3: Does it also mean runtime-filter can only take > > effect > > > > in > > > > > > >>>> blocking > > > > > > >>>>>>>>>>>> shuffle? > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> Best regards, > > > > > > >>>>>>>>>>>> Yuxia > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> ----- 原始邮件 ----- > > > > > > >>>>>>>>>>>> 发件人: "ron9 liu" <ron9....@gmail.com <mailto: > > > > > > ron9....@gmail.com> <mailto:ron9....@gmail.com> > > > > > > >> <mailto:ron9....@gmail.com> > > > > > > >>>> <mailto:ron9....@gmail.com>> > > > > > > >>>>>>>>>>>> 收件人: "dev" <dev@flink.apache.org <mailto: > > > > > dev@flink.apache.org> > > > > > > <mailto:dev@flink.apache.org> > > > > > > >> <mailto:dev@flink.apache.org> > > > > > > >>>> <mailto:dev@flink.apache.org>> > > > > > > >>>>>>>>>>>> 发送时间: 星期三, 2023年 6 月 14日 下午 5:29:28 > > > > > > >>>>>>>>>>>> 主题: Re: [DISCUSS] FLIP-324: Introduce Runtime Filter > > for > > > > > Flink > > > > > > >>>> Batch > > > > > > >>>>>>>>>> Jobs > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> Thanks Lijie start this discussion. Runtime Filter > is > > a > > > > > common > > > > > > >>>>>>>>>>>> optimization > > > > > > >>>>>>>>>>>> to improve the join performance that has been > adopted > > by > > > > > many > > > > > > >>>>>>>>> computing > > > > > > >>>>>>>>>>>> engines such as Spark, Doris, etc... Flink is a > > > streaming > > > > > > batch > > > > > > >>>>>>>>>> computing > > > > > > >>>>>>>>>>>> engine, and we are continuously optimizing the > > > performance > > > > > of > > > > > > >>>>>>>> batches. > > > > > > >>>>>>>>>>>> Runtime filter is a general performance optimization > > > > > technique > > > > > > >>>> that > > > > > > >>>>>>>>> can > > > > > > >>>>>>>>>>>> improve the performance of Flink batch jobs, so we > are > > > > > > >> introducing > > > > > > >>>>>>>> it > > > > > > >>>>>>>>> on > > > > > > >>>>>>>>>>>> batch as well. > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> Looking forward to all feedback. > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> Best, > > > > > > >>>>>>>>>>>> Ron > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>> Lijie Wang <wangdachui9...@gmail.com <mailto: > > > > > > wangdachui9...@gmail.com> <mailto: > > > > > > >> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com>> > > > > <mailto: > > > > > > >>>> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com> > > > > <mailto: > > > > > > wangdachui9...@gmail.com>> <mailto: > > > > > > >>>>>> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com > > > > > > > <mailto: > > > > > > wangdachui9...@gmail.com> <mailto: > > > > > > >> wangdachui9...@gmail.com <mailto:wangdachui9...@gmail.com>>>> > > > > > > >>>> 于2023年6月14日周三 17:17写道： > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>>>> Hi devs > > > > > > >>>>>>>>>>>>> > > > > > > >>>>>>>>>>>>> Ron Liu, Gen Luo and I would like to start a > > discussion > > > > > about > > > > > > >>>>>>>>>> FLIP-324: > > > > > > >>>>>>>>>>>>> Introduce Runtime Filter for Flink Batch Jobs[1] > > > > > > >>>>>>>>>>>>> > > > > > > >>>>>>>>>>>>> Runtime Filter is a common optimization to improve > > join > > > > > > >>>>>>>> performance. > > > > > > >>>>>>>>>> It > > > > > > >>>>>>>>>>>> is > > > > > > >>>>>>>>>>>>> designed to dynamically generate filter conditions > > for > > > > > > certain > > > > > > >>>>>>>> Join > > > > > > >>>>>>>>>>>> queries > > > > > > >>>>>>>>>>>>> at runtime to reduce the amount of scanned or > > shuffled > > > > > data, > > > > > > >>>> avoid > > > > > > >>>>>>>>>>>>> unnecessary I/O and network transmission, and speed > > up > > > > the > > > > > > >> query. > > > > > > >>>>>>>>> Its > > > > > > >>>>>>>>>>>>> working principle is building a filter(e.g. bloom > > > filter) > > > > > > based > > > > > > >>>> on > > > > > > >>>>>>>>> the > > > > > > >>>>>>>>>>>> data > > > > > > >>>>>>>>>>>>> on the small table side(build side) first, then > pass > > > this > > > > > > >> filter > > > > > > >>>>>>>> to > > > > > > >>>>>>>>>> the > > > > > > >>>>>>>>>>>>> large table side(probe side) to filter the > irrelevant > > > > data > > > > > on > > > > > > >> it, > > > > > > >>>>>>>>> this > > > > > > >>>>>>>>>>>> can > > > > > > >>>>>>>>>>>>> reduce the data reaching the join and improve > > > > performance. > > > > > > >>>>>>>>>>>>> > > > > > > >>>>>>>>>>>>> You can find more details in the FLIP-324[1]. > Looking > > > > > forward > > > > > > >> to > > > > > > >>>>>>>>> your > > > > > > >>>>>>>>>>>>> feedback. > > > > > > >>>>>>>>>>>>> > > > > > > >>>>>>>>>>>>> [1] > > > > > > >>>>>>>>>>>>> > > > > > > >>>>>>>>>>>>> > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>> > > > > > > >>>>>>>>> > > > > > > >>>>>>>> > > > > > > >>>>>> > > > > > > >>>> > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://www.google.com/url?q=https://www.google.com/url?q%3Dhttps://www.google.com/url?q%253Dhttps://www.google.com/url?q%25253Dhttps://cwiki.apache.org/confluence/display/FLINK/FLIP-324%252525253A%2525252BIntroduce%2525252BRuntime%2525252BFilter%2525252Bfor%2525252BFlink%2525252BBatch%2525252BJobs%252526source%25253Dgmail-imap%252526ust%25253D1687433776000000%252526usg%25253DAOvVaw0ke1ZHcJ--A1QgsbB84MHA%2526source%253Dgmail-imap%2526ust%253D1687760804000000%2526usg%253DAOvVaw21E3CQyayeBTYztmOnwMcz%26source%3Dgmail-imap%26ust%3D1687773407000000%26usg%3DAOvVaw0xVu0zYYNRmh8u8aq7uSi3&source=gmail-imap&ust=1687781326000000&usg=AOvVaw1LXwtWT177350iKD3sKCEt > > > > > > >>>>>>>>>>>>> > > > > > > >>>>>>>>>>>>> Best, > > > > > > >>>>>>>>>>>>> Ron & Gen & Lijie > > > > > > >>>>>>>>>>>>> > > > > > > >>>>>>>>>>>> > > > > > > >>>>>>>>>>> > > > > > > >>>>>>>>>> > > > > > > >>>>>>>>> > > > > > > >>>>>>>>> > > > > > > >>>>>>>>> -- > > > > > > >>>>>>>>> > > > > > > >>>>>>>>> Best, > > > > > > >>>>>>>>> Benchao Li > > > > > > > > > > > > > > > > > > > > > > > > > > >
