Hi Fabian,
I don't fully understand the question you mentioned:
Any query that relies on the composite type with three fields will fail
after adding a forth field.
I am appreciate if you can give some detail examples ?
Regards，
JIncheng
Fabian Hueske <fhue...@gmail.com> 于2018年11月23日周五 下午4:41写道：
> Hi,
>
> My concerns are about the case when there is no additional select() method,
> i.e.,
>
> tab.window(Tumble ... as 'w)
> .groupBy('w, 'k1, 'k2)
> .flatAgg(tableAgg('a)).as('w, 'k1, 'k2, 'col1, 'col2)
>
> In this case, 'w is a composite field consisting of three fields (end,
> start, rowtime).
> Once we add a new property, it would need to be added to the composite
> type.
> Any query that relies on the composite type with three fields will fail
> after adding a forth field.
>
> Best, Fabian
>
> Am Fr., 23. Nov. 2018 um 02:01 Uhr schrieb jincheng sun <
> sunjincheng...@gmail.com>:
>
> > Thanks Fabian,
> >
> > Thanks a lot for your feedback, and very important and necessary design
> > reminders!
> >
> > Yes, your are right! Spark is the specified grouping columns displayed
> > before 1.3, but the grouping columns are implicitly passed in spark1.4
> and
> > later. The reason for changing this behavior is that due to the user
> > feedback. Although implicit delivery will have the drawbacks you
> mentioned,
> > this approach is really convenient for the user.
> > I agree that grouping on windows we have to pay attention to the handling
> > of the window's properties, because we may introduce new window property.
> > So, from the points of view, We delay the processing of the window
> > property, ie: we pass the complex type 'w on the tableAPI, and provide
> > different property method operations in the SELECT according to the type
> of
> > 'w, such as: 'w.start, 'w.end, 'w.xxx , in the TableAPI will limit and
> > verify the attribute operations that 'w has. An example is as follows:
> >
> > tab.window(Tumble ... as 'w)
> > .groupBy('w, 'k1, 'k2) // 'w should be a group key.
> > .flatAgg(tableAgg('a)).as('w, 'k1, 'k2, 'col1, 'col2) // 'w is
> > composite field
> > .select('k1, 'col1, 'w.rowtime as 'ts, 'w.xxx as 'xx) // In select we
> > will limit and verify that ’w.xx is allowed
> >
> > I am not sure if I fully understand your concerns, if there any
> understand
> > are mistakes, please correct me. Any feedback is appreciate！
> >
> > Bests,
> > Jincheng
> >
> >
> > Fabian Hueske <fhue...@gmail.com> 于2018年11月22日周四 下午10:13写道：
> >
> > > Hi all,
> > >
> > > First of all, it is correct that the flatMap(Expression*) and
> > > flatAggregate(Expression*) methods would mix scalar and table values.
> > > This would be a new concept that is not present in the current API.
> > > From my point of view, the semantics are quite clear, but I understand
> > that
> > > others are more careful and worry about future extensions.
> > >
> > > I am fine with going for single expression arguments for map() and
> > > flatMap(). We can later expand them to Expression* if we feel the need
> > and
> > > are more comfortable about the implications.
> > > Whenever, a time attribute needs to be forwarded, users can fall back
> to
> > > join(TableFunction) as Xiaowei mentioned.
> > > So we restrict the usability of the new methods but don't lose
> > > functionality and don't prevent future extensions.
> > >
> > > The aggregate() and flatAggregate() case is more difficult because
> > implicit
> > > forwarding of grouping fields cannot be changed later without breaking
> > the
> > > API.
> > > There are other APIs (e.g., Spark) that also implicitly forward the
> > > grouping columns. So this is not uncommon.
> > > However, I personally don't like that approach, because it is implicit
> > and
> > > introduces a new behavior that is not present in the current API.
> > >
> > > One thing to consider here is the handling of grouping on windows.
> > > If I understood Xiaowei correctly, a composite field that is named like
> > the
> > > window alias (e.g., 'w) would be implicitly added to the result of
> > > aggregate() or flatAggregate().
> > > The composite field would have fields like (start, end, rowtime) or
> > (start,
> > > end, proctime) depending on the window type.
> > > If we would ever introduce a fourth window property, we might break
> > > existing queries.
> > > Is this something that we should worry about?
> > >
> > > Best,
> > > Fabian
> > >
> > > Am Do., 22. Nov. 2018 um 14:03 Uhr schrieb Piotr Nowojski <
> > > pi...@data-artisans.com>:
> > >
> > > > Hi Jincheng,
> > > >
> > > > #1) ok, got it.
> > > >
> > > > #3)
> > > > > 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?
> > > >
> > > > I don’t think that’s the case here. If we start with single param
> > > > `flatMap(Expression)`, it will need implicit columns to be present in
> > the
> > > > result, which:
> > > >
> > > > a) IMO it brakes SQL convention (that’s why I’m against this)
> > > > b) we can not later easily introduce `flatMap(Expression*)` without
> > those
> > > > implicit columns, without braking the compatibility or at least
> without
> > > > making `flatMap(Expression*)` and `flatMap(Expression)` terribly
> > > > inconsistent.
> > > >
> > > > To elaborate on (a). It’s not nice if our own API is inconsistent and
> > it
> > > > sometimes behaves one way and sometimes another way:
> > > >
> > > > table.groupBy(‘k).select(scalarAggregateFunction(‘v)) => single
> column
> > > > result, just the output of `scalarAggregateFunction`
> > > > vs
> > > > table.groupBy(‘k).flatAggregate(tableAggregateFunction(‘v)) => both
> > > result
> > > > of `tableAggregateFunction` plus key (and an optional window context
> ?)
> > > >
> > > > Thus I think we have to now decide which way we want to jump, since
> > later
> > > > will be too late. Or again, am I missing something? :)
> > > >
> > > > Piotrek
> > > >
> > > > > On 22 Nov 2018, at 02:07, jincheng sun <sunjincheng...@gmail.com>
> > > wrote:
> > > > >
> > > > > 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*
> > > > >>>>>>>
> > > > >>>>>>>
> > > > >>>>>>>
> > > > >>>>>
> > > > >>
> > > >
> > >
> >
> -----------------------------------------------------------------------------------
> > > > >>>>>>>
> > > > >>>>>
> > > > >>>>>
> > > > >>
> > > > >>
> > > >
> > > >
> > > >
> > >
> >
>
