[RFC]feat(pool): add with_pool() to Arrow builders for real-time memory tracking - #1

// Post-hoc: pool is updated AFTER the array is built, not during.
let array = builder.finish();
let buffer = array.into_data().buffers()[0].clone();
buffer.claim(&pool);  // reservation happens here, too late to enforce limits

// Attach a pool to track all buffer growth.
let pool = TrackingMemoryPool::default();
let mut builder = Int32Builder::new()
    .with_pool(&pool)?;

// Infallible append — unchanged API, panics only on OOM.
builder.append_value(42);

// Fallible append — returns Err if pool is exhausted.
builder.try_append_value(42)?;

// finish() transfers the reservation into the array's buffer.
let array = builder.finish();
// pool.used() > 0  ← reservation is alive until the array is dropped
drop(array);
// pool.used() == 0  ← RAII cleanup

let pool = TrackingMemoryPool::default();
let mut builder = StringBuilder::new()
    .with_pool(&pool)?;          // tracks both value and offset buffers

builder.append_value("hello");
builder.try_append_value("world")?;  // fallible
let array = builder.finish();

let pool = TrackingMemoryPool::default();
let values = Int32Builder::new();
let mut builder = ListBuilder::new(values)
    .with_pool(&pool)?;   // propagates pool to child via WithPool trait

builder.values().append_value(1);
builder.append(true);
let array = builder.finish();

use arrow_array::builder::{Int32Builder, WithPool};
use arrow_buffer::{MemoryPool, MemoryReservation, MemoryAllocationError};
use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};

/// A hard-limit pool that rejects allocations once capacity is reached.
#[derive(Debug)]
struct GreedyMemoryPool {
    used: Arc<AtomicUsize>,
    capacity: usize,
}

impl MemoryPool for GreedyMemoryPool {
    fn reserve(&self, size: usize) -> Box<dyn MemoryReservation> {
        // Infallible path: just track, never reject (for backward compat).
        self.used.fetch_add(size, Ordering::Relaxed);
        Box::new(GreedyReservation { size, used: Arc::clone(&self.used) })
    }

    fn try_reserve(&self, size: usize) -> Result<Box<dyn MemoryReservation>, MemoryAllocationError> {
        // Fallible path: enforce the capacity limit.
        let mut current = self.used.load(Ordering::Relaxed);
        loop {
            if current + size > self.capacity {
                return Err(MemoryAllocationError(format!(
                    "GreedyMemoryPool exhausted: {} + {} > {}",
                    current, size, self.capacity
                )));
            }
            match self.used.compare_exchange_weak(
                current, current + size, Ordering::Relaxed, Ordering::Relaxed,
            ) {
                Ok(_) => return Ok(Box::new(GreedyReservation { size, used: Arc::clone(&self.used) })),
                Err(actual) => current = actual,
            }
        }
    }

    fn available(&self) -> isize { self.capacity as isize - self.used() as isize }
    fn used(&self) -> usize { self.used.load(Ordering::Relaxed) }
    fn capacity(&self) -> usize { self.capacity }
}

// In an operator that builds an output batch under memory pressure:
fn build_batch_with_limit(
    pool: &GreedyMemoryPool,
    rows: impl Iterator<Item = i32>,
) -> Result<arrow_array::Int32Array, arrow_schema::ArrowError> {
    let mut builder = Int32Builder::with_capacity(1024)
        .with_pool(pool)?;

    for row in rows {
        match builder.try_append_value(row) {
            Ok(()) => {}
            Err(_oom) => {
                // Pool exhausted — flush what we have and signal to the
                // caller that spilling is needed.
                break;
            }
        }
    }

    Ok(builder.finish())
    // Reservation transfers into the returned array's buffer;
    // it is freed automatically when the batch is dropped.
}