Hi, Jingong, Jark, Jing, Thanks for for the important inputs. Lake storage is a very important scenario, and consider more generic and extended case, I also would like to use "dynamic filtering" concept instead of "dynamic partition".
>maybe the FLIP should also demonstrate the EXPLAIN result, which is also an API. I will add a section to describe the EXPLAIN result. >Does DPP also support streaming queries? Yes, but for bounded source. >it requires the SplitEnumerator must implements new introduced `SupportsHandleExecutionAttemptSourceEvent` interface, +1 I will update the document and the poc code. Best, Godfrey Jing Zhang <beyond1...@gmail.com> 于2022年7月13日周三 20:22写道： > > Hi Godfrey, > Thanks for driving this discussion. > This is an important improvement for batch sql jobs. > I agree with Jingsong to expand the capability to more than just partitions. > Besides, I have two points: > 1. Based on FLIP-248[1], > > > Dynamic partition pruning mechanism can improve performance by avoiding > > reading large amounts of irrelevant data, and it works for both batch and > > streaming queries. > > Does DPP also support streaming queries? > It seems the proposed changes in the FLIP-248 does not work for streaming > queries, > because the dimension table might be an unbounded inputs. > Or does it require all dimension tables to be bounded inputs for streaming > jobs if the job wanna enable DPP? > > 2. I notice there are changes on SplitEnumerator for Hive source and File > source. > And they now depend on SourceEvent to pass PartitionData. > In FLIP-245, if enable speculative execution for sources based on FLIP-27 > which use SourceEvent, > it requires the SplitEnumerator must implements new introduced > `SupportsHandleExecutionAttemptSourceEvent` interface, > otherwise an exception would be thrown out. > Since hive and File sources are commonly used for batch jobs, it's better > to take this point into consideration. > > Best, > Jing Zhang > > [1] FLIP-248: > https://cwiki.apache.org/confluence/display/FLINK/FLIP-248%3A+Introduce+dynamic+partition+pruning > [2] FLIP-245: > https://cwiki.apache.org/confluence/display/FLINK/FLIP-245%3A+Source+Supports+Speculative+Execution+For+Batch+Job > > > Jark Wu <imj...@gmail.com> 于2022年7月12日周二 13:16写道： > > > I agree with Jingsong. DPP is a particular case of Dynamic Filter Pushdown > > that the join key contains partition fields. Extending this FLIP to > > general filter > > pushdown can benefit more optimizations, and they can share the same > > interface. > > > > For example, Trino Hive Connector leverages dynamic filtering to support: > > - dynamic partition pruning for partitioned tables > > - and dynamic bucket pruning for bucket tables > > - and dynamic filter pushed into the ORC and Parquet readers to perform > > stripe > > or row-group pruning and save on disk I/O. > > > > Therefore, +1 to extend this FLIP to Dynamic Filter Pushdown (or Dynamic > > Filtering), > > just like Trino [1]. The interfaces should also be adapted for that. > > > > Besides, maybe the FLIP should also demonstrate the EXPLAIN result, which > > is also an API. > > > > Best, > > Jark > > > > [1]: https://trino.io/docs/current/admin/dynamic-filtering.html > > > > > > > > > > > > > > > > > > > > > > On Tue, 12 Jul 2022 at 09:59, Jingsong Li <jingsongl...@gmail.com> wrote: > > > > > Thanks Godfrey for driving. > > > > > > I like this FLIP. > > > > > > We can restrict this capability to more than just partitions. > > > Here are some inputs from Lake Storage. > > > > > > The format of the splits generated by Lake Storage is roughly as follows: > > > Split { > > > Path filePath; > > > Statistics[] fieldStats; > > > } > > > > > > Stats contain the min and max of each column. > > > > > > If the storage is sorted by a column, this means that the split > > > filtering on that column will be very good, so not only the partition > > > field, but also this column is worthy of being pushed down the > > > RuntimeFilter. > > > This information can only be known by source, so I suggest that source > > > return which fields are worthy of being pushed down. > > > > > > My overall point is: > > > This FLIP can be extended to support Source Runtime Filter push-down > > > for all fields, not just dynamic partition pruning. > > > > > > What do you think? > > > > > > Best, > > > Jingsong > > > > > > On Fri, Jul 8, 2022 at 10:12 PM godfrey he <godfre...@gmail.com> wrote: > > > > > > > > Hi all, > > > > > > > > I would like to open a discussion on FLIP-248: Introduce dynamic > > > > partition pruning. > > > > > > > > Currently, Flink supports static partition pruning: the conditions in > > > > the WHERE clause are analyzed > > > > to determine in advance which partitions can be safely skipped in the > > > > optimization phase. > > > > Another common scenario: the partitions information is not available > > > > in the optimization phase but in the execution phase. > > > > That's the problem this FLIP is trying to solve: dynamic partition > > > > pruning, which could reduce the partition table source IO. > > > > > > > > The query pattern looks like: > > > > select * from store_returns, date_dim where sr_returned_date_sk = > > > > d_date_sk and d_year = 2000 > > > > > > > > We will introduce a mechanism for detecting dynamic partition pruning > > > > patterns in optimization phase > > > > and performing partition pruning at runtime by sending the dimension > > > > table results to the SplitEnumerator > > > > of fact table via existing coordinator mechanism. > > > > > > > > You can find more details in FLIP-248 document[1]. > > > > Looking forward to your any feedback. > > > > > > > > [1] > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-248%3A+Introduce+dynamic+partition+pruning > > > > [2] POC: https://github.com/godfreyhe/flink/tree/FLIP-248 > > > > > > > > > > > > Best, > > > > Godfrey > > > > >
