You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by weijie guo <gu...@gmail.com> on 2024/06/07 01:59:30 UTC
[DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Hi devs,
I'd like to start a discussion about FLIP-462[1]: Support Custom Data
Distribution for Input Stream of Lookup Join.
Lookup Join is an important feature in Flink, It is typically used to
enrich a table with data that is queried from an external system.
If we interact with the external systems for each incoming record, we
incur significant network IO and RPC overhead.
Therefore, most connectors introduce caching to reduce the per-record
level query overhead. However, because the data distribution of Lookup
Join's input stream is arbitrary, the cache hit rate is sometimes
unsatisfactory.
We want to introduce a mechanism for the connector to tell the Flink
planner its desired input stream data distribution or partitioning
strategy. This can significantly reduce the amount of cached data and
improve performance of Lookup Join.
You can find more details in this FLIP[1]. Looking forward to hearing
from you, thanks!
Best regards,
Weijie
[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
Re: Re: Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by weijie guo <gu...@gmail.com>.
Hi all,
Thanks for all the feedback and suggestions so far.
If there is no further comment, we will open the voting thread next monday.
Best regards,
Weijie
weijie guo <gu...@gmail.com> 于2024年6月14日周五 15:49写道：
> Thanks Lincoln for the quick response.
>
> > Since we've decided to extend a new hint option 'shuffle' to the current
> `LOOKUP` join hint, do we support hash shuffle as well?(It seems like it
> shouldn't require a lot of extra work, right?) This will deliver a
> complete new feature to users, also because
> FLIP-204 is stale for now and this new extension will give user a more
> simpler way to achieve the goal, WDYT?
>
> Yes, I think this makes more sense.
>
> In a word: If the target dim table does not
> implement SupportsLookupCustomShuffle, the planner will try best to apply
> customer partitioning for the input stream. Otherwise, the planner will try
> best to apply a hash partitioning.
>
> As for FLIP-204, I think we can discuss whether it should be discarded or
> refactored in a separate thread. TBH, I think the current approach can
> completely cover it and be much easier to use.
> > "upsert mode" should be "updating stream" or "non-insert-only stream".
>
> Thanks, updated the FLIP.
>
>
>
> Best regards,
>
> Weijie
>
>
> Lincoln Lee <li...@gmail.com> 于2024年6月13日周四 23:08写道：
>
>> Thanks Weijie & Wencong for your update including the conclusions of
>> the offline discussion.
>>
>> There's one thing need to be confirmed in the FLIP:
>> > The hint only provides a suggestion to the optimizer, it is not an
>> enforcer. As a result, If the target dim table not implements
>> SupportsLookupCustomShuffle, planner will ignore this newly introduced
>> shuffle option.
>>
>> Since we've decided to extend a new hint option 'shuffle' to the current
>> `LOOKUP` join hint, do we support hash shuffle as well?(It seems like it
>> shouldn't require a lot of extra work, right?)
>> This will deliver a complete new feature to users, also because
>> FLIP-204 is stale for now and this new extension will give user a more
>> simpler way to achieve the goal, WDYT?
>>
>> Another small comment for the new interface:
>> > "... planner may not apply this partitioner in upsert mode ..."
>> > default boolean isDeterministic()
>> "upsert mode" should be "updating stream" or "non-insert-only stream".
>>
>>
>> Best,
>> Lincoln Lee
>>
>>
>> Wencong Liu <li...@163.com> 于2024年6月12日周三 21:43写道：
>>
>> > Hi Jingsong,
>> >
>> >
>> > Some of the points you mentioned are currently clarified in
>> > the updated FLIP. Please check it out.
>> >
>> >
>> > 1. Enabling custom data distribution can be done through the
>> > LOOKUP SQL Hint. There are detailed examples provided in the FLIP.
>> >
>> >
>> > 2. We will add the isDeterministic method to the `InputDataPartitioner`
>> > interface, which will return true by default. If the
>> > `InputDataPartitioner`
>> > is not deterministic, the connector developer need to override the
>> > isDeterministic method to return false. If the connector developer
>> > cannot ensure this protocol, they will need to bear the correctness
>> > issues that arise.
>> >
>> >
>> > 3. Yes, this feature will work in batch mode as well.
>> >
>> >
>> > Best regards,
>> > Wencong
>> >
>> >
>> >
>> >
>> >
>> > At 2024-06-11 23:47:40, "Jingsong Li" <ji...@gmail.com> wrote:
>> > >Hi all,
>> > >
>> > >+1 to this FLIP, very thanks all for your proposal.
>> > >
>> > >isDeterministic looks good to me too.
>> > >
>> > >We can consider stating the following points:
>> > >
>> > >1. How to enable custom data distribution? Is it a dynamic hint? Can
>> > >you provide an SQL example.
>> > >
>> > >2. What impact will it have when the mainstream is changelog? Causing
>> > >disorder? This may need to be emphasized.
>> > >
>> > >3. Does this feature work in batch mode too?
>> > >
>> > >Best,
>> > >Jingsong
>> > >
>> > >On Tue, Jun 11, 2024 at 8:22 PM Wencong Liu <li...@163.com>
>> wrote:
>> > >>
>> > >> Hi Lincoln,
>> > >>
>> > >>
>> > >> Thanks for your reply. Weijie and I discussed these two issues
>> offline,
>> > >> and here are the results of our discussion:
>> > >> 1. When the user utilizes the hash lookup join hint introduced by
>> > FLIP-204[1],
>> > >> the `SupportsLookupCustomShuffle` interface should be ignored. This
>> is
>> > because
>> > >> the hash lookup join hint is directly specified by the user through a
>> > SQL HINT,
>> > >> which is more in line with user intuition. WDYT?
>> > >> 2. We agree with the introduction of the `isDeterministic` method.
>> The
>> > >> `SupportsLookupCustomShuffle` interface introduces a custom shuffle,
>> > which
>> > >> can cause ADD/UPDATE_AFTER events (+I, +U) to appear
>> > >> after UPDATE_BEFORE/DELETE events (-D, -U), thus breaking the current
>> > >> limitations of the Flink Sink Operator[2]. If `isDeterministic`
>> returns
>> > false and the
>> > >> changelog event type is not insert-only, the Planner should not apply
>> > the shuffle
>> > >> provided by `SupportsLookupCustomShuffle`.
>> > >>
>> > >>
>> > >> [1]
>> >
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
>> > >> [2]
>> >
>> https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
>> > >>
>> > >>
>> > >> Best,
>> > >> Wencong
>> > >>
>> > >>
>> > >>
>> > >>
>> > >>
>> > >>
>> > >>
>> > >>
>> > >>
>> > >> At 2024-06-11 00:02:57, "Lincoln Lee" <li...@gmail.com>
>> wrote:
>> > >> >Hi Weijie,
>> > >> >
>> > >> >Thanks for your proposal, this will be a useful advanced
>> optimization
>> > for
>> > >> >connector developers!
>> > >> >
>> > >> >I have two questions:
>> > >> >
>> > >> >1. FLIP-204[1] hash lookup join hint is mentioned in this FLIP,
>> what's
>> > the
>> > >> >apply ordering of the two feature? For example, a connector that
>> > >> >implements the `SupportsLookupCustomShuffle` interface also has a
>> > >> >`SHUFFLE_HASH` lookup join hint specified by the user in sql, what's
>> > >> >the expected behavior?
>> > >> >
>> > >> >2. This FLIP considers the relationship with NDU processing, and I
>> > agree
>> > >> >with the current choice to prioritize NDU first. However, we should
>> > also
>> > >> >consider another issue: out-of-orderness of the changelog events in
>> > >> >streaming[2]. If the connector developer supplies a
>> non-deterministic
>> > >> >partitioner, e.g., a random partitioner for anti-skew purpose, then
>> > it'll
>> > >> >break the assumption relied by current SQL operators in streaming:
>> the
>> > >> >ADD/UDPATE_AFTER events (+I, +U) always occur before its related
>> > >> >UDPATE_BEFORE/DELETE events (-D, -U) and they are always
>> > >> >processed by the same task even if a data shuffle is involved. So a
>> > >> >straightforward approach would be to add method `isDeterministic` to
>> > >> >the `InputDataPartitioner` interface to explicitly tell the planner
>> > whether
>> > >> >the partitioner is deterministic or not(then the planner can reject
>> the
>> > >> >non-deterministic custom partitioner for correctness requirements).
>> > >> >
>> > >> >[1]
>> > >> >
>> >
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
>> > >> >[2]
>> > >> >
>> >
>> https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
>> > >> >
>> > >> >
>> > >> >Best,
>> > >> >Lincoln Lee
>> > >> >
>> > >> >
>> > >> >Xintong Song <to...@gmail.com> 于2024年6月7日周五 13:53写道：
>> > >> >
>> > >> >> +1 for this proposal.
>> > >> >>
>> > >> >> This FLIP will make it possible for each lookup join parallel task
>> > to only
>> > >> >> access and cache a subset of the data. This will significantly
>> > improve the
>> > >> >> performance and reduce the overhead when using Paimon for the
>> > dimension
>> > >> >> table. And it's general enough to also be leveraged by other
>> > connectors.
>> > >> >>
>> > >> >> Best,
>> > >> >>
>> > >> >> Xintong
>> > >> >>
>> > >> >>
>> > >> >>
>> > >> >> On Fri, Jun 7, 2024 at 10:01 AM weijie guo <
>> > guoweijiereswqa@gmail.com>
>> > >> >> wrote:
>> > >> >>
>> > >> >> > Hi devs,
>> > >> >> >
>> > >> >> >
>> > >> >> > I'd like to start a discussion about FLIP-462[1]: Support Custom
>> > Data
>> > >> >> > Distribution for Input Stream of Lookup Join.
>> > >> >> >
>> > >> >> >
>> > >> >> > Lookup Join is an important feature in Flink, It is typically
>> used
>> > to
>> > >> >> > enrich a table with data that is queried from an external
>> system.
>> > >> >> > If we interact with the external systems for each incoming
>> record,
>> > we
>> > >> >> > incur significant network IO and RPC overhead.
>> > >> >> >
>> > >> >> > Therefore, most connectors introduce caching to reduce the
>> > per-record
>> > >> >> > level query overhead. However, because the data distribution of
>> > Lookup
>> > >> >> > Join's input stream is arbitrary, the cache hit rate is
>> sometimes
>> > >> >> > unsatisfactory.
>> > >> >> >
>> > >> >> >
>> > >> >> > We want to introduce a mechanism for the connector to tell the
>> > Flink
>> > >> >> > planner its desired input stream data distribution or
>> partitioning
>> > >> >> > strategy. This can significantly reduce the amount of cached
>> data
>> > and
>> > >> >> > improve performance of Lookup Join.
>> > >> >> >
>> > >> >> >
>> > >> >> > You can find more details in this FLIP[1]. Looking forward to
>> > hearing
>> > >> >> > from you, thanks!
>> > >> >> >
>> > >> >> >
>> > >> >> > Best regards,
>> > >> >> >
>> > >> >> > Weijie
>> > >> >> >
>> > >> >> >
>> > >> >> > [1]
>> > >> >> >
>> > >> >> >
>> > >> >>
>> >
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
>> > >> >> >
>> > >> >>
>> >
>>
>
Re: Re: Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by weijie guo <gu...@gmail.com>.
Thanks Lincoln for the quick response.
> Since we've decided to extend a new hint option 'shuffle' to the current
`LOOKUP` join hint, do we support hash shuffle as well?(It seems like it
shouldn't require a lot of extra work, right?) This will deliver a complete
new feature to users, also because
FLIP-204 is stale for now and this new extension will give user a more
simpler way to achieve the goal, WDYT?
Yes, I think this makes more sense.
In a word: If the target dim table does not
implement SupportsLookupCustomShuffle, the planner will try best to apply
customer partitioning for the input stream. Otherwise, the planner will try
best to apply a hash partitioning.
As for FLIP-204, I think we can discuss whether it should be discarded or
refactored in a separate thread. TBH, I think the current approach can
completely cover it and be much easier to use.
> "upsert mode" should be "updating stream" or "non-insert-only stream".
Thanks, updated the FLIP.
Best regards,
Weijie
Lincoln Lee <li...@gmail.com> 于2024年6月13日周四 23:08写道：
> Thanks Weijie & Wencong for your update including the conclusions of
> the offline discussion.
>
> There's one thing need to be confirmed in the FLIP:
> > The hint only provides a suggestion to the optimizer, it is not an
> enforcer. As a result, If the target dim table not implements
> SupportsLookupCustomShuffle, planner will ignore this newly introduced
> shuffle option.
>
> Since we've decided to extend a new hint option 'shuffle' to the current
> `LOOKUP` join hint, do we support hash shuffle as well?(It seems like it
> shouldn't require a lot of extra work, right?)
> This will deliver a complete new feature to users, also because
> FLIP-204 is stale for now and this new extension will give user a more
> simpler way to achieve the goal, WDYT?
>
> Another small comment for the new interface:
> > "... planner may not apply this partitioner in upsert mode ..."
> > default boolean isDeterministic()
> "upsert mode" should be "updating stream" or "non-insert-only stream".
>
>
> Best,
> Lincoln Lee
>
>
> Wencong Liu <li...@163.com> 于2024年6月12日周三 21:43写道：
>
> > Hi Jingsong,
> >
> >
> > Some of the points you mentioned are currently clarified in
> > the updated FLIP. Please check it out.
> >
> >
> > 1. Enabling custom data distribution can be done through the
> > LOOKUP SQL Hint. There are detailed examples provided in the FLIP.
> >
> >
> > 2. We will add the isDeterministic method to the `InputDataPartitioner`
> > interface, which will return true by default. If the
> > `InputDataPartitioner`
> > is not deterministic, the connector developer need to override the
> > isDeterministic method to return false. If the connector developer
> > cannot ensure this protocol, they will need to bear the correctness
> > issues that arise.
> >
> >
> > 3. Yes, this feature will work in batch mode as well.
> >
> >
> > Best regards,
> > Wencong
> >
> >
> >
> >
> >
> > At 2024-06-11 23:47:40, "Jingsong Li" <ji...@gmail.com> wrote:
> > >Hi all,
> > >
> > >+1 to this FLIP, very thanks all for your proposal.
> > >
> > >isDeterministic looks good to me too.
> > >
> > >We can consider stating the following points:
> > >
> > >1. How to enable custom data distribution? Is it a dynamic hint? Can
> > >you provide an SQL example.
> > >
> > >2. What impact will it have when the mainstream is changelog? Causing
> > >disorder? This may need to be emphasized.
> > >
> > >3. Does this feature work in batch mode too?
> > >
> > >Best,
> > >Jingsong
> > >
> > >On Tue, Jun 11, 2024 at 8:22 PM Wencong Liu <li...@163.com>
> wrote:
> > >>
> > >> Hi Lincoln,
> > >>
> > >>
> > >> Thanks for your reply. Weijie and I discussed these two issues
> offline,
> > >> and here are the results of our discussion:
> > >> 1. When the user utilizes the hash lookup join hint introduced by
> > FLIP-204[1],
> > >> the `SupportsLookupCustomShuffle` interface should be ignored. This is
> > because
> > >> the hash lookup join hint is directly specified by the user through a
> > SQL HINT,
> > >> which is more in line with user intuition. WDYT?
> > >> 2. We agree with the introduction of the `isDeterministic` method. The
> > >> `SupportsLookupCustomShuffle` interface introduces a custom shuffle,
> > which
> > >> can cause ADD/UPDATE_AFTER events (+I, +U) to appear
> > >> after UPDATE_BEFORE/DELETE events (-D, -U), thus breaking the current
> > >> limitations of the Flink Sink Operator[2]. If `isDeterministic`
> returns
> > false and the
> > >> changelog event type is not insert-only, the Planner should not apply
> > the shuffle
> > >> provided by `SupportsLookupCustomShuffle`.
> > >>
> > >>
> > >> [1]
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
> > >> [2]
> >
> https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
> > >>
> > >>
> > >> Best,
> > >> Wencong
> > >>
> > >>
> > >>
> > >>
> > >>
> > >>
> > >>
> > >>
> > >>
> > >> At 2024-06-11 00:02:57, "Lincoln Lee" <li...@gmail.com> wrote:
> > >> >Hi Weijie,
> > >> >
> > >> >Thanks for your proposal, this will be a useful advanced optimization
> > for
> > >> >connector developers!
> > >> >
> > >> >I have two questions:
> > >> >
> > >> >1. FLIP-204[1] hash lookup join hint is mentioned in this FLIP,
> what's
> > the
> > >> >apply ordering of the two feature? For example, a connector that
> > >> >implements the `SupportsLookupCustomShuffle` interface also has a
> > >> >`SHUFFLE_HASH` lookup join hint specified by the user in sql, what's
> > >> >the expected behavior?
> > >> >
> > >> >2. This FLIP considers the relationship with NDU processing, and I
> > agree
> > >> >with the current choice to prioritize NDU first. However, we should
> > also
> > >> >consider another issue: out-of-orderness of the changelog events in
> > >> >streaming[2]. If the connector developer supplies a non-deterministic
> > >> >partitioner, e.g., a random partitioner for anti-skew purpose, then
> > it'll
> > >> >break the assumption relied by current SQL operators in streaming:
> the
> > >> >ADD/UDPATE_AFTER events (+I, +U) always occur before its related
> > >> >UDPATE_BEFORE/DELETE events (-D, -U) and they are always
> > >> >processed by the same task even if a data shuffle is involved. So a
> > >> >straightforward approach would be to add method `isDeterministic` to
> > >> >the `InputDataPartitioner` interface to explicitly tell the planner
> > whether
> > >> >the partitioner is deterministic or not(then the planner can reject
> the
> > >> >non-deterministic custom partitioner for correctness requirements).
> > >> >
> > >> >[1]
> > >> >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
> > >> >[2]
> > >> >
> >
> https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
> > >> >
> > >> >
> > >> >Best,
> > >> >Lincoln Lee
> > >> >
> > >> >
> > >> >Xintong Song <to...@gmail.com> 于2024年6月7日周五 13:53写道：
> > >> >
> > >> >> +1 for this proposal.
> > >> >>
> > >> >> This FLIP will make it possible for each lookup join parallel task
> > to only
> > >> >> access and cache a subset of the data. This will significantly
> > improve the
> > >> >> performance and reduce the overhead when using Paimon for the
> > dimension
> > >> >> table. And it's general enough to also be leveraged by other
> > connectors.
> > >> >>
> > >> >> Best,
> > >> >>
> > >> >> Xintong
> > >> >>
> > >> >>
> > >> >>
> > >> >> On Fri, Jun 7, 2024 at 10:01 AM weijie guo <
> > guoweijiereswqa@gmail.com>
> > >> >> wrote:
> > >> >>
> > >> >> > Hi devs,
> > >> >> >
> > >> >> >
> > >> >> > I'd like to start a discussion about FLIP-462[1]: Support Custom
> > Data
> > >> >> > Distribution for Input Stream of Lookup Join.
> > >> >> >
> > >> >> >
> > >> >> > Lookup Join is an important feature in Flink, It is typically
> used
> > to
> > >> >> > enrich a table with data that is queried from an external system.
> > >> >> > If we interact with the external systems for each incoming
> record,
> > we
> > >> >> > incur significant network IO and RPC overhead.
> > >> >> >
> > >> >> > Therefore, most connectors introduce caching to reduce the
> > per-record
> > >> >> > level query overhead. However, because the data distribution of
> > Lookup
> > >> >> > Join's input stream is arbitrary, the cache hit rate is sometimes
> > >> >> > unsatisfactory.
> > >> >> >
> > >> >> >
> > >> >> > We want to introduce a mechanism for the connector to tell the
> > Flink
> > >> >> > planner its desired input stream data distribution or
> partitioning
> > >> >> > strategy. This can significantly reduce the amount of cached data
> > and
> > >> >> > improve performance of Lookup Join.
> > >> >> >
> > >> >> >
> > >> >> > You can find more details in this FLIP[1]. Looking forward to
> > hearing
> > >> >> > from you, thanks!
> > >> >> >
> > >> >> >
> > >> >> > Best regards,
> > >> >> >
> > >> >> > Weijie
> > >> >> >
> > >> >> >
> > >> >> > [1]
> > >> >> >
> > >> >> >
> > >> >>
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
> > >> >> >
> > >> >>
> >
>
Re: Re: Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by Lincoln Lee <li...@gmail.com>.
Thanks Weijie & Wencong for your update including the conclusions of
the offline discussion.
There's one thing need to be confirmed in the FLIP:
> The hint only provides a suggestion to the optimizer, it is not an
enforcer. As a result, If the target dim table not implements
SupportsLookupCustomShuffle, planner will ignore this newly introduced
shuffle option.
Since we've decided to extend a new hint option 'shuffle' to the current
`LOOKUP` join hint, do we support hash shuffle as well?(It seems like it
shouldn't require a lot of extra work, right?)
This will deliver a complete new feature to users, also because
FLIP-204 is stale for now and this new extension will give user a more
simpler way to achieve the goal, WDYT?
Another small comment for the new interface:
> "... planner may not apply this partitioner in upsert mode ..."
> default boolean isDeterministic()
"upsert mode" should be "updating stream" or "non-insert-only stream".
Best,
Lincoln Lee
Wencong Liu <li...@163.com> 于2024年6月12日周三 21:43写道：
> Hi Jingsong,
>
>
> Some of the points you mentioned are currently clarified in
> the updated FLIP. Please check it out.
>
>
> 1. Enabling custom data distribution can be done through the
> LOOKUP SQL Hint. There are detailed examples provided in the FLIP.
>
>
> 2. We will add the isDeterministic method to the `InputDataPartitioner`
> interface, which will return true by default. If the
> `InputDataPartitioner`
> is not deterministic, the connector developer need to override the
> isDeterministic method to return false. If the connector developer
> cannot ensure this protocol, they will need to bear the correctness
> issues that arise.
>
>
> 3. Yes, this feature will work in batch mode as well.
>
>
> Best regards,
> Wencong
>
>
>
>
>
> At 2024-06-11 23:47:40, "Jingsong Li" <ji...@gmail.com> wrote:
> >Hi all,
> >
> >+1 to this FLIP, very thanks all for your proposal.
> >
> >isDeterministic looks good to me too.
> >
> >We can consider stating the following points:
> >
> >1. How to enable custom data distribution? Is it a dynamic hint? Can
> >you provide an SQL example.
> >
> >2. What impact will it have when the mainstream is changelog? Causing
> >disorder? This may need to be emphasized.
> >
> >3. Does this feature work in batch mode too?
> >
> >Best,
> >Jingsong
> >
> >On Tue, Jun 11, 2024 at 8:22 PM Wencong Liu <li...@163.com> wrote:
> >>
> >> Hi Lincoln,
> >>
> >>
> >> Thanks for your reply. Weijie and I discussed these two issues offline,
> >> and here are the results of our discussion:
> >> 1. When the user utilizes the hash lookup join hint introduced by
> FLIP-204[1],
> >> the `SupportsLookupCustomShuffle` interface should be ignored. This is
> because
> >> the hash lookup join hint is directly specified by the user through a
> SQL HINT,
> >> which is more in line with user intuition. WDYT?
> >> 2. We agree with the introduction of the `isDeterministic` method. The
> >> `SupportsLookupCustomShuffle` interface introduces a custom shuffle,
> which
> >> can cause ADD/UPDATE_AFTER events (+I, +U) to appear
> >> after UPDATE_BEFORE/DELETE events (-D, -U), thus breaking the current
> >> limitations of the Flink Sink Operator[2]. If `isDeterministic` returns
> false and the
> >> changelog event type is not insert-only, the Planner should not apply
> the shuffle
> >> provided by `SupportsLookupCustomShuffle`.
> >>
> >>
> >> [1]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
> >> [2]
> https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
> >>
> >>
> >> Best,
> >> Wencong
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >> At 2024-06-11 00:02:57, "Lincoln Lee" <li...@gmail.com> wrote:
> >> >Hi Weijie,
> >> >
> >> >Thanks for your proposal, this will be a useful advanced optimization
> for
> >> >connector developers!
> >> >
> >> >I have two questions:
> >> >
> >> >1. FLIP-204[1] hash lookup join hint is mentioned in this FLIP, what's
> the
> >> >apply ordering of the two feature? For example, a connector that
> >> >implements the `SupportsLookupCustomShuffle` interface also has a
> >> >`SHUFFLE_HASH` lookup join hint specified by the user in sql, what's
> >> >the expected behavior?
> >> >
> >> >2. This FLIP considers the relationship with NDU processing, and I
> agree
> >> >with the current choice to prioritize NDU first. However, we should
> also
> >> >consider another issue: out-of-orderness of the changelog events in
> >> >streaming[2]. If the connector developer supplies a non-deterministic
> >> >partitioner, e.g., a random partitioner for anti-skew purpose, then
> it'll
> >> >break the assumption relied by current SQL operators in streaming: the
> >> >ADD/UDPATE_AFTER events (+I, +U) always occur before its related
> >> >UDPATE_BEFORE/DELETE events (-D, -U) and they are always
> >> >processed by the same task even if a data shuffle is involved. So a
> >> >straightforward approach would be to add method `isDeterministic` to
> >> >the `InputDataPartitioner` interface to explicitly tell the planner
> whether
> >> >the partitioner is deterministic or not(then the planner can reject the
> >> >non-deterministic custom partitioner for correctness requirements).
> >> >
> >> >[1]
> >> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
> >> >[2]
> >> >
> https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
> >> >
> >> >
> >> >Best,
> >> >Lincoln Lee
> >> >
> >> >
> >> >Xintong Song <to...@gmail.com> 于2024年6月7日周五 13:53写道：
> >> >
> >> >> +1 for this proposal.
> >> >>
> >> >> This FLIP will make it possible for each lookup join parallel task
> to only
> >> >> access and cache a subset of the data. This will significantly
> improve the
> >> >> performance and reduce the overhead when using Paimon for the
> dimension
> >> >> table. And it's general enough to also be leveraged by other
> connectors.
> >> >>
> >> >> Best,
> >> >>
> >> >> Xintong
> >> >>
> >> >>
> >> >>
> >> >> On Fri, Jun 7, 2024 at 10:01 AM weijie guo <
> guoweijiereswqa@gmail.com>
> >> >> wrote:
> >> >>
> >> >> > Hi devs,
> >> >> >
> >> >> >
> >> >> > I'd like to start a discussion about FLIP-462[1]: Support Custom
> Data
> >> >> > Distribution for Input Stream of Lookup Join.
> >> >> >
> >> >> >
> >> >> > Lookup Join is an important feature in Flink, It is typically used
> to
> >> >> > enrich a table with data that is queried from an external system.
> >> >> > If we interact with the external systems for each incoming record,
> we
> >> >> > incur significant network IO and RPC overhead.
> >> >> >
> >> >> > Therefore, most connectors introduce caching to reduce the
> per-record
> >> >> > level query overhead. However, because the data distribution of
> Lookup
> >> >> > Join's input stream is arbitrary, the cache hit rate is sometimes
> >> >> > unsatisfactory.
> >> >> >
> >> >> >
> >> >> > We want to introduce a mechanism for the connector to tell the
> Flink
> >> >> > planner its desired input stream data distribution or partitioning
> >> >> > strategy. This can significantly reduce the amount of cached data
> and
> >> >> > improve performance of Lookup Join.
> >> >> >
> >> >> >
> >> >> > You can find more details in this FLIP[1]. Looking forward to
> hearing
> >> >> > from you, thanks!
> >> >> >
> >> >> >
> >> >> > Best regards,
> >> >> >
> >> >> > Weijie
> >> >> >
> >> >> >
> >> >> > [1]
> >> >> >
> >> >> >
> >> >>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
> >> >> >
> >> >>
>
Re:Re: Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by Wencong Liu <li...@163.com>.
Hi Jingsong,
Some of the points you mentioned are currently clarified in
the updated FLIP. Please check it out.
1. Enabling custom data distribution can be done through the
LOOKUP SQL Hint. There are detailed examples provided in the FLIP.
2. We will add the isDeterministic method to the `InputDataPartitioner`
interface, which will return true by default. If the `InputDataPartitioner`
is not deterministic, the connector developer need to override the
isDeterministic method to return false. If the connector developer
cannot ensure this protocol, they will need to bear the correctness
issues that arise.
3. Yes, this feature will work in batch mode as well.
Best regards,
Wencong
At 2024-06-11 23:47:40, "Jingsong Li" <ji...@gmail.com> wrote:
>Hi all,
>
>+1 to this FLIP, very thanks all for your proposal.
>
>isDeterministic looks good to me too.
>
>We can consider stating the following points:
>
>1. How to enable custom data distribution? Is it a dynamic hint? Can
>you provide an SQL example.
>
>2. What impact will it have when the mainstream is changelog? Causing
>disorder? This may need to be emphasized.
>
>3. Does this feature work in batch mode too?
>
>Best,
>Jingsong
>
>On Tue, Jun 11, 2024 at 8:22 PM Wencong Liu <li...@163.com> wrote:
>>
>> Hi Lincoln,
>>
>>
>> Thanks for your reply. Weijie and I discussed these two issues offline,
>> and here are the results of our discussion:
>> 1. When the user utilizes the hash lookup join hint introduced by FLIP-204[1],
>> the `SupportsLookupCustomShuffle` interface should be ignored. This is because
>> the hash lookup join hint is directly specified by the user through a SQL HINT,
>> which is more in line with user intuition. WDYT?
>> 2. We agree with the introduction of the `isDeterministic` method. The
>> `SupportsLookupCustomShuffle` interface introduces a custom shuffle, which
>> can cause ADD/UPDATE_AFTER events (+I, +U) to appear
>> after UPDATE_BEFORE/DELETE events (-D, -U), thus breaking the current
>> limitations of the Flink Sink Operator[2]. If `isDeterministic` returns false and the
>> changelog event type is not insert-only, the Planner should not apply the shuffle
>> provided by `SupportsLookupCustomShuffle`.
>>
>>
>> [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
>> [2] https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
>>
>>
>> Best,
>> Wencong
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> At 2024-06-11 00:02:57, "Lincoln Lee" <li...@gmail.com> wrote:
>> >Hi Weijie,
>> >
>> >Thanks for your proposal, this will be a useful advanced optimization for
>> >connector developers!
>> >
>> >I have two questions:
>> >
>> >1. FLIP-204[1] hash lookup join hint is mentioned in this FLIP, what's the
>> >apply ordering of the two feature? For example, a connector that
>> >implements the `SupportsLookupCustomShuffle` interface also has a
>> >`SHUFFLE_HASH` lookup join hint specified by the user in sql, what's
>> >the expected behavior?
>> >
>> >2. This FLIP considers the relationship with NDU processing, and I agree
>> >with the current choice to prioritize NDU first. However, we should also
>> >consider another issue: out-of-orderness of the changelog events in
>> >streaming[2]. If the connector developer supplies a non-deterministic
>> >partitioner, e.g., a random partitioner for anti-skew purpose, then it'll
>> >break the assumption relied by current SQL operators in streaming: the
>> >ADD/UDPATE_AFTER events (+I, +U) always occur before its related
>> >UDPATE_BEFORE/DELETE events (-D, -U) and they are always
>> >processed by the same task even if a data shuffle is involved. So a
>> >straightforward approach would be to add method `isDeterministic` to
>> >the `InputDataPartitioner` interface to explicitly tell the planner whether
>> >the partitioner is deterministic or not(then the planner can reject the
>> >non-deterministic custom partitioner for correctness requirements).
>> >
>> >[1]
>> >https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
>> >[2]
>> >https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
>> >
>> >
>> >Best,
>> >Lincoln Lee
>> >
>> >
>> >Xintong Song <to...@gmail.com> 于2024年6月7日周五 13:53写道：
>> >
>> >> +1 for this proposal.
>> >>
>> >> This FLIP will make it possible for each lookup join parallel task to only
>> >> access and cache a subset of the data. This will significantly improve the
>> >> performance and reduce the overhead when using Paimon for the dimension
>> >> table. And it's general enough to also be leveraged by other connectors.
>> >>
>> >> Best,
>> >>
>> >> Xintong
>> >>
>> >>
>> >>
>> >> On Fri, Jun 7, 2024 at 10:01 AM weijie guo <gu...@gmail.com>
>> >> wrote:
>> >>
>> >> > Hi devs,
>> >> >
>> >> >
>> >> > I'd like to start a discussion about FLIP-462[1]: Support Custom Data
>> >> > Distribution for Input Stream of Lookup Join.
>> >> >
>> >> >
>> >> > Lookup Join is an important feature in Flink, It is typically used to
>> >> > enrich a table with data that is queried from an external system.
>> >> > If we interact with the external systems for each incoming record, we
>> >> > incur significant network IO and RPC overhead.
>> >> >
>> >> > Therefore, most connectors introduce caching to reduce the per-record
>> >> > level query overhead. However, because the data distribution of Lookup
>> >> > Join's input stream is arbitrary, the cache hit rate is sometimes
>> >> > unsatisfactory.
>> >> >
>> >> >
>> >> > We want to introduce a mechanism for the connector to tell the Flink
>> >> > planner its desired input stream data distribution or partitioning
>> >> > strategy. This can significantly reduce the amount of cached data and
>> >> > improve performance of Lookup Join.
>> >> >
>> >> >
>> >> > You can find more details in this FLIP[1]. Looking forward to hearing
>> >> > from you, thanks!
>> >> >
>> >> >
>> >> > Best regards,
>> >> >
>> >> > Weijie
>> >> >
>> >> >
>> >> > [1]
>> >> >
>> >> >
>> >> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
>> >> >
>> >>
Re: Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by Jingsong Li <ji...@gmail.com>.
Hi all,
+1 to this FLIP, very thanks all for your proposal.
isDeterministic looks good to me too.
We can consider stating the following points:
1. How to enable custom data distribution? Is it a dynamic hint? Can
you provide an SQL example.
2. What impact will it have when the mainstream is changelog? Causing
disorder? This may need to be emphasized.
3. Does this feature work in batch mode too?
Best,
Jingsong
On Tue, Jun 11, 2024 at 8:22 PM Wencong Liu <li...@163.com> wrote:
>
> Hi Lincoln,
>
>
> Thanks for your reply. Weijie and I discussed these two issues offline,
> and here are the results of our discussion:
> 1. When the user utilizes the hash lookup join hint introduced by FLIP-204[1],
> the `SupportsLookupCustomShuffle` interface should be ignored. This is because
> the hash lookup join hint is directly specified by the user through a SQL HINT,
> which is more in line with user intuition. WDYT?
> 2. We agree with the introduction of the `isDeterministic` method. The
> `SupportsLookupCustomShuffle` interface introduces a custom shuffle, which
> can cause ADD/UPDATE_AFTER events (+I, +U) to appear
> after UPDATE_BEFORE/DELETE events (-D, -U), thus breaking the current
> limitations of the Flink Sink Operator[2]. If `isDeterministic` returns false and the
> changelog event type is not insert-only, the Planner should not apply the shuffle
> provided by `SupportsLookupCustomShuffle`.
>
>
> [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
> [2] https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
>
>
> Best,
> Wencong
>
>
>
>
>
>
>
>
>
> At 2024-06-11 00:02:57, "Lincoln Lee" <li...@gmail.com> wrote:
> >Hi Weijie,
> >
> >Thanks for your proposal, this will be a useful advanced optimization for
> >connector developers!
> >
> >I have two questions:
> >
> >1. FLIP-204[1] hash lookup join hint is mentioned in this FLIP, what's the
> >apply ordering of the two feature? For example, a connector that
> >implements the `SupportsLookupCustomShuffle` interface also has a
> >`SHUFFLE_HASH` lookup join hint specified by the user in sql, what's
> >the expected behavior?
> >
> >2. This FLIP considers the relationship with NDU processing, and I agree
> >with the current choice to prioritize NDU first. However, we should also
> >consider another issue: out-of-orderness of the changelog events in
> >streaming[2]. If the connector developer supplies a non-deterministic
> >partitioner, e.g., a random partitioner for anti-skew purpose, then it'll
> >break the assumption relied by current SQL operators in streaming: the
> >ADD/UDPATE_AFTER events (+I, +U) always occur before its related
> >UDPATE_BEFORE/DELETE events (-D, -U) and they are always
> >processed by the same task even if a data shuffle is involved. So a
> >straightforward approach would be to add method `isDeterministic` to
> >the `InputDataPartitioner` interface to explicitly tell the planner whether
> >the partitioner is deterministic or not(then the planner can reject the
> >non-deterministic custom partitioner for correctness requirements).
> >
> >[1]
> >https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
> >[2]
> >https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
> >
> >
> >Best,
> >Lincoln Lee
> >
> >
> >Xintong Song <to...@gmail.com> 于2024年6月7日周五 13:53写道：
> >
> >> +1 for this proposal.
> >>
> >> This FLIP will make it possible for each lookup join parallel task to only
> >> access and cache a subset of the data. This will significantly improve the
> >> performance and reduce the overhead when using Paimon for the dimension
> >> table. And it's general enough to also be leveraged by other connectors.
> >>
> >> Best,
> >>
> >> Xintong
> >>
> >>
> >>
> >> On Fri, Jun 7, 2024 at 10:01 AM weijie guo <gu...@gmail.com>
> >> wrote:
> >>
> >> > Hi devs,
> >> >
> >> >
> >> > I'd like to start a discussion about FLIP-462[1]: Support Custom Data
> >> > Distribution for Input Stream of Lookup Join.
> >> >
> >> >
> >> > Lookup Join is an important feature in Flink, It is typically used to
> >> > enrich a table with data that is queried from an external system.
> >> > If we interact with the external systems for each incoming record, we
> >> > incur significant network IO and RPC overhead.
> >> >
> >> > Therefore, most connectors introduce caching to reduce the per-record
> >> > level query overhead. However, because the data distribution of Lookup
> >> > Join's input stream is arbitrary, the cache hit rate is sometimes
> >> > unsatisfactory.
> >> >
> >> >
> >> > We want to introduce a mechanism for the connector to tell the Flink
> >> > planner its desired input stream data distribution or partitioning
> >> > strategy. This can significantly reduce the amount of cached data and
> >> > improve performance of Lookup Join.
> >> >
> >> >
> >> > You can find more details in this FLIP[1]. Looking forward to hearing
> >> > from you, thanks!
> >> >
> >> >
> >> > Best regards,
> >> >
> >> > Weijie
> >> >
> >> >
> >> > [1]
> >> >
> >> >
> >> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
> >> >
> >>
Re:Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by Wencong Liu <li...@163.com>.
Hi Lincoln,
Thanks for your reply. Weijie and I discussed these two issues offline,
and here are the results of our discussion:
1. When the user utilizes the hash lookup join hint introduced by FLIP-204[1],
the `SupportsLookupCustomShuffle` interface should be ignored. This is because
the hash lookup join hint is directly specified by the user through a SQL HINT,
which is more in line with user intuition. WDYT?
2. We agree with the introduction of the `isDeterministic` method. The
`SupportsLookupCustomShuffle` interface introduces a custom shuffle, which
can cause ADD/UPDATE_AFTER events (+I, +U) to appear
after UPDATE_BEFORE/DELETE events (-D, -U), thus breaking the current
limitations of the Flink Sink Operator[2]. If `isDeterministic` returns false and the
changelog event type is not insert-only, the Planner should not apply the shuffle
provided by `SupportsLookupCustomShuffle`.
[1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
[2] https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
Best,
Wencong
At 2024-06-11 00:02:57, "Lincoln Lee" <li...@gmail.com> wrote:
>Hi Weijie,
>
>Thanks for your proposal, this will be a useful advanced optimization for
>connector developers!
>
>I have two questions:
>
>1. FLIP-204[1] hash lookup join hint is mentioned in this FLIP, what's the
>apply ordering of the two feature? For example, a connector that
>implements the `SupportsLookupCustomShuffle` interface also has a
>`SHUFFLE_HASH` lookup join hint specified by the user in sql, what's
>the expected behavior?
>
>2. This FLIP considers the relationship with NDU processing, and I agree
>with the current choice to prioritize NDU first. However, we should also
>consider another issue: out-of-orderness of the changelog events in
>streaming[2]. If the connector developer supplies a non-deterministic
>partitioner, e.g., a random partitioner for anti-skew purpose, then it'll
>break the assumption relied by current SQL operators in streaming: the
>ADD/UDPATE_AFTER events (+I, +U) always occur before its related
>UDPATE_BEFORE/DELETE events (-D, -U) and they are always
>processed by the same task even if a data shuffle is involved. So a
>straightforward approach would be to add method `isDeterministic` to
>the `InputDataPartitioner` interface to explicitly tell the planner whether
>the partitioner is deterministic or not(then the planner can reject the
>non-deterministic custom partitioner for correctness requirements).
>
>[1]
>https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
>[2]
>https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
>
>
>Best,
>Lincoln Lee
>
>
>Xintong Song <to...@gmail.com> 于2024年6月7日周五 13:53写道：
>
>> +1 for this proposal.
>>
>> This FLIP will make it possible for each lookup join parallel task to only
>> access and cache a subset of the data. This will significantly improve the
>> performance and reduce the overhead when using Paimon for the dimension
>> table. And it's general enough to also be leveraged by other connectors.
>>
>> Best,
>>
>> Xintong
>>
>>
>>
>> On Fri, Jun 7, 2024 at 10:01 AM weijie guo <gu...@gmail.com>
>> wrote:
>>
>> > Hi devs,
>> >
>> >
>> > I'd like to start a discussion about FLIP-462[1]: Support Custom Data
>> > Distribution for Input Stream of Lookup Join.
>> >
>> >
>> > Lookup Join is an important feature in Flink, It is typically used to
>> > enrich a table with data that is queried from an external system.
>> > If we interact with the external systems for each incoming record, we
>> > incur significant network IO and RPC overhead.
>> >
>> > Therefore, most connectors introduce caching to reduce the per-record
>> > level query overhead. However, because the data distribution of Lookup
>> > Join's input stream is arbitrary, the cache hit rate is sometimes
>> > unsatisfactory.
>> >
>> >
>> > We want to introduce a mechanism for the connector to tell the Flink
>> > planner its desired input stream data distribution or partitioning
>> > strategy. This can significantly reduce the amount of cached data and
>> > improve performance of Lookup Join.
>> >
>> >
>> > You can find more details in this FLIP[1]. Looking forward to hearing
>> > from you, thanks!
>> >
>> >
>> > Best regards,
>> >
>> > Weijie
>> >
>> >
>> > [1]
>> >
>> >
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
>> >
>>
Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by Lincoln Lee <li...@gmail.com>.
Hi Weijie,
Thanks for your proposal, this will be a useful advanced optimization for
connector developers!
I have two questions:
1. FLIP-204[1] hash lookup join hint is mentioned in this FLIP, what's the
apply ordering of the two feature? For example, a connector that
implements the `SupportsLookupCustomShuffle` interface also has a
`SHUFFLE_HASH` lookup join hint specified by the user in sql, what's
the expected behavior?
2. This FLIP considers the relationship with NDU processing, and I agree
with the current choice to prioritize NDU first. However, we should also
consider another issue: out-of-orderness of the changelog events in
streaming[2]. If the connector developer supplies a non-deterministic
partitioner, e.g., a random partitioner for anti-skew purpose, then it'll
break the assumption relied by current SQL operators in streaming: the
ADD/UDPATE_AFTER events (+I, +U) always occur before its related
UDPATE_BEFORE/DELETE events (-D, -U) and they are always
processed by the same task even if a data shuffle is involved. So a
straightforward approach would be to add method `isDeterministic` to
the `InputDataPartitioner` interface to explicitly tell the planner whether
the partitioner is deterministic or not(then the planner can reject the
non-deterministic custom partitioner for correctness requirements).
[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join
[2]
https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-event-out-of-orderness
Best,
Lincoln Lee
Xintong Song <to...@gmail.com> 于2024年6月7日周五 13:53写道：
> +1 for this proposal.
>
> This FLIP will make it possible for each lookup join parallel task to only
> access and cache a subset of the data. This will significantly improve the
> performance and reduce the overhead when using Paimon for the dimension
> table. And it's general enough to also be leveraged by other connectors.
>
> Best,
>
> Xintong
>
>
>
> On Fri, Jun 7, 2024 at 10:01 AM weijie guo <gu...@gmail.com>
> wrote:
>
> > Hi devs,
> >
> >
> > I'd like to start a discussion about FLIP-462[1]: Support Custom Data
> > Distribution for Input Stream of Lookup Join.
> >
> >
> > Lookup Join is an important feature in Flink, It is typically used to
> > enrich a table with data that is queried from an external system.
> > If we interact with the external systems for each incoming record, we
> > incur significant network IO and RPC overhead.
> >
> > Therefore, most connectors introduce caching to reduce the per-record
> > level query overhead. However, because the data distribution of Lookup
> > Join's input stream is arbitrary, the cache hit rate is sometimes
> > unsatisfactory.
> >
> >
> > We want to introduce a mechanism for the connector to tell the Flink
> > planner its desired input stream data distribution or partitioning
> > strategy. This can significantly reduce the amount of cached data and
> > improve performance of Lookup Join.
> >
> >
> > You can find more details in this FLIP[1]. Looking forward to hearing
> > from you, thanks!
> >
> >
> > Best regards,
> >
> > Weijie
> >
> >
> > [1]
> >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
> >
>
Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by Xintong Song <to...@gmail.com>.
+1 for this proposal.
This FLIP will make it possible for each lookup join parallel task to only
access and cache a subset of the data. This will significantly improve the
performance and reduce the overhead when using Paimon for the dimension
table. And it's general enough to also be leveraged by other connectors.
Best,
Xintong
On Fri, Jun 7, 2024 at 10:01 AM weijie guo <gu...@gmail.com>
wrote:
> Hi devs,
>
>
> I'd like to start a discussion about FLIP-462[1]: Support Custom Data
> Distribution for Input Stream of Lookup Join.
>
>
> Lookup Join is an important feature in Flink, It is typically used to
> enrich a table with data that is queried from an external system.
> If we interact with the external systems for each incoming record, we
> incur significant network IO and RPC overhead.
>
> Therefore, most connectors introduce caching to reduce the per-record
> level query overhead. However, because the data distribution of Lookup
> Join's input stream is arbitrary, the cache hit rate is sometimes
> unsatisfactory.
>
>
> We want to introduce a mechanism for the connector to tell the Flink
> planner its desired input stream data distribution or partitioning
> strategy. This can significantly reduce the amount of cached data and
> improve performance of Lookup Join.
>
>
> You can find more details in this FLIP[1]. Looking forward to hearing
> from you, thanks!
>
>
> Best regards,
>
> Weijie
>
>
> [1]
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
>
Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by Zhanghao Chen <zh...@outlook.com>.
Thanks for the clarification, that makes sense. +1 for the proposal.
Best,
Zhanghao Chen
________________________________
From: weijie guo <gu...@gmail.com>
Sent: Wednesday, June 12, 2024 14:20
To: dev@flink.apache.org <de...@flink.apache.org>
Subject: Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Hi Zhanghao,
Thanks for the reply!
> Could you give a more concrete example in production on when a custom
partitioning strategy will outperform partitioning by key
The key point here is partitioning logic cannot be fully expressed with all
or part of the join key. That is, even if we know which fields are used to
calculate buckets, still have to face the following problem:
1. The mapping from the bucket field to the bucket id is not necessarily
done via hashcode, and even if it is, the hash algorithm may be different
from the one used in Flink. The planner can't know how to do this mapping.
2. In order to get the bucket id, we have to mod the bucket number, but
planner has no notion of bucket number.
Best regards,
Weijie
Zhanghao Chen <zh...@outlook.com> 于2024年6月12日周三 13:55写道：
> Thanks for driving this, Weijie. Usually, the data distribution of the
> external system is closely related to the keys, e.g. computing the bucket
> index by key hashcode % bucket num, so I'm not sure about how much
> difference there are between partitioning by key and a custom partitioning
> strategy. Could you give a more concrete example in production on when a
> custom partitioning strategy will outperform partitioning by key? Since
> you've mentioned Paimon in doc, maybe an example on Paimon.
>
> Best,
> Zhanghao Chen
> ________________________________
> From: weijie guo <gu...@gmail.com>
> Sent: Friday, June 7, 2024 9:59
> To: dev <de...@flink.apache.org>
> Subject: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input
> Stream of Lookup Join
>
> Hi devs,
>
>
> I'd like to start a discussion about FLIP-462[1]: Support Custom Data
> Distribution for Input Stream of Lookup Join.
>
>
> Lookup Join is an important feature in Flink, It is typically used to
> enrich a table with data that is queried from an external system.
> If we interact with the external systems for each incoming record, we
> incur significant network IO and RPC overhead.
>
> Therefore, most connectors introduce caching to reduce the per-record
> level query overhead. However, because the data distribution of Lookup
> Join's input stream is arbitrary, the cache hit rate is sometimes
> unsatisfactory.
>
>
> We want to introduce a mechanism for the connector to tell the Flink
> planner its desired input stream data distribution or partitioning
> strategy. This can significantly reduce the amount of cached data and
> improve performance of Lookup Join.
>
>
> You can find more details in this FLIP[1]. Looking forward to hearing
> from you, thanks!
>
>
> Best regards,
>
> Weijie
>
>
> [1]
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
>
Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by weijie guo <gu...@gmail.com>.
Hi Zhanghao,
Thanks for the reply!
> Could you give a more concrete example in production on when a custom
partitioning strategy will outperform partitioning by key
The key point here is partitioning logic cannot be fully expressed with all
or part of the join key. That is, even if we know which fields are used to
calculate buckets, still have to face the following problem:
1. The mapping from the bucket field to the bucket id is not necessarily
done via hashcode, and even if it is, the hash algorithm may be different
from the one used in Flink. The planner can't know how to do this mapping.
2. In order to get the bucket id, we have to mod the bucket number, but
planner has no notion of bucket number.
Best regards,
Weijie
Zhanghao Chen <zh...@outlook.com> 于2024年6月12日周三 13:55写道：
> Thanks for driving this, Weijie. Usually, the data distribution of the
> external system is closely related to the keys, e.g. computing the bucket
> index by key hashcode % bucket num, so I'm not sure about how much
> difference there are between partitioning by key and a custom partitioning
> strategy. Could you give a more concrete example in production on when a
> custom partitioning strategy will outperform partitioning by key? Since
> you've mentioned Paimon in doc, maybe an example on Paimon.
>
> Best,
> Zhanghao Chen
> ________________________________
> From: weijie guo <gu...@gmail.com>
> Sent: Friday, June 7, 2024 9:59
> To: dev <de...@flink.apache.org>
> Subject: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input
> Stream of Lookup Join
>
> Hi devs,
>
>
> I'd like to start a discussion about FLIP-462[1]: Support Custom Data
> Distribution for Input Stream of Lookup Join.
>
>
> Lookup Join is an important feature in Flink, It is typically used to
> enrich a table with data that is queried from an external system.
> If we interact with the external systems for each incoming record, we
> incur significant network IO and RPC overhead.
>
> Therefore, most connectors introduce caching to reduce the per-record
> level query overhead. However, because the data distribution of Lookup
> Join's input stream is arbitrary, the cache hit rate is sometimes
> unsatisfactory.
>
>
> We want to introduce a mechanism for the connector to tell the Flink
> planner its desired input stream data distribution or partitioning
> strategy. This can significantly reduce the amount of cached data and
> improve performance of Lookup Join.
>
>
> You can find more details in this FLIP[1]. Looking forward to hearing
> from you, thanks!
>
>
> Best regards,
>
> Weijie
>
>
> [1]
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join
>
Re: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Posted by Zhanghao Chen <zh...@outlook.com>.
Thanks for driving this, Weijie. Usually, the data distribution of the external system is closely related to the keys, e.g. computing the bucket index by key hashcode % bucket num, so I'm not sure about how much difference there are between partitioning by key and a custom partitioning strategy. Could you give a more concrete example in production on when a custom partitioning strategy will outperform partitioning by key? Since you've mentioned Paimon in doc, maybe an example on Paimon.
Best,
Zhanghao Chen
________________________________
From: weijie guo <gu...@gmail.com>
Sent: Friday, June 7, 2024 9:59
To: dev <de...@flink.apache.org>
Subject: [DISCUSS] FLIP-462: Support Custom Data Distribution for Input Stream of Lookup Join
Hi devs,
I'd like to start a discussion about FLIP-462[1]: Support Custom Data
Distribution for Input Stream of Lookup Join.
Lookup Join is an important feature in Flink, It is typically used to
enrich a table with data that is queried from an external system.
If we interact with the external systems for each incoming record, we
incur significant network IO and RPC overhead.
Therefore, most connectors introduce caching to reduce the per-record
level query overhead. However, because the data distribution of Lookup
Join's input stream is arbitrary, the cache hit rate is sometimes
unsatisfactory.
We want to introduce a mechanism for the connector to tell the Flink
planner its desired input stream data distribution or partitioning
strategy. This can significantly reduce the amount of cached data and
improve performance of Lookup Join.
You can find more details in this FLIP[1]. Looking forward to hearing
from you, thanks!
Best regards,
Weijie
[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-462+Support+Custom+Data+Distribution+for+Input+Stream+of+Lookup+Join