fresh-borzoni commented on code in PR #3630:
URL: https://github.com/apache/fluss/pull/3630#discussion_r3599724998
##########
fluss-rpc/src/main/proto/FlussApi.proto:
##########
@@ -946,6 +946,8 @@ message PbLookupReqForBucket {
optional int64 partition_id = 1;
required int32 bucket_id = 2;
repeated bytes keys = 3;
+ // The original partition name for historical lookup. It is unset for normal
lookup.
+ optional string original_partition_name = 4;
Review Comment:
An old tablet server ignores this field and looks up the empty historical
rocksdb and returns null instead of an error. During rolling upgrades that's
silent misses. Gate on api version instead?
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java:
##########
@@ -978,6 +1124,7 @@ public void stopReplicas(
for (StopReplicaData data : stopReplicaDataList) {
TableBucket tb = data.getTableBucket();
+
historicalLakeLookupManager.invalidateTableLookuper(tb.getTableId());
Review Comment:
minor: This closes the lookuper on the calling thread, and close() waits for
the lock held by an in-flight lookup. So stopping a replica can block on a slow
S3 read. Mb move the close to the ioExecutor?
##########
fluss-client/src/main/java/org/apache/fluss/client/lookup/HistoricalPartitionResolver.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.client.admin.Admin;
+import org.apache.fluss.client.metadata.MetadataUpdater;
+import org.apache.fluss.exception.PartitionNotExistException;
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.fluss.utils.PartitionUtils.toHistoricalPartitionSpec;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Resolves original partitions to their historical system partition ids.
+ *
+ * <p>Historical lookup requests still carry the original partition name, but
the lookup RPC must be
+ * sent to the generated historical system partition. This resolver bridges
that gap on the client
+ * side.
+ */
+@Internal
+class HistoricalPartitionResolver {
+
+ private final MetadataUpdater metadataUpdater;
+ private final Admin admin;
+ private final ConcurrentHashMap<HistoricalPartitionKey,
CompletableFuture<Long>>
+ inflightResolves;
+
+ HistoricalPartitionResolver(MetadataUpdater metadataUpdater, Admin admin) {
+ this.metadataUpdater = checkNotNull(metadataUpdater, "metadataUpdater
must not be null.");
+ this.admin = checkNotNull(admin, "admin must not be null.");
+ this.inflightResolves = new ConcurrentHashMap<>();
+ }
+
+ CompletableFuture<Long> resolveHistoricalPartitionId(
+ TableInfo tableInfo, String originalPartitionName) {
+ HistoricalPartitionKey key =
+ new HistoricalPartitionKey(
+ tableInfo.getTableId(), tableInfo.getTablePath(),
originalPartitionName);
+ // Multiple lookups for the same original partition can be issued
concurrently. Coalesce
+ // them so only one metadata refresh/create path runs for a given
partition.
+ CompletableFuture<Long> result = new CompletableFuture<>();
+ CompletableFuture<Long> previous = inflightResolves.putIfAbsent(key,
result);
+ if (previous != null) {
+ return previous;
+ }
+
+ resolveHistoricalPartitionIdInternal(tableInfo, originalPartitionName)
+ .whenComplete(
+ (partitionId, error) -> {
+ if (error != null) {
+ result.completeExceptionally(error);
+ } else {
+ result.complete(partitionId);
+ }
+ inflightResolves.remove(key, result);
+ });
+ return result;
+ }
+
+ private CompletableFuture<Long> resolveHistoricalPartitionIdInternal(
+ TableInfo tableInfo, String originalPartitionName) {
+ CompletableFuture<Long> result = new CompletableFuture<>();
+ ResolvedPartitionSpec historicalPartitionSpec;
+ PhysicalTablePath historicalPartitionPath;
+ try {
+ // The server-side historical lookup path is keyed by the
historical system partition,
+ // not by the original partition that existed before retention
cleanup.
+ historicalPartitionSpec = toHistoricalPartitionSpec(tableInfo,
originalPartitionName);
+ historicalPartitionPath =
+ PhysicalTablePath.of(
+ tableInfo.getTablePath(),
historicalPartitionSpec.getPartitionName());
+
+ // Prefer cached metadata, then refresh once before creating the
system partition.
+ Long partitionId = getCachedPartitionId(historicalPartitionPath);
+ if (partitionId != null) {
+ result.complete(partitionId);
+ return result;
+ }
+
+ tryRefreshHistoricalPartition(historicalPartitionPath);
+ partitionId = getCachedPartitionId(historicalPartitionPath);
+ if (partitionId != null) {
+ result.complete(partitionId);
+ return result;
+ }
+ } catch (Throwable t) {
+ result.completeExceptionally(t);
+ return result;
+ }
+
+ // If the historical system partition still does not exist, create it
idempotently and
+ // refresh metadata again so the caller can route the lookup request
by partition id.
+ admin.createPartition(
Review Comment:
Creating a partition from the lookup path feels backwards: a read allocates
real replicas and rocksdb that may never see a write. Could we route lookups
without the partition existing, and leave creation to the write path?
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java:
##########
@@ -723,16 +725,36 @@ public CompletableFuture<CreatePartitionResponse>
createPartition(
// first, validate the partition spec, and get resolved partition spec.
PartitionSpec partitionSpec =
getPartitionSpec(request.getPartitionSpec());
- validatePartitionSpec(tablePath, table.partitionKeys, partitionSpec,
true);
+ ResolvedPartitionSpec partitionToCreate;
+ if (isHistoricalPartitionCreate(table, partitionSpec)) {
+ // Historical system partitions are lookup metadata, so creating
one requires the same
+ // permission as reading the table instead of writing table data.
+ authorizeTable(OperationType.READ, tablePath);
Review Comment:
Why READ? A read-only client can now create partitions and for multi-key
tables the prefix values come from the lookup key, so bad keys create real
partitions until max.partition.num is full and auto-partitioning breaks.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java:
##########
@@ -525,8 +526,15 @@ private void dropPartitions(
dropIterator = entry.getValue().iterator();
}
+ boolean shouldRemoveEntry = true;
while (dropIterator.hasNext()) {
String partitionName = dropIterator.next();
+ if (isHistoricalPartitionName(
Review Comment:
Related to dropping here, drop doesn't wait for tiering to catch up. If a
partition is dropped before its tail is tiered, historical lookups quietly
serve older values. Should drop wait for the tiering offset, like local log
deletion waits for remote copy?
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java:
##########
@@ -0,0 +1,359 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+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.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.CompactedRowEncoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.IOUtils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.query.LocalTableQuery;
+import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataField;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
+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;
+
+/**
+ * Paimon implementation of {@link LakeTableLookuper} for primary-key tables.
+ *
+ * <p>The catalog, table, local query, and I/O manager are initialized lazily
on the first lookup.
+ * For each partition and bucket, the lookuper scans the latest Paimon
snapshot once and registers
+ * its data files with {@link LocalTableQuery}. Paimon then creates local
lookup files lazily as
+ * individual remote data files are queried.
+ *
+ * <p>A cached partition-bucket file set can become stale when Paimon
compaction replaces its data
+ * files and snapshot expiration physically deletes the old files. Because
{@code FileIO}
+ * implementations may represent a missing file with different {@link
IOException} types, the first
+ * lookup I/O failure closes the cached query state, reopens the table from
the latest snapshot, and
+ * retries once.
+ *
+ * <p>Lookup and close operations are synchronized because they share mutable
Paimon query, local
+ * cache, and value-encoding state.
+ */
+public class PaimonLakeTableLookuper implements LakeTableLookuper {
+
+ // Hard-code a conservative 1GB default for now to avoid unbounded lookup
cache growth.
+ private static final String DEFAULT_LOOKUP_CACHE_MAX_DISK_SIZE = "1gb";
+
+ private final Configuration paimonConfig;
+ private final TablePath tablePath;
+ private final String ioTmpDir;
+
+ private final Set<PaimonPartitionBucket> initializedBuckets;
+
+ private @Nullable Catalog catalog;
+ private @Nullable FileStoreTable fileStoreTable;
+ private @Nullable IOManager ioManager;
+ private @Nullable LocalTableQuery localTableQuery;
+ private @Nullable RowPartitionKeyExtractor partitionKeyExtractor;
+ private int primaryKeyFieldCount;
+ private boolean hasCachedValueEncoder;
+ private short cachedValueSchemaId;
+ private @Nullable CompactedRowEncoder cachedValueRowEncoder;
+ private @Nullable InternalRow.FieldGetter[] cachedValueFieldGetters;
+ private boolean closed;
+
+ public PaimonLakeTableLookuper(Configuration paimonConfig, TablePath
tablePath) {
+ this(paimonConfig, tablePath, ConfigOptions.IO_TMP_DIR.defaultValue());
+ }
+
+ public PaimonLakeTableLookuper(
+ Configuration paimonConfig, TablePath tablePath, String ioTmpDir) {
+ this.paimonConfig = checkNotNull(paimonConfig, "paimonConfig must not
be null.");
+ this.tablePath = checkNotNull(tablePath, "tablePath must not be
null.");
+ this.ioTmpDir = checkNotNull(ioTmpDir, "ioTmpDir must not be null.");
+ this.initializedBuckets = new HashSet<>();
+ }
+
+ @Override
+ public synchronized @Nullable byte[] lookup(byte[] key, LookupContext
context)
+ throws Exception {
+ checkNotNull(key, "key must not be null.");
+ checkNotNull(context, "context must not be null.");
+ checkNotClosed();
+ ensureInitialized();
+
+ org.apache.paimon.data.BinaryRow partition =
+ convertPartition(context.partitionSpec(),
context.valueRowType());
+ org.apache.paimon.data.BinaryRow keyRow = wrapLookupKey(key);
+ initializeFilesIfNeeded(partition, context.bucketId());
+
+ org.apache.paimon.data.InternalRow paimonRow =
+ lookupWithFileRefresh(partition, context.bucketId(), keyRow);
+ if (paimonRow == null) {
+ return null;
+ }
+ return encodeValue(paimonRow, context.schemaId(),
context.valueRowType());
+ }
+
+ @Override
+ public synchronized void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ IOUtils.closeQuietly(cachedValueRowEncoder, "Fluss value row encoder");
+ IOUtils.closeQuietly(localTableQuery, "Paimon local table query");
+ IOUtils.closeQuietly(ioManager, "Paimon lookup IO manager");
+ IOUtils.closeQuietly(catalog, "Paimon catalog");
+ initializedBuckets.clear();
+ }
+
+ private void checkNotClosed() {
+ if (closed) {
+ throw new IllegalStateException("Paimon lake table lookuper has
been closed.");
+ }
+ }
+
+ private void ensureInitialized() throws Exception {
+ if (localTableQuery != null) {
+ return;
+ }
+
+ catalog =
+ CatalogFactory.createCatalog(
+
CatalogContext.create(Options.fromMap(paimonConfig.toMap())));
+ fileStoreTable =
+ withLookupCacheOptions((FileStoreTable)
catalog.getTable(toPaimon(tablePath)));
+ if (fileStoreTable.primaryKeys().isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Point lookup is only supported for primary-key Paimon
tables.");
+ }
+ ioManager = createIOManager(ioTmpDir);
+ localTableQuery = createLocalTableQuery();
+ partitionKeyExtractor = new
RowPartitionKeyExtractor(fileStoreTable.schema());
+ primaryKeyFieldCount = fileStoreTable.primaryKeys().size();
+ }
+
+ private LocalTableQuery createLocalTableQuery() {
+ return fileStoreTable()
+ .newLocalTableQuery()
+ .withValueProjection(businessFieldProjection(fileStoreTable()))
+ .withIOManager(checkNotNull(ioManager, "ioManager must be
initialized."));
+ }
+
+ private FileStoreTable withLookupCacheOptions(FileStoreTable table) {
+ String key = CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE.key();
+ if (table.options().containsKey(key)) {
+ return table;
+ }
+ return table.copy(Collections.singletonMap(key,
DEFAULT_LOOKUP_CACHE_MAX_DISK_SIZE));
+ }
+
+ private static IOManager createIOManager(String ioTmpDir) {
+ return IOManager.create(ioTmpDir);
+ }
+
+ private static int[] businessFieldProjection(FileStoreTable
fileStoreTable) {
+ List<DataField> fields =
fileStoreTable.schema().logicalRowType().getFields();
+ List<Integer> projectedFields = new ArrayList<>();
+ for (int i = 0; i < fields.size(); i++) {
+ if (!SYSTEM_COLUMNS.containsKey(fields.get(i).name())) {
+ projectedFields.add(i);
+ }
+ }
+
+ int[] projection = new int[projectedFields.size()];
+ for (int i = 0; i < projectedFields.size(); i++) {
+ projection[i] = projectedFields.get(i);
+ }
+ return projection;
+ }
+
+ private org.apache.paimon.data.BinaryRow convertPartition(
+ ResolvedPartitionSpec partitionSpec, RowType valueRowType) {
+ return toPaimonPartition(
+ partitionSpec,
+ valueRowType,
+ fileStoreTable().schema().logicalRowType(),
+ partitionKeyExtractor()::partition);
+ }
+
+ private org.apache.paimon.data.BinaryRow wrapLookupKey(byte[] key) {
+ org.apache.paimon.data.BinaryRow keyRow =
+ new org.apache.paimon.data.BinaryRow(primaryKeyFieldCount);
+ keyRow.pointTo(MemorySegment.wrap(key), 0, key.length);
+ return keyRow;
+ }
+
+ private void initializeFilesIfNeeded(org.apache.paimon.data.BinaryRow
partition, int bucketId) {
+ PaimonPartitionBucket partitionBucket = new
PaimonPartitionBucket(partition, bucketId);
+ if (initializedBuckets.contains(partitionBucket)) {
Review Comment:
Looks like we never refresh the file set after the first scan. If tiering
commits more data into this bucket later, lookups will miss it and the expiry
never fires on a hot partition.
Should we check the snapshot id and rescan when it changes?
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java:
##########
@@ -0,0 +1,346 @@
+/*
+ * 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.server.replica;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.HistoricalLookupThrottledException;
+import org.apache.fluss.exception.InvalidPartitionException;
+import org.apache.fluss.exception.LakeStorageNotConfiguredException;
+import org.apache.fluss.lake.lakestorage.LakeStorage;
+import org.apache.fluss.lake.lakestorage.LakeStoragePlugin;
+import org.apache.fluss.lake.lakestorage.LakeStoragePluginSetUp;
+import org.apache.fluss.lake.lakestorage.LakeTableLookuper;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.plugin.PluginManager;
+import org.apache.fluss.rpc.entity.LookupResultForBucket;
+import org.apache.fluss.rpc.protocol.ApiError;
+import org.apache.fluss.server.entity.LookupDataForBucket;
+import org.apache.fluss.utils.FileUtils;
+import org.apache.fluss.utils.IOUtils;
+import org.apache.fluss.utils.concurrent.Scheduler;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.RemovalCause;
+import com.github.benmanes.caffeine.cache.Ticker;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Semaphore;
+
+import static
org.apache.fluss.server.utils.LakeStorageUtils.extractLakeProperties;
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Handles server-side point lookup for historical partitions stored in lake
storage.
+ *
+ * <p>Accepted requests run on the TabletServer I/O executor. A semaphore
bounds the total number of
+ * accepted historical lookup tasks so slow lake storage cannot create an
unbounded request backlog.
+ *
+ * <p>Creating a lake table lookuper may initialize catalog, table, and query
state and allocate
+ * local lookup files, so lookupers are cached and reused. The cache is keyed
by table ID rather
+ * than table path to prevent a deleted and recreated table from reusing the
old table's lookuper. A
+ * cached lookuper is replaced when its schema ID no longer matches the
requested table schema.
+ *
+ * <p>A lookuper is closed when replaced, explicitly invalidated by a replica
lifecycle event, the
+ * manager shuts down, or after three hours without access. Caffeine
expiration is scheduled on the
+ * shared TabletServer scheduler, allowing idle resources to be released even
if no subsequent
+ * lookup accesses the cache.
+ */
+class HistoricalLakeLookupManager implements AutoCloseable {
+
+ private static final String PAIMON_LOOKUP_DIR_NAME = "paimon-lookup";
+ private static final String LOOKUPER_CACHE_EXPIRATION_TASK_NAME =
+ "historical-lookuper-cache-expiration";
+ private static final Duration LOOKUPER_CACHE_EXPIRATION =
Duration.ofHours(3);
+
+ private final Configuration conf;
+ private final @Nullable PluginManager pluginManager;
+ private final ExecutorService ioExecutor;
+ private final int serverId;
+ private final Semaphore lookupPermits;
+ private final Cache<Long, CachedLakeTableLookuper> lakeTableLookupers;
+ private @Nullable String paimonLookupTempDir;
+
+ HistoricalLakeLookupManager(
+ Configuration conf,
+ @Nullable PluginManager pluginManager,
+ ExecutorService ioExecutor,
+ int serverId,
+ Scheduler scheduler) {
+ this(
+ conf,
+ pluginManager,
+ ioExecutor,
+ serverId,
+ Ticker.systemTicker(),
+ createCacheScheduler(scheduler));
+ }
+
+ @VisibleForTesting
+ HistoricalLakeLookupManager(
+ Configuration conf,
+ @Nullable PluginManager pluginManager,
+ ExecutorService ioExecutor,
+ int serverId,
+ Ticker ticker,
+ com.github.benmanes.caffeine.cache.Scheduler cacheScheduler) {
+ this.conf = checkNotNull(conf, "conf must not be null.");
+ this.pluginManager = pluginManager;
+ this.ioExecutor = checkNotNull(ioExecutor, "ioExecutor must not be
null.");
+ this.serverId = serverId;
+ this.lakeTableLookupers =
+ Caffeine.newBuilder()
+ .expireAfterAccess(LOOKUPER_CACHE_EXPIRATION)
+ .ticker(checkNotNull(ticker, "ticker must not be
null."))
+ .scheduler(checkNotNull(cacheScheduler,
"cacheScheduler must not be null."))
+ .executor(Runnable::run)
+ .removalListener(
+ (Long ignored,
+ CachedLakeTableLookuper cachedLookuper,
+ RemovalCause ignoredCause) -> {
+ if (cachedLookuper != null) {
+ closeLookuper(cachedLookuper);
+ }
+ })
+ .build();
+ int maxQueuedHistoricalRequests =
+
conf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS);
+ checkArgument(
+ maxQueuedHistoricalRequests > 0,
+ "%s must be greater than 0.",
+
ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key());
+ this.lookupPermits = new Semaphore(maxQueuedHistoricalRequests);
+ }
+
+ private static com.github.benmanes.caffeine.cache.Scheduler
createCacheScheduler(
+ Scheduler scheduler) {
+ checkNotNull(scheduler, "scheduler must not be null.");
+ // Schedule expiration maintenance so idle lookupers are closed even
if no more lookups
+ // arrive.
+ return (executor, command, delay, timeUnit) ->
+ scheduler.scheduleOnce(
+ LOOKUPER_CACHE_EXPIRATION_TASK_NAME,
+ () -> executor.execute(command),
+ timeUnit.toMillis(delay));
+ }
+
+ CompletableFuture<LookupResultForBucket> lookup(
+ LookupDataForBucket lookupData, TableInfo tableInfo) {
+ TableBucket tableBucket = lookupData.tableBucket();
+ if (!lookupPermits.tryAcquire()) {
+ return CompletableFuture.completedFuture(
+ new LookupResultForBucket(
+ tableBucket,
+ ApiError.fromThrowable(
+ new HistoricalLookupThrottledException(
+ "Historical lookup is throttled
for "
+ + tableBucket
+ + "."))));
+ }
+
+ CompletableFuture<LookupResultForBucket> future;
+ try {
+ future =
+ CompletableFuture.supplyAsync(
+ () -> lookupInternal(lookupData, tableInfo),
ioExecutor);
+ } catch (RuntimeException e) {
+ lookupPermits.release();
+ throw e;
+ }
+ future.whenComplete((ignored, error) -> lookupPermits.release());
+ return future;
+ }
+
+ @Override
+ public void close() {
+ lakeTableLookupers.invalidateAll();
+ lakeTableLookupers.cleanUp();
+ }
+
+ void invalidateTableLookuper(long tableId) {
+ lakeTableLookupers.invalidate(tableId);
+ }
+
+ private LookupResultForBucket lookupInternal(
+ LookupDataForBucket lookupData, TableInfo tableInfo) {
+ TableBucket tableBucket = lookupData.tableBucket();
+ try {
+ LookupContext context = createLookupContext(lookupData, tableInfo);
+ CachedLakeTableLookuper cachedLookuper =
+ lakeTableLookupers
+ .asMap()
+ .compute(
+ context.tableId,
+ (ignored, currentLookuper) -> {
+ if (currentLookuper != null
+ && currentLookuper.schemaId ==
context.schemaId) {
+ return currentLookuper;
+ }
+ LakeTableLookuper newLookuper =
+ createLakeTableLookuper(
+ context.tablePath,
+
getOrPreparePaimonLookupTempDir());
+ return new CachedLakeTableLookuper(
+ context.schemaId, newLookuper);
+ });
+ List<byte[]> values = new ArrayList<>(lookupData.keys().size());
+ for (byte[] key : lookupData.keys()) {
+ values.add(cachedLookuper.lookuper.lookup(key,
context.lookupContext));
+ }
+ return new LookupResultForBucket(tableBucket, values);
+ } catch (Exception e) {
+ return new LookupResultForBucket(tableBucket,
ApiError.fromThrowable(e));
+ }
+ }
+
+ private LookupContext createLookupContext(LookupDataForBucket lookupData,
TableInfo tableInfo) {
+ TableBucket tableBucket = lookupData.tableBucket();
+ String originalPartitionName = lookupData.originalPartitionName();
+ if (originalPartitionName == null) {
+ throw new InvalidPartitionException(
+ "Historical lookup request must carry the original
partition name.");
+ }
+
+ TablePath tablePath = tableInfo.getTablePath();
+
+ ResolvedPartitionSpec originalPartitionSpec;
+ try {
+ originalPartitionSpec =
+ ResolvedPartitionSpec.fromPartitionName(
+ tableInfo.getPartitionKeys(),
originalPartitionName);
+ } catch (RuntimeException e) {
+ throw new InvalidPartitionException(
+ String.format(
+ "Invalid original partition name %s for historical
lookup on table %s.",
+ originalPartitionName, tablePath));
+ }
+
+ LakeTableLookuper.LookupContext lookupContext =
+ new LakeTableLookuper.LookupContext(
+ originalPartitionSpec,
+ tableBucket.getBucket(),
+ (short) tableInfo.getSchemaInfo().getSchemaId(),
Review Comment:
Replica.tableInfo is frozen when the replica is created. After adding a
column, won't this keep encoding with the old schema and return null for the
new column?
Maybe use replica.getSchemaGetter() instead?
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java:
##########
@@ -774,6 +871,49 @@ public void lookups(
lookups(false, null, null, entriesPerBucket, apiVersion,
responseCallback);
}
+ /** Lookup historical lake data for requests that carry original partition
names. */
+ public void historicalLookups(
Review Comment:
This goes straight to lake, but the FIP says buffer -> historical rocksdb ->
lake. How will the write-path PR plug that in here?
##########
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java:
##########
Review Comment:
Q: Historical lookups share this semaphore with normal ones, and lake
lookups are slow, they can hold all the permits and block normal lookups. The
FIP splits this into two semaphores (client.lookup.historical-inflight-ratio,
historical capped at ~10%). Was dropping that intentional?
##########
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java:
##########
@@ -115,19 +125,89 @@ public CompletableFuture<LookupResult> lookup(InternalRow
lookupKey) {
tableInfo.getTablePath(),
metadataUpdater);
} catch (PartitionNotExistException e) {
- return CompletableFuture.completedFuture(new
LookupResult(Collections.emptyList()));
+ return mayFallbackToHistoricalLookup(
+ bucketingFunction.bucketing(bkBytes, numBuckets),
pkBytes, lookupKey);
}
}
int bucketId = bucketingFunction.bucketing(bkBytes, numBuckets);
TableBucket tableBucket = new TableBucket(tableInfo.getTableId(),
partitionId, bucketId);
+ return lookupBucket(tableBucket, pkBytes, insertIfNotExists, null,
lookupKey);
+ }
+
+ /** Falls back to historical lookup if the lookup key belongs to a
historical partition. */
+ private CompletableFuture<LookupResult> mayFallbackToHistoricalLookup(
+ int bucketId, byte[] keyBytes, InternalRow lookupKey) {
+ String originalPartitionName = partitionGetter.getPartition(lookupKey);
+ if (!isHistoricalLookupCandidatePartition(
+ tableInfo, originalPartitionName, Instant.now())) {
+ return CompletableFuture.completedFuture(new
LookupResult(Collections.emptyList()));
+ }
+ metadataUpdater.invalidPhysicalTableBucketAndPartitionMeta(
Review Comment:
This happens on every lookup to an expired partition, right? Each one does a
failed metadata RPC plus a full Cluster copy here.
Can we cache the "this partition is expired" decision?
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java:
##########
@@ -0,0 +1,359 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+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.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.CompactedRowEncoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.IOUtils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.query.LocalTableQuery;
+import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataField;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
+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;
+
+/**
+ * Paimon implementation of {@link LakeTableLookuper} for primary-key tables.
+ *
+ * <p>The catalog, table, local query, and I/O manager are initialized lazily
on the first lookup.
+ * For each partition and bucket, the lookuper scans the latest Paimon
snapshot once and registers
+ * its data files with {@link LocalTableQuery}. Paimon then creates local
lookup files lazily as
+ * individual remote data files are queried.
+ *
+ * <p>A cached partition-bucket file set can become stale when Paimon
compaction replaces its data
+ * files and snapshot expiration physically deletes the old files. Because
{@code FileIO}
+ * implementations may represent a missing file with different {@link
IOException} types, the first
+ * lookup I/O failure closes the cached query state, reopens the table from
the latest snapshot, and
+ * retries once.
+ *
+ * <p>Lookup and close operations are synchronized because they share mutable
Paimon query, local
+ * cache, and value-encoding state.
+ */
+public class PaimonLakeTableLookuper implements LakeTableLookuper {
+
+ // Hard-code a conservative 1GB default for now to avoid unbounded lookup
cache growth.
+ private static final String DEFAULT_LOOKUP_CACHE_MAX_DISK_SIZE = "1gb";
Review Comment:
nit: this is per table, so 20 lake tables can use 20gb of tmp disk. Do we
need a global cap?
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java:
##########
@@ -0,0 +1,359 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+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.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.CompactedRowEncoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.IOUtils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.query.LocalTableQuery;
+import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataField;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
+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;
+
+/**
+ * Paimon implementation of {@link LakeTableLookuper} for primary-key tables.
+ *
+ * <p>The catalog, table, local query, and I/O manager are initialized lazily
on the first lookup.
+ * For each partition and bucket, the lookuper scans the latest Paimon
snapshot once and registers
+ * its data files with {@link LocalTableQuery}. Paimon then creates local
lookup files lazily as
+ * individual remote data files are queried.
+ *
+ * <p>A cached partition-bucket file set can become stale when Paimon
compaction replaces its data
+ * files and snapshot expiration physically deletes the old files. Because
{@code FileIO}
+ * implementations may represent a missing file with different {@link
IOException} types, the first
+ * lookup I/O failure closes the cached query state, reopens the table from
the latest snapshot, and
+ * retries once.
+ *
+ * <p>Lookup and close operations are synchronized because they share mutable
Paimon query, local
+ * cache, and value-encoding state.
+ */
+public class PaimonLakeTableLookuper implements LakeTableLookuper {
+
+ // Hard-code a conservative 1GB default for now to avoid unbounded lookup
cache growth.
+ private static final String DEFAULT_LOOKUP_CACHE_MAX_DISK_SIZE = "1gb";
+
+ private final Configuration paimonConfig;
+ private final TablePath tablePath;
+ private final String ioTmpDir;
+
+ private final Set<PaimonPartitionBucket> initializedBuckets;
+
+ private @Nullable Catalog catalog;
+ private @Nullable FileStoreTable fileStoreTable;
+ private @Nullable IOManager ioManager;
+ private @Nullable LocalTableQuery localTableQuery;
+ private @Nullable RowPartitionKeyExtractor partitionKeyExtractor;
+ private int primaryKeyFieldCount;
+ private boolean hasCachedValueEncoder;
+ private short cachedValueSchemaId;
+ private @Nullable CompactedRowEncoder cachedValueRowEncoder;
+ private @Nullable InternalRow.FieldGetter[] cachedValueFieldGetters;
+ private boolean closed;
+
+ public PaimonLakeTableLookuper(Configuration paimonConfig, TablePath
tablePath) {
+ this(paimonConfig, tablePath, ConfigOptions.IO_TMP_DIR.defaultValue());
+ }
+
+ public PaimonLakeTableLookuper(
+ Configuration paimonConfig, TablePath tablePath, String ioTmpDir) {
+ this.paimonConfig = checkNotNull(paimonConfig, "paimonConfig must not
be null.");
+ this.tablePath = checkNotNull(tablePath, "tablePath must not be
null.");
+ this.ioTmpDir = checkNotNull(ioTmpDir, "ioTmpDir must not be null.");
+ this.initializedBuckets = new HashSet<>();
+ }
+
+ @Override
+ public synchronized @Nullable byte[] lookup(byte[] key, LookupContext
context)
Review Comment:
This synchronizes all lake lookups of a table on one lock, and each waiting
lookup holds an ioExecutor thread.
That pool is size 10? by default and also does snapshot/remote-log
transfers. Should historical lookups use their own pool?
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java:
##########
@@ -0,0 +1,359 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+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.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.CompactedRowEncoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.IOUtils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.query.LocalTableQuery;
+import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataField;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
+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;
+
+/**
+ * Paimon implementation of {@link LakeTableLookuper} for primary-key tables.
+ *
+ * <p>The catalog, table, local query, and I/O manager are initialized lazily
on the first lookup.
+ * For each partition and bucket, the lookuper scans the latest Paimon
snapshot once and registers
+ * its data files with {@link LocalTableQuery}. Paimon then creates local
lookup files lazily as
+ * individual remote data files are queried.
+ *
+ * <p>A cached partition-bucket file set can become stale when Paimon
compaction replaces its data
+ * files and snapshot expiration physically deletes the old files. Because
{@code FileIO}
+ * implementations may represent a missing file with different {@link
IOException} types, the first
+ * lookup I/O failure closes the cached query state, reopens the table from
the latest snapshot, and
+ * retries once.
+ *
+ * <p>Lookup and close operations are synchronized because they share mutable
Paimon query, local
+ * cache, and value-encoding state.
+ */
+public class PaimonLakeTableLookuper implements LakeTableLookuper {
+
+ // Hard-code a conservative 1GB default for now to avoid unbounded lookup
cache growth.
+ private static final String DEFAULT_LOOKUP_CACHE_MAX_DISK_SIZE = "1gb";
+
+ private final Configuration paimonConfig;
+ private final TablePath tablePath;
+ private final String ioTmpDir;
+
+ private final Set<PaimonPartitionBucket> initializedBuckets;
+
+ private @Nullable Catalog catalog;
+ private @Nullable FileStoreTable fileStoreTable;
+ private @Nullable IOManager ioManager;
+ private @Nullable LocalTableQuery localTableQuery;
+ private @Nullable RowPartitionKeyExtractor partitionKeyExtractor;
+ private int primaryKeyFieldCount;
+ private boolean hasCachedValueEncoder;
+ private short cachedValueSchemaId;
+ private @Nullable CompactedRowEncoder cachedValueRowEncoder;
+ private @Nullable InternalRow.FieldGetter[] cachedValueFieldGetters;
+ private boolean closed;
+
+ public PaimonLakeTableLookuper(Configuration paimonConfig, TablePath
tablePath) {
+ this(paimonConfig, tablePath, ConfigOptions.IO_TMP_DIR.defaultValue());
+ }
+
+ public PaimonLakeTableLookuper(
+ Configuration paimonConfig, TablePath tablePath, String ioTmpDir) {
+ this.paimonConfig = checkNotNull(paimonConfig, "paimonConfig must not
be null.");
+ this.tablePath = checkNotNull(tablePath, "tablePath must not be
null.");
+ this.ioTmpDir = checkNotNull(ioTmpDir, "ioTmpDir must not be null.");
+ this.initializedBuckets = new HashSet<>();
+ }
+
+ @Override
+ public synchronized @Nullable byte[] lookup(byte[] key, LookupContext
context)
+ throws Exception {
+ checkNotNull(key, "key must not be null.");
+ checkNotNull(context, "context must not be null.");
+ checkNotClosed();
+ ensureInitialized();
+
+ org.apache.paimon.data.BinaryRow partition =
+ convertPartition(context.partitionSpec(),
context.valueRowType());
+ org.apache.paimon.data.BinaryRow keyRow = wrapLookupKey(key);
+ initializeFilesIfNeeded(partition, context.bucketId());
+
+ org.apache.paimon.data.InternalRow paimonRow =
+ lookupWithFileRefresh(partition, context.bucketId(), keyRow);
+ if (paimonRow == null) {
+ return null;
+ }
+ return encodeValue(paimonRow, context.schemaId(),
context.valueRowType());
+ }
+
+ @Override
+ public synchronized void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ IOUtils.closeQuietly(cachedValueRowEncoder, "Fluss value row encoder");
+ IOUtils.closeQuietly(localTableQuery, "Paimon local table query");
+ IOUtils.closeQuietly(ioManager, "Paimon lookup IO manager");
+ IOUtils.closeQuietly(catalog, "Paimon catalog");
+ initializedBuckets.clear();
+ }
+
+ private void checkNotClosed() {
+ if (closed) {
+ throw new IllegalStateException("Paimon lake table lookuper has
been closed.");
+ }
+ }
+
+ private void ensureInitialized() throws Exception {
+ if (localTableQuery != null) {
+ return;
+ }
+
+ catalog =
+ CatalogFactory.createCatalog(
+
CatalogContext.create(Options.fromMap(paimonConfig.toMap())));
+ fileStoreTable =
+ withLookupCacheOptions((FileStoreTable)
catalog.getTable(toPaimon(tablePath)));
+ if (fileStoreTable.primaryKeys().isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Point lookup is only supported for primary-key Paimon
tables.");
+ }
+ ioManager = createIOManager(ioTmpDir);
+ localTableQuery = createLocalTableQuery();
+ partitionKeyExtractor = new
RowPartitionKeyExtractor(fileStoreTable.schema());
+ primaryKeyFieldCount = fileStoreTable.primaryKeys().size();
+ }
+
+ private LocalTableQuery createLocalTableQuery() {
+ return fileStoreTable()
+ .newLocalTableQuery()
+ .withValueProjection(businessFieldProjection(fileStoreTable()))
+ .withIOManager(checkNotNull(ioManager, "ioManager must be
initialized."));
+ }
+
+ private FileStoreTable withLookupCacheOptions(FileStoreTable table) {
+ String key = CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE.key();
+ if (table.options().containsKey(key)) {
+ return table;
+ }
+ return table.copy(Collections.singletonMap(key,
DEFAULT_LOOKUP_CACHE_MAX_DISK_SIZE));
+ }
+
+ private static IOManager createIOManager(String ioTmpDir) {
+ return IOManager.create(ioTmpDir);
+ }
+
+ private static int[] businessFieldProjection(FileStoreTable
fileStoreTable) {
+ List<DataField> fields =
fileStoreTable.schema().logicalRowType().getFields();
+ List<Integer> projectedFields = new ArrayList<>();
+ for (int i = 0; i < fields.size(); i++) {
+ if (!SYSTEM_COLUMNS.containsKey(fields.get(i).name())) {
+ projectedFields.add(i);
+ }
+ }
+
+ int[] projection = new int[projectedFields.size()];
+ for (int i = 0; i < projectedFields.size(); i++) {
+ projection[i] = projectedFields.get(i);
+ }
+ return projection;
+ }
+
+ private org.apache.paimon.data.BinaryRow convertPartition(
+ ResolvedPartitionSpec partitionSpec, RowType valueRowType) {
+ return toPaimonPartition(
+ partitionSpec,
+ valueRowType,
+ fileStoreTable().schema().logicalRowType(),
+ partitionKeyExtractor()::partition);
+ }
+
+ private org.apache.paimon.data.BinaryRow wrapLookupKey(byte[] key) {
+ org.apache.paimon.data.BinaryRow keyRow =
+ new org.apache.paimon.data.BinaryRow(primaryKeyFieldCount);
+ keyRow.pointTo(MemorySegment.wrap(key), 0, key.length);
+ return keyRow;
+ }
+
+ private void initializeFilesIfNeeded(org.apache.paimon.data.BinaryRow
partition, int bucketId) {
+ PaimonPartitionBucket partitionBucket = new
PaimonPartitionBucket(partition, bucketId);
+ if (initializedBuckets.contains(partitionBucket)) {
+ return;
+ }
+
+ LinkedHashMap<String, DataFileMeta> beforeFilesByName = new
LinkedHashMap<>();
+ LinkedHashMap<String, DataFileMeta> dataFilesByName = new
LinkedHashMap<>();
+
+ InnerTableScan tableScan =
+ fileStoreTable()
+ .newScan()
+
.withPartitionFilter(Collections.singletonList(partition))
+ .withBucket(bucketId);
+ for (Split split : tableScan.plan().splits()) {
+ if (!(split instanceof DataSplit)) {
+ continue;
+ }
+ DataSplit dataSplit = (DataSplit) split;
+ addFilesByName(beforeFilesByName, dataSplit.beforeFiles());
+ addFilesByName(dataFilesByName, dataSplit.dataFiles());
+ }
+
+ localTableQuery()
+ .refreshFiles(
+ partition,
+ bucketId,
+ new ArrayList<>(beforeFilesByName.values()),
+ new ArrayList<>(dataFilesByName.values()));
+ initializedBuckets.add(partitionBucket);
+ }
+
+ private org.apache.paimon.data.InternalRow lookupWithFileRefresh(
+ org.apache.paimon.data.BinaryRow partition,
+ int bucketId,
+ org.apache.paimon.data.InternalRow keyRow)
+ throws Exception {
+ try {
+ return localTableQuery().lookup(partition, bucketId, keyRow);
+ } catch (IOException e) {
+ // FileIO only guarantees IOException and storage plugins may use
different exception
+ // types for a missing file. The missing old file after compaction
may therefore
+ // surface as any IOException. Refresh and retry only once so
persistent I/O failures
+ // do not repeatedly rebuild Paimon lookup state within one
request.
+ try {
+ refreshFiles(partition, bucketId);
+ return localTableQuery().lookup(partition, bucketId, keyRow);
+ } catch (IOException retryError) {
+ retryError.addSuppressed(e);
+ // Historical Paimon point lookup is part of the Fluss KV
lookup path. Expose a
+ // persistent I/O failure as a retriable KV error so the
existing KV RPC retry
+ // semantics can handle it consistently.
+ throw new KvStorageException(
+ "Failed to lookup historical data from Paimon after
refreshing files for "
+ + tablePath
+ + ".",
+ retryError);
+ }
+ }
+ }
+
+ private void refreshFiles(org.apache.paimon.data.BinaryRow partition, int
bucketId)
+ throws Exception {
+ IOUtils.closeQuietly(localTableQuery, "Paimon local table query");
+ IOUtils.closeQuietly(ioManager, "Paimon lookup IO manager");
+ IOUtils.closeQuietly(catalog, "Paimon catalog");
+ localTableQuery = null;
+ ioManager = null;
+ catalog = null;
+ fileStoreTable = null;
+ partitionKeyExtractor = null;
+ initializedBuckets.clear();
+ ensureInitialized();
+ initializeFilesIfNeeded(partition, bucketId);
+ }
+
+ private static void addFilesByName(
+ LinkedHashMap<String, DataFileMeta> filesByName,
List<DataFileMeta> files) {
+ for (DataFileMeta file : files) {
+ filesByName.put(file.fileName(), file);
+ }
+ }
+
+ private byte[] encodeValue(
+ org.apache.paimon.data.InternalRow paimonRow, short schemaId,
RowType valueRowType) {
+ PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow);
+ try {
+ ensureValueEncoder(schemaId, valueRowType);
+ CompactedRowEncoder rowEncoder =
+ checkNotNull(cachedValueRowEncoder, "cachedValueRowEncoder
must not be null.");
+ InternalRow.FieldGetter[] fieldGetters =
+ checkNotNull(
+ cachedValueFieldGetters, "cachedValueFieldGetters
must not be null.");
+
+ 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);
+ }
+ }
+
+ private void ensureValueEncoder(short schemaId, RowType valueRowType) {
+ if (hasCachedValueEncoder && cachedValueSchemaId == schemaId) {
+ return;
+ }
+
+ IOUtils.closeQuietly(cachedValueRowEncoder, "Fluss value row encoder");
+ DataType[] fieldTypes = valueRowType.getChildren().toArray(new
DataType[0]);
+ cachedValueRowEncoder = new CompactedRowEncoder(fieldTypes);
Review Comment:
nit: The client decodes lookup values with the table's kv format, but here
we always encode compacted.
What happens for a table with table.kv.format = indexed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
