Hi Piotrek，
#1）We have unbounded and bounded group window aggregate, for unbounded case
we should early fire the result with retract message, we can not using
watermark, because unbounded aggregate never finished. (for improvement we
can introduce micro-batch in feature), for bounded window we never support
early fire, so we do not need retract.
#3) About validation of `table.select(F(‘a).unnest(), ‘b,
G(‘c).unnest())/table.flatMap(F(‘a), ‘b, scalarG(‘c))` Fabian had mentioned
above, please look at the prior mail. For `table.flatMap(F(‘a), ‘b,
scalarG(‘c))` that we concerned, i.e.: we should discuss the issue of
`Expression*` vs `Expression`. From points of my view I we can using
`Expression`, and after the discussion decided to use Expression*, then
improve it. In any case, we can use Expression, and there is an opportunity
to become Expression* (compatibility). If we use Expression* directly, it
is difficult for us to become Expression, which will break the
compatibility between versions. What do you think?
If there anything not clearly, welcome any feedback！Agains，thanks for share
your thoughts！
Thanks，
Jincheng
Piotr Nowojski <pi...@data-artisans.com> 于2018年11月21日周三 下午9:37写道：
> Hi Jincheng,
>
> > #1） No，watermark solves the issue of the late event. Here, the
> performance
> > problem is caused by the update emit mode. i.e.: When current calculation
> > result is output, the previous calculation result needs to be retracted.
>
> Hmm, yes I missed this. For time-windowed cases (some
> aggregate/flatAggregate cases) emitting only on watermark should solve the
> problem. For non time windowed cases it would reduce the amount of
> retractions, right? Or am I still missing something?
>
> > #3）I still hope to keep the simplicity that select only support projected
> > scalar, we can hardly tell the semantics of tab.select(flatmap('a), 'b,
> > flatmap('d)).
>
> table.select(F(‘a).unnest(), ‘b, G(‘c).unnest())
>
> Could be rejected during some validation phase. On the other hand:
>
> table.select(F(‘a).unnest(), ‘b, scalarG(‘c))
> or
> table.flatMap(F(‘a), ‘b, scalarG(‘c))
>
> Could work and be more or less a syntax sugar for cross apply.
>
> Piotrek
>
> > On 21 Nov 2018, at 12:16, jincheng sun <sunjincheng...@gmail.com> wrote:
> >
> > Hi shaoxuan & Hequn,
> >
> > Thanks for your suggestion，I'll file the JIRAs later.
> > We can prepare PRs while continuing to move forward the ongoing
> discussion.
> >
> > Regards，
> > Jincheng
> >
> > jincheng sun <sunjincheng...@gmail.com> 于2018年11月21日周三 下午7:07写道：
> >
> >> Hi Piotrek,
> >> Thanks for your feedback, and thanks for share your thoughts!
> >>
> >> #1） No，watermark solves the issue of the late event. Here, the
> performance
> >> problem is caused by the update emit mode. i.e.: When current
> calculation
> >> result is output, the previous calculation result needs to be retracted.
> >> #2） As I mentioned above we should continue the discussion until we
> solve
> >> the problems raised by Xiaowei and Fabian.
> >> #3）I still hope to keep the simplicity that select only support
> projected
> >> scalar, we can hardly tell the semantics of tab.select(flatmap('a), 'b,
> >> flatmap('d)).
> >>
> >> Thanks,
> >> Jincheng
> >>
> >> Piotr Nowojski <pi...@data-artisans.com> 于2018年11月21日周三 下午5:24写道：
> >>
> >>> Hi,
> >>>
> >>> 1.
> >>>
> >>>> In fact, in addition to the design of APIs, there will be various
> >>>> performance optimization details, such as: table Aggregate function
> >>>> emitValue will generate multiple calculation results, in extreme
> cases,
> >>>> each record will trigger a large number of retract messages, this will
> >>> have
> >>>> poor performance
> >>>
> >>> Can this be solved/mitigated by emitting the results only on
> watermarks?
> >>> I think that was the path that we decided to take both for Temporal
> Joins
> >>> and upsert stream conversion. I know that this increases the latency
> and
> >>> there is a place for a future global setting/user preference “emit the
> data
> >>> ASAP mode”, but emitting only on watermarks seems to me as a
> better/more
> >>> sane default.
> >>>
> >>> 2.
> >>>
> >>> With respect to the API discussion and implicit columns. The problem
> for
> >>> me so far is I’m not sure if I like the additionally complexity of
> >>> `append()` solution, while implicit columns are definitely not in the
> >>> spirit of SQL. Neither joins nor aggregations add extra unexpected
> columns
> >>> to the result without asking. This definitely can be confusing for the
> >>> users since it brakes the convention. Thus I would lean towards
> Fabian’s
> >>> proposal of multi-argument `map(Expression*)` from those 3 options.
> >>>
> >>> 3.
> >>>
> >>> Another topic is that I’m not 100% convinced that we should be adding
> new
> >>> api functions for `map`,`aggregate`,`flatMap` and `flatAggregate`. I
> think
> >>> the same could be achieved by changing
> >>>
> >>> table.map(F('x))
> >>>
> >>> into
> >>>
> >>> table.select(F('x)).unnest()
> >>> or
> >>> table.select(F('x).unnest())
> >>>
> >>> Where `unnest()` means unnest row/tuple type into a columnar table.
> >>>
> >>> table.flatMap(F('x))
> >>>
> >>> Could be on the other hand also handled by
> >>>
> >>> table.select(F('x))
> >>>
> >>> By correctly deducing that F(x) is a multi row output function
> >>>
> >>> Same might apply to `aggregate(F('x))`, but this maybe could be
> replaced
> >>> by:
> >>>
> >>> table.groupBy(…).select(F('x).unnest())
> >>>
> >>> Adding scalar functions should also be possible:
> >>>
> >>> table.groupBy('k).select(F('x).unnest(), ‘k)
> >>>
> >>> Maybe such approach would allow us to implement the same features in
> the
> >>> SQL as well?
> >>>
> >>> Piotrek
> >>>
> >>>> On 21 Nov 2018, at 09:43, Hequn Cheng <chenghe...@gmail.com> wrote:
> >>>>
> >>>> Hi,
> >>>>
> >>>> Thank you all for the great proposal and discussion!
> >>>> I also prefer to move on to the next step, so +1 for opening the JIRAs
> >>> to
> >>>> start the work.
> >>>> We can have more detailed discussion there. Btw, we can start with
> JIRAs
> >>>> which we have agreed on.
> >>>>
> >>>> Best,
> >>>> Hequn
> >>>>
> >>>> On Tue, Nov 20, 2018 at 11:38 PM Shaoxuan Wang <wshaox...@gmail.com>
> >>> wrote:
> >>>>
> >>>>> +1. I agree that we should open the JIRAs to start the work. We may
> >>>>> have better ideas on the flavor of the interface when
> implement/review
> >>>>> the code.
> >>>>>
> >>>>> Regards,
> >>>>> shaoxuan
> >>>>>
> >>>>>
> >>>>> On 11/20/18, jincheng sun <sunjincheng...@gmail.com> wrote:
> >>>>>> Hi all,
> >>>>>>
> >>>>>> Thanks all for the feedback.
> >>>>>>
> >>>>>> @Piotr About not using abbreviations naming, +1，I like
> >>>>>> your proposal！Currently both DataSet and DataStream API are using
> >>>>>> `aggregate`,
> >>>>>> BTW，I find other language also not using abbreviations naming，such
> as
> >>> R.
> >>>>>>
> >>>>>> Sometimes the interface of the API is really difficult to perfect,
> we
> >>>>> need
> >>>>>> to spend a lot of time thinking and feedback from a large number of
> >>>>> users,
> >>>>>> and constantly improve, but for backward compatibility issues, we
> >>> have to
> >>>>>> adopt the most conservative approach when designing the API(Of
> >>> course, I
> >>>>> am
> >>>>>> more in favor of developing more rich features, when we discuss
> >>> clearly).
> >>>>>> Therefore, I propose to divide the function implementation of
> >>>>>> map/faltMap/agg/flatAgg into basic functions of JIRAs and JIRAs that
> >>>>>> support time attributes and groupKeys. We can develop the features
> >>> which
> >>>>>> we have already agreed on the design. And we will continue to
> discuss
> >>>>> the
> >>>>>> uncertain design.
> >>>>>>
> >>>>>> In fact, in addition to the design of APIs, there will be various
> >>>>>> performance optimization details, such as: table Aggregate function
> >>>>>> emitValue will generate multiple calculation results, in extreme
> >>> cases,
> >>>>>> each record will trigger a large number of retract messages, this
> will
> >>>>> have
> >>>>>> poor performance，so we will also optimize the interface design, such
> >>> as
> >>>>>> adding the emitWithRetractValue interface (I have updated the google
> >>> doc)
> >>>>>> to allow the user to optionally perform incremental calculations,
> thus
> >>>>>> avoiding a large number of retracts. Details like this are difficult
> >>> to
> >>>>>> fully discuss in the mail list, so I recommend creating JIRAs/FLIP
> >>> first,
> >>>>>> we develop designs that have been agreed upon and continue to
> discuss
> >>>>>> non-deterministic designs! What do you think? @Fabian & Piotr &
> >>> XiaoWei
> >>>>>>
> >>>>>> Best,
> >>>>>> Jincheng
> >>>>>>
> >>>>>> Xiaowei Jiang <xiaow...@gmail.com> 于2018年11月19日周一 上午12:07写道：
> >>>>>>
> >>>>>>> Hi Fabian & Piotr, thanks for the feedback!
> >>>>>>>
> >>>>>>> I appreciate your concerns, both on timestamp attributes as well as
> >>> on
> >>>>>>> implicit group keys. At the same time, I'm also concerned with the
> >>>>>>> proposed
> >>>>>>> approach of allowing Expression* as parameters, especially for
> >>>>>>> flatMap/flatAgg. So far, we never allowed a scalar expression to
> >>> appear
> >>>>>>> together with table expressions. With the Expression* approach,
> this
> >>>>> will
> >>>>>>> happen for the parameters to flatMap/flatAgg. I'm a bit concerned
> on
> >>> if
> >>>>>>> we
> >>>>>>> fully understand the consequences when we try to extend our system
> in
> >>>>> the
> >>>>>>> future. I would be extra cautious in doing this. To avoid this, I
> >>> think
> >>>>>>> an
> >>>>>>> implicit group key for flatAgg is safer. For flatMap, if users want
> >>> to
> >>>>>>> keep
> >>>>>>> the rowtime column, he can use crossApply/join instead. So we are
> not
> >>>>>>> losing any real functionality here.
> >>>>>>>
> >>>>>>> Also a clarification on the following example:
> >>>>>>> tab.window(Tumble ... as 'w)
> >>>>>>> .groupBy('w, 'k1, 'k2) // 'w should be a group key.
> >>>>>>> .flatAgg(tableAgg('a)).as('w, 'k1, 'k2, 'col1, 'col2)
> >>>>>>> .select('k1, 'col1, 'w.rowtime as 'rtime)
> >>>>>>> If we did not have the select clause in this example, we will have
> >>> 'w as
> >>>>>>> a
> >>>>>>> regular column in the output. It should not magically disappear.
> >>>>>>>
> >>>>>>> The concern is not as strong for Table.map/Table.agg because we are
> >>> not
> >>>>>>> mixing scalar and table expressions. But we also want to be a bit
> >>>>>>> consistent with these methods. If we used implicit group keys for
> >>>>>>> Table.flatAgg, we probably should do the same for Table.agg. Now we
> >>> only
> >>>>>>> have to choose what to do with Table.map. I can see good arguments
> >>> from
> >>>>>>> both sides. But starting with a single Expression seems safer
> because
> >>>>>>> that
> >>>>>>> we can always extend to Expression* in the future.
> >>>>>>>
> >>>>>>> While thinking about this problem, it appears that we may need more
> >>> work
> >>>>>>> in
> >>>>>>> our handling of watermarks for SQL/Table API. Our current way of
> >>>>>>> propagating the watermarks from source all the way to sink might
> not
> >>> be
> >>>>>>> optimal. For example, after a tumbling window, the watermark can
> >>>>> actually
> >>>>>>> be advanced to just before the expiring of next window. I think
> that
> >>> in
> >>>>>>> general, each operator may need to generate new watermarks instead
> of
> >>>>>>> simply propagating them. Once we accept that watermarks may change
> >>>>> during
> >>>>>>> the execution, it appears that the timestamp columns may also
> >>> change, as
> >>>>>>> long as we have some way to associate watermark with it. My
> >>> intuition is
> >>>>>>> that once we have a through solution for the watermark issue, we
> may
> >>> be
> >>>>>>> able to solve the problem we encountered for Table.map in a cleaner
> >>> way.
> >>>>>>> But this is a complex issue which deserves a discussion on its own.
> >>>>>>>
> >>>>>>> Regards,
> >>>>>>> Xiaowei
> >>>>>>>
> >>>>>>>
> >>>>>>> On Fri, Nov 16, 2018 at 12:34 AM Piotr Nowojski <
> >>>>> pi...@data-artisans.com>
> >>>>>>> wrote:
> >>>>>>>
> >>>>>>>> Hi,
> >>>>>>>>
> >>>>>>>> Isn’t the problem of multiple expressions limited only to
> `flat***`
> >>>>>>>> functions and to be more specific only to having two (or more)
> >>>>>>>> different
> >>>>>>>> table functions passed as an expressions? `.flatAgg(TableAggA('a),
> >>>>>>>> scalarFunction1(‘b), scalarFunction2(‘c))` seems to be well
> defined
> >>>>>>>> (duplicate result of every scalar function to every record. Or am
> I
> >>>>>>> missing
> >>>>>>>> something?
> >>>>>>>>
> >>>>>>>> Another remark, I would be in favour of not using abbreviations
> and
> >>>>>>> naming
> >>>>>>>> `agg` -> `aggregate`, `flatAgg` -> `flatAggregate`.
> >>>>>>>>
> >>>>>>>> Piotrek
> >>>>>>>>
> >>>>>>>>> On 15 Nov 2018, at 14:15, Fabian Hueske <fhue...@gmail.com>
> wrote:
> >>>>>>>>>
> >>>>>>>>> Hi Jincheng,
> >>>>>>>>>
> >>>>>>>>> I said before, that I think that the append() method is better
> than
> >>>>>>>>> implicitly forwarding keys, but still, I believe it adds
> >>> unnecessary
> >>>>>>>> boiler
> >>>>>>>>> plate code.
> >>>>>>>>>
> >>>>>>>>> Moreover, I haven't seen a convincing argument why
> map(Expression*)
> >>>>>>>>> is
> >>>>>>>>> worse than map(Expression). In either case we need to do all
> kinds
> >>>>> of
> >>>>>>>>> checks to prevent invalid use of functions.
> >>>>>>>>> If the method is not correctly used, we can emit a good error
> >>>>> message
> >>>>>>> and
> >>>>>>>>> documenting map(Expression*) will be easier than
> >>>>>>>> map(append(Expression*)),
> >>>>>>>>> in my opinion.
> >>>>>>>>> I think we should not add unnessary syntax unless there is a good
> >>>>>>> reason
> >>>>>>>>> and to be honest, I haven't seen this reason yet.
> >>>>>>>>>
> >>>>>>>>> Regarding the groupBy.agg() method, I think it should behave just
> >>>>>>>>> like
> >>>>>>>> any
> >>>>>>>>> other method, i.e., not do any implicit forwarding.
> >>>>>>>>> Let's take the example of the windowed group by, that you posted
> >>>>>>> before.
> >>>>>>>>>
> >>>>>>>>> tab.window(Tumble ... as 'w)
> >>>>>>>>> .groupBy('w, 'k1, 'k2) // 'w should be a group key.
> >>>>>>>>> .agg(agg('a)).as('w, 'k1, 'k2, 'col1, 'col2)
> >>>>>>>>> .select('k1, 'col1, 'w.rowtime as 'rtime)
> >>>>>>>>>
> >>>>>>>>> What happens if 'w.rowtime is not selected? What is the data type
> >>> of
> >>>>>>> the
> >>>>>>>>> field 'w in the resulting Table? Is it a regular field at all or
> >>>>> just
> >>>>>>>>> a
> >>>>>>>>> system field that disappears if it is not selected?
> >>>>>>>>>
> >>>>>>>>> IMO, the following syntax is shorter, more explicit, and better
> >>>>>>>>> aligned
> >>>>>>>>> with the regular window.groupBy.select aggregations that are
> >>>>>>>>> supported
> >>>>>>>>> today.
> >>>>>>>>>
> >>>>>>>>> tab.window(Tumble ... as 'w)
> >>>>>>>>> .groupBy('w, 'k1, 'k2) // 'w should be a group key.
> >>>>>>>>> .agg('w.rowtime as 'rtime, 'k1, 'k2, agg('a))
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> Best, Fabian
> >>>>>>>>>
> >>>>>>>>> Am Mi., 14. Nov. 2018 um 08:37 Uhr schrieb jincheng sun <
> >>>>>>>>> sunjincheng...@gmail.com>:
> >>>>>>>>>
> >>>>>>>>>> Hi Fabian/Xiaowei,
> >>>>>>>>>>
> >>>>>>>>>> I am very sorry for my late reply! Glad to see your reply, and
> >>>>>>>>>> sounds
> >>>>>>>>>> pretty good！
> >>>>>>>>>> I agree that the approach with append() which can clearly
> defined
> >>>>>>>>>> the
> >>>>>>>>>> result schema is better which Fabian mentioned.
> >>>>>>>>>> In addition and append() and also contains non-time attributes,
> >>>>>>>>>> e.g.:
> >>>>>>>>>>
> >>>>>>>>>> tab('name, 'age, 'address, 'rowtime)
> >>>>>>>>>> tab.map(append(udf('name), 'address, 'rowtime).as('col1, 'col2,
> >>>>>>>>>> 'address, 'rowtime)
> >>>>>>>>>> .window(Tumble over 5.millis on 'rowtime as 'w)
> >>>>>>>>>> .groupBy('w, 'address)
> >>>>>>>>>>
> >>>>>>>>>> In this way the append() is very useful, and the behavior is
> very
> >>>>>>>> similar
> >>>>>>>>>> to withForwardedFields() in DataSet.
> >>>>>>>>>> So +1 to using append() approach for the map()&flatmap()!
> >>>>>>>>>>
> >>>>>>>>>> But how about the agg() and flatAgg()? In agg/flatAgg case I
> agree
> >>>>>>>>>> Xiaowei's approach that define the keys to be implied in the
> >>> result
> >>>>>>>> table
> >>>>>>>>>> and appears at the beginning, for example as follows:
> >>>>>>>>>> tab.window(Tumble ... as 'w)
> >>>>>>>>>> .groupBy('w, 'k1, 'k2) // 'w should be a group key.
> >>>>>>>>>> .agg(agg('a)).as('w, 'k1, 'k2, 'col1, 'col2)
> >>>>>>>>>> .select('k1, 'col1, 'w.rowtime as 'rtime)
> >>>>>>>>>>
> >>>>>>>>>> What to you think? @Fabian @Xiaowei
> >>>>>>>>>>
> >>>>>>>>>> Thanks,
> >>>>>>>>>> Jincheng
> >>>>>>>>>>
> >>>>>>>>>> Fabian Hueske <fhue...@gmail.com> 于2018年11月9日周五 下午6:35写道：
> >>>>>>>>>>
> >>>>>>>>>>> Hi Jincheng,
> >>>>>>>>>>>
> >>>>>>>>>>> Thanks for the summary!
> >>>>>>>>>>> I like the approach with append() better than the implicit
> >>>>>>>>>>> forwarding
> >>>>>>>> as
> >>>>>>>>>> it
> >>>>>>>>>>> clearly indicates which fields are forwarded.
> >>>>>>>>>>> However, I don't see much benefit over the flatMap(Expression*)
> >>>>>>>> variant,
> >>>>>>>>>> as
> >>>>>>>>>>> we would still need to analyze the full expression tree to
> ensure
> >>>>>>> that
> >>>>>>>> at
> >>>>>>>>>>> most (or exactly?) one Scalar / TableFunction is used.
> >>>>>>>>>>>
> >>>>>>>>>>> Best,
> >>>>>>>>>>> Fabian
> >>>>>>>>>>>
> >>>>>>>>>>> Am Do., 8. Nov. 2018 um 19:25 Uhr schrieb jincheng sun <
> >>>>>>>>>>> sunjincheng...@gmail.com>:
> >>>>>>>>>>>
> >>>>>>>>>>>> Hi all,
> >>>>>>>>>>>>
> >>>>>>>>>>>> We are discussing very detailed content about this proposal.
> We
> >>>>>>>>>>>> are
> >>>>>>>>>>> trying
> >>>>>>>>>>>> to design the API in many aspects (functionality,
> compatibility,
> >>>>>>> ease
> >>>>>>>>>> of
> >>>>>>>>>>>> use, etc.). I think this is a very good process. Only such a
> >>>>>>> detailed
> >>>>>>>>>>>> discussion, In order to develop PR more clearly and smoothly
> in
> >>>>>>>>>>>> the
> >>>>>>>>>> later
> >>>>>>>>>>>> stage. I am very grateful to @Fabian and @Xiaowei for
> sharing a
> >>>>>>>>>>>> lot
> >>>>>>>> of
> >>>>>>>>>>>> good ideas.
> >>>>>>>>>>>> About the definition of method signatures I want to share my
> >>>>>>>>>>>> points
> >>>>>>>>>> here
> >>>>>>>>>>>> which I am discussing with fabian in google doc (not yet
> >>>>>>>>>>>> completed)，
> >>>>>>>> as
> >>>>>>>>>>>> follows:
> >>>>>>>>>>>>
> >>>>>>>>>>>> Assume we have a table:
> >>>>>>>>>>>> val tab = util.addTable[(Long, String)]("MyTable", 'long,
> >>>>> 'string,
> >>>>>>>>>>>> 'proctime.proctime)
> >>>>>>>>>>>>
> >>>>>>>>>>>> Approach 1:
> >>>>>>>>>>>> case1: Map follows Source Table
> >>>>>>>>>>>> val result =
> >>>>>>>>>>>> tab.map(udf('string)).as('proctime, 'col1, 'col2)// proctime
> >>>>>>> implied
> >>>>>>>>>> in
> >>>>>>>>>>>> the output
> >>>>>>>>>>>> .window(Tumble over 5.millis on 'proctime as 'w)
> >>>>>>>>>>>>
> >>>>>>>>>>>> case2: FatAgg follows Window (Fabian mentioned above)
> >>>>>>>>>>>> val result =
> >>>>>>>>>>>> tab.window(Tumble ... as 'w)
> >>>>>>>>>>>> .groupBy('w, 'k1, 'k2) // 'w should be a group key.
> >>>>>>>>>>>> .flatAgg(tabAgg('a)).as('k1, 'k2, 'w, 'col1, 'col2)
> >>>>>>>>>>>> .select('k1, 'col1, 'w.rowtime as 'rtime)
> >>>>>>>>>>>>
> >>>>>>>>>>>> Approach 2: Similar to Fabian‘s approach, which the result
> >>> schema
> >>>>>>>> would
> >>>>>>>>>>> be
> >>>>>>>>>>>> clearly defined, but add a built-in append UDF. That make
> >>>>>>>>>>>> map/flatmap/agg/flatAgg interface only accept one Expression.
> >>>>>>>>>>>> val result =
> >>>>>>>>>>>> tab.map(append(udf('string), 'long, 'proctime)) as ('col1,
> >>>>>>>>>>>> 'col2,
> >>>>>>>>>>>> 'long, 'proctime)
> >>>>>>>>>>>> .window(Tumble over 5.millis on 'proctime as 'w)
> >>>>>>>>>>>>
> >>>>>>>>>>>> Note: Append is a special UDF for built-in that can pass
> through
> >>>>>>>>>>>> any
> >>>>>>>>>>>> column.
> >>>>>>>>>>>>
> >>>>>>>>>>>> So, May be we can defined the as table.map(Expression)
> first,
> >>>>> If
> >>>>>>>>>>>> necessary, we can extend to table.map(Expression*) in the
> >>> future
> >>>>>>>>>>>> ?
> >>>>>>>> Of
> >>>>>>>>>>>> course, I also hope that we can do more perfection in this
> >>>>>>>>>>>> proposal
> >>>>>>>>>>> through
> >>>>>>>>>>>> discussion.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Thanks,
> >>>>>>>>>>>> Jincheng
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>> Xiaowei Jiang <xiaow...@gmail.com> 于2018年11月7日周三 下午11:45写道：
> >>>>>>>>>>>>
> >>>>>>>>>>>>> Hi Fabian,
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> I think that the key question you raised is if we allow extra
> >>>>>>>>>>> parameters
> >>>>>>>>>>>> in
> >>>>>>>>>>>>> the methods map/flatMap/agg/flatAgg. I can see why allowing
> >>> that
> >>>>>>> may
> >>>>>>>>>>>> appear
> >>>>>>>>>>>>> more convenient in some cases. However, it might also cause
> >>> some
> >>>>>>>>>>>> confusions
> >>>>>>>>>>>>> if we do that. For example, do we allow multiple UDFs in
> these
> >>>>>>>>>>>> expressions?
> >>>>>>>>>>>>> If we do, the semantics may be weird to define, e.g. what
> does
> >>>>>>>>>>>>> table.groupBy('k).flatAgg(TableAggA('a), TableAggB('b)) mean?
> >>>>>>>>>>>>> Even
> >>>>>>>>>>> though
> >>>>>>>>>>>>> not allowing it may appear less powerful, but it can make
> >>> things
> >>>>>>> more
> >>>>>>>>>>>>> intuitive too. In the case of agg/flatAgg, we can define the
> >>>>> keys
> >>>>>>> to
> >>>>>>>>>> be
> >>>>>>>>>>>>> implied in the result table and appears at the beginning. You
> >>>>> can
> >>>>>>>>>> use a
> >>>>>>>>>>>>> select method if you want to modify this behavior. I think
> that
> >>>>>>>>>>>> eventually
> >>>>>>>>>>>>> we will have some API which allows other expressions as
> >>>>>>>>>>>>> additional
> >>>>>>>>>>>>> parameters, but I think it's better to do that after we
> >>>>> introduce
> >>>>>>> the
> >>>>>>>>>>>>> concept of nested tables. A lot of things we suggested here
> can
> >>>>>>>>>>>>> be
> >>>>>>>>>>>>> considered as special cases of that. But things are much
> >>> simpler
> >>>>>>>>>>>>> if
> >>>>>>>>>> we
> >>>>>>>>>>>>> leave that to later.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>> Xiaowei
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> On Wed, Nov 7, 2018 at 5:18 PM Fabian Hueske <
> >>> fhue...@gmail.com
> >>>>>>
> >>>>>>>>>>> wrote:
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>> Hi,
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> * Re emit:
> >>>>>>>>>>>>>> I think we should start with a well understood semantics of
> >>>>> full
> >>>>>>>>>>>>>> replacement. This is how the other agg functions work.
> >>>>>>>>>>>>>> As was said before, there are open questions regarding an
> >>>>> append
> >>>>>>>>>> mode
> >>>>>>>>>>>>>> (checkpointing, whether supporting retractions or not and if
> >>>>> yes
> >>>>>>>>>> how
> >>>>>>>>>>> to
> >>>>>>>>>>>>>> declare them, ...).
> >>>>>>>>>>>>>> Since this seems to be an optimization, I'd postpone it.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> * Re grouping keys:
> >>>>>>>>>>>>>> I don't think we should automatically add them because the
> >>>>>>>>>>>>>> result
> >>>>>>>>>>>> schema
> >>>>>>>>>>>>>> would not be intuitive.
> >>>>>>>>>>>>>> Would they be added at the beginning of the tuple or at the
> >>>>> end?
> >>>>>>>>>> What
> >>>>>>>>>>>>>> metadata fields of windows would be added? In which order
> >>> would
> >>>>>>>>>> they
> >>>>>>>>>>> be
> >>>>>>>>>>>>>> added?
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> However, we could support syntax like this:
> >>>>>>>>>>>>>> val t: Table = ???
> >>>>>>>>>>>>>> t
> >>>>>>>>>>>>>> .window(Tumble ... as 'w)
> >>>>>>>>>>>>>> .groupBy('a, 'b)
> >>>>>>>>>>>>>> .flatAgg('b, 'a, myAgg(row('*)), 'w.end as 'wend, 'w.rowtime
> >>>>> as
> >>>>>>>>>>>> 'rtime)
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> The result schema would be clearly defined as [b, a, f1, f2,
> >>>>>>>>>>>>>> ...,
> >>>>>>>>>> fn,
> >>>>>>>>>>>>> wend,
> >>>>>>>>>>>>>> rtime]. (f1, f2, ...fn) are the result attributes of the
> UDF.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> * Re Multi-staged evaluation:
> >>>>>>>>>>>>>> I think this should be an optimization that can be applied
> if
> >>>>>>>>>>>>>> the
> >>>>>>>>>> UDF
> >>>>>>>>>>>>>> implements the merge() method.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Best, Fabian
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Am Mi., 7. Nov. 2018 um 08:01 Uhr schrieb Shaoxuan Wang <
> >>>>>>>>>>>>>> wshaox...@gmail.com
> >>>>>>>>>>>>>>> :
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Hi xiaowei,
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Yes, I agree with you that the semantics of
> >>>>>>>>>> TableAggregateFunction
> >>>>>>>>>>>> emit
> >>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>> much more complex than AggregateFunction. The fundamental
> >>>>>>>>>>> difference
> >>>>>>>>>>>> is
> >>>>>>>>>>>>>>> that TableAggregateFunction emits a "table" while
> >>>>>>>>>> AggregateFunction
> >>>>>>>>>>>>>> outputs
> >>>>>>>>>>>>>>> (a column of) a "row". In the case of AggregateFunction it
> >>>>> only
> >>>>>>>>>> has
> >>>>>>>>>>>> one
> >>>>>>>>>>>>>>> mode which is “replacing” (complete update). But for
> >>>>>>>>>>>>>>> TableAggregateFunction, it could be incremental (only emit
> >>> the
> >>>>>>>>>> new
> >>>>>>>>>>>>>> updated
> >>>>>>>>>>>>>>> results) update or complete update (always emit the entire
> >>>>>>>>>>>>>>> table
> >>>>>>>>>>> when
> >>>>>>>>>>>>>>> “emit" is triggered). From the performance perspective, we
> >>>>>>>>>>>>>>> might
> >>>>>>>>>>>> want
> >>>>>>>>>>>>> to
> >>>>>>>>>>>>>>> use incremental update. But we need review and design this
> >>>>>>>>>>> carefully,
> >>>>>>>>>>>>>>> especially taking into account the cases of the failover
> >>>>>>>>>>>>>>> (instead
> >>>>>>>>>>> of
> >>>>>>>>>>>>> just
> >>>>>>>>>>>>>>> back-up the ACC it may also needs to remember the emit
> >>> offset)
> >>>>>>>>>> and
> >>>>>>>>>>>>>>> retractions, as the semantics of TableAggregateFunction
> emit
> >>>>>>>>>>>>>>> are
> >>>>>>>>>>>>>> different
> >>>>>>>>>>>>>>> than other UDFs. TableFunction also emits a table, but it
> >>> does
> >>>>>>>>>> not
> >>>>>>>>>>>> need
> >>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>> worry this due to the nature of stateless.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>> Shaoxuan
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> On Tue, Nov 6, 2018 at 7:16 PM Xiaowei Jiang
> >>>>>>>>>>>>>>> <xiaow...@gmail.com
> >>>>>>>>>>>
> >>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Hi Jincheng,
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Thanks for adding the public interfaces! I think that
> it's a
> >>>>>>>>>> very
> >>>>>>>>>>>>> good
> >>>>>>>>>>>>>>>> start. There are a few points that we need to have more
> >>>>>>>>>>>> discussions.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> - TableAggregateFunction - this is a very complex beast,
> >>>>>>>>>>>>> definitely
> >>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>> most complex user defined objects we introduced so far. I
> >>>>>>>>>>> think
> >>>>>>>>>>>>>> there
> >>>>>>>>>>>>>>>> are
> >>>>>>>>>>>>>>>> quite some interesting questions here. For example, do we
> >>>>>>>>>>> allow
> >>>>>>>>>>>>>>>> multi-staged TableAggregate in this case? What is the
> >>>>>>>>>>> semantics
> >>>>>>>>>>>> of
> >>>>>>>>>>>>>>>> emit? Is
> >>>>>>>>>>>>>>>> it amendments to the previous output, or replacing it? I
> >>>>>>>>>> think
> >>>>>>>>>>>>> that
> >>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>> subject itself is worth a discussion to make sure we get
> >>>>> the
> >>>>>>>>>>>>> details
> >>>>>>>>>>>>>>>> right.
> >>>>>>>>>>>>>>>> - GroupedTable.agg - does the group keys automatically
> >>>>>>>>>> appear
> >>>>>>>>>>> in
> >>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>> output? how about the case of windowing aggregation?
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>> Xiaowei
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> On Tue, Nov 6, 2018 at 6:25 PM jincheng sun <
> >>>>>>>>>>>>> sunjincheng...@gmail.com>
> >>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Hi, Xiaowei,
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Thanks for bring up the discuss of Table API Enhancement
> >>>>>>>>>>> Outline
> >>>>>>>>>>>> !
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> I quickly looked at the overall content, these are good
> >>>>>>>>>>>> expressions
> >>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>> our
> >>>>>>>>>>>>>>>>> offline discussions. But from the points of my view, we
> >>>>>>>>>> should
> >>>>>>>>>>>> add
> >>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>> usage of public interfaces that we will introduce in this
> >>>>>>>>>>>> propose.
> >>>>>>>>>>>>>>> So, I
> >>>>>>>>>>>>>>>>> added the following usage description of interface and
> >>>>>>>>>>> operators
> >>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>> google doc：
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> 1. Map Operator
> >>>>>>>>>>>>>>>>> Map operator is a new operator of Table, Map operator
> can
> >>>>>>>>>>>>> apply a
> >>>>>>>>>>>>>>>>> scalar function, and can return multi-column. The usage
> as
> >>>>>>>>>>>> follows:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> val res = tab
> >>>>>>>>>>>>>>>>> .map(fun: ScalarFunction).as(‘a, ‘b, ‘c)
> >>>>>>>>>>>>>>>>> .select(‘a, ‘c)
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> 2. FlatMap Operator
> >>>>>>>>>>>>>>>>> FaltMap operator is a new operator of Table, FlatMap
> >>>>>>>>>>> operator
> >>>>>>>>>>>>> can
> >>>>>>>>>>>>>>>> apply
> >>>>>>>>>>>>>>>>> a table function, and can return multi-row. The usage as
> >>>>>>>>>>> follows:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> val res = tab
> >>>>>>>>>>>>>>>>> .flatMap(fun: TableFunction).as(‘a, ‘b, ‘c)
> >>>>>>>>>>>>>>>>> .select(‘a, ‘c)
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> 3. Agg Operator
> >>>>>>>>>>>>>>>>> Agg operator is a new operator of Table/GroupedTable,
> Agg
> >>>>>>>>>>>>>> operator
> >>>>>>>>>>>>>>>> can
> >>>>>>>>>>>>>>>>> apply a aggregate function, and can return multi-column.
> >>> The
> >>>>>>>>>>>> usage
> >>>>>>>>>>>>> as
> >>>>>>>>>>>>>>>>> follows:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> val res = tab
> >>>>>>>>>>>>>>>>> .groupBy(‘a) // leave groupBy-Clause out to define
> >>>>>>>>>> global
> >>>>>>>>>>>>>>>> aggregates
> >>>>>>>>>>>>>>>>> .agg(fun: AggregateFunction).as(‘a, ‘b, ‘c)
> >>>>>>>>>>>>>>>>> .select(‘a, ‘c)
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> 4. FlatAgg Operator
> >>>>>>>>>>>>>>>>> FlatAgg operator is a new operator of
> Table/GroupedTable,
> >>>>>>>>>>>>> FaltAgg
> >>>>>>>>>>>>>>>>> operator can apply a table aggregate function, and can
> >>>>> return
> >>>>>>>>>>>>>>> multi-row.
> >>>>>>>>>>>>>>>>> The usage as follows:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> val res = tab
> >>>>>>>>>>>>>>>>> .groupBy(‘a) // leave groupBy-Clause out to define
> >>>>>>>>>>> global
> >>>>>>>>>>>>>> table
> >>>>>>>>>>>>>>>>> aggregates
> >>>>>>>>>>>>>>>>> .flatAgg(fun: TableAggregateFunction).as(‘a, ‘b, ‘c)
> >>>>>>>>>>>>>>>>> .select(‘a, ‘c)
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> 5. TableAggregateFunction
> >>>>>>>>>>>>>>>>> The behavior of table aggregates is most like
> >>>>>>>>>>>>>> GroupReduceFunction
> >>>>>>>>>>>>>>>> did,
> >>>>>>>>>>>>>>>>> which computed for a group of elements, and output a
> group
> >>>>>>>>>> of
> >>>>>>>>>>>>>>> elements.
> >>>>>>>>>>>>>>>>> The TableAggregateFunction can be applied on
> >>>>>>>>>>>>> GroupedTable.flatAgg() .
> >>>>>>>>>>>>>>> The
> >>>>>>>>>>>>>>>>> interface of TableAggregateFunction has a lot of content,
> >>> so
> >>>>>>>>>> I
> >>>>>>>>>>>>> don't
> >>>>>>>>>>>>>>> copy
> >>>>>>>>>>>>>>>>> it here， Please look at the detail in google doc:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>
> >>>
> https://docs.google.com/document/d/19rVeyqveGtV33UZt72GV-DP2rLyNlfs0QNGG0xWjayY/edit
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> I will be very appreciate to anyone for reviewing and
> >>>>>>>>>>> commenting.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Best,
> >>>>>>>>>>>>>>>>> Jincheng
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>
> >>>>>
> >>>>>
> >>>>> --
> >>>>>
> >>>>>
> >>>
> -----------------------------------------------------------------------------------
> >>>>>
> >>>>> *Rome was not built in one day*
> >>>>>
> >>>>>
> >>>>>
> >>>
> -----------------------------------------------------------------------------------
> >>>>>
> >>>
> >>>
>
>
