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> 于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> 于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://issues.apache.org/jira/browse/FLINK-25318 > > > > Lijie Wang <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> 于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> 于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> > > > >> 收件人: "dev" <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> 于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://cwiki.apache.org/confluence/display/FLINK/FLIP-324%3A+Introduce+Runtime+Filter+for+Flink+Batch+Jobs > > > >> > > > > >> > Best, > > > >> > Ron & Gen & Lijie > > > >> > > > > >> > > > > > > > > > > > > > -- > > > > Best, > > Benchao Li > > >
