luoyuxia commented on code in PR #3630: URL: https://github.com/apache/fluss/pull/3630#discussion_r3602145810
########## 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: I agree that creating real replicas(no rocksdb for read now) from a read path is not ideal. However, the historical system partition is currently also the routing and ownership anchor: the client needs its partition ID to resolve a bucket leader, and the server uses the corresponding replica to validate the request and obtain table/schema metadata. Routing historical lookups without creating the partition would require a separate virtual-routing path: selecting a TabletServer without partition leadership, resolving table and schema metadata without a replica, handling retries, and gating rolling upgrades. That is broader than removing the createPartition call. -- 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]
