[lake/paimon] Introduce scan-based lake table lookuper

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -24,6 +24,7 @@
 import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.DeleteBehavior;
 import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.metadata.LakeLookupMode;
 import org.apache.fluss.metadata.LogFormat;
 import org.apache.fluss.metadata.MergeEngineType;
 import org.apache.fluss.rpc.protocol.FetchLogReadPreference;
@@ -1933,6 +1934,19 @@ public class ConfigOptions {
                                     + "to look up historical partition data so that their clients load the "
                                     + "updated table configuration.");
 
+    /** Lookup strategy for historical partitions stored in lake storage. */
+    public static final ConfigOption<LakeLookupMode>
+            TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE =
+                    key("table.datalake.historical-partition.lookup-mode")
+                            .enumType(LakeLookupMode.class)
+                            .defaultValue(LakeLookupMode.SST)
+                            .withDescription(
+                                    "The lookup mode for historical partitions stored in Paimon. "
+                                            + "SST uses local lookup files cached from lake storage. "
+                                            + "SCAN scans the requested partition and bucket with primary-key filters "
+                                            + "and a limit of one row, without creating local lookup files. "
+                                            + "This option can only be set when creating the table and cannot be altered.");
+
     public static final ConfigOption<DataLakeFormat> TABLE_DATALAKE_FORMAT =
             key("table.datalake.format")
                     .enumType(DataLakeFormat.class)

fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java

@@ -23,6 +23,7 @@
 import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.DeleteBehavior;
 import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.metadata.LakeLookupMode;
 import org.apache.fluss.metadata.LogFormat;
 import org.apache.fluss.metadata.MergeEngineType;
 import org.apache.fluss.utils.AutoPartitionStrategy;
@@ -134,6 +135,11 @@ public boolean isHistoricalPartitionEnabled() {
         return config.get(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED);
     }
 
+    /** Gets the lookup mode for historical partitions of the table. */
+    public LakeLookupMode getHistoricalLookupMode() {
+        return config.get(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE);
+    }
+
     /**
      * Return the data lake format of the table. It'll be the datalake format configured in Fluss
      * whiling creating the table. Return empty if no datalake format configured while creating.

fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java

@@ -21,6 +21,7 @@
 import org.apache.fluss.config.TableConfig;
 import org.apache.fluss.lake.source.LakeSource;
 import org.apache.fluss.lake.writer.LakeTieringFactory;
+import org.apache.fluss.metadata.LakeLookupMode;
 import org.apache.fluss.metadata.TablePath;
 
 import static org.apache.fluss.utils.Preconditions.checkArgument;
@@ -74,6 +75,7 @@ final class LookuperContext {
         private final TableConfig tableConfig;
         private final long lookupCacheMaxDiskBytes;
         private final Runnable diskWriteGuard;
+        private final LakeLookupMode lookupMode;
 
         /**
          * Creates a lookuper context.
@@ -94,6 +96,7 @@ public LookuperContext(
                     lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0.");
             this.lookupCacheMaxDiskBytes = lookupCacheMaxDiskBytes;
             this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null.");
+            this.lookupMode = tableConfig.getHistoricalLookupMode();
         }
 
         /** Returns the local directory for temporary files used by the lookuper. */
@@ -115,5 +118,10 @@ public long lookupCacheMaxDiskBytes() {
         public Runnable diskWriteGuard() {
             return diskWriteGuard;
         }
+
+        /** Returns the mode used to look up historical data. */
+        public LakeLookupMode lookupMode() {
+            return lookupMode;
+        }
     }
 }

fluss-common/src/main/java/org/apache/fluss/metadata/LakeLookupMode.java

@@ -0,0 +1,27 @@
+/*
+ * 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.metadata;
+
+/** Mode used to look up historical partitions in lake storage. */
+public enum LakeLookupMode {
+    /** Use local lookup files cached from lake storage. */
+    SST,
+
+    /** Scan lake storage with primary-key filters. */
+    SCAN
+}

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java

@@ -21,13 +21,15 @@
 import org.apache.fluss.lake.lakestorage.LakeStorage;
 import org.apache.fluss.lake.lakestorage.LakeTableLookuper;
 import org.apache.fluss.lake.paimon.lookup.PaimonLakeTableLookuper;
