examples/flight: add OomGuard allocator-level OOM circuit breaker to sql_server

[avantgardnerio]

Which issue does this PR close?

Not closing an issue directly. This is a follow-up to the
MemoryPool accounting discussion (#22723, #22758) — demonstrates an alternative approach.
Instead of catching MemoryPool divergences from allocations, this uses the flight_sql_server example
to demonstrate how downstream applications can install a custom allocator to abort queries rather
than OOMing when MemoryTracker under-reports.

Rationale for this change

MemoryPool is voluntary. Operators that don't call try_grow —
in-flight batches, arrow-row decode buffers, Arrow kernel scratch
space, allocator slab overhead — allocate freely past the declared
budget. On a process with a hard memory ceiling (k8s pod cgroup,
ulimit), the result is an OS-level kill that takes down every
concurrent query, not just the offender.

This PR adds OomGuard, a GlobalAlloc wrapper. Implementation details:

1. Every Rust allocation is debited from a global BALANCE atomic.
2. A dedicated OS thread polls jemalloc's stats.resident every 10ms
   and syncs BALANCE = threshold - resident, catching slab pages
   that the kernel sees but GlobalAlloc doesn't.
3. When BALANCE goes negative on a stamped query worker, the
   wrapper returns NULL from alloc.
4. An alloc_error_hook originates panic_any(OomGuardPanic) from
   a safe-Rust frame outside the allocator.
5. The panic unwinds through df.collect(), gets caught in the
   example, and is converted to Status::resource_exhausted. The
   query dies; the server survives.

A/B demo on the same 1GiB cgroup:

• OOM_GUARD=off: client sees broken pipe, server SIGKILLed.
• OOM_GUARD on (default): client sees ResourceExhausted,
  follow-up SELECT 1 against the same server returns immediately.

What changes are included in this PR?

Everything is scoped to datafusion-examples/examples/flight/:

• oom_guard.rs (new) — the GlobalAlloc wrapper, per-thread
  drift settling, jemalloc-stats-based poll,
  cgroup-v2/v1 memory-limit discovery.
• sql_server.rs — wraps the global allocator with
  OomGuard<Jemalloc>, arms via OOM_GUARD env var with
  configurable headroom (OOM_GUARD_HEADROOM, default 5%), wraps
  df.collect() in catch_unwind to convert OomGuardPanic to
  a Status::resource_exhausted response.
• main.rs — installs the alloc_error_hook, installs a global
  panic hook that logs OomGuardPanic distinctly
• .cargo/config.toml (new) — sets RUSTC_BOOTSTRAP=1 so
  #![feature(alloc_error_hook)] (tracking Tracking issue for the OOM hook (alloc_error_hook) rust-lang/rust#51245)
  builds on stable.
• README.md (new) — A/B walkthrough.

Are these changes tested?

Manually via the A/B in the README. No automated tests
added: the example's behavior depends on a cgroup limit being in
effect, which doesn't fit the existing test harness.

Are there any user-facing changes?

Only to the flight example. No core / library API surface changed.

Caveats

• alloc_error_hook is currently unstable. The workspace
  .cargo/config.toml sets RUSTC_BOOTSTRAP=1 to compile it on a
  stable toolchain.
• The poll currently reads jemalloc-specific stats. Making it
  allocator-agnostic by reading cgroup's memory.current instead is
  a straightforward follow-up if there's interest.

[alamb]

Thanks @avantgardnerio -- is there anything DataFusion specific about this example? It looks like it doesn't use a memory pool at all. I thought your solution hooked the allocator into the memory pool infrastructure -- this seems like it is even deeper and hooks the allocator itself

If that is the case, perhaps it would be better as a standalone crate?

[alamb]

Rather than extending an existing example, what would you think about making this its own example ? That might keep the basic arrow-flight example slimpler

[avantgardnerio]

it doesn't use a memory pool at all

It does not. And at this point I have an honest question: what is the MemoryPool for?

We've established it doesn't prevent OOMs, and that it's not easy to keep accurate.

So cost=high, benefit=low - Is there a reason for tracking in-flight batches I'm not seeing?

[avantgardnerio]

would be better as a standalone crate?

acknowledged