[lookup] Support historical partition lookup for lake tables

[luoyuxia]

Purpose

Linked issue: close #3632

Part of #3631.

This PR adds historical partition lookup support for datalake-enabled primary key tables. It lets clients resolve an expired Fluss partition to the corresponding lake partition and serve point lookups from lake storage when the partition is no longer available in Fluss.

Brief change log

• Add historical partition metadata utilities and client-side lookup plumbing for requests carrying partition names.
• Add lake lookup SPI plus the Paimon implementation, including key conversion, row encoding, schema evolution padding, and lookup cache temp context.
• Route historical lookup requests through TabletServer/ReplicaManager to lake lookup instead of normal KV lookup.
• Add dedicated historical lookup admission control with netty.server.max-queued-historical-requests and a retriable throttle error.
• Add client-side backoff for historical lookup throttling without blocking normal lookup dispatch.
• Keep the Rust lookup client compatible with the updated lookup RPC schema.

Tests

• ./mvnw test -pl fluss-rpc -am -Dtest=ApiErrorTest -DfailIfNoTests=false
• ./mvnw test -pl fluss-client -am -Dtest=LookupSenderTest -DfailIfNoTests=false
• ./mvnw test -pl fluss-server -am -Dtest=HistoricalLakeLookupManagerTest -DfailIfNoTests=false
• ./mvnw test -pl fluss-server -am -Dtest=ReplicaFetcherThreadTest -DfailIfNoTests=false
• ./mvnw spotless:check -pl fluss-common,fluss-rpc,fluss-client,fluss-server
• cargo fmt --all -- --check from fluss-rust/
• cargo check -p fluss-rs from fluss-rust/

Full affected-module verify was started locally, but the environment hit unrelated local-state issues before the replica fetcher test was made independent from host disk usage.

API and Format

• Adds LakeTableLookuper/lookup context APIs for lake lookup implementations.
• Adds partition_name to lookup RPC bucket requests for historical lookup routing.
• Adds HISTORICAL_LOOKUP_THROTTLED RPC error mapping.
• Adds netty.server.max-queued-historical-requests server config.

Documentation

N/A. The FIP-28 design documents are intentionally not included in this PR diff.

Generative AI disclosure

Yes. Generated with OpenAI Codex and reviewed by the human developer.

[Copilot]

Pull request overview

This PR implements end-to-end “historical partition point lookup” for datalake-enabled primary-key tables: when an auto-partition has expired in Fluss, the client resolves a corresponding historical system partition and the server serves the point lookup from lake storage (currently Paimon), with dedicated throttling and client retry/backoff.

Changes:

• Extend lookup RPC + client/server routing to carry an original partition_name and route those requests through a historical lake lookup path.
• Add historical partition utilities/validation and adjust partition lifecycle behavior (e.g., retention cleanup, historical system partition creation authorization).
• Introduce a lake lookup SPI (LakeTableLookuper) and Paimon implementation (key conversion, value encoding, schema evolution padding) plus tests and IT coverage.

Reviewed changes

Copilot reviewed 41 out of 41 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java	Unit tests for server-side historical lookup admission control / permit release.
fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java	Verifies auto-partition cleanup does not drop historical system partitions.
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java	Adds conversion helper for historical lookup bucket requests (keeps partition name).
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java	Routes lookup requests to historical path when partition_name is present; rejects insert-if-not-exists for historical.
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java	Wires PluginManager through to ReplicaManager for lake lookup plugin setup.
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Adds HistoricalLakeLookupManager and a historicalLookups dispatch path; closes manager on shutdown.
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java	Implements server-side lake lookup execution, lookup throttling, and per-table lookuper caching.
fluss-server/src/main/java/org/apache/fluss/server/entity/LookupDataForBucket.java	New request data container including optional original partition name.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java	Allows creating historical system partitions with READ auth and specific validation.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java	Skips historical system partitions during retention-based drop logic.
fluss-rust/crates/fluss/src/rpc/message/lookup.rs	Keeps Rust lookup request construction compatible with new optional partition_name field.
fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/ApiErrorTest.java	Adds mapping test for HistoricalLookupThrottledException → HISTORICAL_LOOKUP_THROTTLED.
fluss-rpc/src/main/proto/FlussApi.proto	Adds optional partition_name to PbLookupReqForBucket.
fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java	Introduces HISTORICAL_LOOKUP_THROTTLED error code and exception mapping.
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java	Unit tests for Paimon lake lookuper (PK lookup, partition types, schema evolution).
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/HistoricalPartitionLookupITCase.java	End-to-end IT validating lookup of an expired Fluss partition from Paimon with schema evolution.
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java	Adds shared helper to convert resolved partition specs into Paimon partition rows.
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java	Refactors partition resolution to use PaimonConversions.toPaimonPartition.
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java	Implements createLakeTableLookuper for Paimon storage.
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java	New Paimon-based point lookup implementation with value encoding and lookup cache handling.
fluss-common/src/test/java/org/apache/fluss/utils/PartitionUtilsTest.java	Adds tests for historical partition naming/spec conversion and lookup candidacy.
fluss-common/src/main/java/org/apache/fluss/utils/PartitionUtils.java	Adds historical partition constants/utilities and validation helpers.
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/PluginLakeStorageWrapper.java	Wraps LakeTableLookuper with plugin classloader context.
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java	New SPI for lake table point lookups.
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java	Adds default createLakeTableLookuper + lookuper context.
fluss-common/src/main/java/org/apache/fluss/exception/HistoricalLookupThrottledException.java	New retriable exception for historical lookup admission control.
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Adds io.tmpdir and netty.server.max-queued-historical-requests options.
fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java	Verifies historical system partition creation requires READ permission.
fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupSenderTest.java	Adds tests for batching/splitting historical lookups and throttle backoff behavior.
fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java	Sets partition_name in lookup RPC when present on batch.
fluss-client/src/main/java/org/apache/fluss/client/table/FlussTable.java	Passes Admin into lookup construction for historical partition resolution.
fluss-client/src/main/java/org/apache/fluss/client/lookup/TableLookup.java	Wires Admin into lookup creation and adds historical resolver into PK lookuper.
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java	Adds historical partition fallback path and Paimon key encoding for lake lookups.
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java	Splits/packs historical vs normal lookup requests and adds throttling backoff logic.
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupQueue.java	Adds delayed retry scheduling via nextRetryTimeMs to avoid blocking normal lookups.
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupQuery.java	Adds optional partition name to carry original partition name for historical lookup.
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupClient.java	Extends lookup API to accept optional partition name.
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupBatchKey.java	New key type to batch by (bucket, partitionName).
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupBatch.java	Tracks optional partition name for a batch to populate RPC.
fluss-client/src/main/java/org/apache/fluss/client/lookup/HistoricalPartitionResolver.java	Resolves original partition → historical system partition id (refresh/create/refresh).
fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookupQuery.java	Adds partition name + next retry timestamp to support historical routing/backoff.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[zuston]

