Thanks for that feedback Lincoln, Only one question with the async `timeout` parameter[1](since I > haven't seen the POC code), current description is: 'The time which can > pass before a restart strategy is triggered', > but in the previous flip-232[2] and flip-234[3], in retry scenario, this > timeout is the total time, do we keep the behavior of the parameter > consistent?
That's a good catch. I was intending to use *AsyncWaitOperator*, and to pass this timeout directly. Looking through the code a bit, it appears that it doesn't restart the timer on a retry, and this timeout is total, as you're saying. I do intend on being consistent with the other FLIPs and retaining this behavior, so I will update the wording on my FLIP to reflect that. -Alan On Wed, Dec 20, 2023 at 1:36 AM Lincoln Lee <lincoln.8...@gmail.com> wrote: > +1 for this useful feature! > Hope this reply isn't too late. Agree that we start with global > async-scalar configuration and ordered mode first. > > @Alan Only one question with the async `timeout` parameter[1](since I > haven't seen the POC code), current description is: 'The time which can > pass before a restart strategy is triggered', > but in the previous flip-232[2] and flip-234[3], in retry scenario, this > timeout is the total time, do we keep the behavior of the parameter > consistent? > > [1] > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-400%3A+AsyncScalarFunction+for+asynchronous+scalar+function+support > [2] > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=211883963 > [3] > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-234%3A+Support+Retryable+Lookup+Join+To+Solve+Delayed+Updates+Issue+In+External+Systems > > Best, > Lincoln Lee > > > Alan Sheinberg <asheinb...@confluent.io.invalid> 于2023年12月20日周三 08:41写道： > > > Thanks for the comments Timo. > > > > > > > Can you remove the necessary parts? Esp.: > > > > @Override > > > public Set<FunctionRequirement> getRequirements() { > > > return Collections.singleton(FunctionRequirement.ORDERED); > > > } > > > > > > I removed this section from the FLIP since presumably, there's no use in > > adding to the public API if it's ignored, with handling just ORDERED for > > the first version. I'm not sure how quickly I'll want to add UNORDERED > > support, but I guess I can always do another FLIP. > > > > Otherwise I have no objections to start a VOTE soonish. If others are > > > fine as well? > > > > That would be great. Any areas that people are interested in discussing > > further before a vote? > > > > -Alan > > > > On Tue, Dec 19, 2023 at 5:49 AM Timo Walther <twal...@apache.org> wrote: > > > > > > I would be totally fine with the first version only having ORDERED > > > > mode. For a v2, we could attempt to do the next most conservative > > > > thing > > > > > > Sounds good to me. > > > > > > I also cheked AsyncWaitOperator and could not find n access of > > > StreamRecord's timestamp but only watermarks. But as we said, let's > > > focus on ORDERED first. > > > > > > Can you remove the necessary parts? Esp.: > > > > > > @Override > > > public Set<FunctionRequirement> getRequirements() { > > > return Collections.singleton(FunctionRequirement.ORDERED); > > > } > > > > > > Otherwise I have no objections to start a VOTE soonish. If others are > > > fine as well? > > > > > > Regards, > > > Timo > > > > > > > > > On 19.12.23 07:32, Alan Sheinberg wrote: > > > > Thanks for the helpful comments, Xuyang and Timo. > > > > > > > > @Timo, @Alan: IIUC, there seems to be something wrong here. Take > kafka > > as > > > >> source and mysql as sink as an example. > > > >> Although kafka is an append-only source, one of its fields is used > as > > pk > > > >> when writing to mysql. If async udx is executed > > > >> in an unordered mode, there may be problems with the data in mysql > > in > > > the > > > >> end. In this case, we need to ensure that > > > >> the sink-based pk is in order actually. > > > > > > > > > > > > @Xuyang: That's a great point. If some node downstream of my > operator > > > > cares about ordering, there's no way for it to reconstruct the > original > > > > ordering of the rows as they were input to my operator. So even if > > they > > > > want to preserve ordering by key, the order in which they see it may > > > > already be incorrect. Somehow I thought that maybe the analysis of > the > > > > changelog mode at a given operator was aware of downstream > operations, > > > but > > > > it seems not. > > > > > > > > Clear "no" on this. Changelog semantics make the planner complex and > we > > > >> need to be careful. Therefore I would strongly suggest we introduce > > > >> ORDERED and slowly enable UNORDERED whenever we see a good fit for > it > > in > > > >> plans with appropriate planner rules that guard it. > > > > > > > > > > > > @Timo: The better I understand the complexity, the more I agree with > > > this. > > > > I would be totally fine with the first version only having ORDERED > > mode. > > > > For a v2, we could attempt to do the next most conservative thing and > > > only > > > > allow UNORDERED when the whole graph is in *INSERT *changelog mode. > > The > > > > next best type of optimization might understand what's the key > required > > > > downstream, and allow breaking the original order only between > > unrelated > > > > keys, but maintaining it between rows of the same key. Of course if > > the > > > > key used downstream is computed in some manner, that makes it all the > > > > harder to know this beforehand. > > > > > > > > So unordering should be fine *within* watermarks. This is also what > > > >> watermarks are good for, a trade-off between strict ordering and > > making > > > >> progress. The async operator from DataStream API also supports this > > if I > > > >> remember correctly. However, it assumes a timestamp is present in > > > >> StreamRecord on which it can work. But this is not the case within > the > > > >> SQL engine. > > > > > > > > > > > > *AsyncWaitOperator* and *UnorderedStreamElementQueue* (the > > > implementations > > > > I plan on using) seem to support exactly this behavior. I don't > think > > it > > > > makes assumptions about the record's timestamp, but just preserves > > > whatever > > > > the input order is w.r.t watermarks. I'd be curious to understand > the > > > > timestamp use in more detail and see if it's required with the > > mentioned > > > > classes. > > > > > > > > TLDR: Let's focus on ORDERED first. > > > > > > > > > > > > I'm more than happy to start here and we can consider UNORDERED as a > > > > followup. Then maybe we consider only INSERT mode graphs and ones > > where > > > we > > > > can solve the watermark constraints. > > > > > > > > Thanks, > > > > Alan > > > > > > > > > > > > On Mon, Dec 18, 2023 at 2:36 AM Timo Walther <twal...@apache.org> > > wrote: > > > > > > > >> Hi Xuyang and Alan, > > > >> > > > >> thanks for this productive discussion. > > > >> > > > >> > Would it make a difference if it were exposed by the explain > > > >> > > > >> @Alan: I think this is great idea. +1 on exposing the sync/async > > > >> behavior thought EXPLAIN. > > > >> > > > >> > > > >> > Is there an easy way to determine if the output of an async > > function > > > >> > would be problematic or not? > > > >> > > > >> Clear "no" on this. Changelog semantics make the planner complex and > > we > > > >> need to be careful. Therefore I would strongly suggest we introduce > > > >> ORDERED and slowly enable UNORDERED whenever we see a good fit for > it > > in > > > >> plans with appropriate planner rules that guard it. > > > >> > > > >> > If the input to the operator is append-only, it seems fine, > > because > > > >> > this implies that each row is effectively independent and > ordering > > > is > > > >> > unimportant. > > > >> > > > >> As @Xuyang pointed out, it's not only the input that decides whether > > > >> append-only is safe. It's also the subsequent operators in the > > pipeline. > > > >> The example of Xuyang is a good one, when the sink operates in > upsert > > > >> mode. Append-only source, append-only operators, and append-only > sink > > > >> are safer. > > > >> > > > >> However, even in this combination, a row is not fully "independent" > > > >> there are still watermarks flowing between rows: > > > >> > > > >> R(5), W(4), R(3), R(4), R(2), R(1), W(0) > > > >> > > > >> So unordering should be fine *within* watermarks. This is also what > > > >> watermarks are good for, a trade-off between strict ordering and > > making > > > >> progress. The async operator from DataStream API also supports this > > if I > > > >> remember correctly. However, it assumes a timestamp is present in > > > >> StreamRecord on which it can work. But this is not the case within > the > > > >> SQL engine. > > > >> > > > >> TLDR: Let's focus on ORDERED first. > > > >> > > > >> If we want to use UNORDERED, I would suggest to check the input > > operator > > > >> for exactly 1 time attribute column. If there is exactly 1 time > > > >> attribute column, we could insert it into the StreamRecord and allow > > > >> UNORDERED mode. If this condition is not met, we go with ORDERED. > > > >> > > > >> Regards, > > > >> Timo > > > >> > > > >> > > > >> > > > >> > > > >> On 18.12.23 07:05, Xuyang wrote: > > > >>> Hi, Alan and Timo. Thanks for your reply. > > > >>>> Would it make a difference if it were exposed by the explain > > > >>>> method (the operator having "syncMode" vs not)? > > > >>> @Alan: I think this is a good way to tell the user what mode these > > > async > > > >> udx are currently in. > > > >>>> A regular SQL user doesn't care whether the function is sync or > > async. > > > >>> @Timo: I agree that the planner should throw as few exceptions as > > > >> possible rather than confusing users. So I think > > > >>> it is a good way to expose syncMode through explain syntax. > > > >>>> If the input to the operator is append-only, it seems fine, > > > >>>> because this implies that each row is effectively independent and > > > >> ordering is unimportant. > > > >>> > > > >>> > > > >>>> For example, if the query is > an append-only `SELECT FUNC(c) FROM > > t`, > > > >>>> I don't see a reason why the > operator is not allowed to produce > > > >> unordered results. > > > >>> > > > >>> > > > >>> @Timo, @Alan: IIUC, there seems to be something wrong here. Take > > kafka > > > >> as source and mysql as sink as an example. > > > >>> Although kafka is an append-only source, one of its fields is used > as > > > pk > > > >> when writing to mysql. If async udx is executed > > > >>> in an unordered mode, there may be problems with the data in > mysql > > > in > > > >> the end. In this case, we need to ensure that > > > >>> the sink-based pk is in order actually. > > > >>> > > > >>> > > > >>> > > > >>> -- > > > >>> > > > >>> Best！ > > > >>> Xuyang > > > >>> > > > >>> > > > >>> > > > >>> > > > >>> > > > >>> At 2023-12-16 03:33:55, "Alan Sheinberg" <asheinb...@confluent.io > > > .INVALID> > > > >> wrote: > > > >>>> Thanks for the replies everyone. My responses are inline: > > > >>>> > > > >>>> About the configs, what do you think using hints as mentioned in > > [1]. > > > >>>> > > > >>>> @Aitozi: I think hints could be a good way to do this, similar to > > > lookup > > > >>>> joins or the proposal in FLIP-313. One benefit of hints is that > > they > > > >> allow > > > >>>> for the highest granularity of configuration because you can > decide > > at > > > >>>> each and every call site just what parameters to use. The > downside > > of > > > >>>> hints is that there's more syntax to learn and more verbosity. > I'm > > > >>>> somewhat partial to a configuration like this with a class > > definition > > > >> level > > > >>>> of granularity (similar to how metrics reporters are defined [1]): > > > >>>> > > > >>>> table.exec.async-scalar.myfunc.class: > > > >> org.apache.flink.MyAsyncScalarFunction > > > >>>> table.exec.async-scalar.myfunc.buffer-capacity: 10 > > > >>>> ... > > > >>>> > > > >>>> As Timo mentioned, the downside to this is that there's not a nice > > > >> static > > > >>>> way to do this at the moment, unless you extend ConfigOption. It > > > would > > > >> be > > > >>>> good ultimately if Lookup joins, async scalar functions, and other > > > >> future > > > >>>> configurable UDFs shared the same methodology, but maybe a unified > > > >> approach > > > >>>> is a followup discussion. > > > >>>> > > > >>>> I’m just curious why you don’t use conf(global) and query > > > >> hint(individual > > > >>>>> async udx) to mark the output > > > >>>>> mode 'order' or 'unorder' like async look join [1] and async > > udtf[2], > > > >> but > > > >>>>> chose to introduce a new enum > > > >>>>> in AsyncScalarFunction. > > > >>>> > > > >>>> > > > >>>> @Xuyang: I'm open to adding hints. I think the important part is > > that > > > we > > > >>>> have some method for the user to have a class definition level way > > to > > > >>>> define whether ORDERED or ALLOW_UNORDERED is most appropriate. I > > > don't > > > >>>> have a strong sense yet for what would be most appropriately > exposed > > > as > > > >> a > > > >>>> FunctionRequirement vs a simple configuration/hint. > > > >>>> > > > >>>> What about throwing an exception to make it clear to users that > > using > > > >> async > > > >>>>> scalar functions in this situation > > > >>>>> is problematic instead of executing silently in sync mode? > Because > > > >> users > > > >>>>> may be confused about > > > >>>>> the final actual job graph. > > > >>>> > > > >>>> > > > >>>> @Xuyang: Would it make a difference if it were exposed by the > > explain > > > >>>> method (the operator having "syncMode" vs not)? I'd be fine to do > > it > > > >>>> either way -- certainly throwing an error is a bit simpler. > > > >>>> > > > >>>> You are right. Actually it should be the planner that fully > decides > > > >>>>> whether ORDERED or UNORDERED is safe to do. For example, if the > > query > > > >> is > > > >>>>> an append-only `SELECT FUNC(c) FROM t`, I don't see a reason why > > the > > > >>>>> operator is not allowed to produce unordered results. By global > > > >>>>> configuration, we can set ORDERED such that users don't get > > confused > > > >>>>> about the unordered output. > > > >>>> > > > >>>> > > > >>>> @Timo: Is there an easy way to determine if the output of an async > > > >> function > > > >>>> would be problematic or not? If the input to the operator is > > > >> append-only, > > > >>>> it seems fine, because this implies that each row is effectively > > > >>>> independent and ordering is unimportant. For upsert mode with +U > > rows, > > > >> you > > > >>>> wouldn't want to swap order with other +U rows for the same key > > > because > > > >> the > > > >>>> last one should win. For -D or -U rows, you wouldn't want to swap > > > with > > > >>>> other rows for the same key for similar reasons. Is it as simple > as > > > >>>> looking for the changlelog mode to determine whether it's safe to > > run > > > >> async > > > >>>> functions UNORDERED? I had considered analyzing various query > forms > > > >> (join > > > >>>> vs aggregation vs whatever), but it seems like changelog mode > could > > be > > > >>>> sufficient to understand what works and what would be an issue. > Any > > > >> code > > > >>>> pointers and explanation for similar analysis would be great to > > > >> understand > > > >>>> this more. > > > >>>> > > > >>>> The mode UNORDERED however should only have > > > >>>>> effect for these simply use cases and throw an exception if > > UNORDERED > > > >>>>> would mess up a changelog or other subsequent operators. > > > >>>> > > > >>>> @Timo: Should we throw errors or run in sync mode? It seems like > > > >> running > > > >>>> in sync mode is an option to ensure correctness in all changelog > > > modes. > > > >>>> > > > >>>> Let's go with global configuration first and later introduce > > > >>>>> hints. I feel the more hints we introduce, the harder SQL queries > > get > > > >>>>> when maintaining them. > > > >>>> > > > >>>> @Timo: That seems like a reasonable approach to me. > > > >>>> > > > >>>> -Alan > > > >>>> > > > >>>> [1] > > > >>>> > > > >> > > > > > > https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/deployment/metric_reporters/ > > > >>>> > > > >>>> On Fri, Dec 15, 2023 at 2:56 AM Timo Walther <twal...@apache.org> > > > >> wrote: > > > >>>> > > > >>>>> 1. Override the function `getRequirements` in > `AsyncScalarFunction` > > > >>>>> > > > >>>>> > If the user overrides `requirements()` to omit the `ORDERED` > > > >>>>> > requirement, do we allow the operator to return out-of-order > > > >> results > > > >>>>> > or should it fall back on `AsyncOutputMode.ALLOW_UNORDERED` > > type > > > >>>>> > behavior (where we allow out-of-order only if it's deemed > > > correct)? > > > >>>>> > > > >>>>> You are right. Actually it should be the planner that fully > decides > > > >>>>> whether ORDERED or UNORDERED is safe to do. For example, if the > > query > > > >> is > > > >>>>> an append-only `SELECT FUNC(c) FROM t`, I don't see a reason why > > the > > > >>>>> operator is not allowed to produce unordered results. By global > > > >>>>> configuration, we can set ORDERED such that users don't get > > confused > > > >>>>> about the unordered output. The mode UNORDERED however should > only > > > have > > > >>>>> effect for these simply use cases and throw an exception if > > UNORDERED > > > >>>>> would mess up a changelog or other subsequent operators. > > > >>>>> > > > >>>>> 2. In some scenarios with semantic correctness, async operators > > must > > > be > > > >>>>> executed in sync mode. > > > >>>>> > > > >>>>> > What about throwing an exception to make it clear to users > > that > > > >> using > > > >>>>> async scalar functions > > > >>>>> > > > >>>>> @Xuyang: A regular SQL user doesn't care whether the function is > > sync > > > >> or > > > >>>>> async. The planner should simply give its best to make the > > execution > > > >>>>> performant. I would not throw an exception here. There more > > > exceptions > > > >>>>> the, the more struggles and questions from the user. > Conceptually, > > we > > > >>>>> can run async code also sync, and that's why we should also do it > > to > > > >>>>> avoid errors. > > > >>>>> > > > >>>>> 3. Hints > > > >>>>> > > > >>>>> @Aitozi: Let's go with global configuration first and later > > introduce > > > >>>>> hints. I feel the more hints we introduce, the harder SQL queries > > get > > > >>>>> when maintaining them. > > > >>>>> > > > >>>>> Regards, > > > >>>>> Timo > > > >>>>> > > > >>>>> > > > >>>>> > > > >>>>> > > > >>>>> On 15.12.23 04:51, Xuyang wrote: > > > >>>>>> Hi, Alan. Thanks for driving this. > > > >>>>>> > > > >>>>>> > > > >>>>>> Using async to improve throughput has been done on look join, > and > > > the > > > >>>>> improvement > > > >>>>>> effect is obvious, so I think it makes sense to support async > > scalar > > > >>>>> function. Big +1 for this flip. > > > >>>>>> I have some questions below. > > > >>>>>> > > > >>>>>> > > > >>>>>> 1. Override the function `getRequirements` in > > `AsyncScalarFunction` > > > >>>>>> > > > >>>>>> > > > >>>>>> I’m just curious why you don’t use conf(global) and query > > > >>>>> hint(individual async udx) to mark the output > > > >>>>>> mode 'order' or 'unorder' like async look join [1] and async > > > udtf[2], > > > >>>>> but chose to introduce a new enum > > > >>>>>> in AsyncScalarFunction. > > > >>>>>> > > > >>>>>> > > > >>>>>> 2. In some scenarios with semantic correctness, async operators > > must > > > >> be > > > >>>>> executed in sync mode. > > > >>>>>> > > > >>>>>> > > > >>>>>> What about throwing an exception to make it clear to users that > > > using > > > >>>>> async scalar functions in this situation > > > >>>>>> is problematic instead of executing silently in sync mode? > Because > > > >> users > > > >>>>> may be confused about > > > >>>>>> the final actual job graph. > > > >>>>>> > > > >>>>>> > > > >>>>>> [1] > > > >>>>> > > > >> > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-234%3A+Support+Retryable+Lookup+Join+To+Solve+Delayed+Updates+Issue+In+External+Systems > > > >>>>>> [2] > > > >>>>> > > > >> > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-313%3A+Add+support+of+User+Defined+AsyncTableFunction > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> -- > > > >>>>>> > > > >>>>>> Best！ > > > >>>>>> Xuyang > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> > > > >>>>>> 在 2023-12-15 11:20:24，"Aitozi" <gjying1...@gmail.com> 写道： > > > >>>>>>> Hi Alan, > > > >>>>>>> Nice FLIP, I also explore leveraging the async table > > > >> function[1] to > > > >>>>>>> improve the throughput before. > > > >>>>>>> > > > >>>>>>> About the configs, what do you think using hints as mentioned > in > > > [1]. > > > >>>>>>> > > > >>>>>>> [1]: > > > >>>>>>> > > > >>>>> > > > >> > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-313%3A+Add+support+of+User+Defined+AsyncTableFunction > > > >>>>>>> > > > >>>>>>> Thanks, > > > >>>>>>> Aitozi. > > > >>>>>>> > > > >>>>>>> Timo Walther <twal...@apache.org> 于2023年12月14日周四 17:29写道： > > > >>>>>>> > > > >>>>>>>> Hi Alan, > > > >>>>>>>> > > > >>>>>>>> thanks for proposing this FLIP. It's a great addition to Flink > > and > > > >> has > > > >>>>>>>> been requested multiple times. It will be in particular > > > interesting > > > >> for > > > >>>>>>>> accessing REST endpoints and other remote services. > > > >>>>>>>> > > > >>>>>>>> Great that we can generalize and reuse parts of the Python > > planner > > > >>>>> rules > > > >>>>>>>> and code for this. > > > >>>>>>>> > > > >>>>>>>> I have some feedback regarding the API: > > > >>>>>>>> > > > >>>>>>>> 1) Configuration > > > >>>>>>>> > > > >>>>>>>> Configuration keys like > > > >>>>>>>> > > > >>>>>>>> `table.exec.async-scalar.catalog.db.func-name.buffer-capacity` > > > >>>>>>>> > > > >>>>>>>> are currently not supported in the configuration stack. The > key > > > >> space > > > >>>>>>>> should remain constant. Only a constant key space enables the > > use > > > of > > > >>>>> the > > > >>>>>>>> ConfigOption class which is required in the layered > > configuration. > > > >> For > > > >>>>>>>> now I would suggest to only allow a global setting for buffer > > > >> capacity, > > > >>>>>>>> timeout, and retry-strategy. We can later work on a > per-function > > > >>>>>>>> configuration (potentially also needed for other use cases). > > > >>>>>>>> > > > >>>>>>>> 2) Semantical declaration > > > >>>>>>>> > > > >>>>>>>> Regarding > > > >>>>>>>> > > > >>>>>>>> `table.exec.async-scalar.catalog.db.func-name.output-mode` > > > >>>>>>>> > > > >>>>>>>> this is a semantical property of a function and should be > > defined > > > >>>>>>>> per-function. It impacts the query result and potentially the > > > >> behavior > > > >>>>>>>> of planner rules. > > > >>>>>>>> > > > >>>>>>>> I see two options for this either: (a) an additional method in > > > >>>>>>>> AsyncScalarFunction or (b) adding this to the function's > > > >> requirements. > > > >>>>> I > > > >>>>>>>> vote for (b), because a FunctionDefinition should be fully > self > > > >>>>>>>> contained and sufficient for planning. > > > >>>>>>>> > > > >>>>>>>> Thus, for `FunctionDefinition.getRequirements(): > > > >>>>>>>> Set<FunctionRequirement>` we can add a new requirement > `ORDERED` > > > >> which > > > >>>>>>>> should also be the default for AsyncScalarFunction. > > > >> `getRequirements()` > > > >>>>>>>> can be overwritten and return a set without this requirement > if > > > the > > > >>>>> user > > > >>>>>>>> intents to do this. > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> Thanks, > > > >>>>>>>> Timo > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> On 11.12.23 18:43, Piotr Nowojski wrote: > > > >>>>>>>>> +1 to the idea, I don't have any comments. > > > >>>>>>>>> > > > >>>>>>>>> Best, > > > >>>>>>>>> Piotrek > > > >>>>>>>>> > > > >>>>>>>>> czw., 7 gru 2023 o 07:15 Alan Sheinberg < > > asheinb...@confluent.io > > > >>>>>>>> .invalid> > > > >>>>>>>>> napisał(a): > > > >>>>>>>>> > > > >>>>>>>>>>> > > > >>>>>>>>>>> Nicely written and makes sense. The only feedback I have > is > > > >> around > > > >>>>> the > > > >>>>>>>>>>> naming of the generalization, e.g. "Specifically, > > > >>>>>>>> PythonCalcSplitRuleBase > > > >>>>>>>>>>> will be generalized into RemoteCalcSplitRuleBase." This > > naming > > > >>>>> seems > > > >>>>>>>> to > > > >>>>>>>>>>> imply/suggest that all Async functions are remote. I > wonder > > if > > > >> we > > > >>>>> can > > > >>>>>>>>>> find > > > >>>>>>>>>>> another name which doesn't carry that connotation; maybe > > > >>>>>>>>>>> AsyncCalcSplitRuleBase. (An AsyncCalcSplitRuleBase which > > > handles > > > >>>>>>>> Python > > > >>>>>>>>>>> and Async functions seems reasonable.) > > > >>>>>>>>>>> > > > >>>>>>>>>> Thanks. That's fair. I agree that "Remote" isn't always > > > >> accurate. > > > >>>>> I > > > >>>>>>>>>> believe that the python calls are also done asynchronously, > so > > > >> that > > > >>>>>>>> might > > > >>>>>>>>>> be a reasonable name, so long as there's no confusion > between > > > the > > > >>>>> base > > > >>>>>>>> and > > > >>>>>>>>>> async child class. > > > >>>>>>>>>> > > > >>>>>>>>>> On Wed, Dec 6, 2023 at 3:48 PM Jim Hughes > > > >>>>> <jhug...@confluent.io.invalid > > > >>>>>>>>> > > > >>>>>>>>>> wrote: > > > >>>>>>>>>> > > > >>>>>>>>>>> Hi Alan, > > > >>>>>>>>>>> > > > >>>>>>>>>>> Nicely written and makes sense. The only feedback I have > is > > > >> around > > > >>>>> the > > > >>>>>>>>>>> naming of the generalization, e.g. "Specifically, > > > >>>>>>>> PythonCalcSplitRuleBase > > > >>>>>>>>>>> will be generalized into RemoteCalcSplitRuleBase." This > > naming > > > >>>>> seems > > > >>>>>>>> to > > > >>>>>>>>>>> imply/suggest that all Async functions are remote. I > wonder > > if > > > >> we > > > >>>>> can > > > >>>>>>>>>> find > > > >>>>>>>>>>> another name which doesn't carry that connotation; maybe > > > >>>>>>>>>>> AsyncCalcSplitRuleBase. (An AsyncCalcSplitRuleBase which > > > handles > > > >>>>>>>> Python > > > >>>>>>>>>>> and Async functions seems reasonable.) > > > >>>>>>>>>>> > > > >>>>>>>>>>> Cheers, > > > >>>>>>>>>>> > > > >>>>>>>>>>> Jim > > > >>>>>>>>>>> > > > >>>>>>>>>>> On Wed, Dec 6, 2023 at 5:45 PM Alan Sheinberg > > > >>>>>>>>>>> <asheinb...@confluent.io.invalid> wrote: > > > >>>>>>>>>>> > > > >>>>>>>>>>>> I'd like to start a discussion of FLIP-400: > > > AsyncScalarFunction > > > >> for > > > >>>>>>>>>>>> asynchronous scalar function support [1] > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> This feature proposes adding a new UDF type > > > AsyncScalarFunction > > > >>>>> which > > > >>>>>>>>>> is > > > >>>>>>>>>>>> invoked just like a normal ScalarFunction, but is > > implemented > > > >> with > > > >>>>> an > > > >>>>>>>>>>>> asynchronous eval method. I had brought this up including > > the > > > >>>>>>>>>> motivation > > > >>>>>>>>>>>> in a previous discussion thread [2]. > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> The purpose is to achieve high throughput scalar function > > UDFs > > > >>>>> while > > > >>>>>>>>>>>> allowing that an individual call may have high latency. > It > > > >> allows > > > >>>>>>>>>>> scaling > > > >>>>>>>>>>>> up the parallelism of just these calls without having to > > > >> increase > > > >>>>> the > > > >>>>>>>>>>>> parallelism of the whole query (which could be rather > > resource > > > >>>>>>>>>>>> inefficient). > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> In practice, it should enable SQL integration with > external > > > >>>>> services > > > >>>>>>>>>> and > > > >>>>>>>>>>>> systems, which Flink has limited support for at the > moment. > > It > > > >>>>> should > > > >>>>>>>>>>> also > > > >>>>>>>>>>>> allow easier integration with existing libraries which use > > > >>>>>>>> asynchronous > > > >>>>>>>>>>>> APIs. > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> Looking forward to your feedback and suggestions. > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> [1] > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> > > > >>>>>>>>>>> > > > >>>>>>>>>> > > > >>>>>>>> > > > >>>>> > > > >> > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-400%3A+AsyncScalarFunction+for+asynchronous+scalar+function+support > > > >>>>>>>>>>>> < > > > >>>>>>>>>>>> > > > >>>>>>>>>>> > > > >>>>>>>>>> > > > >>>>>>>> > > > >>>>> > > > >> > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-400%3A+AsyncScalarFunction+for+asynchronous+scalar+function+support > > > >>>>>>>>>>>>> > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> [2] > > > >>>>> https://lists.apache.org/thread/bn153gmcobr41x2nwgodvmltlk810hzs > > > >>>>>>>>>>>> < > > > >> https://lists.apache.org/thread/bn153gmcobr41x2nwgodvmltlk810hzs> > > > >>>>>>>>>>>> > > > >>>>>>>>>>>> Thanks, > > > >>>>>>>>>>>> Alan > > > >>>>>>>>>>>> > > > >>>>>>>>>>> > > > >>>>>>>>>> > > > >>>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>> > > > >>>>> > > > >> > > > >> > > > > > > > > > > > > >
