[kv] Supports compacted row as change log

[luoyuxia]

Purpose

Linked issue: close #2107

Brief change log

CP from #1605 , only keep the code of supporting compacted row as change log

Tests

API and Format

Documentation

[Copilot]

Pull request overview

This pull request adds support for using compacted row format as a change log in primary key tables. The change enables a new COMPACTED log format option alongside existing ARROW and INDEXED formats, providing a space-optimized row-oriented format that trades CPU for reduced storage.

Key changes:

• Introduces COMPACTED as a new log format option for primary key tables with COMPACTED kv format
• Implements compacted row encoding/decoding infrastructure with variable-length integer encoding (VLQ)
• Refactors existing indexed log builder to extract common functionality into a shared abstract base class

Reviewed changes

Copilot reviewed 15 out of 15 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java	Updates validation logic to allow both ARROW and COMPACTED log formats when kv format is COMPACTED
fluss-server/src/main/java/org/apache/fluss/server/kv/wal/CompactedWalBuilder.java	New WAL builder implementation for COMPACTED log format with row encoding support
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java	Integrates CompactedWalBuilder into KV tablet for handling COMPACTED log format
fluss-common/src/main/java/org/apache/fluss/record/CompactedLogRecord.java	New log record class implementing space-optimized encoding for CompactedRow format
fluss-common/src/main/java/org/apache/fluss/record/MemoryLogRecordsCompactedBuilder.java	Builder for creating memory log records in COMPACTED format
fluss-common/src/main/java/org/apache/fluss/record/AbstractRowMemoryLogRecordsBuilder.java	New abstract base class extracting common logic from row-based log record builders
fluss-common/src/main/java/org/apache/fluss/record/MemoryLogRecordsIndexedBuilder.java	Refactored to extend AbstractRowMemoryLogRecordsBuilder, reducing code duplication
fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java	Adds support for creating read contexts for COMPACTED format with projection validation
fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java	Implements iterator for reading COMPACTED log records from batches
fluss-common/src/main/java/org/apache/fluss/metadata/LogFormat.java	Adds COMPACTED enum value and updates documentation
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Updates configuration documentation to include COMPACTED format
fluss-common/src/test/java/org/apache/fluss/record/MemoryLogRecordsCompactedBuilderTest.java	Comprehensive unit tests for CompactedBuilder including append, abort, and offset semantics
fluss-common/src/test/java/org/apache/fluss/record/CompactedLogRecordTest.java	Unit tests for CompactedLogRecord serialization/deserialization
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java	Integration tests for COMPACTED log format including upsert, delete, and projection validation
fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java	Updates test assertion to match new validation error message

---

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

[luoyuxia]

@polyzos Hi, I append a small commit to imrove some code. PTAL

[polyzos]

@luoyuxia Looks good to me 👍