feat(table): expire snapshots and add sys.expire_snapshots

[zhuxiangyi]

Purpose

Linked issue: #964 (part 1 of 4)

Tables written through paimon-rust (Rust API, pypaimon native, C FFI) never expire snapshots, so
their snapshot, manifest, and data files only ever grow. This PR adds snapshot expiration to the
core crate and exposes it as CALL sys.expire_snapshots, following Java's ExpireSnapshotsImpl
and the Flink/Spark expire_snapshots procedure.

Brief change log

• Table::new_expire_snapshots() / ExpireSnapshots (table/expire_snapshots.rs):

  • Chooses snapshots to expire with Java's rules:

    • never more than snapshot.num-retained.max snapshots are kept, and never fewer than
      snapshot.num-retained.min;
    • a snapshot expires only once the snapshot after it is older than the cut-off
      (snapshot.time-retained, or an explicit older_than);
    • snapshots that a consumer (consumer/consumer-*) still reads from are kept;
    • one run expires at most snapshot.expire.limit snapshots.

    Each rule can be overridden per call.

  • Expiring [earliest, end) deletes, in Java's order:

    1. data files deleted by the delta manifests of (earliest, end] that the closest earlier
       tag does not still read;
    2. changelog files of [earliest, end);
    3. manifest lists, manifests and their extra files, index manifests, index files, statistics
       files, and row-id reassign plans that neither end nor a tag in the range references;
    4. the snapshot files, and finally the EARLIEST hint.
• table/snapshot_deletion.rs (Java FileDeletionBase / SnapshotDeletion) holds the planning
  and deletion. Every read failure makes the plan delete less, never more:
  • an unreadable delta manifest skips that snapshot's data files;
  • a tag that cannot be read skips the files it might protect;
  • a skipping set that cannot be built skips manifest deletion;
  • if the end snapshot disappears concurrently, the run stops before deleting any manifest.
• Unlike Java, the check "a file the same delta re-adds (a level upgrade) is not deleted" does not
  depend on the DELETE entry coming before the ADD entry.
• New CoreOptions: snapshot.num-retained.min / .max, snapshot.time-retained,
  snapshot.expire.limit, with Java's defaults. Invalid values fail instead of silently falling
  back, since this path deletes files.
• SnapshotManager::try_get_snapshot and earliest_hint_exists.
• DataFusion CALL sys.expire_snapshots(table, retain_max, retain_min, older_than, max_deletes, options) returns deleted_snapshots_count, like Java. older_than accepts epoch milliseconds
  or a yyyy-MM-dd HH:mm:ss[.SSS] timestamp in the local time zone (Java's format). options
  are dynamic table options for the call.
• Uses table.snapshot_manager(), so REST-catalog tables resolve their latest snapshot through the
  catalog, as in Java.

Scope notes:

• Like Java's procedure, which builds its ExpireConfig without changelog options, changelog
  files expire with their snapshots. changelog.num-retained.* / changelog.time-retained
  (changelog decoupling) need a changelog manager and are left for later.
• snapshot.clean-empty-directories is not implemented yet.

Tests

• table::expire_snapshots::tests (23 tests). The central check is
  assert_files_match_references: after a run, the data, manifest, and index files on disk must
  be exactly the files some remaining snapshot or tag references. That includes changelog files
  and index files kept in bucket directories. Live data files come from the scan planner, not
  from the deletion code. The scenarios:
  • an empty table;
  • an append-only table, where every data file is kept;
  • overwrite, where the replaced files are deleted;
  • a partitioned overwrite, which checks partition path resolution;
  • a tag keeps its data files and manifests and stays readable through scan.tag-name;
  • a consumer keeps the snapshot it reads, which stays readable through scan.snapshot-id;
  • each retention rule on its own: num-retained.max, older_than (successor rule),
    max_deletes, retain_min;
  • invalid retention arguments and options are rejected;
  • a level upgrade (DELETE and ADD of the same file in one delta) keeps the file;
  • rewritten files are deleted;
  • a dropped global index's files are deleted while an index still in use is kept;
  • a missing end snapshot keeps all manifests;
  • the EARLIEST hint is written even when nothing expires;
  • previous_tag and find_skipping_tags;
  • deletion-vector files follow their snapshots, with index-file-in-data-file-dir both off
    and on;
  • changelog files and dynamic-bucket hash index files of a primary-key table
    (changelog-producer = input);
  • external data files (external_path) are deleted at their own path;
  • when a delta, a tag, or the skipping set cannot be read, less is deleted, never more;
  • with several tags, each snapshot is protected by the closest earlier one;
  • a snapshot missing from the range;
  • the slowest of several consumers limits the run.
• The tests catch the key mistakes. Each of these changes makes a test fail: disabling tag
  protection or upgrade protection, skipping index or changelog deletion, ignoring
  external_path, and treating an unreadable tag as "no tag".
• CoreOptions defaults, overrides, and invalid values.
• DataFusion tests/procedures.rs:
  • an end-to-end CALL sys.expire_snapshots test covering retain_max, older_than,
    max_deletes, and options, checked against $snapshots and the table data;
  • argument validation: an unknown argument, a non-integer, retain_max below retain_min, and
    a bad timestamp;
  • older_than as a local timestamp string.
• A unit test for older_than parsing.
• cargo test -p paimon --all-targets --features fulltext,vortex and cargo clippy --all-targets --workspace --features fulltext,vortex -- -D warnings pass.

API and Format

• New public API: Table::new_expire_snapshots() / ExpireSnapshots; new CoreOptions
  accessors; new sys.expire_snapshots procedure.
• No storage format change.

Documentation

docs/src/sql.md documents expire_snapshots: its arguments, the table options behind them, and
what the procedure keeps.