[core][flink] Support parallelism snapshot expire

[wzhero1]

Purpose

This PR implements parallel snapshot expiration using Flink distributed execution to improve the performance of large-scale cleanup operations.

Motivation:

• Serial file deletion becomes a performance bottleneck for tables with large amounts of data
• Current implementation cannot leverage Flink's distributed computing capabilities

Architecture:

The expire process is split into two phases:

• Worker phase (parallel flatMap): Deletes data files and changelog files in parallel using RangePartitionedExpireFunction. Tasks are partitioned by snapshot ID range to maximize cache locality (adjacent snapshots share manifest files and tag data file skippers).
• Sink phase (serial): Deletes manifests and snapshot metadata serially using SnapshotExpireSink, because manifestSkippingSet is global mutable state and snapshot deletion must be contiguous.

Key classes:

• ExpireSnapshotsPlanner: Computes expiration plan including snapshot range, four types of tasks, protection set, and snapshot cache
• SnapshotExpireTask: Abstract base with 4 polymorphic subclasses (DeleteDataFilesTask, DeleteChangelogFilesTask, DeleteManifestsTask, DeleteSnapshotTask)
• ExpireContext: Holds runtime dependencies (SnapshotManager, SnapshotDeletion, ChangelogManager) and shared state (taggedSnapshots, snapshotCache, skippingSet)
• ExpireSnapshotsPlan: Contains task lists, protection set, snapshot cache, and range-based task partitioning logic
• DeletionReport: Carries deletion bucket info from workers to sink for empty directory cleanup

Execution modes:

• Local mode: forceStartFlinkJob=false and parallelism <= 1 → executes locally via ExpireSnapshotsProcedure
• Parallel mode: forceStartFlinkJob=true or parallelism > 1 → uses Flink distributed execution

Other changes:

• Moved planner/plan/task/report classes from paimon-core to paimon-flink (per reviewer suggestion)
• Planner pre-collects snapshots in parallel using CompletableFuture and caches them, serialized to both workers and sink to avoid redundant reads
• ExpireSnapshotsImpl (serial mode) reuses ExpireSnapshotsPlanner for planning, keeping serial execution path unchanged
• Added failure recovery test: verifies that if expire job fails midway (data files partially deleted but snapshot metadata intact), a subsequent run completes successfully

Tests

Unit Tests:

• ExpireSnapshotsPlanTest - Tests range-based task partitioning logic
• DeletionReportTest - Tests deletion report serialization

Integration Tests:

• ParallelExpireSnapshotsActionITCase - Extends ExpireSnapshotsTest to validate parallel mode produces same results as serial mode, plus failure recovery test
• ExpireSnapshotsProcedureITCase - Tests both local and parallel modes via procedure

API and Format

New CLI parameter for expire-snapshots action:

Parameter	Type	Default	Description
--parallelism	Integer	env parallelism	Parallelism for parallel expire workers

No storage format changes.

Backward compatible: Default behavior (local mode) remains unchanged.

Documentation

Yes, this introduces a new feature: parallel snapshot expiration.

Documentation should cover:

• New --parallelism parameter for expire-snapshots action
• Either --parallelism > 1 or --force_start_flink_job triggers parallel mode

[yunfengzhou-hub]

Thanks for the PR. Left some comments as below.

[yunfengzhou-hub]

In case this method is overridden, this class needs not implement LocalAction anymore. Still you can keep the executeLocally method for internal use.

[wzhero1]

Prefer to keep LocalAction here. By default (without forceStartFlinkJob), the action runs locally via super.run() → executeLocally(), which is sufficient for normal scenarios and safer. Only when
forceStartFlinkJob is set, we start a Flink job with multi-parallelism for large-scale expiration. This also leaves room for future adjustments.

[yunfengzhou-hub]

Better to configure an empty Flink job, even if the job would complete immediately after submission. This is what "force_start_flink_job" means.

[wzhero1]

Prefer to keep the early return here. Submitting an empty Flink job still involves cluster resource allocation and scheduling overhead for zero useful work. I've improved the log message to make the skip behavior clear to users.

[wzhero1]

Based on actual implementation:

PR Comment work items:

1. Empty plan logging: Adjusted log message to "No snapshots to expire, skipping Flink job submission." (still returns early, not submitting an empty job)
2. Sink operator rename: Renamed sink operator from "SnapshotExpire" to "SnapshotExpireCommit" to distinguish from worker phase
3. Inline executeTask: Removed single-use method, inlined task execution directly into flatMap
4. Progress logging: Added per-task progress log ("Processing expire task {}/{}, {}"), start/end logs with elapsed time
5. Failure recovery test: Added testExpireRecoveryAfterPartialFailure with failure injection via volatile int failAfterTasks
6. Test class rename: ExpireSnapshotsActionITCase → ParallelExpireSnapshotsActionITCase
7. Disabled test comments: Added @disabled annotations with explanations for tests not suitable for parallel mode
8. Polymorphic refactoring: Replaced TaskType enum + ExpireSnapshotsExecutor switch dispatch with 4 polymorphic SnapshotExpireTask subclasses, deleted ExpireSnapshotsExecutor
9. Logic migration: Moved planner/plan/task/report classes from paimon-core to paimon-flink package

Additional changes (not from comments):

• ExpireContext simplification: Removed delegate methods, callers access underlying objects directly (context.snapshotDeletion().xxx()). Made taggedSnapshots and snapshotCache
  final constructor parameters, only skippingSet retains a setter (mutated during manifest deletion)
• snapshotCache in parallel mode: Propagated planner's pre-collected snapshot cache to both workers and sink via serialization, avoiding redundant snapshot file reads
• Trigger condition: Changed to forceStartFlinkJob || parallelism > 1 — either condition enters Flink parallel pipeline