jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!800 [I3DV3A] added concurrent update protection for the finishedQuery::iterator 

Merge pull request !800 from Nitin-Kashyap/defectFix.dynamicFilterCache
This commit is contained in:
i-robot 2021-03-29 20:48:39 +08:00 committed by Gitee
parent c03b77b861 7b571ca3e1
commit ecb3a52048
1 changed files with 19 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
presto-main/src/main/java/io/prestosql/dynamicfilter/DynamicFilterService.java
View File

@ -226,24 +226,26 @@ public class DynamicFilterService
List<String> handledQuery = new ArrayList<>();
StateMap mergedStateCollection = (StateMap) stateStoreProvider.getStateStore().getOrCreateStateCollection(DynamicFilterUtils.MERGED_DYNAMIC_FILTERS, MAP);
// Clear registered dynamic filter tasks
for (String queryId : finishedQuery) {
Map<String, DynamicFilterRegistryInfo> filters = dynamicFilters.get(queryId);
if (filters != null) {
for (Entry<String, DynamicFilterRegistryInfo> entry : filters.entrySet()) {
String filterId = entry.getKey();
clearPartialResults(filterId, queryId);
if (entry.getValue().isMerged()) {
String filterKey = createKey(DynamicFilterUtils.FILTERPREFIX, filterId, queryId);
mergedStateCollection.remove(filterKey);
synchronized (finishedQuery) {
for (String queryId : finishedQuery) {
Map<String, DynamicFilterRegistryInfo> filters = dynamicFilters.get(queryId);
if (filters != null) {
for (Entry<String, DynamicFilterRegistryInfo> entry : filters.entrySet()) {
String filterId = entry.getKey();
clearPartialResults(filterId, queryId);
if (entry.getValue().isMerged()) {
String filterKey = createKey(DynamicFilterUtils.FILTERPREFIX, filterId, queryId);
mergedStateCollection.remove(filterKey);
}
}
}
}
dynamicFilters.remove(queryId);
dynamicFilters.remove(queryId);
cachedDynamicFilters.remove(queryId);
handledQuery.add(queryId);
cachedDynamicFilters.remove(queryId);
handledQuery.add(queryId);
}
finishedQuery.removeAll(handledQuery);
}
finishedQuery.removeAll(handledQuery);
}
private static BloomFilter mergeBloomFilters(Collection<Object> partialBloomFilters)
@ -355,7 +357,9 @@ public class DynamicFilterService
*/
public void clearDynamicFiltersForQuery(String queryId)
{
finishedQuery.add(queryId);
synchronized (finishedQuery) {
finishedQuery.add(queryId);
}
}
/**