Add undo recovery support for aggregation tables

[platinumhamburg]

This PR introduces the undo recovery mechanism for Flink sink writer to handle failure recovery scenarios with aggregation merge engine tables.

Key components:

• ByteArrayWrapper: Utility class for using byte arrays as map keys
• UndoComputer: Computes undo operations by comparing checkpoint state with current log records, supporting both full row and partial update modes
• UndoRecoveryExecutor: Executes undo operations using UpsertWriter
• UndoRecoveryCoordinator: Coordinates the recovery process across buckets, managing log scanning, undo computation, and execution
• BucketRecoveryContext: Holds per-bucket recovery state

The undo recovery works by:

1. Scanning log records from checkpoint offset to current end offset
2. Computing inverse operations for uncommitted records
3. Executing undo operations to restore table state

For partial update mode, INSERT records require full row deletion (not partial column deletion), which is handled by using a separate delete writer with null target columns.

Purpose

Linked issue: close #2544 2544

Brief change log

Tests

API and Format

Documentation

[wuchong]

Thanks @platinumhamburg , I left some comments.

[wuchong]

Btw, it seems the UNDO recovery for aggregate merge engine table relies on the full changelog of the table. However, table.changelog.image=WAL will break this and lead to in-correct results. I think we need to forbid such case, a validation check can be done during table creation and throw exception for tables with agg merge engine and WAL mode. We can do that in another issue. What do you think @platinumhamburg ?

[platinumhamburg]

@wuchong Thank you for your detailed review. I’ve addressed all the comments above—please take a look when you have a moment.

[platinumhamburg]

Btw, it seems the UNDO recovery for aggregate merge engine table relies on the full changelog of the table. However, table.changelog.image=WAL will break this and lead to in-correct results. I think we need to forbid such case, a validation check can be done during table creation and throw exception for tables with agg merge engine and WAL mode. We can do that in another issue. What do you think @platinumhamburg ?

Yes, this restriction will be addressed in a separate issue [https://github.com//issues/2608].

[wuchong]

@platinumhamburg , I only have one remaining comment. I will merge this PR first. Please open follow-up pull requests to fix this.

[wuchong]

I think we still need to check the zeroOffsetBucketKeysAfterCheckpoint related rows after restoring. Could you add this test back?