[luoyuxia]

Generated-by: Codex (GPT-5) following the guidelines

Purpose

Linked issue: close #3633

Allow tablet servers to process primary-key writes that target the shared historical partition. The RPC carries the original partition name so the server can preserve partition identity while multiple original partitions share one physical bucket.

This PR provides the server-side write path only. Client-side routing is deferred to a follow-up PR. Historical KV recovery, snapshots, and cleanup are also left as explicit follow-up work.

Brief change log

• Extend the PutKv bucket RPC with optional original-partition context and preserve it during server decoding.
• Dispatch historical writes to the bounded historical executor while keeping normal writes on the existing path.
• Reuse the existing KvTablet merge, WAL, backpressure, segmented-flush, and retry behavior.
• Namespace local historical keys by original partition and primary key, fall back to lake data on a local miss, and retain delete tombstones locally.
• Include the local historical overlay in historical lookups.

Tests

• ./mvnw clean install -DskipTests -DskipITs -pl fluss-protogen,fluss-rpc
• ./mvnw -pl fluss-server -DskipITs -Dtest=HistoricalPkWriteProcessorTest,HistoricalKvKeyEncoderTest,ServerRpcMessageUtilsTest test
• Result: 6 tests passed; Checkstyle reported 0 violations and Spotless passed.

API and Format

• Extends the PutKv RPC bucket message with the optional original_partition_name field.
• Introduces a local historical KV key format: length-prefixed UTF-8 original partition name followed by the physical primary key.
• Does not change the public Java client API or enable client-side historical routing.

Documentation

The change is covered by FIP-28 and issue #3633. User-facing documentation is deferred until the client write path is enabled.

[Copilot]

Pull request overview

This PR extends the tablet-server write path to support primary-key writes targeting the shared historical partition by carrying the original partition name through the PutKv bucket RPC, dispatching historical writes onto the bounded historical executor, and maintaining a local “historical overlay” (partition-scoped key encoding + tombstones) that is consulted during historical lookups before falling back to lake storage.

Changes:

• Extend PbPutKvReqForBucket with optional original_partition_name and preserve it when decoding PutKv requests server-side.
• Route historical PutKv requests through a dedicated historical write manager/processor (bounded executor) and introduce a composite key format for local historical KV overlay state.
• Enhance historical lookup flow to consult local historical overlay first, then lake lookup, plus add targeted unit tests.

Reviewed changes

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

Show a summary per file

File	Description
fluss-rpc/src/main/proto/FlussApi.proto	Adds optional original_partition_name to PutKv bucket RPC.
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java	Decodes PutKv requests into PutKvDataForBucket while preserving original partition context.
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java	Switches PutKv handling to dispatchPutRecordsToKv with historical context.
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Adds dispatch + historical write execution path; passes local KV overlay into historical lookup manager.
fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java	Adds historical-partition handling (KV creation, write rejection for normal path, historical put path).
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalPkWriteProcessor.java	Validates/decodes original partition name and executes historical PK write with lake fallback lookup.
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalPkWriteManager.java	Submits historical write tasks to bounded executor with throttling behavior.
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java	Generalizes bounded executor submission and consults local overlay before lake lookup.
fluss-server/src/main/java/org/apache/fluss/server/kv/NormalKvStateAccessor.java	Introduces a state-access abstraction for normal KV tablets (buffer + RocksDB).
fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java	Extracts shared merge+WAL write logic for normal and historical KV writes.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java	Integrates KvWriteProcessor + state accessor; implements historical tombstone persistence and lookup.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateLookupResult.java	Adds a typed lookup result to distinguish not-found vs deleted vs present.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateAccessor.java	Defines the state accessor interface used by KV write processing.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java	Adjusts KV drop behavior (log message removal).
fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalValueLookup.java	Defines fallback lookup interface for historical writes.
fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalKvStateAccessor.java	Implements partition-scoped key encoding + lake fallback on local miss.
fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalKvKeyEncoder.java	Implements length-prefixed UTF-8 partition name + primary key encoding.
fluss-server/src/main/java/org/apache/fluss/server/entity/PutKvDataForBucket.java	Adds request container type holding records + optional original partition name.
fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java	Tests decoding of historical PutKv request and duplicate bucket rejection.
fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalPkWriteProcessorTest.java	Tests historical insert/update/delete behavior and lake fallback interactions.
fluss-server/src/test/java/org/apache/fluss/server/kv/historical/HistoricalKvKeyEncoderTest.java	Tests encoding properties and invalid-input validation.
fluss-rust/crates/fluss/src/rpc/message/put_kv.rs	Initializes new protobuf field in Rust PutKv request builder.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

