Hi Jark, Cool, following your suggestions I have created three related subtasks under Flink-20791. Hope to assign these subtasks to me too, when you have time. And I will push forward the relevant implementation.
Jark Wu <imj...@gmail.com> 于2021年1月8日周五 下午12:30写道： > Hi Sebastian, > > I assigned the issue to you. But I suggest creating sub-tasks under this > issue. Because I think this would be a big contribution. > For example, you can split it into: > 1. Introduce SupportsAggregatePushDown interface > 2. Support SupportsAggregatePushDown in planner > 3. Support SupportsAggregatePushDown for JDBC source > 4. ... > > Best, > Jark > > On Thu, 7 Jan 2021 at 23:27, Sebastian Liu <liuyang0...@gmail.com> wrote: > >> Hi Jark, >> >> Seems that we have reached the agreement on the proposal. Could you >> please help to assign the below jira ticket to me? >> https://issues.apache.org/jira/browse/FLINK-20791 >> >> Jark Wu <imj...@gmail.com> 于2021年1月7日周四 上午10:25写道： >> >>> Thanks for updating the design doc. >>> It looks good to me. >>> >>> Best, >>> Jark >>> >>> On Thu, 7 Jan 2021 at 10:16, Jingsong Li <jingsongl...@gmail.com> wrote: >>> >>>> Sounds good to me. >>>> >>>> We don't have to worry about future changes, because it has covered all >>>> the capabilities of calcite aggregation. >>>> >>>> Best, >>>> Jingsong >>>> >>>> On Thu, Jan 7, 2021 at 12:14 AM Sebastian Liu <liuyang0...@gmail.com> >>>> wrote: >>>> >>>>> Hi Jark, >>>>> >>>>> Sounds good to me. For better scalability in the future, we could add >>>>> the AggregateExpression. >>>>> ``` >>>>> >>>>> public class AggregateExpression implements ResolvedExpression { >>>>> >>>>> private final FunctionDefinition functionDefinition; >>>>> >>>>> private final List<FieldReferenceExpression> args; >>>>> >>>>> private final @Nullable CallExpression filterExpression; >>>>> >>>>> private final DataType resultType; >>>>> >>>>> private final boolean distinct; >>>>> >>>>> private final boolean approximate; >>>>> >>>>> >>>>> >>>>> private final boolean ignoreNulls; >>>>> >>>>> } >>>>> ``` >>>>> >>>>> And we really only need one GroupingSets parameter for grouping. I >>>>> have updated the related interface in the proposal. >>>>> Appreciate the continued feedback and help. >>>>> >>>>> Jark Wu <imj...@gmail.com> 于2021年1月6日周三 下午9:34写道： >>>>> >>>>>> Hi Liu, Jingsong, >>>>>> >>>>>> Regarding the agg with filter, I think in theory we can support >>>>>> pushing such a pattern into source. >>>>>> We don't need to support it in the first version, but in the long >>>>>> term, we can support it. >>>>>> The designed interface should be future proof. >>>>>> >>>>>> Considering filter arg and distinct flag should be part of the >>>>>> aggregate expression. >>>>>> I'm wondering if CallExpression is a good representation for it. >>>>>> What do you think about proposing the following `AggregateExpression` >>>>>> to replace the `CallExpression`? >>>>>> >>>>>> class AggregateExpression implements ResolvedExpression { >>>>>> private final FunctionDefinition functionDefinition; >>>>>> private final List<FieldReferenceExpression> args; >>>>>> private final @Nullable CallExpression filterExpr; >>>>>> private final boolean distinct; >>>>>> } >>>>>> >>>>>> Besides, we don't need both groupingFields and groupingSets. >>>>>> `groupingSets` should be a superset of groupingFields. >>>>>> Then the interface of SupportsAggregatePushDown can be: >>>>>> >>>>>> interface SupportsAggregatePushDown { >>>>>> >>>>>> boolean applyAggregates( >>>>>> List<int[]> groupingSets, >>>>>> List<AggregateExpression> aggregates, >>>>>> DataType producedDataType); >>>>>> } >>>>>> >>>>>> What do you think? >>>>>> >>>>>> Best, >>>>>> Jark >>>>>> >>>>>> On Wed, 6 Jan 2021 at 19:56, Sebastian Liu <liuyang0...@gmail.com> >>>>>> wrote: >>>>>> >>>>>>> Hi Jingsong, Jark, >>>>>>> >>>>>>> Thx so much for our discussion, and the cases mentioned above are >>>>>>> really worthy for further discussion. >>>>>>> >>>>>>> 1. For aggregate with filter expressions: eg: select COUNT(1) >>>>>>> FILTER(WHERE cc_call_center_sk > 3) from call_center; >>>>>>> For the current Blink Planner, the optimized plan will be: >>>>>>> TableSourceScan -> Calc(IS TRUE(>(cc_call_center_sk, 3))) -> >>>>>>> LocalAgg -> Exchange -> FinalAgg. >>>>>>> As there is a Calc above the TableSource, this pattern can't match >>>>>>> the LocalAggPushDownRule in the current design. >>>>>>> >>>>>>> 2. For the grouping set or rollup use case: eg: select COUNT(1) from >>>>>>> call_center group by rollup(cc_class, cc_employees); >>>>>>> For the current Blink Planner, the optimized plan will be: >>>>>>> TableSourceScan -> Expand -> LocalAgg -> Exchange -> FinalAgg -> >>>>>>> Calc. >>>>>>> It's also not covered by the current LocalAggPushDownRule design. >>>>>>> >>>>>>> 3. I want to add a case which we haven't discussed yet. >>>>>>> Aggregate with Having clause. >>>>>>> eg: select COUNT(1) from call_center group by cc_class having >>>>>>> max(cc_tax_percentage) > 0.2; >>>>>>> For the current Blink Planner, the optimized plan will be: >>>>>>> TableSourceScan -> LocalAgg -> Exchange -> FinalAgg -> >>>>>>> Calc(where=[>($f2, 0.2:DECIMAL(2, 1))]). >>>>>>> >>>>>>> The core discussion points are summarized as follows: >>>>>>> a) Aggregate is a more complex scenario than predicates or limits, >>>>>>> and also depends on the different underlying storages. >>>>>>> b) One rule seems can't completely cover all aggregate scenario, but >>>>>>> whether SupportSAggregatePushDown interface can be a bit more general? >>>>>>> c) Could the CallExpression express the semantics of CalCite >>>>>>> AggregateCall? >>>>>>> >>>>>>> IMO: Completely push down aggregate is generally hard for >>>>>>> distributed systems. Usually we need a GROUP BY and exactly >>>>>>> matches the partition mode in downstream storage. At the same time, >>>>>>> the benefit of remove the final aggregate is actually limited. >>>>>>> The LocalAggPushDown generally yields more than 80% of the CPU and >>>>>>> IO benefits. But I also agree that >>>>>>> the SupportsAggregatePushDown interface should be as generic as >>>>>>> possible for future extensions, and meanwhile keep confidence >>>>>>> in the interface we design. >>>>>>> >>>>>>> For core points (a): As the complexity of aggregate, one >>>>>>> LogicalAggregate node may extend to "Expand / Calc / LocalXXAgg / >>>>>>> Exchange >>>>>>> / FinalXXAgg" >>>>>>> in physical phase. Seems that we can't solve all cases with only one >>>>>>> rule. So I suggest PushLocalAggIntoTableSourceScanRule focus only >>>>>>> on the pattern of TableSourceScan + LocalXXAggregate at present. >>>>>>> >>>>>>> For core points (b & c): I think we can change the interface to be: >>>>>>> ``` >>>>>>> >>>>>>> boolean applyAggregates(int[] groupingFields, List<CallExpression> >>>>>>> aggregateExpressions, DataType producedDataType, List<int[]> >>>>>>> groupingSets); >>>>>>> >>>>>>> ``` >>>>>>> >>>>>>> >>>>>>> Simple Group: groupingSets.size() == 1 && >>>>>>> groupingSets.get(0).equals(groupingFields) >>>>>>> >>>>>>> Cube Group: groupingSets.size() == IntMath.pow(2, >>>>>>> groupingFields.cardinality()) >>>>>>> Rollup: >>>>>>> Refernece org.apache.calcite.rel.core.Aggregate.Group#isRollup >>>>>>> >>>>>>> Then we can handle the complex grouping case. The Connector >>>>>>> developer of the downstream storage should determine >>>>>>> whether it supports the associated grouping type. For the filter and >>>>>>> having clause, they will convert to be related Calc RelNode, >>>>>>> and no longer in the LocalAggregate node, the CallExpression may be >>>>>>> sufficient to express the semantics of AggregateCall. >>>>>>> >>>>>>> What do you think? Looking forward to our further discussion. >>>>>>> >>>>>>> >>>>>>> Jingsong Li <jingsongl...@gmail.com> 于2021年1月6日周三 下午2:24写道： >>>>>>> >>>>>>>> > I think filter expressions and grouping sets are semantic >>>>>>>> arguments instead of utilities. If we want to push them into sources, >>>>>>>> the >>>>>>>> connector developers should be aware of them.Wrapping them in a context >>>>>>>> implicitly is error-prone that the existing connector will produce >>>>>>>> wrong >>>>>>>> results when upgrading to new Flink versions. >>>>>>>> >>>>>>>> We can have some mechanism to check the upgrading. >>>>>>>> >>>>>>>> > I think for these cases, providing a new default method to >>>>>>>> override might be a better choice. >>>>>>>> >>>>>>>> Then we will have three or more methods. For the API level, I >>>>>>>> really don't like it... >>>>>>>> >>>>>>>> Best, >>>>>>>> Jingsong >>>>>>>> >>>>>>>> On Wed, Jan 6, 2021 at 2:10 PM Jark Wu <imj...@gmail.com> wrote: >>>>>>>> >>>>>>>>> I think filter expressions and grouping sets are semantic >>>>>>>>> arguments instead of utilities. >>>>>>>>> If we want to push them into sources, the connector developers >>>>>>>>> should be aware of them. >>>>>>>>> Wrapping them in a context implicitly is error-prone that the >>>>>>>>> existing connector will produce wrong results >>>>>>>>> when upgrading to new Flink versions (as we are pushing >>>>>>>>> grouping_sets/filter_args, but connector ignores it). >>>>>>>>> I think for these cases, providing a new default method to >>>>>>>>> override might be a better choice. >>>>>>>>> >>>>>>>>> Best, >>>>>>>>> Jark >>>>>>>>> >>>>>>>>> On Wed, 6 Jan 2021 at 13:56, Jingsong Li <jingsongl...@gmail.com> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>>> Hi, >>>>>>>>>> >>>>>>>>>> I'm also curious about aggregate with filter (COUNT(1) >>>>>>>>>> FILTER(WHERE d > 1)). Can we push it down? I'm not sure that a >>>>>>>>>> single call >>>>>>>>>> expression can express it, and how we should embody it and convey it >>>>>>>>>> to >>>>>>>>>> users. >>>>>>>>>> >>>>>>>>>> Best, >>>>>>>>>> Jingsong >>>>>>>>>> >>>>>>>>>> On Wed, Jan 6, 2021 at 1:36 PM Jingsong Li < >>>>>>>>>> jingsongl...@gmail.com> wrote: >>>>>>>>>> >>>>>>>>>>> Hi Jark, >>>>>>>>>>> >>>>>>>>>>> I don't want to limit this interface to LocalAgg Push down. >>>>>>>>>>> Actually, sometimes, we can push whole aggregation to source too. >>>>>>>>>>> >>>>>>>>>>> So, this rule can do something more advanced. For example, we >>>>>>>>>>> can push down group sets to source too, for the SQL: "GROUP BY >>>>>>>>>>> GROUPING >>>>>>>>>>> SETS (f1, f2)". Then, we need to add more information to push down. >>>>>>>>>>> >>>>>>>>>>> Best, >>>>>>>>>>> Jingsong >>>>>>>>>>> >>>>>>>>>>> On Wed, Jan 6, 2021 at 11:02 AM Jark Wu <imj...@gmail.com> >>>>>>>>>>> wrote: >>>>>>>>>>> >>>>>>>>>>>> I think this may be over designed. We should have confidence in >>>>>>>>>>>> the interface we design, the interface should be stable. >>>>>>>>>>>> Wrapping things in a big context has a cost of losing user >>>>>>>>>>>> convenience. >>>>>>>>>>>> Foremost, we don't see any parameters to add in the future. Do >>>>>>>>>>>> you know any potential parameters? >>>>>>>>>>>> >>>>>>>>>>>> Best, >>>>>>>>>>>> Jark >>>>>>>>>>>> >>>>>>>>>>>> On Wed, 6 Jan 2021 at 10:28, Jingsong Li < >>>>>>>>>>>> jingsongl...@gmail.com> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> Hi Sebastian, >>>>>>>>>>>>> >>>>>>>>>>>>> Well, I mean: >>>>>>>>>>>>> >>>>>>>>>>>>> `boolean applyAggregates(int[] groupingFields, >>>>>>>>>>>>> List<CallExpression> aggregateExpressions, DataType >>>>>>>>>>>>> producedDataType);` >>>>>>>>>>>>> VS >>>>>>>>>>>>> ``` >>>>>>>>>>>>> boolean applyAggregates(Aggregation agg); >>>>>>>>>>>>> >>>>>>>>>>>>> interface Aggregation { >>>>>>>>>>>>> int[] groupingFields(); >>>>>>>>>>>>> List<CallExpression> aggregateExpressions(); >>>>>>>>>>>>> DataType producedDataType(); >>>>>>>>>>>>> } >>>>>>>>>>>>> ``` >>>>>>>>>>>>> >>>>>>>>>>>>> Maybe I've over considered it, but I think Aggregation is a >>>>>>>>>>>>> complicated thing. Maybe we need to extend its parameters in the >>>>>>>>>>>>> future, so >>>>>>>>>>>>> make the parameters interface, which is conducive to the future >>>>>>>>>>>>> expansion >>>>>>>>>>>>> without destroying the compatibility of user implementation. If >>>>>>>>>>>>> it is the >>>>>>>>>>>>> way before, users need to modify the code. >>>>>>>>>>>>> >>>>>>>>>>>>> Best, >>>>>>>>>>>>> Jingsong >>>>>>>>>>>>> >>>>>>>>>>>>> On Wed, Jan 6, 2021 at 12:52 AM Sebastian Liu < >>>>>>>>>>>>> liuyang0...@gmail.com> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>>> Hi Jinsong, >>>>>>>>>>>>>> >>>>>>>>>>>>>> Thx a lot for your suggestion. These points really need to be >>>>>>>>>>>>>> clear in the proposal. >>>>>>>>>>>>>> >>>>>>>>>>>>>> For the semantic problem, I think the main point is the >>>>>>>>>>>>>> different returned data types >>>>>>>>>>>>>> for the target aggregate function and the row format returned >>>>>>>>>>>>>> by the underlying storage. >>>>>>>>>>>>>> That's why we provide the producedDataType in the >>>>>>>>>>>>>> SupportsAggregatePushDown interface. >>>>>>>>>>>>>> Need to let developers know that we need to handle the >>>>>>>>>>>>>> semantic differences between >>>>>>>>>>>>>> the underlying storage system and Flink in related >>>>>>>>>>>>>> connectors. [Supplemented in proposal] >>>>>>>>>>>>>> >>>>>>>>>>>>>> For the phase of the new PushLocalAggIntoTableSourceScanRule >>>>>>>>>>>>>> rule, it's also a key point. >>>>>>>>>>>>>> As you suggested, we should put it into the PHYSICAL_REWRITE >>>>>>>>>>>>>> rule set, and better to put it >>>>>>>>>>>>>> behind the EnforceLocalXXAggRule. [Supplemented in proposal] >>>>>>>>>>>>>> >>>>>>>>>>>>>> For the scalability of the interface, actually I don't >>>>>>>>>>>>>> exactly understand your suggestion. Is it to add >>>>>>>>>>>>>> an abstract class, to implement the SupportsAggregatePushDown >>>>>>>>>>>>>> interface, and holds the >>>>>>>>>>>>>> `List < CallExpression > aggregateExpressions, int[] >>>>>>>>>>>>>> GroupingFields, DataType producedDataType` >>>>>>>>>>>>>> fields? >>>>>>>>>>>>>> >>>>>>>>>>>>>> Looking forward to your further feedback or guidance. >>>>>>>>>>>>>> >>>>>>>>>>>>>> Jingsong Li <jingsongl...@gmail.com> 于2021年1月5日周二 下午2:44写道： >>>>>>>>>>>>>> >>>>>>>>>>>>>>> Thanks for your proposal! Sebastian. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> +1 for SupportsAggregatePushDown. The above wonderful >>>>>>>>>>>>>>> discussion has solved >>>>>>>>>>>>>>> many of my concerns. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> ## Semantic problems >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> We may need to add some mechanisms or comments, because as >>>>>>>>>>>>>>> far as I know, >>>>>>>>>>>>>>> the semantics of each database is actually different, which >>>>>>>>>>>>>>> may need to be >>>>>>>>>>>>>>> reflected in your specific implementation. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> For example, the AVG output types of various databases may >>>>>>>>>>>>>>> be different. >>>>>>>>>>>>>>> For example, MySQL outputs double, this is different from >>>>>>>>>>>>>>> Flink. What >>>>>>>>>>>>>>> should we do? (Lucky, avg will be splitted into sum and >>>>>>>>>>>>>>> count, But we also >>>>>>>>>>>>>>> need care about decimal and others) >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> ## The phase of push-down rule >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> I strongly recommend that you do not put it in the Volcano >>>>>>>>>>>>>>> phase, which may >>>>>>>>>>>>>>> make the cost calculation very troublesome. >>>>>>>>>>>>>>> So in PHYSICAL_REWRITE? >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> ## About interface >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> For scalability, I slightly recommend that we introduce an >>>>>>>>>>>>>>> `Aggregate` >>>>>>>>>>>>>>> interface, it contains `List<CallExpression> >>>>>>>>>>>>>>> aggregateExpressions, int[] >>>>>>>>>>>>>>> groupingFields, DataType producedDataType` fields. In this >>>>>>>>>>>>>>> way, we can add >>>>>>>>>>>>>>> fields easily without breaking compatibility. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> I think the current design is very good, just put forward >>>>>>>>>>>>>>> some ideas. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Best, >>>>>>>>>>>>>>> Jingsong >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> On Tue, Jan 5, 2021 at 1:55 PM Sebastian Liu < >>>>>>>>>>>>>>> liuyang0...@gmail.com> wrote: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> > Hi Jark, >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > Thx for your further feedback and help. The interface of >>>>>>>>>>>>>>> > SupportsAggregatePushDown may indeed need some adjustments. >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > For (1) Agree: Yeah, the upstream only need to know if the >>>>>>>>>>>>>>> TableSource can >>>>>>>>>>>>>>> > handle all of the aggregates. >>>>>>>>>>>>>>> > It's better to just return a boolean type to indicate >>>>>>>>>>>>>>> whether all of >>>>>>>>>>>>>>> > aggregates push down was successful or not. [Resolved in >>>>>>>>>>>>>>> proposal] >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > For (2) Agree: The aggOutputDataType represent the >>>>>>>>>>>>>>> produced data type of >>>>>>>>>>>>>>> > the new table source to make sure that the new table >>>>>>>>>>>>>>> source can >>>>>>>>>>>>>>> > connect with the related exchange node. The format of this >>>>>>>>>>>>>>> > aggOutputDataType is groupedFields's type + agg function's >>>>>>>>>>>>>>> return type. >>>>>>>>>>>>>>> > The reason for adding this parameter in this function is >>>>>>>>>>>>>>> also to facilitate >>>>>>>>>>>>>>> > the user to build the final output type. I have changed >>>>>>>>>>>>>>> this parameter >>>>>>>>>>>>>>> > to be producedDataType. [Resolved in proposal] >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > For (3) Agree: Indeed, groupSet may mislead users, I have >>>>>>>>>>>>>>> changed to use >>>>>>>>>>>>>>> > groupingFields. [Resolved in proposal] >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > Thx again for the suggestion, looking for the further >>>>>>>>>>>>>>> discussion. >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > Jark Wu <imj...@gmail.com> 于2021年1月5日周二 下午12:05写道： >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > > I'm also +1 for idea#2. >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > Regarding to the updated interface, >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > Result applyAggregates(List<CallExpression> >>>>>>>>>>>>>>> aggregateExpressions, >>>>>>>>>>>>>>> > > int[] groupSet, DataType aggOutputDataType); >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > final class Result { >>>>>>>>>>>>>>> > > private final List<CallExpression> >>>>>>>>>>>>>>> acceptedAggregates; >>>>>>>>>>>>>>> > > private final List<CallExpression> >>>>>>>>>>>>>>> remainingAggregates; >>>>>>>>>>>>>>> > > } >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > I have following comments: >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > 1) Do we need the composite Result return type? Is a >>>>>>>>>>>>>>> boolean return type >>>>>>>>>>>>>>> > > enough? >>>>>>>>>>>>>>> > > From my understanding, all of the aggregates should >>>>>>>>>>>>>>> be accepted, >>>>>>>>>>>>>>> > > otherwise the pushdown should fail. >>>>>>>>>>>>>>> > > Therefore, users don't need to distinguish which >>>>>>>>>>>>>>> aggregates are >>>>>>>>>>>>>>> > > "accepted". >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > 2) Does the `aggOutputDataType` represent the produced >>>>>>>>>>>>>>> data type of the >>>>>>>>>>>>>>> > > new source, or just the return type of all the agg >>>>>>>>>>>>>>> functions? >>>>>>>>>>>>>>> > > I would prefer to `producedDataType` just like >>>>>>>>>>>>>>> > > `SupportsReadingMetadata` to reduce the effort for users >>>>>>>>>>>>>>> to concat a >>>>>>>>>>>>>>> > final >>>>>>>>>>>>>>> > > output type. >>>>>>>>>>>>>>> > > The return type of each agg function can be obtained >>>>>>>>>>>>>>> from the >>>>>>>>>>>>>>> > > `CallExpression`. >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > 3) What do you think about renaming `groupSet` to >>>>>>>>>>>>>>> `grouping` or >>>>>>>>>>>>>>> > > `groupedFields` ? >>>>>>>>>>>>>>> > > The `groupSet` may confuse users that it relates to >>>>>>>>>>>>>>> "grouping sets". >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > What do you think? >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > Best, >>>>>>>>>>>>>>> > > Jark >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > > On Tue, 5 Jan 2021 at 11:04, Kurt Young < >>>>>>>>>>>>>>> ykt...@gmail.com> wrote: >>>>>>>>>>>>>>> > > >>>>>>>>>>>>>>> > >> Sorry for the typo -_-! >>>>>>>>>>>>>>> > >> I meant idea #2. >>>>>>>>>>>>>>> > >> >>>>>>>>>>>>>>> > >> Best, >>>>>>>>>>>>>>> > >> Kurt >>>>>>>>>>>>>>> > >> >>>>>>>>>>>>>>> > >> >>>>>>>>>>>>>>> > >> On Tue, Jan 5, 2021 at 10:59 AM Sebastian Liu < >>>>>>>>>>>>>>> liuyang0...@gmail.com> >>>>>>>>>>>>>>> > >> wrote: >>>>>>>>>>>>>>> > >> >>>>>>>>>>>>>>> > >>> Hi Kurt, >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>> Thx a lot for your feedback. If local aggregation is >>>>>>>>>>>>>>> more like a >>>>>>>>>>>>>>> > >>> physical operator rather than logical >>>>>>>>>>>>>>> > >>> operator, I think your suggestion should be idea #2 >>>>>>>>>>>>>>> which handle all in >>>>>>>>>>>>>>> > >>> the physical optimization phase? >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>> Looking forward for the further discussion. >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>> Kurt Young <ykt...@gmail.com> 于2021年1月5日周二 上午9:52写道： >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>>> Local aggregation is more like a physical operator >>>>>>>>>>>>>>> rather than logical >>>>>>>>>>>>>>> > >>>> operator. I would suggest going with idea #1. >>>>>>>>>>>>>>> > >>>> >>>>>>>>>>>>>>> > >>>> Best, >>>>>>>>>>>>>>> > >>>> Kurt >>>>>>>>>>>>>>> > >>>> >>>>>>>>>>>>>>> > >>>> >>>>>>>>>>>>>>> > >>>> On Wed, Dec 30, 2020 at 8:31 PM Sebastian Liu < >>>>>>>>>>>>>>> liuyang0...@gmail.com> >>>>>>>>>>>>>>> > >>>> wrote: >>>>>>>>>>>>>>> > >>>> >>>>>>>>>>>>>>> > >>>> > Hi Jark, Thx a lot for your quick reply and >>>>>>>>>>>>>>> valuable suggestions. >>>>>>>>>>>>>>> > >>>> > For (1): Agree: Since we are in the period of >>>>>>>>>>>>>>> upgrading the new >>>>>>>>>>>>>>> > table >>>>>>>>>>>>>>> > >>>> > source api, >>>>>>>>>>>>>>> > >>>> > we really should consider the new interface for the >>>>>>>>>>>>>>> new optimize >>>>>>>>>>>>>>> > >>>> rule. If >>>>>>>>>>>>>>> > >>>> > the new rule >>>>>>>>>>>>>>> > >>>> > doesn't use the new api, we'll have to upgrade it >>>>>>>>>>>>>>> sooner or later. I >>>>>>>>>>>>>>> > >>>> have >>>>>>>>>>>>>>> > >>>> > change to use >>>>>>>>>>>>>>> > >>>> > the ability interface for the >>>>>>>>>>>>>>> SupportsAggregatePushDown definition >>>>>>>>>>>>>>> > in >>>>>>>>>>>>>>> > >>>> above >>>>>>>>>>>>>>> > >>>> > proposal. >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > For (2): Agree: Change to use CallExpression is a >>>>>>>>>>>>>>> better choice, and >>>>>>>>>>>>>>> > >>>> have >>>>>>>>>>>>>>> > >>>> > resolved this >>>>>>>>>>>>>>> > >>>> > comment in the proposal. >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > For (3): I suggest we first support the JDBC >>>>>>>>>>>>>>> connector, as we don't >>>>>>>>>>>>>>> > >>>> have >>>>>>>>>>>>>>> > >>>> > Druid connector >>>>>>>>>>>>>>> > >>>> > and ES connector just has sink api at present. >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > But perhaps the biggest question may be whether we >>>>>>>>>>>>>>> should use idea 1 >>>>>>>>>>>>>>> > >>>> or >>>>>>>>>>>>>>> > >>>> > idea 2 in proposal. >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > What do you think? After we reach the agreement on >>>>>>>>>>>>>>> the proposal, >>>>>>>>>>>>>>> > our >>>>>>>>>>>>>>> > >>>> team >>>>>>>>>>>>>>> > >>>> > can drive to >>>>>>>>>>>>>>> > >>>> > complete this feature. >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > Jark Wu <imj...@gmail.com> 于2020年12月29日周二 下午2:58写道： >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > > Hi Sebastian, >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > > Thanks for the proposal. I think this is a great >>>>>>>>>>>>>>> improvement for >>>>>>>>>>>>>>> > >>>> Flink >>>>>>>>>>>>>>> > >>>> > SQL. >>>>>>>>>>>>>>> > >>>> > > I went through the design doc and have following >>>>>>>>>>>>>>> thoughts: >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > > 1) Flink has deprecated the legacy TableSource in >>>>>>>>>>>>>>> 1.11 and >>>>>>>>>>>>>>> > proposed >>>>>>>>>>>>>>> > >>>> a new >>>>>>>>>>>>>>> > >>>> > > set of DynamicTableSource interfaces. Could you >>>>>>>>>>>>>>> update your >>>>>>>>>>>>>>> > >>>> proposal to >>>>>>>>>>>>>>> > >>>> > > use the new interfaces? >>>>>>>>>>>>>>> > >>>> > > Follow the existing ability interfaces, e.g. >>>>>>>>>>>>>>> > >>>> > > SupportsFilterPushDown, >>>>>>>>>>>>>>> SupportsProjectionPushDown. >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > > 2) Personally, I think CallExpression would be a >>>>>>>>>>>>>>> better >>>>>>>>>>>>>>> > >>>> representation >>>>>>>>>>>>>>> > >>>> > than >>>>>>>>>>>>>>> > >>>> > > separate `FunctionDefinition` and args. Because, >>>>>>>>>>>>>>> it would be >>>>>>>>>>>>>>> > easier >>>>>>>>>>>>>>> > >>>> to >>>>>>>>>>>>>>> > >>>> > know >>>>>>>>>>>>>>> > >>>> > > what's the index and type of the arguments. >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > > 3) It would be better to list which connectors >>>>>>>>>>>>>>> will be supported >>>>>>>>>>>>>>> > in >>>>>>>>>>>>>>> > >>>> the >>>>>>>>>>>>>>> > >>>> > > plan? >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > > Best, >>>>>>>>>>>>>>> > >>>> > > Jark >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > > On Tue, 29 Dec 2020 at 00:49, Sebastian Liu < >>>>>>>>>>>>>>> > liuyang0...@gmail.com> >>>>>>>>>>>>>>> > >>>> > wrote: >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > > > Hi all, >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> > >>>> > > > I'd like to discuss a new feature for the Blink >>>>>>>>>>>>>>> Planner. >>>>>>>>>>>>>>> > >>>> > > > Aggregate operator of Flink SQL is currently >>>>>>>>>>>>>>> fully done at Flink >>>>>>>>>>>>>>> > >>>> layer. >>>>>>>>>>>>>>> > >>>> > > > With the developing of storage, many downstream >>>>>>>>>>>>>>> storage of Flink >>>>>>>>>>>>>>> > >>>> SQL >>>>>>>>>>>>>>> > >>>> > has >>>>>>>>>>>>>>> > >>>> > > > the ability to deal with Aggregation operator. >>>>>>>>>>>>>>> > >>>> > > > Pushing down Aggregate to data source layer >>>>>>>>>>>>>>> will improve >>>>>>>>>>>>>>> > >>>> performance >>>>>>>>>>>>>>> > >>>> > from >>>>>>>>>>>>>>> > >>>> > > > the perspective of the network IO and >>>>>>>>>>>>>>> computation overhead. >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> > >>>> > > > I have drafted a design doc for this new >>>>>>>>>>>>>>> feature. >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> https://docs.google.com/document/d/1kGwC_h4qBNxF2eMEz6T6arByOB8yilrPLqDN0QBQXW4/edit?usp=sharing >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> > >>>> > > > Any comment or discussion is welcome. >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> > >>>> > > > -- >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> > >>>> > > > *With kind regards >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> ------------------------------------------------------------ >>>>>>>>>>>>>>> > >>>> > > > Sebastian Liu 刘洋 >>>>>>>>>>>>>>> > >>>> > > > Institute of Computing Technology, Chinese >>>>>>>>>>>>>>> Academy of Science >>>>>>>>>>>>>>> > >>>> > > > Mobile\WeChat: +86—15201613655 >>>>>>>>>>>>>>> > >>>> > > > E-mail: liuyang0...@gmail.com < >>>>>>>>>>>>>>> liuyang0...@gmail.com> >>>>>>>>>>>>>>> > >>>> > > > QQ: 3239559* >>>>>>>>>>>>>>> > >>>> > > > >>>>>>>>>>>>>>> > >>>> > > >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > -- >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> > *With kind regards >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> ------------------------------------------------------------ >>>>>>>>>>>>>>> > >>>> > Sebastian Liu 刘洋 >>>>>>>>>>>>>>> > >>>> > Institute of Computing Technology, Chinese Academy >>>>>>>>>>>>>>> of Science >>>>>>>>>>>>>>> > >>>> > Mobile\WeChat: +86—15201613655 >>>>>>>>>>>>>>> > >>>> > E-mail: liuyang0...@gmail.com < >>>>>>>>>>>>>>> liuyang0...@gmail.com> >>>>>>>>>>>>>>> > >>>> > QQ: 3239559* >>>>>>>>>>>>>>> > >>>> > >>>>>>>>>>>>>>> > >>>> >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>> -- >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>> *With kind regards >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> ------------------------------------------------------------ >>>>>>>>>>>>>>> > >>> Sebastian Liu 刘洋 >>>>>>>>>>>>>>> > >>> Institute of Computing Technology, Chinese Academy of >>>>>>>>>>>>>>> Science >>>>>>>>>>>>>>> > >>> Mobile\WeChat: +86—15201613655 >>>>>>>>>>>>>>> > >>> E-mail: liuyang0...@gmail.com <liuyang0...@gmail.com> >>>>>>>>>>>>>>> > >>> QQ: 3239559* >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>> >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > -- >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > *With kind regards >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> ------------------------------------------------------------ >>>>>>>>>>>>>>> > Sebastian Liu 刘洋 >>>>>>>>>>>>>>> > Institute of Computing Technology, Chinese Academy of >>>>>>>>>>>>>>> Science >>>>>>>>>>>>>>> > Mobile\WeChat: +86—15201613655 >>>>>>>>>>>>>>> > E-mail: liuyang0...@gmail.com <liuyang0...@gmail.com> >>>>>>>>>>>>>>> > QQ: 3239559* >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> -- >>>>>>>>>>>>>>> Best, Jingsong Lee >>>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> -- >>>>>>>>>>>>>> >>>>>>>>>>>>>> *With kind regards >>>>>>>>>>>>>> ------------------------------------------------------------ >>>>>>>>>>>>>> Sebastian Liu 刘洋 >>>>>>>>>>>>>> Institute of Computing Technology, Chinese Academy of Science >>>>>>>>>>>>>> Mobile\WeChat: +86—15201613655 >>>>>>>>>>>>>> E-mail: liuyang0...@gmail.com <liuyang0...@gmail.com> >>>>>>>>>>>>>> QQ: 3239559* >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> -- >>>>>>>>>>>>> Best, Jingsong Lee >>>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> -- >>>>>>>>>>> Best, Jingsong Lee >>>>>>>>>>> >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> -- >>>>>>>>>> Best, Jingsong Lee >>>>>>>>>> >>>>>>>>> >>>>>>>> >>>>>>>> -- >>>>>>>> Best, Jingsong Lee >>>>>>>> >>>>>>> >>>>>>> >>>>>>> -- >>>>>>> >>>>>>> *With kind regards >>>>>>> ------------------------------------------------------------ >>>>>>> Sebastian Liu 刘洋 >>>>>>> Institute of Computing Technology, Chinese Academy of Science >>>>>>> Mobile\WeChat: +86—15201613655 >>>>>>> E-mail: liuyang0...@gmail.com <liuyang0...@gmail.com> >>>>>>> QQ: 3239559* >>>>>>> >>>>>>> >>>>> >>>>> -- >>>>> >>>>> *With kind regards >>>>> ------------------------------------------------------------ >>>>> Sebastian Liu 刘洋 >>>>> Institute of Computing Technology, Chinese Academy of Science >>>>> Mobile\WeChat: +86—15201613655 >>>>> E-mail: liuyang0...@gmail.com <liuyang0...@gmail.com> >>>>> QQ: 3239559* >>>>> >>>>> >>>> >>>> -- >>>> Best, Jingsong Lee >>>> >>> >> >> -- >> >> *With kind regards >> ------------------------------------------------------------ >> Sebastian Liu 刘洋 >> Institute of Computing Technology, Chinese Academy of Science >> Mobile\WeChat: +86—15201613655 >> E-mail: liuyang0...@gmail.com <liuyang0...@gmail.com> >> QQ: 3239559* >> >> -- *With kind regards ------------------------------------------------------------ Sebastian Liu 刘洋 Institute of Computing Technology, Chinese Academy of Science Mobile\WeChat: +86—15201613655 E-mail: liuyang0...@gmail.com <liuyang0...@gmail.com> QQ: 3239559*
