[Core]Support async lookup in hash store #4423
Conversation
neuyilan
changed the title
|
@JingsongLi PTAL, thanks |
There was a problem hiding this comment.
Hi @neuyilan , I took a rough look and found that there are many thread safety risks in certain areas. Can we go back to this requirement? Is it necessary for us to do multi-threaded access? Is this effective? Why not increase Flink's parallelism?
| } | ||
|
|
||
| private synchronized Triple<BinaryRow, Integer, InternalRow> extractPartitionAndBucket( | ||
| InternalRow key) { | ||
| InternalRow adjustedKey = key; | ||
| if (keyRearrange != null) { | ||
| adjustedKey = keyRearrange.replaceRow(adjustedKey); |
There was a problem hiding this comment.
I have encapsulated these as a separate function called extractPartartitionAndBucket. And it has been declared as synchronized, so there will be no thread safety issues.
| } | ||
|
|
||
| private synchronized Triple<BinaryRow, Integer, InternalRow> extractPartitionAndBucket( | ||
| InternalRow key) { | ||
| InternalRow adjustedKey = key; | ||
| if (keyRearrange != null) { | ||
| adjustedKey = keyRearrange.replaceRow(adjustedKey); | ||
| } | ||
| extractor.setRecord(adjustedKey); | ||
| int bucket = extractor.bucket(); | ||
| BinaryRow partition = extractor.partition(); |
| InternalRow adjustedKey = key; | ||
| if (keyRearrange != null) { | ||
| adjustedKey = keyRearrange.replaceRow(adjustedKey); | ||
| } | ||
| extractor.setRecord(adjustedKey); | ||
| int bucket = extractor.bucket(); | ||
| BinaryRow partition = extractor.partition(); | ||
|
|
||
| InternalRow trimmedKey = key; | ||
| if (trimmedKeyRearrange != null) { | ||
| trimmedKey = trimmedKeyRearrange.replaceRow(trimmedKey); |
Yes, there are too many thread safety issues in the current design. So essentially, access is synchronous. Even if ' I think supporting asynchronous multi-threaded access is necessary. And in our testing, it was effective. Asynchronous multi-threaded access has the following benefits compared to increasing concurrency in Flink:
|
|
Hi, @JingsongLi Could you please review it again. |
Currently, lookup join does not really support asynchronous, and the purpose of this PR is to support asynchronous lookup join.
This pr only support the hash store async lookup, in the next pr, I will do the rocksdb store async lookup.