Thanks for proposing this. @luoyuxia I’m also exploring this area with Fluss + Paimon.

Regarding the proposed client fallback to server-side historical partition lookup, I’m a bit concerned that this could be risky. The lake lookup operation can be much slower than a local RocksDB lookup, especially when Paimon is backed by HDFS with poor performance. If this lookup is executed on the tablet server side, it may slow down Netty worker threads and affect the latency of other RPC requests.

For the case without historical data ingestion, I think this could potentially be handled by direct client-side lake lookup instead. On the Fluss side, we could consider introducing a Flink hybrid lake lookup async function for this purpose.
For the case with historical data ingestion, client-side lookup could still be used as a fallback when the historical partition + key is not found on the tablet server side.

Overall, I would prefer using client-side PK lookup where possible, to avoid overloading tablet servers during historical data ingestion bursts.

[luoyuxia]

Thanks for proposing this. @luoyuxia I’m also exploring this area with Fluss + Paimon.

Regarding the proposed client fallback to server-side historical partition lookup, I’m a bit concerned that this could be risky. The lake lookup operation can be much slower than a local RocksDB lookup, especially when Paimon is backed by HDFS with poor performance. If this lookup is executed on the tablet server side, it may slow down Netty worker threads and affect the latency of other RPC requests.

For the case without historical data ingestion, I think this could potentially be handled by direct client-side lake lookup instead. On the Fluss side, we could consider introducing a Flink hybrid lake lookup async function for this purpose. For the case with historical data ingestion, client-side lookup could still be used as a fallback when the historical partition + key is not found on the tablet server side.

Overall, I would prefer using client-side PK lookup where possible, to avoid overloading tablet servers during historical data ingestion bursts.

Thanks for sharing your thoughts. First, I’d like to clarify that the expensive Paimon lookup does not run on the Netty worker thread. It is handed off to the TabletServer IO executor through CompletableFuture.supplyAsync. Its concurrency is bounded by a semaphore, and excess requests receive a retriable throttling error with client-side backoff.

I agree that historical lookup still consumes additional TabletServer CPU, memory, network IO, and local disk for the Paimon lookup cache. This is a resource trade-off, even though it does not block the Netty event loop.

An important motivation for server-side historical lookup is historical data ingestion. We want to support writing into historical partitions while still producing the correct changelog. The server-side write path therefore needs to retrieve the previous value from historical storage before applying a write. Some server-accessible historical lookup infrastructure is required independently of the client fallback in this PR; the client lookup path reuses the same infrastructure.

Server-side lookup also keeps the generic Fluss lookup API independent of a particular lake implementation. Clients do not need to manage Paimon/Hadoop dependencies, lake credentials, catalog configuration, key encoding, snapshot selection, and fallback semantics.

A Flink hybrid lake lookup async function would be useful for Flink workloads, but Fluss also has Java, Rust, CPP, Python clients. Requiring every client to implement its own lake lookup and fallback logic would duplicate behavior and would be particularly difficult for Rust and Python clients when the lake implementation is JVM-based.

I think the two approaches can coexist, similar to Paimon’s local lookup and remote query service models. A remote lookup service can provide generic lookup capability for historical ingestion and all Fluss clients. The current PR runs this capability on TabletServers, but it could potentially be moved to a dedicated service later for better resource isolation. Client-side local lookup can then be introduced as an optional optimization where practical, with Flink async lookup as one possible first integration.

For this PR, my preference is to keep the bounded asynchronous server-side path because it also supports historical ingestion semantics, and explore client-side lookup as an optional strategy in a follow-up.