[spark] Support remove_orphan_blobs procedure.

[Stephen0421]

Purpose

Stacked on #9609. Adds Spark CALL sys.remove_orphan_blobs for unreferenced primary-key .managed.blob packs.

• local reuses LocalManagedBlobOrphanFilesClean.executeDatabase
• distributed runs a Spark DAG on the same used-pack / sidecar scan as Core (emitUsedPacks, pack identity, SKIP_MANAGED_BLOB_GC)
• Abort if topology or the used-pack set changes; persist mark caches and keep abort keys in the deletion DAG so Spark cache recomputation cannot drop a live pack
• database.* is one Spark job per table (serial + unpersist), not a union of every table, so abort gates stay per-table
• parallelism is validated before mode dispatch; docs describe per-mode concurrency (distributed stays within the per-table value, local may exceed it when several tables run at once)

Tests

• local / distributed: delete unreferenced packs, dry-run, skip GC when sidecar is missing
• database.* for both modes
• reject non-positive parallelism
• distributed abort / cache-recompute / used-set freeze / sidecar identity

[JingsongLi]

Requirement fit: SUPPORTED. Implementation: FINDINGS.

Reviewed 20b75be51fc9. A Spark procedure for reclaiming unreferenced managed BLOB packs has clear operator value. The distributed mark must, however, preserve already validated references through Spark cache and shuffle recomputation before this can safely delete files.

Validation: traced the exact-head Spark DAG and unchanged core collector/local cleaner, and independently reproduced the mark-loss behavior in a real Spark 3.5.8 cache/shuffle model. The model emits candidate names and models the compaction transition; it is not a full Paimon table integration test. The core test testStaleSnapshotListMissesReusedPackAfterCompactBeforeDeleted establishes the reused-pack/expired-sidecar transition. Current CI is green but its cache tests do not cover partial shuffle loss.

[JingsongLi]

[P1] Freeze the validated used-pack mark before the deletion join

After both initial marks protect pack P and frozenAbort is false, compaction can reuse P in a replacement data file and expiration can remove the old data file plus its .blobref. If Spark then loses the mark caches/downstream shuffle outputs while retaining the old sidecarWorkItems shuffle outputs, both marks can reread only those old sidecars. ManagedBlobReachabilityCollector.fromSidecar(dataFile, sidecar) returns safe-empty for the removed data/sidecar pair, so both recomputed sets lose P without emitting SKIP; they still agree, and the topology comparison is the earlier driver-side false value. This anti-join then admits the still-live P for deletion. MEMORY_AND_DISK does not freeze that filesystem-reading lineage. Please durably materialize and cut the validated mark lineage, or abort/restart validation when it must be recomputed, and add a regression preserving stale upstream sidecar shuffle outputs while losing the final marks.

[Stephen0421]

Thanks for the catch.

persist(MEMORY_AND_DISK) does not cut sidecar lineage. After a cache miss both marks can recompute from stale sidecarWorkItems shuffle, lose a reused pack via fromSidecar → Result.empty(), still agree, and pass the abort gates. The deletion join now collects the validated used names onto a driver-backed Dataset and anti-joins that, so a later recompute cannot reread sidecars or silently drop P.

frozenAbort now returns an empty deletion Dataset on the driver. abortKeys is gone because it only helped when the mark could still be recomputed.

Regression: frozen used mark survives stale sidecar recompute unpersists marks, clears shuffle, and asserts collectUsedPacksDf is not called a third time. frozen used mark still deletes after sidecar reads start failing asserts no sidecar reread after the freeze, while still deleting true orphans.

Driver memory now scales with the used-pack set; an OOM fails the procedure instead of deleting live packs. I added a comment not to join the persisted used2Packs Dataset again, and a log of the frozen name count. There is no hard cap and no "use local mode" hint — local holds the same set on the calling JVM. No forced broadcast(); a large LocalRelation can still shuffle-join without putting sidecar reads back.