Copilot reviewed 29 out of 29 changed files in this pull request and generated 1 comment.

Suppressed comments (2)

fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java:255

• ResolvedPartitionSpec.fromPartitionName() may throw RuntimeException for invalid originalPartitionName values (e.g., wrong number of '$' segments). Right now this is returned as a generic ApiError, rather than a clear InvalidPartitionException that will map to the expected INVALID_PARTITION-style RPC error.

            TableInfo tableInfo = replica.getTableInfo();
            if (originalPartitionName == null) {
                throw new InvalidPartitionException(
                        "Historical lookup request must carry the original partition name.");
            }
            ResolvedPartitionSpec originalPartitionSpec =
                    ResolvedPartitionSpec.fromPartitionName(
                            tableInfo.getPartitionKeys(), originalPartitionName);

fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java:231

• ResolvedPartitionSpec.fromPartitionName() can throw RuntimeException (for example when the number of '$'-separated values doesn't match the table's partition keys). In the historical write path this currently bubbles up as a generic exception type/message, which may be mapped to a non-specific RPC error instead of InvalidPartitionException with a clear message.

This issue also appears on line 247 of the same file.

        TableInfo tableInfo = replica.getTableInfo();
        String originalPartitionName =
                checkNotNull(
                        putData.originalPartitionName(), "originalPartitionName must not be null");
        ResolvedPartitionSpec originalPartitionSpec =
                ResolvedPartitionSpec.fromPartitionName(
                        tableInfo.getPartitionKeys(), originalPartitionName);
        return replica.putHistoricalRecordsToLeader(

[luoyuxia]

@wuchong Could you please help review this pr?

[wuchong]

Because putHistoricalRecordsToLeader holds leaderIsrUpdateLock across kv.putHistoricalAsLeader, and that call performs synchronous fallback lake/file I/O after acquiring kvLock, a slow or stuck cache miss can block leader transitions indefinitely. Please perform the fallback outside these locks and revalidate the leader epoch immediately before applying the write so that an old leader cannot commit.

[wuchong]

Would a lazy-miss, unlock-and-retry flow work here? Keep the normal attempt entirely local so WAL-mode full overwrites and prewrite/RocksDB hits never query the lake. If the historical accessor reaches a genuine local miss, surface an internal NEEDS_LAKE_LOOKUP outcome and let the existing failure path roll back any staged prewrite mutations. Capture the leader epoch, leave both kvLock and leaderIsrUpdateLock, resolve only the missing keys from the lake asynchronously, and memoize both FOUND and ABSENT for this logical request. Then reacquire the locks and retry only if the captured epoch is still current; on retry, consult local state first and use the memoized lake result only while the key remains absent. The admission and ordering token should remain held across the asynchronous gap so later writes for the same historical key cannot overtake it. This preserves the fast paths, avoids lake I/O under the locks, and fences stale old-leader commits.

[wuchong]

The abstraction of KvStateAccessor is excellent as it hides the differences between local and historical access. However, hardcoding LocalKvStateAccessor here breaks this abstraction and makes the code fragile. For instance, a historical partition tablet might mistakenly use this local accessor (e.g., via encodeKey()), leading to errors. I suggest generalizing this to KvStateAccessor and determining whether it is a local or historical accessor in the constructor parameters, thereby avoiding hardcoding within the logic.

Given that we cannot perform lake lookups within org.apache.fluss.server.kv.KvStateAccessor#lookup (as it holds the kvLock), the only aspect that may need abstraction and generalization is encodeKey. Could we consider passing the partition name as a nullable parameter? This would allow us to use a unified KvStateAccessor within KvTablet.

[luoyuxia]

@wuchong Thanks for your review. Comments has been addressed. Can you please help review again?

[wuchong]

This duplicates org.apache.fluss.client.table.getter.PartitionGetter: both resolve partition fields from a RowType, convert their values with PartitionUtils.convertValueOfType, and build a ResolvedPartitionSpec. Could we move or extract PartitionGetter into fluss-common under a neutral package and reuse it here? fluss-server should not depend on fluss-client, but all of PartitionGetter's current dependencies already live in fluss-common. Please keep the recovery-specific non-empty partition-key check at this call site, since the existing getter permits an empty key list. If moving the existing public class is a compatibility concern, we can retain a delegating wrapper in its current client package.

[wuchong]

Why using write lock here? IIUC, this is read-only operation.

[wuchong]

Please add javadoc on this method to explain the difference between encodedPrimaryKey, primaryKey. I think the encodedPrimaryKey` is not accurate now, better to rename it.

[luoyuxia]

Let me merge this since I still have one pull request based on it.