Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[flink][lookup] Support async refresh of lookup table #3297

Merged
merged 4 commits into from
May 22, 2024
Merged

[flink][lookup] Support async refresh of lookup table #3297

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fee498c
[flink][lookup] Support async refresh of lookup table
FangYongs May 6, 2024
56e036e
[flink][lookup] Add abstract refresh row method
FangYongs May 11, 2024
f243c72
[flink][lookup] Check max pending snapshot count for lookup join
FangYongs May 22, 2024
e833a96
[flink][lookup] Use directly executor and shutdown executor in close
FangYongs May 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
[flink][lookup] Use directly executor and shutdown executor in close
  • Loading branch information
@FangYongs
FangYongs committed May 22, 2024
commit e833a9664d5fb002571c39803e97d2ecd44c17c2
61 changes: 30 additions & 31 deletions ...aimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@
import org.apache.paimon.utils.TypeUtils;
import org.apache.paimon.utils.UserDefinedSeqComparator;

import org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -123,7 +125,7 @@ public FullCacheLookupTable(Context context) {
String.format(
"%s-lookup-refresh",
Thread.currentThread().getName())))
: null;
: MoreExecutors.newDirectExecutorService();
this.cachedException = new AtomicReference<>();
this.maxPendingSnapshotCount = options.get(LOOKUP_REFRESH_ASYNC_PENDING_SNAPSHOT_COUNT);
}
Expand Down Expand Up @@ -196,33 +198,29 @@ public void refresh() throws Exception {
}
doRefresh();
} else {
if (refreshAsync) {
Future<?> currentFuture = null;
try {
currentFuture =
refreshExecutor.submit(
() -> {
try {
doRefresh();
} catch (Exception e) {
LOG.error(
"Refresh lookup table {} failed",
context.table.name(),
e);
cachedException.set(e);
}
});
} catch (RejectedExecutionException ignored) {
LOG.warn(
"Add refresh task for lookup table {} failed",
context.table.name(),
ignored);
}
if (currentFuture != null) {
refreshFuture = currentFuture;
}
} else {
doRefresh();
Future<?> currentFuture = null;
try {
currentFuture =
refreshExecutor.submit(
() -> {
try {
doRefresh();
} catch (Exception e) {
LOG.error(
"Refresh lookup table {} failed",
context.table.name(),
e);
cachedException.set(e);
}
});
} catch (RejectedExecutionException ignored) {
LOG.warn(
"Add refresh task for lookup table {} failed",
context.table.name(),
ignored);
}
if (currentFuture != null) {
refreshFuture = currentFuture;
}
}
}
Expand Down Expand Up @@ -292,11 +290,12 @@ public Predicate projectedPredicate() {

@Override
public void close() throws IOException {
stateFactory.close();
if (refreshExecutor != null) {
try {
stateFactory.close();
FileIOUtils.deleteDirectory(context.tempPath);
} finally {
refreshExecutor.shutdown();
FangYongs marked this conversation as resolved.
Show resolved Hide resolved
}
FileIOUtils.deleteDirectory(context.tempPath);
}

/** Bulk loader for the table. */
Expand Down
Loading