[core][flink][spark] Introduce remove_orphan_blobs to clean orphan blobs.

[Stephen0421]

Purpose

PK managed BLOB packs (.managed.blob) are shared payloads, not first-class data files. Compaction reuses pack bytes without copying them, so snapshot expire only deletes the data file and its .blobref extra file. Unreferenced packs previously accumulated forever.

This change reclaims those packs through a dedicated remove_orphan_blobs procedure (local / Flink / Spark):

• Collect reachability from live ADD data files' .blobref sidecars.
• Join candidates by full pack identity (storageRootId + relativePath), not basename, so same-named packs in different directories are not mixed.
• Collect used packs twice and abort the run (delete nothing) if the snapshot topology or used-pack set changed between those collections. That shrinks the compaction-reuse race; it is not a commit lease.
• Delete unused .managed.blob files older than older_than (1 day by default).
• If a listed sidecar on a still-existing data file is missing, corrupt, or unsupported, skip all pack deletes for that run. An empty valid sidecar is not missing: it means the file references no pack.
• Unmerged ADD entries whose data files are already gone after expire are ignored, so they do not abort pack GC.

remove_orphan_files never deletes .managed.blob packs.

This is a conservative, best-effort first version. There is still no commit lease. Compaction can reuse a pack without refreshing mtime, so older_than does not fence a compact that commits after the second collection and before delete. Keep a non-zero older_than. Do not treat this as strictly proven concurrent-safe delete semantics.

Out of scope: expire-time incremental pack delete, FileStoreCommitImpl.abort pack cleanup, and Append / Data Evolution .blob files (already handled as ordinary data files).

Tests

• ManagedBlobReachabilityCollectorTest: empty extras / empty sidecar, referenced packs, missing sidecar (live vs already gone), corrupt / unsupported version, unsafe merge.
• ManagedBlobOrphanFilesCleanTest: delete unreferenced pack, keep referenced pack, join by full pack path, empty sidecar does not block others, missing / corrupt / unsupported sidecar skips all packs, unreferenced pack after update+expire, abort when used set changes between the two collections, the deterministic compact+expire interleaving that shows a stale snapshot list neither skips nor retains a reused live pack, and a stale compaction that cannot commit after the final mark.
• LocalOrphanFilesCleanTest: remove_orphan_files keeps managed packs (fail-safe).
• Flink RemoveOrphanBlobsActionITCaseBase: local and distributed — delete unreferenced pack; missing sidecar skips pack GC and does not delete other orphan files.
• Spark RemoveOrphanBlobsProcedureTest: same two cases × local / distributed.

[JingsongLi]

I think we should introduce new procedure to clean blob orphan files.

[Stephen0421]

I think we should introduce new procedure to clean blob orphan files.

ok, I will extract the function to a new procedure.