[lookup] Support historical partition lookup for lake tables

fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookupQuery.java

@@ -21,6 +21,8 @@
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TablePath;
 
+import javax.annotation.Nullable;
+
 import java.util.concurrent.CompletableFuture;
 
 /** Abstract Class to represent a lookup operation. */
@@ -30,13 +32,31 @@ public abstract class AbstractLookupQuery<T> {
     private final TablePath tablePath;
     private final TableBucket tableBucket;
     private final byte[] key;
+
+    /**
+     * Null for normal and prefix lookups. Historical lookups use this to carry the original
+     * partition name.
+     */
+    private final @Nullable String originalPartitionName;
+
     private int retries;
+    private long nextRetryTimeMs;
 
     public AbstractLookupQuery(TablePath tablePath, TableBucket tableBucket, byte[] key) {
+        this(tablePath, tableBucket, key, null);
+    }
+
+    public AbstractLookupQuery(
+            TablePath tablePath,
+            TableBucket tableBucket,
+            byte[] key,
+            @Nullable String originalPartitionName) {
         this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.key = key;
+        this.originalPartitionName = originalPartitionName;
         this.retries = 0;
+        this.nextRetryTimeMs = 0;
     }
 
     public byte[] key() {
@@ -51,6 +71,10 @@ public TableBucket tableBucket() {
         return tableBucket;
     }
 
+    public @Nullable String originalPartitionName() {
+        return originalPartitionName;
+    }
+
     public int retries() {
         return retries;
     }
@@ -59,6 +83,14 @@ public void incrementRetries() {
         retries++;
     }
 
+    public long nextRetryTimeMs() {
+        return nextRetryTimeMs;
+    }
+
+    public void setNextRetryTimeMs(long nextRetryTimeMs) {
+        this.nextRetryTimeMs = nextRetryTimeMs;
+    }
+
     public abstract LookupType lookupType();
 
     public abstract CompletableFuture<T> future();

fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupBatch.java

@@ -21,20 +21,21 @@
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.metadata.TableBucket;
 
+import javax.annotation.Nullable;
+
 import java.util.ArrayList;
 import java.util.List;
 
 /** A batch that contains the lookup operations that send to same tablet bucket together. */
 @Internal
 public class LookupBatch {
 
-    /** The table bucket that the lookup operations should fall into. */
-    private final TableBucket tableBucket;
+    private final LookupBatchKey lookupBatchKey;
 
     private final List<LookupQuery> lookups;
 
-    public LookupBatch(TableBucket tableBucket) {
-        this.tableBucket = tableBucket;
+    LookupBatch(LookupBatchKey lookupBatchKey) {
+        this.lookupBatchKey = lookupBatchKey;
         this.lookups = new ArrayList<>();
     }
 
@@ -47,7 +48,15 @@ public List<LookupQuery> lookups() {
     }
 
     public TableBucket tableBucket() {
-        return tableBucket;
+        return lookupBatchKey.tableBucket();
+    }
+
+    public @Nullable String originalPartitionName() {
+        return lookupBatchKey.originalPartitionName();
+    }
+
+    LookupBatchKey lookupBatchKey() {
+        return lookupBatchKey;
     }
 
     /** Complete the lookup operations using given values . */

fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupBatchKey.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.lookup;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metadata.TableBucket;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+/** A key that identifies lookup batches by route bucket and original partition name. */
+@Internal
+class LookupBatchKey {
+
+    /** The table bucket to which the lookup batch is routed. */
+    private final TableBucket tableBucket;
+
+    /** Null for normal lookup batches; non-null for historical batches. */
+    private final @Nullable String originalPartitionName;
+
+    LookupBatchKey(TableBucket tableBucket, @Nullable String originalPartitionName) {
+        this.tableBucket = tableBucket;
+        this.originalPartitionName = originalPartitionName;
+    }
+
+    TableBucket tableBucket() {
+        return tableBucket;
+    }
+
+    @Nullable
+    String originalPartitionName() {
+        return originalPartitionName;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        LookupBatchKey that = (LookupBatchKey) o;
+        return Objects.equals(tableBucket, that.tableBucket)
+                && Objects.equals(originalPartitionName, that.originalPartitionName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tableBucket, originalPartitionName);
+    }
+
+    @Override
+    public String toString() {
+        return "LookupBatchKey{"
+                + "tableBucket="
+                + tableBucket
+                + ", originalPartitionName='"
+                + originalPartitionName
+                + '\''
+                + '}';
+    }
+}

fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupClient.java

@@ -28,6 +28,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
 import java.time.Duration;
@@ -44,8 +45,8 @@
  * that is responsible for turning these lookup operations into network requests and transmitting
  * them to the cluster.
  *
- * <p>The {@link #lookup(TablePath, TableBucket, byte[], boolean)} method is asynchronous, when
- * called, it adds the lookup operation to a queue of pending lookup operations and immediately
+ * <p>The {@link #lookup(TablePath, TableBucket, byte[], boolean, String)} method is asynchronous,
+ * when called, it adds the lookup operation to a queue of pending lookup operations and immediately
  * returns. This allows the lookup operations to batch together individual lookup operations for
  * efficiency.
  */
@@ -95,12 +96,27 @@ private ExecutorService createThreadPool() {
         return Executors.newFixedThreadPool(1, new ExecutorThreadFactory(LOOKUP_THREAD_PREFIX));
     }
 
+    /**
+     * Looks up a key from the specified table bucket.
+     *
+     * @param tablePath path of the table to look up
+     * @param tableBucket bucket to look up
+     * @param keyBytes serialized key to look up
+     * @param insertIfNotExists whether to insert the key when it does not exist
+     * @param originalPartitionName null for a regular lookup against an active Fluss partition; the
+     *     original partition name for a historical lookup against a partition whose Fluss data has
+     *     expired and is looked up from lake storage
+     * @return a future containing the serialized value, or null if the key does not exist
+     */
     public CompletableFuture<byte[]> lookup(
             TablePath tablePath,
             TableBucket tableBucket,
             byte[] keyBytes,
-            boolean insertIfNotExists) {
-        LookupQuery lookup = new LookupQuery(tablePath, tableBucket, keyBytes, insertIfNotExists);
+            boolean insertIfNotExists,
+            @Nullable String originalPartitionName) {
+        LookupQuery lookup =
+                new LookupQuery(
+                        tablePath, tableBucket, keyBytes, insertIfNotExists, originalPartitionName);
         lookupQueue.appendLookup(lookup);
         return lookup.future();
     }

fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupQuery.java

@@ -18,9 +18,12 @@
 package org.apache.fluss.client.lookup;
 
 import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TablePath;
 
+import javax.annotation.Nullable;
+
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -33,17 +36,22 @@ public class LookupQuery extends AbstractLookupQuery<byte[]> {
     private final CompletableFuture<byte[]> future;
     private final boolean insertIfNotExists;
 
-    LookupQuery(TablePath tablePath, TableBucket tableBucket, byte[] key) {
-        this(tablePath, tableBucket, key, false);
-    }
-
     LookupQuery(
-            TablePath tablePath, TableBucket tableBucket, byte[] key, boolean insertIfNotExists) {
-        super(tablePath, tableBucket, key);
+            TablePath tablePath,
+            TableBucket tableBucket,
+            byte[] key,
+            boolean insertIfNotExists,
+            @Nullable String originalPartitionName) {
+        super(tablePath, tableBucket, key, originalPartitionName);
         this.future = new CompletableFuture<>();
         this.insertIfNotExists = insertIfNotExists;
     }
 
+    @VisibleForTesting
+    LookupQuery(TablePath tablePath, TableBucket tableBucket, byte[] key) {
+        this(tablePath, tableBucket, key, false, null);
+    }
+
     @Override
     public LookupType lookupType() {
         return insertIfNotExists ? LookupType.LOOKUP_WITH_INSERT_IF_NOT_EXISTS : LookupType.LOOKUP;

fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupQueue.java

@@ -99,17 +99,32 @@ List<AbstractLookupQuery<?>> drain() throws Exception {
                 break;
             }
 
-            while (!reEnqueuedLookupQueue.isEmpty() && count < maxBatchSize) {
-                AbstractLookupQuery<?> lookup =
-                        reEnqueuedLookupQueue.poll(waitNanos, TimeUnit.NANOSECONDS);
+            long nextRetryDelayNanos = Long.MAX_VALUE;
+            int reEnqueuedToCheck = reEnqueuedLookupQueue.size();
+            while (reEnqueuedToCheck > 0 && count < maxBatchSize) {
+                AbstractLookupQuery<?> lookup = reEnqueuedLookupQueue.poll();
                 if (lookup == null) {
                     break;
                 }
-                lookupOperations.add(lookup);
-                count++;
+                long retryDelayMs = lookup.nextRetryTimeMs() - System.currentTimeMillis();
+                if (retryDelayMs <= 0) {
+                    lookupOperations.add(lookup);
+                    count++;
+                } else {
+                    nextRetryDelayNanos =
+                            Math.min(
+                                    nextRetryDelayNanos,
+                                    TimeUnit.MILLISECONDS.toNanos(retryDelayMs));
+                    reEnqueuedLookupQueue.add(lookup);
+                }
+                reEnqueuedToCheck--;
             }
 
-            AbstractLookupQuery<?> lookup = lookupQueue.poll(waitNanos, TimeUnit.NANOSECONDS);
+            long lookupWaitNanos = waitNanos;
+            if (count == 0 && nextRetryDelayNanos != Long.MAX_VALUE) {
+                lookupWaitNanos = Math.min(waitNanos, Math.max(1L, nextRetryDelayNanos));
+            }
+            AbstractLookupQuery<?> lookup = lookupQueue.poll(lookupWaitNanos, TimeUnit.NANOSECONDS);
             if (lookup == null) {
                 break;
             }