[server] Support primary-key writes to historical partitions

[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(