Hi, Jing, Sorry for the late reply! The previous discussion for the hint syntax left a minor difference there: whether to use both sides of join table names or just one 'build' side table name only. I would prefer the later one. Users only need to pass the `build` side table(usually the smaller one) into `SHUFFLE_HASH(build_table)` join hint, more concisely than `SHUFFLE_HASH(left_table, right_table)`, WDYT?
Best, Lincoln Lee Jing Zhang <beyond1...@gmail.com> 于2022年1月15日周六 17:22写道: > Hi all, > Thanks for all the feedback so far. > If there is no more suggestions, I would like to drive a vote in Tuesday > next week (18 Jan). > > Best, > Jing Zhang > > Jing Zhang <beyond1...@gmail.com> 于2022年1月5日周三 11:33写道: > > > Hi Francesco, > > Thanks a lot for the feedback. > > > > > does it makes sense for a lookup join to use hash distribution whenever > > is possible by default? > > I prefer to enable the hash lookup join only find the hint in the query > > for the following reason: > > 1. Plan compatibility > > Add a hash shuffle by default would leads to the change of plan after > > users upgrade the flink version. > > Besides, lookup join is commonly used feature in flink SQL. > > 2. Not all flink jobs could benefit from this improvement. > > It is a trade off for the lookup join with dimension connectors which > > has cache inside. > > We hope the raise the cache hit ratio by Hash Lookup Join, however it > > would leads to an extra shuffle at the same time. > > It is not always a positive optimization, especially for the > > connectors which does not have cache inside. > > > > > Shouldn't the hint take the table alias as the "table name"? What if > > you do two lookup joins in cascade within the same query with the same > > table (once > > on a key, then on another one), where you use two different aliases for > > the table? > > In theory, it's better to support both table names and alias names. > > But in calcite, the alias name of subquery or table would not be lost in > > the sql conversion phase and sql optimization phase. > > So here we only support table names. > > > > Best, > > Jing Zhang > > > > > > Francesco Guardiani <france...@ververica.com> 于2022年1月3日周一 18:38写道: > > > >> Hi Jing, > >> > >> Thanks for the FLIP. I'm not very knowledgeable about the topic, but > going > >> through both the FLIP and the discussion here, I wonder, does it makes > >> sense for a lookup join to use hash distribution whenever is possible by > >> default? > >> > >> The point you're explaining here: > >> > >> > Many Lookup table sources introduce cache in order > >> to reduce the RPC call, such as JDBC, CSV, HBase connectors. > >> For those connectors, we could raise cache hit ratio by routing the same > >> lookup keys to the same task instance > >> > >> Seems something we can infer automatically, rather than manually asking > >> the > >> user to add this hint to the query. Note that I'm not talking against > the > >> hint syntax, which might still make sense to be introduced, but I feel > >> like > >> this optimization makes sense in the general case when using the > >> connectors > >> you have quoted. Perhaps there is some downside I'm not aware of? > >> > >> Talking about the hint themselves, taking this example as reference: > >> > >> SELECT /*+ SHUFFLE_HASH('Orders', 'Customers') */ o.order_id, o.total, > >> c.country, c.zip > >> FROM Orders AS o > >> JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c > >> ON o.customer_id = c.id; > >> > >> Shouldn't the hint take the table alias as the "table name"? What If you > >> do > >> two lookup joins in cascade within the same query with the same table > >> (once > >> on a key, then on another one), where you use two different aliases for > >> the > >> table? > >> > >> > >> On Fri, Dec 31, 2021 at 9:56 AM Jing Zhang <beyond1...@gmail.com> > wrote: > >> > >> > Hi Lincoln, > >> > Thanks for the feedback. > >> > > >> > > 1. For the hint name, +1 for WenLong's proposal. > >> > > >> > I've added add 'SHUFFLE_HASH' to other alternatives in FLIP. Let's > >> waiting > >> > for more voices here. > >> > > >> > > Regarding the `SKEW` hint, agree with you that it can be used > widely, > >> and > >> > I > >> > prefer to treat it as a metadata hint, a new category differs from a > >> join > >> > hint. > >> > For your example: > >> > ``` > >> > SELECT /*+ USE_HASH('Orders', 'Customers'), SKEW('Orders') */ > >> o.order_id, > >> > o.total, c.country, c.zip > >> > FROM Orders AS o > >> > JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c > >> > ON o.customer_id = c.id; > >> > ``` > >> > I would prefer another form: > >> > ``` > >> > -- provide the skew info to let the engine choose the optimal plan > >> > SELECT /*+ SKEW('Orders') */ o.order_id, ... > >> > > >> > -- or introduce a new hint for the join case, e.g., > >> > SELECT /*+ REPLICATED_SHUFFLE_HASH('Orders') */ o.order_id, ... > >> > ``` > >> > > >> > Maybe there is misunderstanding here. > >> > I just use a syntax sugar here. > >> > > >> > SELECT /*+ USE_HASH('Orders', 'Customers'), SKEW('Orders') */ > >> o.order_id, > >> > .... > >> > > >> > is just a syntax with > >> > > >> > SELECT /*+ USE_HASH('Orders', 'Customers') */ /*+SKEW('Orders') */ > >> > o.order_id, > >> > .... > >> > > >> > Although I list 'USE_HASH' and 'SKEW' hint in a query hint clause, it > >> does > >> > not mean they must appear together as a whole. > >> > Based on calcite syntax doc [1], you could list more than one hint in > >> > a /*+' hint [, hint ]* '*/ clause. > >> > > >> > Each hint has different function. > >> > The'USE_HASH' hint suggests the optimizer use hash partitioner for > >> Lookup > >> > Join for table 'Orders' and table 'Customers' while the 'SKEW' hint > >> tells > >> > the optimizer the skew metadata about the table 'Orders'. > >> > > >> > Best, > >> > Jing Zhang > >> > > >> > [1] https://calcite.apache.org/docs/reference.html#sql-hints > >> > > >> > > >> > > >> > > >> > Jing Zhang <beyond1...@gmail.com> 于2021年12月31日周五 16:39写道: > >> > > >> > > Hi Martijn, > >> > > Thanks for the feedback. > >> > > > >> > > Glad to hear that we reached a consensus on the first and second > >> point. > >> > > > >> > > About whether to use `use_hash` as a term, I think your concern > makes > >> > > sense. > >> > > Although the hash lookup join is similar to Hash join in oracle that > >> they > >> > > all require hash distribution on input, there exists a little > >> difference > >> > > between them. > >> > > About this point, Lincoln and WenLong both prefer the term > >> > 'SHUFFLE_HASH', > >> > > WDYT? > >> > > > >> > > Best, > >> > > Jing Zhang > >> > > > >> > > > >> > > Lincoln Lee <lincoln.8...@gmail.com> 于2021年12月30日周四 11:21写道: > >> > > > >> > >> Hi Jing, > >> > >> Thanks for your explanation! > >> > >> > >> > >> 1. For the hint name, +1 for WenLong's proposal. I think the > >> `SHUFFLE` > >> > >> keyword is important in a classic distributed computing system, > >> > >> a hash-join usually means there's a shuffle stage(include shuffle > >> > >> hash-join, broadcast hash-join). Users only need to pass the > `build` > >> > side > >> > >> table(usually the smaller one) into `SHUFFLE_HASH` join hint, more > >> > >> concisely than `USE_HASH(left_table, right_table)`. Please correct > >> me if > >> > >> my > >> > >> understanding is wrong. > >> > >> Regarding the `SKEW` hint, agree with you that it can be used > widely, > >> > and > >> > >> I > >> > >> prefer to treat it as a metadata hint, a new category differs from > a > >> > join > >> > >> hint. > >> > >> For your example: > >> > >> ``` > >> > >> SELECT /*+ USE_HASH('Orders', 'Customers'), SKEW('Orders') */ > >> > o.order_id, > >> > >> o.total, c.country, c.zip > >> > >> FROM Orders AS o > >> > >> JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c > >> > >> ON o.customer_id = c.id; > >> > >> ``` > >> > >> I would prefer another form: > >> > >> ``` > >> > >> -- provide the skew info to let the engine choose the optimal plan > >> > >> SELECT /*+ SKEW('Orders') */ o.order_id, ... > >> > >> > >> > >> -- or introduce a new hint for the join case, e.g., > >> > >> SELECT /*+ REPLICATED_SHUFFLE_HASH('Orders') */ o.order_id, ... > >> > >> ``` > >> > >> > >> > >> 2. Agree with Martin adding the feature to 1.16, we need time to > >> > complete > >> > >> the change in calcite and also the upgrading work. > >> > >> > >> > >> 3. I misunderstood the 'Other Alternatives' part as the 'Rejected' > >> ones > >> > in > >> > >> the FLIP doc. And my point is avoiding the hacky way with our best > >> > effort. > >> > >> The potential issues for calcite's hint propagation, e.g., join > hints > >> > >> correctly propagate into proper join scope include subquery or > views > >> > which > >> > >> may have various sql operators, so we should check all kinds of > >> > operators > >> > >> for the correct propagation. Hope this may help. And also cc @Shuo > >> Cheng > >> > >> may > >> > >> offer more help. > >> > >> > >> > >> > >> > >> Best, > >> > >> Lincoln Lee > >> > >> > >> > >> > >> > >> Martijn Visser <mart...@ververica.com> 于2021年12月29日周三 22:21写道: > >> > >> > >> > >> > Hi Jing, > >> > >> > > >> > >> > Thanks for explaining this in more detail and also to others > >> > >> > participating. > >> > >> > > >> > >> > > I think using query hints in this case is more natural for > users, > >> > >> WDYT? > >> > >> > > >> > >> > Yes, I agree. As long as we properly explain in our documentation > >> that > >> > >> we > >> > >> > support both Query Hints and Table Hints, what's the difference > >> > between > >> > >> > them and how to use them, I think our users can understand this > >> > >> perfectly. > >> > >> > > >> > >> > > I admit upgrading from Calcite 1.26 to 1.30 would be a big > >> change. > >> > >> > However we could not always avoid upgrade for the following > reason > >> > >> > > >> > >> > We have to upgrade Calcite. We actually considered putting that > in > >> the > >> > >> > Flink 1.15 scope but ultimately had to drop it, but I definitely > >> think > >> > >> this > >> > >> > needs to be done for 1.16. It's not only because of new features > >> that > >> > >> are > >> > >> > depending on Calcite upgrades, but also because newer versions > have > >> > >> > resolved bugs that also hurt our users. That's why we also > already > >> > have > >> > >> > tickets for upgrading to Calcite 1.27 [1] and 1.28 [2]. > >> > >> > > >> > >> > With regards to using `use_hash` as a term, I think the most > >> important > >> > >> part > >> > >> > is that if we re-use a term like Oracle is using, is that the > >> > behaviour > >> > >> and > >> > >> > outcome should be the same/comparable to the one from (in this > >> case) > >> > >> > Oracle. If their behaviour and outcome are not the same or > >> > comparable, I > >> > >> > would probably introduce our own term to avoid that users get > >> > confused. > >> > >> > > >> > >> > Best regards, > >> > >> > > >> > >> > Martijn > >> > >> > > >> > >> > [1] https://issues.apache.org/jira/browse/FLINK-20873 > >> > >> > [2] https://issues.apache.org/jira/browse/FLINK-21239 > >> > >> > > >> > >> > On Wed, 29 Dec 2021 at 14:18, Jing Zhang <beyond1...@gmail.com> > >> > wrote: > >> > >> > > >> > >> > > Hi Jian gang, > >> > >> > > Thanks for the feedback. > >> > >> > > > >> > >> > > > When it comes to hive, how do you load partial data instead > of > >> the > >> > >> > > whole data? Any change related with hive? > >> > >> > > > >> > >> > > The question is same as Yuan mentioned before. > >> > >> > > I prefer to drive another FLIP on this topic to further > >> discussion > >> > >> > > individually because this point involves many extension on API. > >> > >> > > Here I would like to share the implementation in our internal > >> > version > >> > >> > > firstly, it maybe very different with the final solution which > >> > merged > >> > >> to > >> > >> > > community. > >> > >> > > The core idea is push the partitioner information down to the > >> lookup > >> > >> > table > >> > >> > > source. > >> > >> > > Hive connector need also upgrades. When loading data into > caches, > >> > each > >> > >> > task > >> > >> > > could only store records which look keys are sent to current > >> task. > >> > >> > > > >> > >> > > > How to define the cache configuration? For example, the size > >> and > >> > the > >> > >> > ttl. > >> > >> > > > >> > >> > > I'm afraid there is no a unify caching configuration and cache > >> > >> > > implementation of different connectors yet. > >> > >> > > You could find cache size and ttl config of JDBC in doc [1], > >> HBase > >> > in > >> > >> doc > >> > >> > > [2] > >> > >> > > > >> > >> > > > Will this feature add another shuffle phase compared with > the > >> > >> default > >> > >> > > behavior? In what situations will user choose this feature? > >> > >> > > > >> > >> > > Yes, if user specify hash hint in query, optimizer would prefer > >> to > >> > >> choose > >> > >> > > Hash Lookup Join, which would add a Hash Shuffle. > >> > >> > > If lookup table source has cache inside (for example > HBase/Jdbc) > >> and > >> > >> the > >> > >> > > benefit of increasing cache hit ratio is bigger than add an > extra > >> > >> shuffle > >> > >> > > cost, the user could use Hash Lookup Join. > >> > >> > > > >> > >> > > > For the keys, the default implementation will be ok. But I > >> wonder > >> > >> > > whether we can support more flexible strategies. > >> > >> > > > >> > >> > > The question is same as Yuan mentioned before. > >> > >> > > > >> > >> > > I'm afraid there is no plan to support flexible strategies yet > >> > because > >> > >> > the > >> > >> > > feature involves many things, for example: > >> > >> > > 1. sql syntax > >> > >> > > 2. user defined partitioner API > >> > >> > > 3. RelDistribution type extension and Flink RelDistribution > >> > extension > >> > >> > > 4. FlinkExpandConversionRule > >> > >> > > 5. Exchange execNode extension > >> > >> > > 6. .... > >> > >> > > It needs well designed and more discussion. If this is a strong > >> > >> > > requirement, we would drive another discussion on this point > >> > >> > individually. > >> > >> > > In this FLIP, I would first support hash shuffle. WDYT? > >> > >> > > > >> > >> > > Best, > >> > >> > > Jing Zhang > >> > >> > > > >> > >> > > [1] > >> > >> > > > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/jdbc/#connector-options > >> > >> > > [2] > >> > >> > > > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/hbase/#connector-options > >> > >> > > > >> > >> > > Jing Zhang <beyond1...@gmail.com> 于2021年12月29日周三 20:37写道: > >> > >> > > > >> > >> > > > Hi Wenlong, > >> > >> > > > Thanks for the feedback. > >> > >> > > > I've checked similar syntax in other systems, they are all > >> > different > >> > >> > from > >> > >> > > > each other. It seems to be without consensus. > >> > >> > > > As mentioned in FLIP-204, oracle uses a query hint, the hint > >> name > >> > is > >> > >> > > > 'use_hash' [1]. > >> > >> > > > Spark also uses a query hint, its name is 'SHUFFLE_HASH' [2]. > >> > >> > > > SQL Server uses keyword 'HASH' instead of query hint [3]. > >> > >> > > > Note, the purposes of hash shuffle in [1][2][3] are a little > >> > >> different > >> > >> > > > from the purpose of FLIP-204, we just discuss syntax here. > >> > >> > > > > >> > >> > > > I've added this part to FLIP waiting for further discussion. > >> > >> > > > > >> > >> > > > Best, > >> > >> > > > Jing Zhang > >> > >> > > > > >> > >> > > > [1] > >> > >> > > > > >> > >> > > >> > >> > >> > > >> > https://docs.oracle.com/cd/B12037_01/server.101/b10752/hintsref.htm#5683 > >> > >> > > > [2] > >> > >> > > > > >> > >> > > >> > >> > >> > > >> > https://spark.apache.org/docs/3.0.0/sql-ref-syntax-qry-select-hints.html > >> > >> > > > [3] > >> > >> > > > > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://docs.microsoft.com/en-us/sql/t-sql/queries/hints-transact-sql-join?view=sql-server-ver15 > >> > >> > > > > >> > >> > > > > >> > >> > > > wenlong.lwl <wenlong88....@gmail.com> 于2021年12月29日周三 > 17:18写道: > >> > >> > > > > >> > >> > > >> Hi, Jing, thanks for driving the discussion. > >> > >> > > >> > >> > >> > > >> Have you made some investigation on the syntax of join hint? > >> > >> > > >> Why do you choose USE_HASH from oracle instead of the style > of > >> > >> spark > >> > >> > > >> SHUFFLE_HASH, they are quite different. > >> > >> > > >> People in the big data world may be more familiar with > >> > spark/hive, > >> > >> if > >> > >> > we > >> > >> > > >> need to choose one, personally, I prefer the style of spark. > >> > >> > > >> > >> > >> > > >> > >> > >> > > >> Best, > >> > >> > > >> Wenlong > >> > >> > > >> > >> > >> > > >> On Wed, 29 Dec 2021 at 16:48, zst...@163.com < > zst...@163.com> > >> > >> wrote: > >> > >> > > >> > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > Hi Jing, > >> > >> > > >> > Thanks for your detail reply. > >> > >> > > >> > 1) In the last suggestion, hash by primary key is not use > >> for > >> > >> > raising > >> > >> > > >> the > >> > >> > > >> > cache hit, but handling with skew of left source. Now that > >> you > >> > >> have > >> > >> > > >> 'skew' > >> > >> > > >> > hint and other discussion about it, I'm looking forward to > >> it. > >> > >> > > >> > 2) I mean to support user defined partitioner function. We > >> > have a > >> > >> > case > >> > >> > > >> > that joining a datalake source with special way of > >> partition, > >> > and > >> > >> > have > >> > >> > > >> > implemented not elegantly in our internal version. As you > >> said, > >> > >> it > >> > >> > > needs > >> > >> > > >> > more design. > >> > >> > > >> > 3) I thing so-called 'HashPartitionedCache' is usefull, > >> > otherwise > >> > >> > > >> loading > >> > >> > > >> > all data such as hive lookup table source is almost not > >> > >> available in > >> > >> > > big > >> > >> > > >> > data. > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > Best regards, > >> > >> > > >> > Yuan > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > > >> > >> > > >> > 在 2021-12-29 14:52:11,"Jing Zhang" <beyond1...@gmail.com> > >> 写道: > >> > >> > > >> > >Hi, Lincoln > >> > >> > > >> > >Thanks a lot for the feedback. > >> > >> > > >> > > > >> > >> > > >> > >> Regarding the hint name ‘USE_HASH’, could we consider > >> more > >> > >> > > >> candidates? > >> > >> > > >> > >Things are a little different from RDBMS in the > distributed > >> > >> world, > >> > >> > > and > >> > >> > > >> we > >> > >> > > >> > >also aim to solve the data skew problem, so all these > >> incoming > >> > >> > hints > >> > >> > > >> names > >> > >> > > >> > >should be considered together. > >> > >> > > >> > > > >> > >> > > >> > >About skew problem, I would discuss this in next FLIP > >> > >> > individually. I > >> > >> > > >> > would > >> > >> > > >> > >like to share hint proposal for skew here. > >> > >> > > >> > >I want to introduce 'skew' hint which is a query hint, > >> similar > >> > >> with > >> > >> > > >> skew > >> > >> > > >> > >hint in spark [1] and MaxCompute[2]. > >> > >> > > >> > >The 'skew' hint could only contain the name of the table > >> with > >> > >> skew. > >> > >> > > >> > >Besides, skew hint could accept table name and column > >> names. > >> > >> > > >> > >In addition, skew hint could accept table name, column > >> names > >> > and > >> > >> > skew > >> > >> > > >> > >values. > >> > >> > > >> > >For example: > >> > >> > > >> > > > >> > >> > > >> > >SELECT /*+ USE_HASH('Orders', 'Customers'), > SKEW('Orders') > >> */ > >> > >> > > >> o.order_id, > >> > >> > > >> > >o.total, c.country, c.zip > >> > >> > > >> > >FROM Orders AS o > >> > >> > > >> > >JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c > >> > >> > > >> > >ON o.customer_id = c.id; > >> > >> > > >> > > > >> > >> > > >> > >The 'skew' hint is not only used for look up join here, > but > >> > also > >> > >> > > could > >> > >> > > >> be > >> > >> > > >> > >used for other types of join later, for example, batch > hash > >> > >> join or > >> > >> > > >> > >streaming regular join. > >> > >> > > >> > >Go back to better name problem for hash look up join. > Since > >> > the > >> > >> > > 'skew' > >> > >> > > >> > hint > >> > >> > > >> > >is a separate hint, so 'use_hash' is still an > alternative. > >> > >> > > >> > >WDYT? > >> > >> > > >> > >I don't have a good idea about the better hint name yet. > I > >> > would > >> > >> > like > >> > >> > > >> to > >> > >> > > >> > >heard more suggestions about hint names. > >> > >> > > >> > > > >> > >> > > >> > >> As you mentioned in the flip, this solution depends on > >> > future > >> > >> > > >> changes > >> > >> > > >> > to > >> > >> > > >> > >calcite (and also upgrading calcite would be another > >> possible > >> > >> big > >> > >> > > >> change: > >> > >> > > >> > >at least calicite-1.30 vs 1.26, are we preparing to > accept > >> > this > >> > >> big > >> > >> > > >> > >change?). > >> > >> > > >> > > > >> > >> > > >> > >Indeed, solution 1 depends on calcite upgrade. > >> > >> > > >> > >I admit upgrade from Calcite 1.26 to 1.30 would be a big > >> > >> change. I > >> > >> > > >> still > >> > >> > > >> > >remember what we have suffered from last upgrade to > Calcite > >> > >> 1.26. > >> > >> > > >> > >However we could not always avoid upgrade for the > following > >> > >> reason: > >> > >> > > >> > >1. Other features also depends on the Calcite upgrade. > For > >> > >> example, > >> > >> > > >> > Session > >> > >> > > >> > >Window and Count Window. > >> > >> > > >> > >2. If we always avoid Calcite upgrade, there would be > more > >> gap > >> > >> with > >> > >> > > the > >> > >> > > >> > >latest version. One day, if upgrading becomes a thing > which > >> > has > >> > >> to > >> > >> > be > >> > >> > > >> > done, > >> > >> > > >> > >the pain is more. > >> > >> > > >> > > > >> > >> > > >> > >WDYT? > >> > >> > > >> > > > >> > >> > > >> > >> Is there another possible way to minimize the change > in > >> > >> calcite? > >> > >> > > >> > > > >> > >> > > >> > >Do you check the 'Other Alternatives' part in the > >> FLIP-204? It > >> > >> > gives > >> > >> > > >> > >another solution which does not depend on calcite upgrade > >> and > >> > do > >> > >> > not > >> > >> > > >> need > >> > >> > > >> > >to worry about the hint would be missed in the > propagation. > >> > >> > > >> > >This is also what we have done in the internal version. > >> > >> > > >> > >The core idea is propagating 'use_hash' hint to TableScan > >> with > >> > >> > > matched > >> > >> > > >> > >table names. However, it is a little hacky. > >> > >> > > >> > > > >> > >> > > >> > >> As I know there're more limitations than `Correlate`. > >> > >> > > >> > > > >> > >> > > >> > >As mentioned before, in our external version, I choose > the > >> the > >> > >> > 'Other > >> > >> > > >> > >Alternatives' part in the FLIP-204. > >> > >> > > >> > >Although I do a POC in the solution 1 and lists all > >> changes I > >> > >> found > >> > >> > > in > >> > >> > > >> the > >> > >> > > >> > >FLIP, there may still be something I missed. > >> > >> > > >> > >I'm very happy to hear that you point out there're more > >> > >> limitations > >> > >> > > >> except > >> > >> > > >> > >for `Correlate`, would you please give more details on > this > >> > >> part? > >> > >> > > >> > > > >> > >> > > >> > >Best, > >> > >> > > >> > >Jing Zhang > >> > >> > > >> > > > >> > >> > > >> > >[1] > >> > >> > > > >> https://docs.databricks.com/delta/join-performance/skew-join.html > >> > >> > > >> > >[2] > >> > >> > > >> > > > >> > >> > > >> > > >> > >> > > >> > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://help.aliyun.com/apsara/enterprise/v_3_13_0_20201215/odps/enterprise-ascm-user-guide/hotspot-tilt.html?spm=a2c4g.14484438.10001.669 > >> > >> > > >> > > > >> > >> > > >> > >Jing Zhang <beyond1...@gmail.com> 于2021年12月29日周三 > 14:40写道: > >> > >> > > >> > > > >> > >> > > >> > >> Hi Yuan and Lincoln, > >> > >> > > >> > >> thanks a lot for the attention. I would answer the > email > >> one > >> > >> by > >> > >> > > one. > >> > >> > > >> > >> > >> > >> > > >> > >> To Yuan > >> > >> > > >> > >> > How shall we deal with CDC data? If there is CDC data > >> in > >> > the > >> > >> > > >> pipeline, > >> > >> > > >> > >> IMHO, shuffle by join key will cause CDC data disorder. > >> Will > >> > >> it > >> > >> > be > >> > >> > > >> > better > >> > >> > > >> > >> to use primary key in this case? > >> > >> > > >> > >> > >> > >> > > >> > >> Good question. > >> > >> > > >> > >> The problem could not only exists in CDC data source, > but > >> > also > >> > >> > > exists > >> > >> > > >> > when > >> > >> > > >> > >> the input stream is not insert-only stream (for > example, > >> the > >> > >> > result > >> > >> > > >> of > >> > >> > > >> > >> unbounded aggregate or regular join). > >> > >> > > >> > >> I think use hash by primary key is not a good choise. > It > >> > could > >> > >> > not > >> > >> > > >> raise > >> > >> > > >> > >> the cache hit because cache key is look up key instead > of > >> > >> primary > >> > >> > > >> key of > >> > >> > > >> > >> input. > >> > >> > > >> > >> > >> > >> > > >> > >> To avoid wrong result, hash lookup Join requires that > the > >> > >> input > >> > >> > > >> stream > >> > >> > > >> > >> should be insert_only stream or its upsert keys > contains > >> > >> lookup > >> > >> > > keys. > >> > >> > > >> > >> > >> > >> > > >> > >> I've added this limitation to FLIP, thanks a lot for > >> > >> reminding. > >> > >> > > >> > >> > >> > >> > > >> > >> > If the shuffle keys can be customized when users > have > >> the > >> > >> > > >> knowledge > >> > >> > > >> > >> about distribution of data? > >> > >> > > >> > >> > >> > >> > > >> > >> I'm not sure I understand your question. > >> > >> > > >> > >> > >> > >> > > >> > >> Do you mean to support user defined partitioner > function > >> on > >> > >> keys > >> > >> > > just > >> > >> > > >> > like > >> > >> > > >> > >> flink DataStream sql? > >> > >> > > >> > >> If yes, I'm afraid there is no plan to support this > >> feature > >> > >> yet > >> > >> > > >> because > >> > >> > > >> > >> the feature involves many things, for example: > >> > >> > > >> > >> 1. sql syntax > >> > >> > > >> > >> 2. user defined partitioner API > >> > >> > > >> > >> 3. RelDistribution type extension and Flink > >> RelDistribution > >> > >> > > extension > >> > >> > > >> > >> 4. FlinkExpandConversionRule > >> > >> > > >> > >> 5. Exchange execNode extension > >> > >> > > >> > >> 6. .... > >> > >> > > >> > >> It needs well designed and more discussion. If this is > a > >> > >> strong > >> > >> > > >> > >> requirement, we would drive another discussion on this > >> point > >> > >> > > >> > individually. > >> > >> > > >> > >> In this FLIP, I would first support hash shuffle. WDYT? > >> > >> > > >> > >> > >> > >> > > >> > >> Or do you mean support hash by other keys instead of > >> lookup > >> > >> key? > >> > >> > > >> > >> If yes, would you please tell me a specific user case? > >> > >> > > >> > >> We need to fetch the record from external storage of > >> > dimension > >> > >> > > table > >> > >> > > >> by > >> > >> > > >> > >> look up key, so those dimension table source uses look > up > >> > >> keys as > >> > >> > > >> cache > >> > >> > > >> > >> key. > >> > >> > > >> > >> We could only increase the cache ratio by shuffle > lookup > >> > >> keys. > >> > >> > > >> > >> I need more use cases to understand this requirement. > >> > >> > > >> > >> > >> > >> > > >> > >> > Some connectors such as hive, caches all data in > >> > >> > LookupFunction. > >> > >> > > >> How > >> > >> > > >> > to > >> > >> > > >> > >> decrease the valid cache data size if data can be > >> shuffled? > >> > >> > > >> > >> > >> > >> > > >> > >> Very good idea. > >> > >> > > >> > >> There are two types of cache. > >> > >> > > >> > >> For Key-Value storage, such as Redis/HBase, the lookup > >> table > >> > >> > source > >> > >> > > >> > stores > >> > >> > > >> > >> the visited lookup keys and it's record into cache > >> lazily. > >> > >> > > >> > >> For other storage without keys, such as hive, each task > >> > loads > >> > >> all > >> > >> > > >> data > >> > >> > > >> > >> into cache eagerly in the initialize phase. > >> > >> > > >> > >> After introduce hash partitioner, for key-value > storages, > >> > >> there > >> > >> > is > >> > >> > > no > >> > >> > > >> > need > >> > >> > > >> > >> to change; for hive, each task could only load part of > >> cache > >> > >> > > instead > >> > >> > > >> of > >> > >> > > >> > >> load all cache. > >> > >> > > >> > >> > >> > >> > > >> > >> We have implemented this optimization in our internal > >> > version. > >> > >> > > >> > >> The core idea is push the partitioner information down > to > >> > the > >> > >> > > lookup > >> > >> > > >> > table > >> > >> > > >> > >> source. When loading data into caches, each task could > >> only > >> > >> store > >> > >> > > >> those > >> > >> > > >> > >> records which look keys are sent to current task. > >> > >> > > >> > >> We called this 'HashPartitionedCache'. > >> > >> > > >> > >> > >> > >> > > >> > >> I have added this point into the Lookup Join > requirements > >> > >> list in > >> > >> > > the > >> > >> > > >> > >> motivation of the FLIP, but I would not do this point > in > >> > this > >> > >> > FLIP > >> > >> > > >> right > >> > >> > > >> > >> now. > >> > >> > > >> > >> If this is a strong requirement, we need drive another > >> > >> discussion > >> > >> > > on > >> > >> > > >> > this > >> > >> > > >> > >> topic individually because this point involves many > >> > extension > >> > >> on > >> > >> > > API. > >> > >> > > >> > >> > >> > >> > > >> > >> Best, > >> > >> > > >> > >> Jing Zhang > >> > >> > > >> > >> > >> > >> > > >> > >> > >> > >> > > >> > >> Lincoln Lee <lincoln.8...@gmail.com> 于2021年12月29日周三 > >> > 10:01写道: > >> > >> > > >> > >> > >> > >> > > >> > >>> Hi Jing, > >> > >> > > >> > >>> Thanks for bringing up this discussion! Agree > that > >> > this > >> > >> > join > >> > >> > > >> hints > >> > >> > > >> > >>> should benefit both bounded and unbounded cases as > >> Martin > >> > >> > > mentioned. > >> > >> > > >> > >>> I also agree that implementing the query hint is the > >> right > >> > >> way > >> > >> > > for a > >> > >> > > >> > more > >> > >> > > >> > >>> general purpose since the dynamic table options has a > >> > limited > >> > >> > > scope. > >> > >> > > >> > >>> Some points I'd like to share are: > >> > >> > > >> > >>> 1. Regarding the hint name ‘USE_HASH’, could we > consider > >> > more > >> > >> > > >> > candidates? > >> > >> > > >> > >>> Things are a little different from RDBMS in the > >> distributed > >> > >> > world, > >> > >> > > >> and > >> > >> > > >> > we > >> > >> > > >> > >>> also aim to solve the data skew problem, so all these > >> > >> incoming > >> > >> > > hints > >> > >> > > >> > names > >> > >> > > >> > >>> should be considered together. > >> > >> > > >> > >>> 2. As you mentioned in the flip, this solution depends > >> on > >> > >> future > >> > >> > > >> > changes > >> > >> > > >> > >>> to > >> > >> > > >> > >>> calcite (and also upgrading calcite would be another > >> > possible > >> > >> > big > >> > >> > > >> > change: > >> > >> > > >> > >>> at least calicite-1.30 vs 1.26, are we preparing to > >> accept > >> > >> this > >> > >> > > big > >> > >> > > >> > >>> change?). Is there another possible way to minimize > the > >> > >> change > >> > >> > in > >> > >> > > >> > calcite? > >> > >> > > >> > >>> As I know there're more limitations than `Correlate`. > >> > >> > > >> > >>> > >> > >> > > >> > >>> Best, > >> > >> > > >> > >>> Lincoln Lee > >> > >> > > >> > >>> > >> > >> > > >> > >>> > >> > >> > > >> > >>> Jing Zhang <beyond1...@gmail.com> 于2021年12月28日周二 > >> 23:04写道: > >> > >> > > >> > >>> > >> > >> > > >> > >>> > Hi Martijn, > >> > >> > > >> > >>> > Thanks a lot for your attention. > >> > >> > > >> > >>> > I'm sorry I didn't explain the motivation clearly. I > >> > would > >> > >> > like > >> > >> > > to > >> > >> > > >> > >>> explain > >> > >> > > >> > >>> > it in detail, and then give response on your > >> questions. > >> > >> > > >> > >>> > A lookup join is typically used to enrich a table > with > >> > data > >> > >> > that > >> > >> > > >> is > >> > >> > > >> > >>> queried > >> > >> > > >> > >>> > from an external system. Many Lookup table sources > >> > >> introduce > >> > >> > > >> cache in > >> > >> > > >> > >>> order > >> > >> > > >> > >>> > to reduce the RPC call, such as JDBC, CSV, HBase > >> > >> connectors. > >> > >> > > >> > >>> > For those connectors, we could raise cache hit ratio > >> by > >> > >> > routing > >> > >> > > >> the > >> > >> > > >> > same > >> > >> > > >> > >>> > lookup keys to the same task instance. This is the > >> > purpose > >> > >> of > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > >> > >> > > >> > > >> > >> > > >> > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join > >> > >> > > >> > >>> > . > >> > >> > > >> > >>> > Other cases might benefit from Hash distribution, > >> such as > >> > >> > batch > >> > >> > > >> hash > >> > >> > > >> > >>> join > >> > >> > > >> > >>> > as you mentioned. It is a cool idea, however it is > not > >> > the > >> > >> > > >> purpose of > >> > >> > > >> > >>> this > >> > >> > > >> > >>> > FLIP, we could discuss this in FLINK-20670 > >> > >> > > >> > >>> > <https://issues.apache.org/jira/browse/FLINK-20670 > >. > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > > - When I was reading about this topic [1] I was > >> > >> wondering if > >> > >> > > >> this > >> > >> > > >> > >>> feature > >> > >> > > >> > >>> > would be more beneficial for bounded use cases and > >> not so > >> > >> much > >> > >> > > for > >> > >> > > >> > >>> > unbounded use cases. What do you think? > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > As mentioned before, the purpose of Hash Lookup Join > >> is > >> > to > >> > >> > > >> increase > >> > >> > > >> > the > >> > >> > > >> > >>> > cache hit ratio which is different from Oracle Hash > >> Join. > >> > >> > > However > >> > >> > > >> we > >> > >> > > >> > >>> could > >> > >> > > >> > >>> > use the similar hint syntax. > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > > - If I look at the current documentation for SQL > >> Hints > >> > in > >> > >> > > Flink > >> > >> > > >> > [2], I > >> > >> > > >> > >>> > notice that all of the hints there are located at > the > >> end > >> > >> of > >> > >> > the > >> > >> > > >> SQL > >> > >> > > >> > >>> > statement. In the FLIP, the use_hash is defined > >> directly > >> > >> after > >> > >> > > the > >> > >> > > >> > >>> 'SELECT' > >> > >> > > >> > >>> > keyword. Can we somehow make this consistent for the > >> > user? > >> > >> Or > >> > >> > > >> should > >> > >> > > >> > the > >> > >> > > >> > >>> > user be able to specify hints anywhere in its SQL > >> > >> statement? > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > Calcite supports hints in two locations [3]: > >> > >> > > >> > >>> > Query Hint: right after the SELECT keyword; > >> > >> > > >> > >>> > Table Hint: right after the referenced table name. > >> > >> > > >> > >>> > Now Flink has supported dynamic table options based > on > >> > the > >> > >> > Hint > >> > >> > > >> > >>> framework > >> > >> > > >> > >>> > of Calcite which is mentioned in doc[2]. > >> > >> > > >> > >>> > Besides, query hints are also important, it could > >> give a > >> > >> hint > >> > >> > > for > >> > >> > > >> > >>> > optimizers to choose a better plan. Almost all > popular > >> > >> > databases > >> > >> > > >> and > >> > >> > > >> > >>> > big-data engines support sql query hints, such as > >> oracle, > >> > >> > hive, > >> > >> > > >> spark > >> > >> > > >> > >>> and > >> > >> > > >> > >>> > so on. > >> > >> > > >> > >>> > I think using query hints in this case is more > natural > >> > for > >> > >> > > users, > >> > >> > > >> > WDYT? > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > I have updated the motivation part in the FLIP, > >> > >> > > >> > >>> > Thanks for the feedback! > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > [1] > >> > >> > > >> > >> https://logicalread.com/oracle-11g-hash-joins-mc02/#.V5Wm4_mnoUI > >> > >> > > >> > >>> > [2] > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > >> > >> > > >> > > >> > >> > > >> > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/hints/ > >> > >> > > >> > >>> > [3] > >> > >> https://calcite.apache.org/docs/reference.html#sql-hints > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > Best, > >> > >> > > >> > >>> > Jing Zhang > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > Martijn Visser <mart...@ververica.com> > 于2021年12月28日周二 > >> > >> > 22:02写道: > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > > Hi Jing, > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > Thanks a lot for the explanation and the FLIP. I > >> > >> definitely > >> > >> > > >> learned > >> > >> > > >> > >>> > > something when reading more about `use_hash`. My > >> > >> > > interpretation > >> > >> > > >> > would > >> > >> > > >> > >>> be > >> > >> > > >> > >>> > > that the primary benefit of a hash lookup join > >> would be > >> > >> > > improved > >> > >> > > >> > >>> > > performance by allowing the user to explicitly > >> optimise > >> > >> the > >> > >> > > >> > planner. > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > I have a couple of questions: > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > - When I was reading about this topic [1] I was > >> > >> wondering if > >> > >> > > >> this > >> > >> > > >> > >>> feature > >> > >> > > >> > >>> > > would be more beneficial for bounded use cases and > >> not > >> > so > >> > >> > much > >> > >> > > >> for > >> > >> > > >> > >>> > > unbounded use cases. What do you think? > >> > >> > > >> > >>> > > - If I look at the current documentation for SQL > >> Hints > >> > in > >> > >> > > Flink > >> > >> > > >> > [2], I > >> > >> > > >> > >>> > > notice that all of the hints there are located at > >> the > >> > >> end of > >> > >> > > the > >> > >> > > >> > SQL > >> > >> > > >> > >>> > > statement. In the FLIP, the use_hash is defined > >> > directly > >> > >> > after > >> > >> > > >> the > >> > >> > > >> > >>> > 'SELECT' > >> > >> > > >> > >>> > > keyword. Can we somehow make this consistent for > the > >> > >> user? > >> > >> > Or > >> > >> > > >> > should > >> > >> > > >> > >>> the > >> > >> > > >> > >>> > > user be able to specify hints anywhere in its SQL > >> > >> statement? > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > Best regards, > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > Martijn > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > [1] > >> > >> > > >> > > >> > https://logicalread.com/oracle-11g-hash-joins-mc02/#.V5Wm4_mnoUI > >> > >> > > >> > >>> > > [2] > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > >> > >> > > >> > > >> > >> > > >> > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/hints/ > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > On Tue, 28 Dec 2021 at 08:17, Jing Zhang < > >> > >> > > beyond1...@gmail.com> > >> > >> > > >> > >>> wrote: > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > > Hi everyone, > >> > >> > > >> > >>> > > > Look up join > >> > >> > > >> > >>> > > > < > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > >> > >> > > >> > > >> > >> > > >> > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/joins/#lookup-join > >> > >> > > >> > >>> > > > >[1] > >> > >> > > >> > >>> > > > is > >> > >> > > >> > >>> > > > commonly used feature in Flink SQL. We have > >> received > >> > >> many > >> > >> > > >> > >>> optimization > >> > >> > > >> > >>> > > > requirements on look up join. For example: > >> > >> > > >> > >>> > > > 1. Enforces left side of lookup join do a hash > >> > >> partitioner > >> > >> > > to > >> > >> > > >> > raise > >> > >> > > >> > >>> > cache > >> > >> > > >> > >>> > > > hint ratio > >> > >> > > >> > >>> > > > 2. Solves the data skew problem after introduces > >> hash > >> > >> > lookup > >> > >> > > >> join > >> > >> > > >> > >>> > > > 3. Enables mini-batch optimization to reduce RPC > >> call > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > Next we will solve these problems one by one. > >> > >> Firstly, we > >> > >> > > >> would > >> > >> > > >> > >>> focus > >> > >> > > >> > >>> > on > >> > >> > > >> > >>> > > > point 1, and continue to discuss point 2 and > >> point 3 > >> > >> > later. > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > There are many similar requirements from user > mail > >> > list > >> > >> > and > >> > >> > > >> JIRA > >> > >> > > >> > >>> about > >> > >> > > >> > >>> > > hash > >> > >> > > >> > >>> > > > Lookup Join, for example: > >> > >> > > >> > >>> > > > 1. FLINK-23687 < > >> > >> > > >> > https://issues.apache.org/jira/browse/FLINK-23687> > >> > >> > > >> > >>> - > >> > >> > > >> > >>> > > > Introduce partitioned lookup join to enforce > >> input of > >> > >> > > >> LookupJoin > >> > >> > > >> > to > >> > >> > > >> > >>> > hash > >> > >> > > >> > >>> > > > shuffle by lookup keys > >> > >> > > >> > >>> > > > 2. FLINK-25396 < > >> > >> > > >> > https://issues.apache.org/jira/browse/FLINK-25396> > >> > >> > > >> > >>> - > >> > >> > > >> > >>> > > > lookupjoin source table for pre-partitioning > >> > >> > > >> > >>> > > > 3. FLINK-25262 < > >> > >> > > >> > https://issues.apache.org/jira/browse/FLINK-25262> > >> > >> > > >> > >>> - > >> > >> > > >> > >>> > > > Support to send data to lookup table for > >> > >> > > >> > KeyGroupStreamPartitioner > >> > >> > > >> > >>> way > >> > >> > > >> > >>> > > for > >> > >> > > >> > >>> > > > SQL. > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > In this FLIP, I would like to start a discussion > >> > about > >> > >> > Hash > >> > >> > > >> > Lookup > >> > >> > > >> > >>> > Join. > >> > >> > > >> > >>> > > > The core idea is introducing a 'USE_HASH' hint > in > >> > >> query. > >> > >> > > This > >> > >> > > >> > >>> syntax > >> > >> > > >> > >>> > is > >> > >> > > >> > >>> > > > directly user-oriented and therefore requires > >> careful > >> > >> > > design. > >> > >> > > >> > >>> > > > There are two ways about how to propagate this > >> hint > >> > to > >> > >> > > >> > LookupJoin in > >> > >> > > >> > >>> > > > optimizer. We need further discussion to do > final > >> > >> decide. > >> > >> > > >> Anyway, > >> > >> > > >> > >>> the > >> > >> > > >> > >>> > > > difference between the two solution is only > about > >> the > >> > >> > > internal > >> > >> > > >> > >>> > > > implementation and has no impact on the user. > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > For more detail on the proposal: > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > >> > >> > > >> > > >> > >> > > >> > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-204%3A+Introduce+Hash+Lookup+Join > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > Looking forward to your feedback, thanks. > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > Best, > >> > >> > > >> > >>> > > > Jing Zhang > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > [1] > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > >> > >> > > >> > > >> > >> > > >> > >> > >> > > > >> > >> > > >> > >> > >> > > >> > https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/joins/#lookup-join > >> > >> > > >> > >>> > > > > >> > >> > > >> > >>> > > > >> > >> > > >> > >>> > > >> > >> > > >> > >>> > >> > >> > > >> > >> > >> > >> > > >> > > >> > >> > > >> > >> > >> > > > > >> > >> > > > >> > >> > > >> > >> > >> > > > >> > > >> > >> >