Flink: Add RemoveDanglingDeleteFiles to maintenance API

[sachinnn99]

Summary

Fixes #16138.

The RemoveDanglingDeleteFiles action currently exists only for Spark. Flink users have no equivalent, causing equality delete files to accumulate indefinitely in V2 tables managed by the Flink maintenance pipeline.

This PR adds RemoveDanglingDeleteFiles support to the Flink maintenance API across all three Flink versions (v1.20, v2.0, v2.1).

Approach:

• Follows the ExpireSnapshots pattern (single ProcessFunction operator)
• Mirrors RemoveDanglingDeletesSparkAction's two-method structure:
  • findDanglingDeletes — sequence-number-based detection using an O(partitions) minDataSeqByPartition map (position, equality, and DV deletes)
  • findDanglingDvs — reference-based detection for deletion vectors via a targeted data manifest scan with O(DV candidates) memory
• Avoids collecting all live data file paths in memory, preventing OOM on large tables

New files (identical across all three Flink versions):

• RemoveDanglingDeleteFiles.java — API builder extending MaintenanceTaskBuilder
• RemoveDanglingDeleteFilesProcessor.java — Core operator with dangling detection logic
• TestRemoveDanglingDeleteFiles.java — Tests covering partitioned deletes, equality delete edge cases, unpartitioned tables, and no-op scenarios

Test plan

• TestRemoveDanglingDeleteFiles passes on Flink v1.20, v2.0, v2.1
• All existing maintenance tests pass (no regressions)
• Compilation succeeds across all Flink versions

[Guosmilesmile]

Thanks for the PR! It looks more like a standalone version that doesn’t need Flink . It can be done directly with the Java API. If the table is large, there could be a bottleneck.

[sachinnn99]

Thanks for the review! The single-operator approach follows the same pattern as ExpireSnapshotsProcessor. A distributed approach like DeleteOrphanFiles could be a follow-up for very large tables.

[Guosmilesmile]

In the current implementation of this feature, there are places where DataFiles/DeleteFiles need to be read one by one. It also requires building a table-level Set and constructing partition-level minDataSeqByPartition. In large table scenarios, this will have performance issues and OOM. So I hope the Flink implementation can leverage distributed computing capabilities just like Spark does.