[paimon] Fix union read paimon dv table issue

[luoyuxia]

Purpose

Linked issue: close #2121

The pr fixes the union read functionality for Paimon tables with deletion vectors (DV) enabled.
The fix enables Flink to correctly perform union reads that combine data from both:

• Paimon lake storage (compacted historical data via readable snapshot)
• Fluss log store (incremental changes after the readable snapshot)

Brief change log

The basic idea is:

1. Determine when a readable snapshot exists by checking compaction status across all buckets
2. Calculate the correct log offsets for each bucket to start reading incremental changes

The full change log is as follows:

• Add DvTableReadableSnapshotRetriever to determine readable snapshots for DV tables by analyzing
  Paimon snapshot compaction history and L0 file status across all buckets/partitions
• Add PaimonDvTableUtils with utilities for working with Paimon DV tables, including partition
  name conversion and bucket analysis
• Add LakeCommitResult to represent lake commit results with readable snapshot information so that the fluss cluster can record it
• Extend Admin API with getReadableLakeSnapshot() method to retrieve readable snapshots
• Update FlussTableLakeSnapshotCommitter to commit readable snapshot information along with
  tiered snapshots
• Update RPC protocol and message handling to support readable snapshot operations
• Update coordinator and ZooKeeper storage to handle readable snapshot metadata

Tests

• Add FlinkUnionReadDvTableITCase with comprehensive integration tests:

  • testUnionReadDvTable(): Tests union read on non-partitioned DV table with 3 buckets
  • testUnionReadDvPartitionedTable(): Tests union read on partitioned DV table with 2
    partitions × 2 buckets

• Add DvTableReadableSnapshotRetrieverTest with unit tests covering:

  • Readable snapshot calculation for non-partitioned tables
  • Readable snapshot calculation for partitioned tables
  • Edge cases with partial compaction and L0 file handling

• Add PaimonDvTableUtilsTest for utility function testing

API and Format

• New API: Admin.getReadableLakeSnapshot(TablePath) - Returns a readable snapshot if one
  exists, throws LakeTableSnapshotNotExistException otherwise

• New API: Admin.getLakeSnapshotById(TablePath, long) - Retrieves a specific historical
  snapshot by ID

• RPC Changes: Added new RPC methods for readable snapshot operations in FlussApi.proto

Documentation

[Copilot]

Pull request overview

This PR fixes union read issues for Paimon deletion-vectors enabled tables by implementing a mechanism to track multiple lake snapshots and distinguish between tiered and readable snapshots. The changes enable proper handling of L0 files in Paimon DV tables, preventing data loss during union reads.

Key changes:

• Introduces readable snapshot concept separate from tiered snapshot for Paimon DV tables
• Implements snapshot retention policy with configurable earliest snapshot ID to keep
• Adds logic to compute readable offsets based on L0 file compaction status

Reviewed changes

Copilot reviewed 46 out of 46 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
FlussApi.proto	Added optional fields for snapshot_id, readable flag, ignore_previous_table_offsets, and earliest_snapshot_id_to_keep
LakeCommitResult.java	New class encapsulating commit results with readable snapshot information
CommitLakeTableSnapshotsData.java	Refactored from CommitLakeTableSnapshotData to support per-table snapshot metadata
DvTableReadableSnapshotRetriever.java	New utility to compute readable snapshots and offsets for DV tables
PaimonDvTableUtils.java	New utility for tracking L0 file lifecycle in Paimon snapshots
LakeTableHelper.java	Enhanced with snapshot retention logic based on earliestSnapshotIDToKeep
PaimonLakeCommitter.java	Updated to retrieve and return readable snapshot information for DV tables
Admin.java/FlussAdmin.java	Added getLakeSnapshot(snapshotId) and getReadableLakeSnapshot() methods
Test files	Comprehensive tests for DV table readable snapshot retrieval and retention

---

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

[luoyuxia]

@wuchong The pr to fix union read paimon dv table is ready. Please help review when you got time