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

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

@@ -20,6 +20,7 @@
 import org.apache.fluss.annotation.Internal;
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.compression.ArrowCompressionType;
+import org.apache.fluss.lake.lakestorage.LakeStorage.LookupMode;
 import org.apache.fluss.metadata.ChangelogImage;
 import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.DeleteBehavior;
@@ -1902,6 +1903,18 @@ 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<LookupMode> TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE =
+            key("table.datalake.historical-partition.lookup-mode")
+                    .enumType(LookupMode.class)
+                    .defaultValue(LookupMode.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

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.compression.ArrowCompressionInfo;
+import org.apache.fluss.lake.lakestorage.LakeStorage.LookupMode;
 import org.apache.fluss.metadata.ChangelogImage;
 import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.DeleteBehavior;
@@ -112,6 +113,11 @@ public boolean isHistoricalPartitionEnabled() {
         return config.get(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED);
     }
 
+    /** Gets the lookup mode for historical partitions of the table. */
+    public LookupMode 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

@@ -68,12 +68,22 @@ default LakeTableLookuper createLakeTableLookuper(
                 "Point lookup is not supported for this lake storage.");
     }
 
+    /** Mode used to look up historical data in lake storage. */
+    enum LookupMode {
+        /** Use local lookup files cached from lake storage. */
+        SST,
+
+        /** Scan lake storage with primary-key filters and return at most one row. */
+        SCAN
+    }
+
     /** Runtime context for creating a lake table lookuper. */
     final class LookuperContext {
         private final String ioTmpDir;
         private final TableConfig tableConfig;
         private final long lookupCacheMaxDiskBytes;
         private final Runnable diskWriteGuard;
+        private final LookupMode lookupMode;
 
         /**
          * Creates a lookuper context.
@@ -94,6 +104,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 +126,10 @@ public long lookupCacheMaxDiskBytes() {
         public Runnable diskWriteGuard() {
             return diskWriteGuard;
         }
+
+        /** Returns the mode used to look up historical data. */
+        public LookupMode lookupMode() {
+            return lookupMode;
+        }
     }
 }

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

@@ -21,6 +21,7 @@
 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;
