XComp commented on code in PR #20919:
URL: https://github.com/apache/flink/pull/20919#discussion_r1084051452
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/fullcache/inputformat/InputFormatCacheLoader.java:
##########
@@ -71,56 +72,56 @@ public void open(Configuration parameters) throws Exception
{
}
@Override
- protected void reloadCache() throws Exception {
+ protected boolean updateCache() throws Exception {
InputSplit[] inputSplits = createInputSplits();
int numSplits = inputSplits.length;
+ int concurrencyLevel = getConcurrencyLevel(numSplits);
// load data into the another copy of cache
- // notice: it requires twice more memory, but on the other hand we
don't need any blocking
+ // notice: it requires twice more memory, but on the other hand we
don't need any blocking;
// cache has default initialCapacity and loadFactor, but overridden
concurrencyLevel
ConcurrentHashMap<RowData, Collection<RowData>> newCache =
- new ConcurrentHashMap<>(16, 0.75f,
getConcurrencyLevel(numSplits));
- this.cacheLoadTasks =
+ new ConcurrentHashMap<>(16, 0.75f, concurrencyLevel);
+ Deque<InputSplitCacheLoadTask> cacheLoadTasks =
Arrays.stream(inputSplits)
.map(split -> createCacheLoadTask(split, newCache))
- .collect(Collectors.toList());
- if (isStopped) {
- // check for cases when #close was called during reload before
creating cacheLoadTasks
- return;
- }
- // run first task or create numSplits threads to run all tasks
+ .collect(Collectors.toCollection(ArrayDeque::new));
+ // run first task and create concurrencyLevel - 1 threads to run
remaining tasks
ExecutorService cacheLoadTaskService = null;
+ boolean wasInterrupted;
Review Comment:
```suggestion
boolean wasInterrupted = false;
```
I know that `false` is the default value but for readability purposes it's
good to make this explicit, I think.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingLookupFunction.java:
##########
@@ -107,12 +107,12 @@ public void open(FunctionContext context) throws
Exception {
numLoadFailuresCounter = new SimpleCounter();
cacheMetricGroup.numLoadFailuresCounter(numLoadFailuresCounter);
}
- // Initialize cache and the delegating function
- cache.open(cacheMetricGroup);
if (cache instanceof LookupFullCache) {
// TODO add Configuration into FunctionContext
- ((LookupFullCache) cache).open(new Configuration());
+ ((LookupFullCache) cache).setParameters(new Configuration());
Review Comment:
What value does this call bring here? Can't we get rid of the if block
entirely? We're setting an empty Conflguration here which is already the
default within `CachingLookupFunction`. I don't see
`LookupFullCache.setParameter` being called anywhere else? :thinking:
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingLookupFunction.java:
##########
@@ -107,12 +107,12 @@ public void open(FunctionContext context) throws
Exception {
numLoadFailuresCounter = new SimpleCounter();
cacheMetricGroup.numLoadFailuresCounter(numLoadFailuresCounter);
}
- // Initialize cache and the delegating function
- cache.open(cacheMetricGroup);
if (cache instanceof LookupFullCache) {
// TODO add Configuration into FunctionContext
Review Comment:
I know that this is not part of the current PR. But just as a side-note:
It's better to create follow-up Jira issues rather than hiding planned changes
in a TODO comment.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/fullcache/inputformat/InputFormatCacheLoader.java:
##########
@@ -71,58 +72,58 @@ public void open(Configuration parameters) throws Exception
{
}
@Override
- protected void reloadCache() throws Exception {
+ protected void updateCache() throws Exception {
InputSplit[] inputSplits = createInputSplits();
int numSplits = inputSplits.length;
+ int concurrencyLevel = getConcurrencyLevel(numSplits);
// load data into the another copy of cache
- // notice: it requires twice more memory, but on the other hand we
don't need any blocking
+ // notice: it requires twice more memory, but on the other hand we
don't need any blocking;
// cache has default initialCapacity and loadFactor, but overridden
concurrencyLevel
ConcurrentHashMap<RowData, Collection<RowData>> newCache =
- new ConcurrentHashMap<>(16, 0.75f,
getConcurrencyLevel(numSplits));
- this.cacheLoadTasks =
+ new ConcurrentHashMap<>(16, 0.75f, concurrencyLevel);
+ Deque<InputSplitCacheLoadTask> cacheLoadTasks =
Arrays.stream(inputSplits)
.map(split -> createCacheLoadTask(split, newCache))
- .collect(Collectors.toList());
- if (isStopped) {
- // check for cases when #close was called during reload before
creating cacheLoadTasks
- return;
- }
- // run first task or create numSplits threads to run all tasks
+ .collect(Collectors.toCollection(ArrayDeque::new));
+ // run first task and create concurrencyLevel - 1 threads to run
remaining tasks
ExecutorService cacheLoadTaskService = null;
try {
- if (numSplits > 1) {
- int numThreads = getConcurrencyLevel(numSplits);
- cacheLoadTaskService =
Executors.newFixedThreadPool(numThreads);
- ExecutorService finalCacheLoadTaskService =
cacheLoadTaskService;
- List<Future<?>> futures =
+ InputSplitCacheLoadTask firstTask = cacheLoadTasks.pop();
+ CompletableFuture<?> otherTasksFuture = null;
+ if (!cacheLoadTasks.isEmpty()) {
+ cacheLoadTaskService =
Executors.newFixedThreadPool(concurrencyLevel - 1);
+ ExecutorService finalExecutor = cacheLoadTaskService;
Review Comment:
ah true, I overlooked that one. 8)
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/fullcache/LookupFullCache.java:
##########
@@ -109,7 +116,9 @@ public long size() {
@Override
public void close() throws Exception {
- reloadTrigger.close(); // firstly try to interrupt reload thread
+ // in default triggers shutdowns scheduled thread pool used for
periodic triggers of reloads
Review Comment:
```suggestion
// stops scheduled thread pool that's responsible for scheduling
cache updates
```
Just as a proposal
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/fullcache/LookupFullCache.java:
##########
@@ -109,7 +116,9 @@ public long size() {
@Override
public void close() throws Exception {
- reloadTrigger.close(); // firstly try to interrupt reload thread
+ // in default triggers shutdowns scheduled thread pool used for
periodic triggers of reloads
+ reloadTrigger.close();
+ // shutdowns the reload thread and interrupts active reload task, if
present
Review Comment:
```suggestion
// stops thread pool that's responsible for executing the actual
cache update
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
