[upstream/lance] Add a direct pending-flushed-generations count API for a MemWAL shard

[beinan]

[upstream/lance] — this tracks a change that ideally lands in Lance (the MemWAL owner). Filed here for internal triage; move upstream when ready.

Summary

To decide whether a shard needs compaction (or to report backlog), callers need the count of flushed generations pending in a shard's manifest. Today this requires constructing a ShardManifestStore, calling read_latest(), and reading flushed_generations.len() by hand — a pattern repeated in multiple places. Lance should expose a direct count API so callers don't re-implement manifest plumbing just to get a number.

Motivation

lance-context does exactly this in three separate places, each re-deriving the same manifest read:

let manifest_store = ShardManifestStore::new(
    object_store, &branch_location.path, shard, DEFAULT_MANIFEST_SCAN_BATCH_SIZE,
);
let Some(manifest) = manifest_store.read_latest().await? else { return Ok(0) };
let pending = manifest.flushed_generations.len();

It appears in crates/lance-context-core/src/rollout_store.rs at:

• merge_own_shard_if_ready (:509) — a merge-threshold check
• observe (:930) — a metrics path
• wal_shard_snapshots (:1265) — shard-snapshot enumeration

Each hand-rolls the store construction, the read_latest null-check, and the length read. Lance knows the manifest layout best and should provide the count directly.

Goal

A one-call way to get a shard's pending flushed-generation count (and ideally the set of pending generation ids/paths) without the caller touching ShardManifestStore or ShardManifest internals.

Proposed API

impl Dataset {
    /// Number of flushed MemWAL generations currently pending (not yet merged)
    /// in this shard's manifest. 0 if the shard has no manifest yet.
    pub async fn mem_wal_pending_generations(&self, shard_id: Uuid) -> Result<usize>;

    /// Optional richer form: the pending generations (id + path), for callers
    /// that also need to enumerate them (e.g. to build a read snapshot).
    pub async fn mem_wal_pending_generation_list(
        &self, shard_id: Uuid,
    ) -> Result<Vec<FlushedGeneration>>;
}

An all-shards convenience (mem_wal_pending_generations_by_shard() -> Result<Vec<(Uuid, usize)>>) would also help callers that scan every shard, but the per-shard form is the minimum.

Where to look (Lance side)

• ShardManifestStore::new / read_latest, DEFAULT_MANIFEST_SCAN_BATCH_SIZE.
• ShardManifest::flushed_generations and the FlushedGeneration type.
• Dataset::list_mem_wal_latest_shard_ids (already exists) — pairs naturally with a by-shard count for the all-shards convenience form.

Acceptance criteria

• mem_wal_pending_generations(shard) returns the count without the caller constructing a ShardManifestStore (test: flush N generations, assert count == N; after compaction, assert it drops).
• Returns 0 (not an error) when the shard has no manifest yet.
• The richer list form, if provided, returns the same generations the manifest lists (ids + paths), matching what a read snapshot would use.

Non-goals

• Counting rows inside generations (that's a heavier metadata/scan operation and stays separate).
• Any merge/compaction side effects — these are pure reads.

Downstream follow-up (not part of this issue)

lance-context replaces its three hand-rolled ShardManifestStore::new().read_latest().flushed_generations.len() sites with dataset.mem_wal_pending_generations(shard), and uses the list form to build read snapshots.