+import org.apache.fluss.lake.paimon.lookup.PaimonScanBasedTableLookuper;
 import org.apache.fluss.lake.paimon.source.PaimonLakeSource;
 import org.apache.fluss.lake.paimon.source.PaimonSplit;
 import org.apache.fluss.lake.paimon.tiering.PaimonCommittable;
 import org.apache.fluss.lake.paimon.tiering.PaimonLakeTieringFactory;
 import org.apache.fluss.lake.paimon.tiering.PaimonWriteResult;
 import org.apache.fluss.lake.source.LakeSource;
 import org.apache.fluss.lake.writer.LakeTieringFactory;
+import org.apache.fluss.metadata.LakeLookupMode;
 import org.apache.fluss.metadata.TablePath;
 
 /** Paimon implementation of {@link LakeStorage}. */
@@ -56,6 +58,9 @@ public LakeSource<PaimonSplit> createLakeSource(TablePath tablePath) {
 
     @Override
     public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, LookuperContext context) {
+        if (context.lookupMode() == LakeLookupMode.SCAN) {
+            return new PaimonScanBasedTableLookuper(paimonConfig, tablePath, context.tableConfig());
+        }
         return new PaimonLakeTableLookuper(
                 paimonConfig,
                 tablePath,

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java

@@ -24,13 +24,9 @@
 import org.apache.fluss.exception.KvStorageException;
 import org.apache.fluss.lake.lakestorage.LakeTableLookuper;
 import org.apache.fluss.lake.paimon.utils.PaimonPartitionBucket;
-import org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow;
 import org.apache.fluss.metadata.TablePath;
-import org.apache.fluss.row.BinaryRow;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.decode.CompactedKeyDecoder;
-import org.apache.fluss.row.encode.RowEncoder;
-import org.apache.fluss.row.encode.ValueEncoder;
 import org.apache.fluss.row.encode.paimon.PaimonKeyEncoder;
 import org.apache.fluss.types.RowType;
 import org.apache.fluss.utils.ExceptionUtils;
@@ -68,6 +64,7 @@
 
 import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
 import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS;
+import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toFlussValue;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonPartition;
 import static org.apache.fluss.utils.Preconditions.checkArgument;
@@ -330,7 +327,15 @@ private org.apache.paimon.data.BinaryRow getKey(byte[] key, LookupContext contex
         if (paimonRow == null) {
             return null;
         }
-        return encodeValue(paimonRow, context.schemaId(), context.valueRowType());
+        try {
+            return toFlussValue(
+                    paimonRow,
+                    context.schemaId(),
+                    context.valueRowType(),
+                    tableConfig.getKvFormat());
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to encode Paimon lookup row as Fluss value.", e);
+        }
     }
 
     private @Nullable org.apache.paimon.data.InternalRow lookupPaimon(
@@ -418,22 +423,6 @@ private List<DataFileMeta> scanDataFiles(
         return Collections.unmodifiableList(new ArrayList<>(dataFilesByName.values()));
     }
 
-    private byte[] encodeValue(
-            org.apache.paimon.data.InternalRow paimonRow, short schemaId, RowType valueRowType) {
-        PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
-        InternalRow.FieldGetter[] fieldGetters = InternalRow.createFieldGetters(valueRowType);
-        try (RowEncoder rowEncoder = RowEncoder.create(tableConfig.getKvFormat(), valueRowType)) {
-            rowEncoder.startNewRow();
-            for (int i = 0; i < fieldGetters.length; i++) {
-                rowEncoder.encodeField(i, fieldGetters[i].getFieldOrNull(flussRow));
-            }
-            BinaryRow row = rowEncoder.finishRow();
-            return ValueEncoder.encodeValue(schemaId, row);
-        } catch (Exception e) {
-            throw new RuntimeException("Failed to encode Paimon lookup row as Fluss value.", e);
-        }
-    }
-
     /** Tracks creation of Paimon lookup files while delegating all local I/O operations. */
     private final class TrackingIOManager implements IOManager {