[flink] Support remove_orphan_blobs action and procedure.

[Stephen0421]

Purpose

Add Flink remove_orphan_blobs (action + procedure) for unreferenced primary-key .managed.blob packs, stacked on the landed core cleaner and Spark procedure.

• local reuses LocalManagedBlobOrphanFilesClean
• distributed freezes the validated used-pack set with executeAndCollect before listing candidates, then filters with a captured HashSet. Deletion does not reconnect the sidecar/mark DAG, so a lost used-pack shuffle cannot drop a live pack
• Abort if the used-pack set changes (SKIP_MANAGED_BLOB_GC)
• database.* runs tables one after another so one table’s freeze cannot execute another table’s uncommitted deletion graph
• Bounded used/candidate joins keep ALL_EXCHANGES_BLOCKING
• Docs: docs/docs/flink/procedures/repair.md, procedure index, and blob-storage.md

Tests

• RemoveOrphanBlobsActionITCase / RemoveOrphanBlobsActionITCaseBase
  • testDeleteUnreferencedManagedBlobPack
  • testMissingManagedBlobSidecarSkipsPackGc
  • testDistributedDeleteWithEmptyUsedPackSet
  • testDistributedUnresolvableRelativeCandidateSkipsGc
  • testDistributedMarkReadsSharedMetadataOncePerPass
  • testDistributedCleanupDoesNotHangWithPipelinedShuffle
  • testDistributedDeletionAccountingSurvivesCleanupTaskRestartAfterDelete
  • testDistributedCanonicalAliasCandidatesAreCountedOnce
  • testActionFactoryDryRunParsing
  • testActionFactoryRejectsInvalidDryRun
  • testActionFactoryRejectsNonPositiveParallelism
  • testProcedureRejectsNonPositiveParallelism
  • testFrozenUsedMarkIsNotRecomputedDuringDeletion
  • testUsedPackSetChangeSkipsGc
  • testRemoveDatabaseOrphanBlobs
  • testProcedureDryRun
• RemoveOrphanFilesActionITCaseBase
  • testDistributedCleanupForcesBlockingShuffle
• Flink 1.18 / 1.19 subclasses of RemoveOrphanBlobsActionITCase

[JingsongLi]

Requirement fit: SUPPORTED (triage: GO)\nImplementation: FINDINGS\n\nThe Flink action and procedure have clear end-to-end operational value, but I found one scalability blocker in the distributed path.\n\n[P1] Avoid copying the complete used-pack set into every cleanup task\n\nFlinkManagedBlobOrphanFilesClean.java freezes all live pack identities into a client HashSet, then deleteUnused copies that set and captures it in the candidate filter. Because that filter runs at the requested parallelism, each subtask receives the complete N-entry set, making worker heap usage O(P×N) for N live packs and P cleanup tasks. Live packs scale with retained snapshots and files, so increasing parallelism can make the distributed cleaner OOM on exactly the large tables it is intended to handle.\n\nPlease turn the frozen identities into a lineage-detached bounded source and use a keyed bounded anti-join with candidates, as FlinkOrphanFilesClean already does, so each task retains only its key partition while preserving the no-recomputed-mark safety property. The current 256-orphan scale case has a small live set and does not exercise this amplification.

[Stephen0421]

Requirement fit: SUPPORTED (triage: GO)\nImplementation: FINDINGS\n\nThe Flink action and procedure have clear end-to-end operational value, but I found one scalability blocker in the distributed path.\n\n[P1] Avoid copying the complete used-pack set into every cleanup task\n\nFlinkManagedBlobOrphanFilesClean.java freezes all live pack identities into a client HashSet, then deleteUnused copies that set and captures it in the candidate filter. Because that filter runs at the requested parallelism, each subtask receives the complete N-entry set, making worker heap usage O(P×N) for N live packs and P cleanup tasks. Live packs scale with retained snapshots and files, so increasing parallelism can make the distributed cleaner OOM on exactly the large tables it is intended to handle.\n\nPlease turn the frozen identities into a lineage-detached bounded source and use a keyed bounded anti-join with candidates, as FlinkOrphanFilesClean already does, so each task retains only its key partition while preserving the no-recomputed-mark safety property. The current 256-orphan scale case has a small live set and does not exercise this amplification.

Thanks for the catch.

The deletion filter captured the full frozen HashSet, so each cleanup subtask held all N identities (O(P×N)).

Deletion now re-sources the frozen names with fromCollection (lineage-detached, parallelism 1) and keyed-joins them with candidates, same pattern as FlinkOrphanFilesClean. Each join task only keeps its key partition. The freeze executeAndCollect is unchanged, so the deletion job still cannot recompute sidecars.

An empty used set skips fromCollection (Flink rejects an empty collection) and treats every candidate as unused.