@@ -56,6 +57,9 @@ public LakeSource<PaimonSplit> createLakeSource(TablePath tablePath) {
 
     @Override
     public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, LookuperContext context) {
+        if (context.lookupMode() == LookupMode.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/PaimonScanBasedTableLookuper.java

@@ -0,0 +1,231 @@
+/*
+ * 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.lake.paimon.lookup;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.exception.KvStorageException;
+import org.apache.fluss.lake.lakestorage.LakeTableLookuper;
+import org.apache.fluss.lake.paimon.source.FlussRowAsPaimonRow;
+import org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.decode.KeyDecoder;
+import org.apache.fluss.row.encode.RowEncoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+import org.apache.fluss.utils.IOUtils;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReader.RecordIterator;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+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.checkNotNull;
+import static org.apache.fluss.utils.Preconditions.checkState;
+import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock;
+import static org.apache.fluss.utils.concurrent.LockUtils.inWriteLock;
+
+/**
+ * Looks up a primary key by scanning the latest Paimon snapshot with a limit of one row.
+ *
+ * <p>Each scan is restricted to the requested partition, bucket, and complete primary key. It does
+ * not create local lookup files. Lookups use independent readers and encoders and may run in
+ * parallel. The catalog and table are initialized once, and close waits for active lookups to
+ * finish.
+ */
+public class PaimonScanBasedTableLookuper implements LakeTableLookuper {
+
+    private final Configuration paimonConfig;
+    private final TablePath tablePath;
+    private final TableConfig tableConfig;
+    private final ReadWriteLock lifecycleLock = new ReentrantReadWriteLock();
+    private final Object initializationLock = new Object();
+
+    private @Nullable Catalog catalog;
+    private @Nullable FileStoreTable fileStoreTable;
+    private boolean closed;
+
+    /** Creates a scan-based lookuper for the specified Paimon table. */
+    public PaimonScanBasedTableLookuper(
+            Configuration paimonConfig, TablePath tablePath, TableConfig tableConfig) {
+        this.paimonConfig = checkNotNull(paimonConfig, "paimonConfig must not be null.");
+        this.tablePath = checkNotNull(tablePath, "tablePath must not be null.");
+        this.tableConfig = checkNotNull(tableConfig, "tableConfig must not be null.");
+    }
+
+    @Override
+    public @Nullable byte[] lookup(byte[] key, LookupContext context) throws Exception {
+        checkNotNull(key, "key must not be null.");
+        checkNotNull(context, "context must not be null.");
+        return inReadLock(
+                lifecycleLock,
+                () -> {
+                    checkState(!closed, "Paimon scan-based lookuper has been closed.");
+                    long lookupStartNanos = System.nanoTime();
+                    try {
+                        FileStoreTable table = table();
+                        // Paimon tables contain mutable lazy store state; isolate it per lookup.
+                        return scanLookup(table.copy(table.schema()), key, context);
+                    } catch (IOException | UncheckedIOException e) {
+                        // The next RPC retry plans a fresh scan after compaction or expiration.
+                        throw new KvStorageException(
+                                "Failed to scan historical data from Paimon for " + tablePath + ".",
+                                e);
+                    } finally {
+                        context.lookupMetricRecorder()
+                                .recordLookup(System.nanoTime() - lookupStartNanos, false);
+                    }
+                });
+    }
+
+    @Override
+    public void close() {
+        inWriteLock(
+                lifecycleLock,
+                () -> {
+                    if (!closed) {
+                        closed = true;
+                        IOUtils.closeQuietly(catalog, "Paimon catalog");
+                    }
+                });
+    }
+
+    private FileStoreTable table() throws Exception {
+        synchronized (initializationLock) {
+            if (fileStoreTable == null) {
+                Catalog newCatalog =
+                        CatalogFactory.createCatalog(
+                                CatalogContext.create(Options.fromMap(paimonConfig.toMap())));
+                try {
+                    FileStoreTable table =
+                            (FileStoreTable) newCatalog.getTable(toPaimon(tablePath));
+                    if (table.primaryKeys().isEmpty()) {
+                        throw new UnsupportedOperationException(
+                                "Point lookup is only supported for primary-key Paimon tables.");
+                    }
+                    catalog = newCatalog;
+                    fileStoreTable = table;
+                } finally {
+                    if (fileStoreTable == null) {
+                        IOUtils.closeQuietly(newCatalog, "Paimon catalog");
+                    }
+                }
+            }
+            return fileStoreTable;
+        }
+    }
+
+    private @Nullable byte[] scanLookup(FileStoreTable table, byte[] key, LookupContext context)
+            throws Exception {
+        RowType rowType = table.rowType();
+        List<String> primaryKeys = table.schema().trimmedPrimaryKeys();
+        KeyDecoder keyDecoder =
+                KeyDecoder.ofPrimaryKeyDecoder(
+                        context.valueRowType(),
+                        primaryKeys,
+                        tableConfig.getKvFormatVersion().orElse(1).shortValue(),
+                        DataLakeFormat.PAIMON,
+                        table.schema().bucketKeys().equals(primaryKeys));
+        FlussRowAsPaimonRow keyRow =
+                new FlussRowAsPaimonRow(keyDecoder.decodeKey(key), rowType.project(primaryKeys));
+        PredicateBuilder predicateBuilder = new PredicateBuilder(rowType);
+        List<Predicate> predicates = new ArrayList<>(primaryKeys.size());
+        for (int i = 0; i < primaryKeys.size(); i++) {
+            int fieldIndex = predicateBuilder.indexOf(primaryKeys.get(i));
+            Object value =
+                    org.apache.paimon.data.InternalRow.createFieldGetter(
+                                    rowType.getTypeAt(fieldIndex), i)
+                            .getFieldOrNull(keyRow);
+            predicates.add(predicateBuilder.equal(fieldIndex, value));
+        }
+
+        RowPartitionKeyExtractor partitionKeyExtractor =
+                new RowPartitionKeyExtractor(table.schema());
+        BinaryRow partition =
+                toPaimonPartition(
+                        context.partitionSpec(),
+                        context.valueRowType(),
+                        rowType,
+                        partitionKeyExtractor::partition);
+        ReadBuilder readBuilder =
+                table.newReadBuilder()
+                        .withFilter(predicates)
+                        .withPartitionFilter(
+                                PartitionPredicate.fromMultiple(
+                                        rowType.project(table.partitionKeys()),
+                                        Collections.singletonList(partition)))
+                        .withBucket(context.bucketId())
+                        .withReadType(rowType.project(context.valueRowType().getFieldNames()))
+                        .withLimit(1);
+        // Pushdown alone may only prune files. Filter each row before applying the limit.
+        try (RecordReader<org.apache.paimon.data.InternalRow> reader =
+                readBuilder.newRead().executeFilter().createReader(readBuilder.newScan().plan())) {
+            RecordIterator<org.apache.paimon.data.InternalRow> batch;
+            while ((batch = reader.readBatch()) != null) {
+                try {
+                    org.apache.paimon.data.InternalRow row = batch.next();
+                    if (row != null) {
+                        // Encode while the batch still owns the row's backing storage.
+                        return encodeValue(row, context);
+                    }
+                } finally {
+                    batch.releaseBatch();
+                }
+            }
+        }
+        return null;
+    }
+
+    private byte[] encodeValue(org.apache.paimon.data.InternalRow row, LookupContext context)
+            throws Exception {
+        PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(row);
+        InternalRow.FieldGetter[] fieldGetters =
+                InternalRow.createFieldGetters(context.valueRowType());
+        try (RowEncoder encoder =
+                RowEncoder.create(tableConfig.getKvFormat(), context.valueRowType())) {
+            encoder.startNewRow();
+            for (int i = 0; i < fieldGetters.length; i++) {
+                encoder.encodeField(i, fieldGetters[i].getFieldOrNull(flussRow));
+            }
+            return ValueEncoder.encodeValue(context.schemaId(), encoder.finishRow());
+        }
+    }
+}