[lake] Fix lake encoder causing prefix lookup to miss data when bucket key is a subset of primary key

[luoyuxia]

Purpose

Linked issue: close #1412

Fix lake encoder causing prefix lookup to miss data when bucket key is a subset of primary key.

When a primary key table has a non-default bucket key (bucket key is a subset of primary key), the previous implementation used lake's encoder (e.g., Paimon) to encode the primary key. However, lake encoders like Paimon store strings longer than 7 characters in a variable-length area, which breaks prefix lookup because the encoded bucket key bytes are no longer a prefix of the encoded primary key bytes.

Brief change log

This PR:

1. Introduces kv format version attribute for tables to distinguish new tables from legacy ones.
2. Refactors KeyEncoder with separate factory methods:
   • ofPrimaryKeyEncoder: For encoding primary keys. New tables (kv format version 2) with non-default bucket keys use CompactedKeyEncoder to ensure bucket key bytes are always a prefix of primary key bytes, enabling prefix lookup.
   • ofBucketKeyEncoder: For encoding bucket keys used in bucket calculation. Always uses lake's encoder when lake format is present.
3. Maintains backward compatibility: Legacy tables (without kv format version) continue using the original lake encoder to ensure data compatibility.
4. Rejects old clients for new format tables: When old client (API version 0) tries to access new format tables (kv format version 2) with non-default bucket keys, the server returns UNSUPPORTED_VERSION error to prevent data corruption.
5. Performance optimization: When bucket key equals primary key (isDefaultBucketKey = true), prefix lookup is not needed, so we can use lake's encoder for both. This allows reusing the encoded primary key bytes for bucket calculation, saving one encoding operation.

Tests

• Added testLegacyPKTableUsePaimonEncoder in FlussLakeTableITCase to verify legacy tables continue using Paimon encoder.
• Added testKvFormatV2TableRejectV0Client in ReplicaManagerTest to verify old clients are rejected for new format tables.
• Added testUnionReadNotDefaultBucketKeyTable in FlinkUnionReadPrimaryKeyTableITCase to verify union read works correctly with new encoding.
• Updated existing prefix lookup tests to use longer strings (length > 7) to properly catch encoding issues.

Also verify the back compatibility:

• use client v0.8 to write new table, expected UnsupportedVersionException is thrown
• use main branch client to write old table, still write will old encode strategy. Both main branch client and v0.8 client can get the key

API and Format

Added new config option TABLE_KV_FORMAT_VERSION to track table's kv encoding format version.

Documentation

[Copilot]

Pull request overview

This PR implements backward compatibility support for key encoding in data lake tables, preventing old clients from accessing new tables that use incompatible encoding strategies.

Changes:

• Introduces API version validation for KV operations (PUT_KV, LOOKUP, PREFIX_LOOKUP) to reject old clients (version 0) when accessing new format tables
• Refactors KeyEncoder with separate factory methods for primary key encoding and bucket key encoding to support different encoding strategies
• Adds kv format version 2 configuration that automatically applies to new primary key tables

Reviewed changes

Copilot reviewed 23 out of 23 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java	Bumps API versions for PUT_KV, LOOKUP, and PREFIX_LOOKUP from 0 to 1
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Adds client version validation for KV operations and updates method signatures
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java	Passes API version from session to ReplicaManager operations
fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java	Adds tableInfo field to enable version validation
fluss-common/src/main/java/org/apache/fluss/row/encode/KeyEncoder.java	Splits key encoding into primary key and bucket key factory methods
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Adds TABLE_KV_FORMAT_VERSION configuration option
fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java	Adds getKvFormatVersion() method
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java	Automatically sets kv format version 2 for new primary key tables
fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java	Updates to use new primary key encoder factory method
fluss-client/src/main/java/org/apache/fluss/client/table/writer/*	Updates UpsertWriter and AppendWriter to use new encoder factory methods
fluss-client/src/main/java/org/apache/fluss/client/lookup/*	Updates PrimaryKeyLookuper and PrefixKeyLookuper to use new encoder factory methods
fluss-server/src/test/java/.../ReplicaManagerTest.java	Adds test for old client version rejection and updates existing tests
fluss-client/src/test/java/.../FlussLakeTableITCase.java	Adds backward compatibility test for legacy tables
fluss-lake/.../FlinkUnionReadPrimaryKeyTableITCase.java	Adds test for non-default bucket key table union read

---

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

[luoyuxia]

@wuchong Could you please help review when you got some time?