[flink] Make FileStoreLookupFunction.refreshBlacklist nullable to avo…

…id performance regression
Tan-JiaLiang · Nov 5, 2024 · c7170e6 · c7170e6
1 parent c17cdd5
commit c7170e6
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 8 deletions.
diff --git a/...on-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java b/...on-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
@@ -88,7 +88,7 @@ public class FileStoreLookupFunction implements Serializable, Closeable {
     private final List<String> projectFields;
     private final List<String> joinKeys;
     @Nullable private final Predicate predicate;
-    private final RefreshBlacklist refreshBlacklist;
+    @Nullable private final RefreshBlacklist refreshBlacklist;
 
     private transient File path;
     private transient LookupTable lookupTable;
@@ -136,7 +136,7 @@ public FileStoreLookupFunction(
         this.predicate = predicate;
 
         this.refreshBlacklist =
-                new RefreshBlacklist(
+                RefreshBlacklist.create(
                         table.options().get(LOOKUP_REFRESH_TIME_PERIODS_BLACKLIST.key()));
     }
 
@@ -284,7 +284,7 @@ private void reopen() {
     @VisibleForTesting
     void tryRefresh() throws Exception {
         // 1. check if this time is in black list
-        if (!refreshBlacklist.canRefresh()) {
+        if (refreshBlacklist != null && !refreshBlacklist.canRefresh()) {
             return;
         }
 
@@ -335,7 +335,7 @@ LookupTable lookupTable() {
 
     @VisibleForTesting
     long nextBlacklistCheckTime() {
-        return refreshBlacklist.nextBlacklistCheckTime();
+        return refreshBlacklist == null ? -1 : refreshBlacklist.nextBlacklistCheckTime();
     }
 
     @Override

diff --git a/...nk/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/RefreshBlacklist.java b/...nk/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/RefreshBlacklist.java
@@ -25,6 +25,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.time.format.DateTimeParseException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -40,12 +42,22 @@ public class RefreshBlacklist {
 
     private long nextBlacklistCheckTime;
 
-    public RefreshBlacklist(String blacklist) {
-        this.timePeriodsBlacklist = parseTimePeriodsBlacklist(blacklist);
+    public RefreshBlacklist(List<Pair<Long, Long>> timePeriodsBlacklist) {
+        this.timePeriodsBlacklist = timePeriodsBlacklist;
         this.nextBlacklistCheckTime = -1;
     }
 
-    private List<Pair<Long, Long>> parseTimePeriodsBlacklist(String blacklist) {
+    @Nullable
+    public static RefreshBlacklist create(String blacklist) {
+        List<Pair<Long, Long>> timePeriodsBlacklist = parseTimePeriodsBlacklist(blacklist);
+        if (timePeriodsBlacklist.isEmpty()) {
+            return null;
+        }
+
+        return new RefreshBlacklist(timePeriodsBlacklist);
+    }
+
+    private static List<Pair<Long, Long>> parseTimePeriodsBlacklist(String blacklist) {
         if (StringUtils.isNullOrWhitespaceOnly(blacklist)) {
             return Collections.emptyList();
         }
@@ -69,7 +81,7 @@ private List<Pair<Long, Long>> parseTimePeriodsBlacklist(String blacklist) {
         return result;
     }
 
-    private long parseToMillis(String dateTime) {
+    private static long parseToMillis(String dateTime) {
         try {
             return DateTimeUtils.parseTimestampData(dateTime + ":00", 3, TimeZone.getDefault())
                     .getMillisecond();