hawdb-executor 0.15.0

// Copyright 2026 Nowledge
// Licensed under the Apache License, Version 2.0.
//!
//! Public transform and optimized-producer contracts with real owning sources.
//! Allocation observations cover execution on this test thread, after fixtures
//! and prebuilt plans have been constructed.

use hawdb_core::{
    Catalog, HawDBError, PropertyType, Result, RuntimeCancellationToken, RuntimeTaskContext,
    TableKind, Value,
};
use hawdb_executor::batch::{execute_binding_batches, BatchReadContext};
use hawdb_executor::binding::{binding_memory_bytes, Binding};
use hawdb_executor::external::seed::BatchExternalRead;
use hawdb_executor::external::{
    TextSeedExecutionOutput, TextSeedExecutionRequest, VectorSeedExecutionOutput,
    VectorSeedExecutionRequest,
};
use hawdb_executor::observer::QueryExecutionObserver;
use hawdb_executor::pipeline::{
    BatchControl, BatchExecutionContext, BindingBatch, BindingBatchSource,
};
use hawdb_executor::transform::stream_projection_batches;
use hawdb_executor::{ExecutionLimit, ExecutionMemoryConfig, QueryMemoryLedger};
use hawdb_plan_cypher::{
    NodeProjectionAccess, PhysicalPlan, Predicate, Projection, ProjectionExpression,
    VectorExecutionResourceProfile,
};
use hawdb_storage::store::GraphStore;
use std::alloc::{GlobalAlloc, Layout, System};
use std::cell::Cell;
use std::collections::{BTreeMap, BTreeSet};
use std::num::NonZeroUsize;

thread_local! {
    static WATCH_SIZE: Cell<usize> = const { Cell::new(0) };
    static WATCH_COUNT: Cell<usize> = const { Cell::new(0) };
}

#[test]
fn point_work_accounting_does_not_admit_extra_payload() {
    use hawdb_executor::store::admit_graph_read;
    use hawdb_storage::read_view::{
        AdmittedNodeRead, AdmittedRelationshipRead, ControlledGraphReadAllocator,
    };
    const OVERHEAD: usize = 4096;
    for persisted in [false, true] {
        let directory = persisted.then(StoredPropertyDirectory::new);
        let mut catalog = Catalog::default();
        let mut store = directory
            .as_ref()
            .map_or_else(GraphStore::default, |directory| {
                GraphStore::open(&directory.0, &mut catalog).unwrap()
            });
        let node = store
            .create_node(&mut catalog, "Item", BTreeMap::new())
            .unwrap();
        let other = store
            .create_node(&mut catalog, "Other", BTreeMap::new())
            .unwrap();
        let relationship = store
            .create_relationship(&mut catalog, node, other, "LINK", BTreeMap::new())
            .unwrap();
        if let Some(directory) = &directory {
            store.checkpoint(&catalog).unwrap();
            drop(store);
            catalog = Catalog::default();
            store = GraphStore::open_with_durability_and_replay_config(
                &directory.0,
                &mut catalog,
                Default::default(),
                hawdb_storage::store::WalReplayConfig {
                    residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
                    ..Default::default()
                },
            )
            .unwrap();
        }
        for node_read in [true, false] {
            let bytes = if node_read {
                hawdb_core::ids::node_allocation_bytes(&store.node_owned(node).unwrap().unwrap())
            } else {
                hawdb_core::ids::relationship_allocation_bytes(
                    &store.relationship_owned(relationship).unwrap().unwrap(),
                )
            };
            let limit = nz(bytes + OVERHEAD);
            let ledger = QueryMemoryLedger::new(limit);
            let account = ledger.account(
                hawdb_executor::QueryMemoryClass::ExternalRead,
                "point source",
                limit,
            );
            let calls = Cell::new(0);
            let task = RuntimeTaskContext::with_timeout(std::time::Duration::from_secs(30));
            let mut allocate = |bytes| {
                assert!(bytes > 0, "work must not call payload admission");
                calls.set(calls.get() + 1);
                admit_graph_read(&account, Some(&task), bytes + OVERHEAD).map(Some)
            };
            let mut policy = ControlledGraphReadAllocator::new(&mut allocate, Some(&task));
            for expected_calls in 1..=16 {
                if node_read {
                    let read = store.node_with_allocation(node, None, &mut policy).unwrap();
                    assert!(matches!(read, AdmittedNodeRead::Node(_)));
                    assert_eq!(ledger.snapshot().used_bytes, limit.get());
                    drop(read);
                } else {
                    let read = store
                        .relationship_with_allocation(relationship, &mut policy)
                        .unwrap();
                    assert!(matches!(read, AdmittedRelationshipRead::Relationship(_)));
                    assert_eq!(ledger.snapshot().used_bytes, limit.get());
                    drop(read);
                }
                assert_eq!(calls.get(), expected_calls);
                assert_eq!(ledger.snapshot().used_bytes, 0);
            }
            let other_label = catalog.label_id("Other").unwrap();
            assert!(matches!(
                store
                    .node_with_allocation(node, Some(&[other_label]), &mut policy)
                    .unwrap(),
                AdmittedNodeRead::Missing
            ));
            assert!(matches!(
                store
                    .node_with_allocation(hawdb_storage::NodeId(u64::MAX), None, &mut policy)
                    .unwrap(),
                AdmittedNodeRead::Missing
            ));
            assert!(matches!(
                store
                    .relationship_with_allocation(hawdb_storage::RelId(u64::MAX), &mut policy)
                    .unwrap(),
                AdmittedRelationshipRead::Missing
            ));
            assert_eq!(calls.get(), 16, "absent reads must not admit payload");
            assert_eq!(ledger.snapshot().peak_bytes, limit.get());
            assert_eq!(task.work_units(), 0);
        }
    }
}

struct ProjectionAllocator;

fn observe(layout: Layout) {
    let _ = WATCH_BYTES.try_with(|total| {
        if WATCH_SIZE.try_with(|size| size.get() != 0).unwrap_or(false) {
            total.set(total.get().saturating_add(layout.size()));
        }
    });
    let _ = WATCH_SIZE.try_with(|size| {
        if size.get() != 0 && (size.get() == usize::MAX || size.get() == layout.size()) {
            let _ = WATCH_COUNT.try_with(|count| count.set(count.get().saturating_add(1)));
        }
    });
}

// This isolated integration binary forwards every allocation unchanged.
// Thread-local observation is enabled only around the actual transform callback.
unsafe impl GlobalAlloc for ProjectionAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        observe(layout);
        unsafe { System.alloc(layout) }
    }
    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
        observe(layout);
        unsafe { System.alloc_zeroed(layout) }
    }
    unsafe fn realloc(&self, pointer: *mut u8, old: Layout, size: usize) -> *mut u8 {
        observe(Layout::from_size_align(size, old.align()).unwrap());
        unsafe { System.realloc(pointer, old, size) }
    }
    unsafe fn dealloc(&self, pointer: *mut u8, layout: Layout) {
        unsafe { System.dealloc(pointer, layout) }
    }
}

#[global_allocator]
static ALLOCATOR: ProjectionAllocator = ProjectionAllocator;

thread_local! { static WATCH_BYTES: Cell<usize> = const { Cell::new(0) }; }

struct AllocationWindow;

impl AllocationWindow {
    fn start(size: usize) -> Self {
        WATCH_COUNT.with(|count| count.set(0));
        WATCH_BYTES.with(|total| total.set(0));
        WATCH_SIZE.with(|watch| {
            assert_eq!(watch.get(), 0, "nested allocation observation");
            watch.set(size);
        });
        Self
    }
    fn allocations(&self) -> usize {
        WATCH_COUNT.with(Cell::get)
    }
}

impl Drop for AllocationWindow {
    fn drop(&mut self) {
        let _ = WATCH_SIZE.try_with(|watch| watch.set(0));
    }
}

struct TextSource {
    external_id: String,
    calls: Cell<usize>,
}

impl BatchExternalRead for TextSource {
    fn execute_vector_seed(
        &self,
        _: VectorSeedExecutionRequest<'_>,
    ) -> Result<VectorSeedExecutionOutput> {
        panic!("unexpected vector source")
    }
    fn execute_text_seed(
        &self,
        request: TextSeedExecutionRequest<'_>,
    ) -> Result<TextSeedExecutionOutput> {
        self.calls.set(self.calls.get() + 1);
        let mut output =
            TextSeedExecutionOutput::new(request.result_account, request.resources.result)?;
        for _ in 0..request.resources.result.max_rows {
            output.push("document:a", Some(&self.external_id), 1.0)?;
        }
        Ok(output)
    }
}

struct RuntimeSource<'a> {
    context: BatchReadContext<'a>,
    watched_bytes: usize,
    allocations: &'a Cell<usize>,
}

impl BindingBatchSource for RuntimeSource<'_> {
    fn execute(
        &mut self,
        input: &PhysicalPlan,
        limit: ExecutionLimit,
        emit: &mut dyn FnMut(BindingBatch) -> Result<BatchControl>,
    ) -> Result<BatchControl> {
        execute_binding_batches(input, self.context, limit, &mut |batch| {
            // The real child has already admitted/constructed its owned rows.
            // Fixture, plan, source and input allocation are outside this window.
            let window = AllocationWindow::start(self.watched_bytes);
            let result = emit(batch);
            self.allocations
                .set(self.allocations.get() + window.allocations());
            result
        })
    }
}

fn nz(value: usize) -> NonZeroUsize {
    NonZeroUsize::new(value).unwrap()
}

fn seed(window: usize) -> PhysicalPlan {
    PhysicalPlan::TextSeedScan {
        complete: false,
        query_parameter: "text".into(),
        top_k: window,
        output_external_id: true,
        metadata_filters: BTreeMap::new(),
        resource_profile: VectorExecutionResourceProfile {
            priority: 1,
            max_parallelism: 1,
            max_working_memory_bytes: Some(16 * 1024),
        },
    }
}

fn run_refused(items: Vec<Projection>, external_id: String, watched_bytes: usize) {
    let catalog = Catalog::default();
    let store = GraphStore::default();
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(64 * 1024),
        blocking_operator_bytes: nz(16 * 1024),
        batch_payload_bytes: nz(16 * 1024),
        batch_rows: nz(8192),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let input = seed(1);
    let observer = QueryExecutionObserver::new(&input);
    let parameters = BTreeMap::from([("text".into(), Value::String("graph".into()))]);
    let external = TextSource {
        external_id,
        calls: Cell::new(0),
    };
    let allocations = Cell::new(0);
    let mut source = RuntimeSource {
        context: BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
            host_scorer: None,
        },
        watched_bytes,
        allocations: &allocations,
    };
    let context = BatchExecutionContext {
        catalog: &catalog,
        memory: &memory,
        memory_ledger: &ledger,
        task_context: None,
        observer: &observer,
    };
    let callbacks = Cell::new(0);
    let result = stream_projection_batches(
        &items,
        &input,
        &mut source,
        context,
        ExecutionLimit::unlimited(),
        &mut |_| {
            callbacks.set(callbacks.get() + 1);
            Ok(BatchControl::Continue)
        },
    );
    assert_eq!(
        allocations.get(),
        0,
        "oversized projection copied payload before row admission"
    );
    assert!(matches!(
        result,
        Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
    ));
    assert_eq!(callbacks.get(), 0);
    assert_eq!(external.calls.get(), 1);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert!(ledger.snapshot().peak_bytes <= memory.query_memory_bytes.get());
}

#[test]
fn text_projection_refuses_large_literal_before_copying_payload() {
    let size = 1024 * 1024 + 137;
    run_refused(
        vec![Projection {
            name: "blob".into(),
            expression: ProjectionExpression::Literal(Value::String("x".repeat(size))),
        }],
        "a".into(),
        size,
    );
}

#[test]
fn text_projection_refuses_repeated_columns_before_copying_payload() {
    let size = 4 * 1024 + 137;
    run_refused(
        (0..5)
            .map(|alias| Projection {
                name: format!("blob{alias}"),
                expression: ProjectionExpression::Column("external_id".into()),
            })
            .collect(),
        "x".repeat(size),
        size,
    );
}

#[test]
fn text_projection_refuses_null_list_storage_before_copying() {
    let count = 131_073;
    run_refused(
        vec![Projection {
            name: "blob".into(),
            expression: ProjectionExpression::Literal(Value::List(vec![Value::Null; count])),
        }],
        "a".into(),
        count * std::mem::size_of::<Value>(),
    );
}

fn run_optimized_literal_refused(numeric: bool, typed: bool) {
    let size = 1024 * 1024 + 137;
    let mut catalog = Catalog::default();
    let mut store = GraphStore::default();
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "score",
            PropertyType::Int,
            false,
        )
        .unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "kind",
            PropertyType::String,
            false,
        )
        .unwrap();
    store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([
                ("score".into(), Value::Int(1)),
                ("kind".into(), Value::String("fixture".into())),
            ]),
        )
        .unwrap();
    let mut items = vec![Projection {
        name: "blob".into(),
        expression: ProjectionExpression::Literal(Value::String("X".repeat(size))),
    }];
    let mut required_properties = Vec::new();
    if numeric {
        required_properties.push("score".into());
    }
    if numeric && !typed {
        required_properties.push("kind".into());
        items.push(Projection {
            name: "kind".into(),
            expression: ProjectionExpression::Property {
                variable: "n".into(),
                property: "kind".into(),
            },
        });
    }
    let plan = PhysicalPlan::NodeProjectionScanExec {
        variable: "n".into(),
        label: "Memory".into(),
        access: NodeProjectionAccess::LabelScan,
        required_properties,
        predicate: numeric.then(|| Predicate::PropertyEq {
            variable: "n".into(),
            property: "score".into(),
            value: Value::Int(1),
        }),
        items,
    };
    let memory = ExecutionMemoryConfig {
        // Leave room for the real numeric morsel admission. The selected
        // output must still be refused by its independent 16 KiB row budget.
        query_memory_bytes: nz(if numeric { 16 * 1024 * 1024 } else { 64 * 1024 }),
        blocking_operator_bytes: nz(16 * 1024),
        batch_payload_bytes: nz(16 * 1024),
        batch_rows: nz(8192),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    let context = BatchReadContext {
        catalog: &catalog,
        store: &store,
        parameters: &parameters,
        external: &external,
        memory: &memory,
        memory_ledger: &ledger,
        task_context: None,
        observer: &observer,
        host_scorer: None,
    };
    let callbacks = Cell::new(0);
    // Schema, fixture, prebuilt plan and literal are outside observation.
    // This executes the actual optimized producer, including numeric selection.
    let window = AllocationWindow::start(size);
    let result = execute_binding_batches(&plan, context, ExecutionLimit::unlimited(), &mut |_| {
        callbacks.set(callbacks.get() + 1);
        Ok(BatchControl::Continue)
    });
    assert_eq!(
        window.allocations(),
        0,
        "optimized projection copied payload before row admission"
    );
    drop(window);
    let error = result.unwrap_err();
    assert!(error.to_string().contains("batch_payload_bytes"), "{error}");
    assert_eq!(callbacks.get(), 0);
    assert_eq!(external.calls.get(), 0);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert!(ledger.snapshot().peak_bytes <= memory.query_memory_bytes.get());
}

#[test]
fn fused_node_projection_refuses_large_literal_before_copying() {
    run_optimized_literal_refused(false, false);
}

#[test]
fn numeric_lending_projection_refuses_large_literal_before_copying() {
    run_optimized_literal_refused(true, true);
}

#[test]
fn numeric_rows_projection_refuses_large_literal_before_copying() {
    run_optimized_literal_refused(true, false);
}

fn run_stored_property_projection(numeric: bool, requested: bool) {
    run_stored_property_projection_with_access(numeric, requested, NodeProjectionAccess::LabelScan);
}

struct StoredPropertyDirectory(std::path::PathBuf);

impl StoredPropertyDirectory {
    fn new() -> Self {
        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let path = std::env::temp_dir().join(format!(
            "hawdb-projection-admission-{}-{}",
            std::process::id(),
            NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        ));
        assert!(!path.exists(), "fresh fixture already exists");
        Self(path)
    }
}

impl Drop for StoredPropertyDirectory {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}

fn run_stored_property_projection_with_access(
    numeric: bool,
    requested: bool,
    access: NodeProjectionAccess,
) {
    run_stored_property_projection_with_source(numeric, requested, access, false);
}

fn run_stored_property_projection_with_source(
    numeric: bool,
    requested: bool,
    access: NodeProjectionAccess,
    persisted: bool,
) {
    let size = 1024 * 1024 + 137;
    let directory = persisted.then(StoredPropertyDirectory::new);
    let mut catalog = Catalog::default();
    let mut store = if let Some(directory) = &directory {
        GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            hawdb_storage::store::WalReplayConfig {
                residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
                ..Default::default()
            },
        )
        .unwrap()
    } else {
        GraphStore::default()
    };
    store.create_node_table(&mut catalog, "Memory").unwrap();
    for (property, property_type) in [("score", PropertyType::Int), ("body", PropertyType::String)]
    {
        store
            .create_property_descriptor(
                &mut catalog,
                TableKind::Node,
                "Memory",
                property,
                property_type,
                false,
            )
            .unwrap();
    }
    store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([
                ("score".into(), Value::Int(1)),
                ("body".into(), Value::String("X".repeat(size))),
            ]),
        )
        .unwrap();
    if let Some(directory) = &directory {
        store.checkpoint(&catalog).unwrap();
        drop(store);
        catalog = Catalog::default();
        store = GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            hawdb_storage::store::WalReplayConfig {
                residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
                ..Default::default()
            },
        )
        .unwrap();
        assert!(store.is_out_of_core());
    }
    let mut required_properties = Vec::new();
    if numeric {
        required_properties.push("score".into());
    }
    if requested {
        required_properties.push("body".into());
    }
    let mut plan = PhysicalPlan::NodeProjectionScanExec {
        variable: "n".into(),
        label: "Memory".into(),
        access,
        required_properties,
        predicate: numeric.then(|| Predicate::PropertyEq {
            variable: "n".into(),
            property: "score".into(),
            value: Value::Int(1),
        }),
        items: vec![Projection {
            name: "result".into(),
            expression: if requested {
                ProjectionExpression::Property {
                    variable: "n".into(),
                    property: "body".into(),
                }
            } else {
                ProjectionExpression::Literal(Value::Int(7))
            },
        }],
    };
    if numeric && persisted {
        // The fused out-of-core NodeProjectionScan deliberately declines the
        // numeric specialization. Its ordinary Project/Filter/Scan entry point
        // exercises the owned numeric and owned typed-numeric producers.
        let PhysicalPlan::NodeProjectionScanExec {
            predicate: Some(predicate),
            items,
            ..
        } = plan
        else {
            unreachable!("numeric fixture retains its selection predicate");
        };
        plan = PhysicalPlan::ProjectExec {
            items,
            input: Box::new(PhysicalPlan::FilterExec {
                predicate,
                input: Box::new(PhysicalPlan::SeqNodeScan {
                    variable: "n".into(),
                    label: "Memory".into(),
                }),
            }),
        };
    }
    let memory = ExecutionMemoryConfig {
        // Numeric selection owns an admitted morsel; the ordinary fused
        // source must also obey its smaller query and working envelopes.
        query_memory_bytes: nz(if numeric { 16 * 1024 * 1024 } else { 64 * 1024 }),
        blocking_operator_bytes: nz(16 * 1024),
        batch_payload_bytes: nz(16 * 1024),
        batch_rows: nz(8192),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    let context = BatchReadContext {
        catalog: &catalog,
        store: &store,
        parameters: &parameters,
        external: &external,
        memory: &memory,
        memory_ledger: &ledger,
        task_context: None,
        observer: &observer,
        host_scorer: None,
    };
    let rows = Cell::new(0);
    // Store ownership and plan preparation precede the observation window.
    // Observe the actual producer, including property selection/hydration.
    let window = AllocationWindow::start(size);
    let result =
        execute_binding_batches(&plan, context, ExecutionLimit::unlimited(), &mut |batch| {
            assert!(!requested, "oversized stored property reached the consumer");
            for row in batch {
                assert_eq!(row.values["result"], Value::Int(7));
                rows.set(rows.get() + 1);
            }
            Ok(BatchControl::Continue)
        });
    assert_eq!(
        window.allocations(),
        usize::from(numeric && persisted && requested),
        "an admitted numeric source may copy once; refused output and unrequested properties must not copy"
    );
    drop(window);
    if requested {
        let error = result.unwrap_err();
        if numeric {
            assert!(error.to_string().contains("batch_payload_bytes"), "{error}");
        } else {
            assert!(matches!(
                error,
                HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_)
            ));
        }
        assert_eq!(rows.get(), 0);
    } else {
        assert_eq!(result.unwrap(), BatchControl::Continue);
        assert_eq!(rows.get(), 1);
    }
    assert_eq!(external.calls.get(), 0);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert!(ledger.snapshot().peak_bytes <= memory.query_memory_bytes.get());
}

#[test]
fn fused_projection_refuses_large_stored_property_before_copying() {
    run_stored_property_projection(false, true);
}

fn run_pruning_budget(predicate: Predicate, count: usize, refusal: bool) {
    let mut catalog = Catalog::default();
    let mut store = GraphStore::default();
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "kind",
            PropertyType::String,
            false,
        )
        .unwrap();
    for _ in 0..count {
        store
            .create_node(
                &mut catalog,
                "Memory",
                BTreeMap::from([("kind".into(), Value::String("a".into()))]),
            )
            .unwrap();
    }
    store
        .create_property_index(&mut catalog, "Memory", "kind")
        .unwrap();
    let label_id = catalog.label_id("Memory").unwrap();
    assert!(catalog.property_index_id(label_id, "kind").is_some());
    // Verify the fixture exercises index pruning before the budgeted run.
    let probe = store.scan_nodes_with_filter_pruning(
        &catalog,
        Some(label_id),
        Some(&hawdb_storage::mutation::PropertyFilter::Eq {
            property: "kind".into(),
            value: Value::String("a".into()),
        }),
    );
    assert!(probe.report.pruned);
    assert_eq!(probe.report.candidate_count_before_filter, count);
    assert_eq!(probe.nodes.len(), count);
    drop(probe);
    let plan = PhysicalPlan::NodeProjectionScanExec {
        variable: "n".into(),
        label: "Memory".into(),
        access: NodeProjectionAccess::LabelScan,
        required_properties: vec!["kind".into()],
        predicate: Some(predicate),
        items: vec![Projection {
            name: "result".into(),
            expression: ProjectionExpression::Literal(Value::Int(7)),
        }],
    };
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(64 * 1024),
        blocking_operator_bytes: nz(4096),
        batch_payload_bytes: nz(4096),
        batch_rows: nz(8192),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    let mut rows = 0;
    let result = execute_binding_batches(
        &plan,
        BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
            host_scorer: None,
        },
        ExecutionLimit::unlimited(),
        &mut |batch| {
            for row in batch {
                assert_eq!(row.values["result"], Value::Int(7));
                rows += 1;
            }
            Ok(BatchControl::Continue)
        },
    );
    if refusal {
        assert!(
            matches!(
                result,
                Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
            ),
            "indexed candidate storage escaped its working budget: {result:?}"
        );
        assert_eq!(rows, 0, "refused pruning exposed a partial result");
    } else {
        assert_eq!(result.unwrap(), BatchControl::Continue);
        assert_eq!(rows, count);
    }
    assert_eq!(external.calls.get(), 0);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert!(ledger.snapshot().peak_bytes <= memory.query_memory_bytes.get());
}

fn pruning_equality() -> Predicate {
    Predicate::PropertyEq {
        variable: "n".into(),
        property: "kind".into(),
        value: Value::String("a".into()),
    }
}

#[test]
fn pruning_dense_index_candidates_refuse_before_result_delivery() {
    run_pruning_budget(pruning_equality(), 128, true);
}

#[test]
fn pruning_and_charges_simultaneous_candidate_sets() {
    run_pruning_budget(
        Predicate::And(vec![pruning_equality(), pruning_equality()]),
        32,
        true,
    );
}

#[test]
fn pruning_or_charges_simultaneous_candidate_sets() {
    run_pruning_budget(
        Predicate::Or(vec![pruning_equality(), pruning_equality()]),
        32,
        true,
    );
}

#[test]
fn pruning_single_candidate_set_within_budget_preserves_all_rows() {
    run_pruning_budget(pruning_equality(), 32, false);
}

#[test]
fn fused_projection_skips_unrequested_large_stored_property() {
    run_stored_property_projection(false, false);
}

#[test]
fn numeric_rows_projection_refuses_large_stored_property_before_copying() {
    run_stored_property_projection(true, true);
}

#[test]
fn property_access_projection_refuses_large_stored_property_before_copying() {
    run_stored_property_projection_with_access(
        false,
        true,
        NodeProjectionAccess::PropertyValues {
            property: "score".into(),
            values: vec![Value::Int(1)],
        },
    );
}

#[test]
fn property_access_projection_skips_unrequested_large_stored_property() {
    run_stored_property_projection_with_access(
        false,
        false,
        NodeProjectionAccess::PropertyValues {
            property: "score".into(),
            values: vec![Value::Int(1)],
        },
    );
}

#[test]
fn range_projection_refuses_large_selected_value_before_ownership() {
    run_stored_property_projection_with_source(
        false,
        true,
        NodeProjectionAccess::PropertyRange {
            property: "score".into(),
            lower: Some((Value::Int(1), true)),
            upper: Some((Value::Int(1), true)),
        },
        false,
    );
}

#[test]
fn range_projection_skips_unrequested_large_value() {
    run_stored_property_projection_with_source(
        false,
        false,
        NodeProjectionAccess::PropertyRange {
            property: "score".into(),
            lower: Some((Value::Int(1), true)),
            upper: Some((Value::Int(1), true)),
        },
        false,
    );
}

#[test]
fn composite_projection_refuses_large_selected_value_before_ownership() {
    run_stored_property_projection_with_source(
        false,
        true,
        NodeProjectionAccess::CompositeEquality {
            predicates: vec![("score".into(), Value::Int(1))],
        },
        false,
    );
}

#[test]
fn composite_projection_skips_unrequested_large_value() {
    run_stored_property_projection_with_source(
        false,
        false,
        NodeProjectionAccess::CompositeEquality {
            predicates: vec![("score".into(), Value::Int(1))],
        },
        false,
    );
}

#[test]
fn persisted_property_projection_refuses_large_selected_value_before_ownership() {
    run_stored_property_projection_with_source(
        false,
        true,
        NodeProjectionAccess::PropertyValues {
            property: "score".into(),
            values: vec![Value::Int(1)],
        },
        true,
    );
}

#[test]
fn persisted_property_projection_skips_unrequested_large_value() {
    run_stored_property_projection_with_source(
        false,
        false,
        NodeProjectionAccess::PropertyValues {
            property: "score".into(),
            values: vec![Value::Int(1)],
        },
        true,
    );
}

#[test]
fn persisted_label_projection_refuses_large_selected_value_before_ownership() {
    run_stored_property_projection_with_source(false, true, NodeProjectionAccess::LabelScan, true);
}

#[test]
fn persisted_label_projection_skips_unrequested_large_value() {
    run_stored_property_projection_with_source(false, false, NodeProjectionAccess::LabelScan, true);
}

#[test]
fn persisted_numeric_projection_refuses_large_selected_value_before_output_ownership() {
    run_stored_property_projection_with_source(true, true, NodeProjectionAccess::LabelScan, true);
}

#[test]
fn persisted_numeric_projection_skips_unrequested_large_value() {
    run_stored_property_projection_with_source(true, false, NodeProjectionAccess::LabelScan, true);
}

#[test]
fn text_projection_refuses_selected_nested_payloads_before_copying() {
    let size = 1024 * 1024 + 137;
    let blob = || ProjectionExpression::Literal(Value::String("X".repeat(size)));
    let expressions = vec![
        ProjectionExpression::Coalesce(vec![ProjectionExpression::Literal(Value::Null), blob()]),
        ProjectionExpression::Case {
            operand: None,
            branches: vec![(ProjectionExpression::Literal(Value::Bool(true)), blob())],
            otherwise: None,
        },
        ProjectionExpression::Lower(Box::new(blob())),
        ProjectionExpression::Left {
            expression: Box::new(blob()),
            length: size,
        },
        ProjectionExpression::Literal(Value::List(vec![Value::String("X".repeat(size))])),
        ProjectionExpression::Literal(Value::Map(BTreeMap::from([(
            "body".into(),
            Value::String("X".repeat(size)),
        )]))),
        ProjectionExpression::ColumnValueCasePropertyNotNullOrEq {
            column: "external_id".into(),
            empty: Value::Null,
            non_empty: Value::String("X".repeat(size)),
            null_or_empty: Value::Int(0),
        },
    ];
    for expression in expressions {
        run_refused(
            vec![Projection {
                name: "blob".into(),
                expression,
            }],
            "a".into(),
            size,
        );
    }
}

#[test]
fn text_projection_shrinks_borrowed_payload_and_skips_unused_branches() {
    let size = 1024 * 1024 + 137;
    let catalog = Catalog::default();
    let store = GraphStore::default();
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(64 * 1024),
        blocking_operator_bytes: nz(16 * 1024),
        batch_payload_bytes: nz(16 * 1024),
        batch_rows: nz(8192),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let input = seed(1);
    let observer = QueryExecutionObserver::new(&input);
    let parameters = BTreeMap::from([("text".into(), Value::String("graph".into()))]);
    let external = TextSource {
        external_id: "a".into(),
        calls: Cell::new(0),
    };
    let allocations = Cell::new(0);
    let mut source = RuntimeSource {
        context: BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
            host_scorer: None,
        },
        watched_bytes: size,
        allocations: &allocations,
    };
    let blob = || ProjectionExpression::Literal(Value::String("X".repeat(size)));
    let items = vec![
        Projection {
            name: "short".into(),
            expression: ProjectionExpression::Left {
                expression: Box::new(blob()),
                length: 3,
            },
        },
        Projection {
            name: "case".into(),
            expression: ProjectionExpression::Case {
                operand: None,
                branches: vec![(ProjectionExpression::Literal(Value::Bool(false)), blob())],
                otherwise: Some(Box::new(ProjectionExpression::Literal(Value::Int(7)))),
            },
        },
        Projection {
            name: "coalesce".into(),
            expression: ProjectionExpression::Coalesce(vec![
                ProjectionExpression::Literal(Value::String("ok".into())),
                blob(),
            ]),
        },
    ];
    let context = BatchExecutionContext {
        catalog: &catalog,
        memory: &memory,
        memory_ledger: &ledger,
        task_context: None,
        observer: &observer,
    };
    let rows = Cell::new(0);
    let result = stream_projection_batches(
        &items,
        &input,
        &mut source,
        context,
        ExecutionLimit::unlimited(),
        &mut |batch| {
            for row in batch {
                assert_eq!(row.values["short"], Value::String("XXX".into()));
                assert_eq!(row.values["case"], Value::Int(7));
                assert_eq!(row.values["coalesce"], Value::String("ok".into()));
                rows.set(rows.get() + 1);
            }
            Ok(BatchControl::Continue)
        },
    );
    assert_eq!(result.unwrap(), BatchControl::Continue);
    assert_eq!(rows.get(), 1);
    assert_eq!(
        allocations.get(),
        0,
        "shrinking projection copied the unneeded full payload"
    );
    assert_eq!(ledger.snapshot().used_bytes, 0);
}

#[test]
fn text_projection_batches_actual_payload_with_exact_rows_and_stop() {
    for stop in [false, true] {
        let catalog = Catalog::default();
        let store = GraphStore::default();
        let memory = ExecutionMemoryConfig {
            query_memory_bytes: nz(64 * 1024),
            blocking_operator_bytes: nz(16 * 1024),
            batch_payload_bytes: nz(16 * 1024),
            batch_rows: nz(8192),
            ..Default::default()
        };
        let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
        let input = seed(3);
        let observer = QueryExecutionObserver::new(&input);
        let parameters = BTreeMap::from([("text".into(), Value::String("graph".into()))]);
        let external = TextSource {
            external_id: "x".repeat(1024),
            calls: Cell::new(0),
        };
        let allocations = Cell::new(0);
        let mut source = RuntimeSource {
            context: BatchReadContext {
                catalog: &catalog,
                store: &store,
                parameters: &parameters,
                external: &external,
                memory: &memory,
                memory_ledger: &ledger,
                task_context: None,
                observer: &observer,
                host_scorer: None,
            },
            watched_bytes: 0,
            allocations: &allocations,
        };
        let items: Vec<_> = (0..8)
            .map(|alias| Projection {
                name: format!("blob{alias}"),
                expression: ProjectionExpression::Column("external_id".into()),
            })
            .collect();
        let context = BatchExecutionContext {
            catalog: &catalog,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
        };
        let rows = Cell::new(0);
        let result = stream_projection_batches(
            &items,
            &input,
            &mut source,
            context,
            ExecutionLimit::unlimited(),
            &mut |batch| {
                let bytes: usize = batch.iter().map(binding_memory_bytes).sum();
                assert!(
                    bytes <= memory.batch_payload_bytes.get(),
                    "projected batch actual payload {bytes} exceeds admitted {}",
                    memory.batch_payload_bytes
                );
                for Binding { values, .. } in &batch {
                    assert_eq!(values.len(), 8);
                    assert!(values
                        .values()
                        .all(|value| value == &Value::String(external.external_id.clone())));
                }
                rows.set(rows.get() + batch.len());
                Ok(if stop {
                    BatchControl::Stop
                } else {
                    BatchControl::Continue
                })
            },
        );
        assert_eq!(
            result.unwrap(),
            if stop {
                BatchControl::Stop
            } else {
                BatchControl::Continue
            }
        );
        assert_eq!(rows.get(), if stop { 1 } else { 3 });
        assert_eq!(ledger.snapshot().used_bytes, 0);
    }
}

struct DirectProjectionSource(Option<BindingBatch>);

impl BindingBatchSource for DirectProjectionSource {
    fn execute(
        &mut self,
        _input: &PhysicalPlan,
        limit: ExecutionLimit,
        emit: &mut dyn FnMut(BindingBatch) -> Result<BatchControl>,
    ) -> Result<BatchControl> {
        let rows: Vec<_> = self
            .0
            .take()
            .unwrap()
            .into_iter()
            .take(limit.output_rows.unwrap_or(usize::MAX))
            .collect();
        if rows.is_empty() {
            Ok(BatchControl::Continue)
        } else {
            emit(rows)
        }
    }
}

fn projection_cancel_guard(optimized: bool) {
    let mut catalog = Catalog::default();
    let mut store = GraphStore::default();
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(128 * 1024),
        blocking_operator_bytes: nz(32 * 1024),
        batch_payload_bytes: nz(16 * 1024),
        batch_rows: nz(128),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let token = RuntimeCancellationToken::new();
    let task = RuntimeTaskContext::without_deadline(token.clone());
    let items: Vec<_> = (0..8)
        .map(|index| Projection {
            name: format!("blob{index}"),
            expression: if optimized {
                ProjectionExpression::Property {
                    variable: "n".into(),
                    property: "blob".into(),
                }
            } else {
                ProjectionExpression::Column("blob".into())
            },
        })
        .collect();
    let input = PhysicalPlan::EmptyExec;
    let observer = QueryExecutionObserver::new(&input);
    let mut calls = 0;
    let mut after_cancel = None;
    let mut emit = |batch: BindingBatch| {
        calls += 1;
        assert_eq!(batch.len(), 1);
        assert!(
            batch.iter().map(binding_memory_bytes).sum::<usize>()
                <= memory.batch_payload_bytes.get()
        );
        if calls == 1 {
            assert!(token.cancel());
            after_cancel = Some(AllocationWindow::start(1024));
        }
        Ok(BatchControl::Continue)
    };
    let result = if optimized {
        store.create_node_table(&mut catalog, "Memory").unwrap();
        store
            .create_property_descriptor(
                &mut catalog,
                TableKind::Node,
                "Memory",
                "blob",
                PropertyType::String,
                false,
            )
            .unwrap();
        for _ in 0..3 {
            store
                .create_node(
                    &mut catalog,
                    "Memory",
                    BTreeMap::from([("blob".into(), Value::String("x".repeat(1024)))]),
                )
                .unwrap();
        }
        let source = ledger.account(
            hawdb_executor::QueryMemoryClass::ExternalRead,
            "test projection source",
            memory.blocking_operator_bytes,
        );
        let output = ledger.account(
            hawdb_executor::QueryMemoryClass::PipelineBatch,
            "test projection output",
            memory.batch_payload_bytes,
        );
        hawdb_executor::scan::stream_node_projection_scan_batches(
            hawdb_executor::scan::NodeProjectionScanSpec {
                variable: "n",
                label: "Memory",
                access: &NodeProjectionAccess::LabelScan,
                required_properties: &["blob".into()],
                predicate: None,
                items: &items,
            },
            hawdb_executor::scan::NodeScanContext {
                catalog: &catalog,
                store: &store,
                execution_limit: ExecutionLimit::unlimited(),
                memory_budget: memory.blocking_operator_bytes,
                memory_account: &source,
                batch_memory_budget: memory.batch_payload_bytes,
                batch_memory_account: &output,
                batch_rows: memory.batch_rows.get(),
                task_context: Some(&task),
            },
            &observer,
            &mut emit,
        )
    } else {
        let mut source = DirectProjectionSource(Some(
            (0..3)
                .map(|_| {
                    Binding::values(BTreeMap::from([(
                        "blob".into(),
                        Value::String("x".repeat(1024)),
                    )]))
                })
                .collect(),
        ));
        stream_projection_batches(
            &items,
            &input,
            &mut source,
            BatchExecutionContext {
                catalog: &catalog,
                memory: &memory,
                memory_ledger: &ledger,
                task_context: Some(&task),
                observer: &observer,
            },
            ExecutionLimit::unlimited(),
            &mut emit,
        )
    };
    assert_eq!(
        after_cancel
            .as_ref()
            .expect("the first byte-boundary callback must run")
            .allocations(),
        0,
        "cancellation after Continue must precede any next-row payload ownership"
    );
    assert_eq!(calls, 1);
    assert!(result
        .unwrap_err()
        .to_string()
        .contains("runtime task stopped: cancelled"));
    assert_eq!(ledger.snapshot().used_bytes, 0);
}

#[test]
fn public_projection_cancel_after_continue_prevents_next_payload_copy() {
    projection_cancel_guard(false);
}

#[test]
fn optimized_projection_cancel_after_continue_prevents_next_payload_copy() {
    projection_cancel_guard(true);
}

#[derive(Clone, Copy)]
enum NumericBufferCase {
    Stop,
    Error,
    Refusal,
    RetainedRows,
    TypedScratch,
}

fn run_persisted_numeric_buffer(case: NumericBufferCase) {
    let directory = StoredPropertyDirectory::new();
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = GraphStore::open_with_durability_and_replay_config(
        &directory.0,
        &mut catalog,
        Default::default(),
        replay,
    )
    .unwrap();
    store.create_node_table(&mut catalog, "Memory").unwrap();
    for (property, kind) in [("score", PropertyType::Int), ("body", PropertyType::String)] {
        store
            .create_property_descriptor(
                &mut catalog,
                TableKind::Node,
                "Memory",
                property,
                kind,
                false,
            )
            .unwrap();
    }
    let count = if matches!(case, NumericBufferCase::RetainedRows) {
        40
    } else {
        2
    };
    let size = 1024 * 1024 + 137;
    let mut expected_ids = Vec::new();
    for index in 0..count {
        let body = if index == 1
            && matches!(
                case,
                NumericBufferCase::Stop | NumericBufferCase::Error | NumericBufferCase::Refusal
            ) {
            "X".repeat(size)
        } else {
            "a".repeat(1041)
        };
        expected_ids.push(
            store
                .create_node(
                    &mut catalog,
                    "Memory",
                    BTreeMap::from([
                        ("score".into(), Value::Int(1)),
                        ("body".into(), Value::String(body)),
                    ]),
                )
                .unwrap(),
        );
    }
    store.checkpoint(&catalog).unwrap();
    drop(store);
    catalog = Catalog::default();
    let store = GraphStore::open_with_durability_and_replay_config(
        &directory.0,
        &mut catalog,
        Default::default(),
        replay,
    )
    .unwrap();
    assert!(store.is_out_of_core());
    let typed = matches!(case, NumericBufferCase::TypedScratch);
    let plan = PhysicalPlan::ProjectExec {
        items: vec![
            Projection {
                name: "id".into(),
                expression: ProjectionExpression::Id {
                    variable: "n".into(),
                },
            },
            Projection {
                name: "result".into(),
                expression: if typed {
                    ProjectionExpression::Literal(Value::Int(7))
                } else {
                    ProjectionExpression::Property {
                        variable: "n".into(),
                        property: "body".into(),
                    }
                },
            },
        ],
        input: Box::new(PhysicalPlan::FilterExec {
            predicate: Predicate::PropertyEq {
                variable: "n".into(),
                property: "score".into(),
                value: Value::Int(1),
            },
            input: Box::new(PhysicalPlan::SeqNodeScan {
                variable: "n".into(),
                label: "Memory".into(),
            }),
        }),
    };
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(if matches!(case, NumericBufferCase::Refusal) {
            16 * 1024 * 1024
        } else {
            size + 128 * 1024
        }),
        blocking_operator_bytes: nz(8192),
        batch_payload_bytes: nz(4096),
        batch_rows: nz(if matches!(case, NumericBufferCase::RetainedRows) {
            1
        } else if typed {
            2
        } else {
            8192
        }),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    let mut ids = Vec::new();
    let mut callbacks = 0;
    let window = AllocationWindow::start(size);
    let result = execute_binding_batches(
        &plan,
        BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
            host_scorer: None,
        },
        ExecutionLimit::unlimited(),
        &mut |batch| {
            callbacks += 1;
            if matches!(
                case,
                NumericBufferCase::RetainedRows | NumericBufferCase::TypedScratch
            ) {
                let class = if typed {
                    hawdb_executor::QueryMemoryClass::PipelineBatch
                } else {
                    hawdb_executor::QueryMemoryClass::ExternalRead
                };
                let retained = ledger
                    .snapshot()
                    .classes
                    .iter()
                    .find(|entry| entry.class == class)
                    .map_or(0, |entry| entry.used_bytes);
                let output_bytes = batch.iter().map(binding_memory_bytes).sum::<usize>();
                assert!(
                    retained >= if typed { output_bytes + 32 } else { 2065 },
                    "numeric buffer lost admission while its input remained owned: {retained}"
                );
            }
            for row in batch {
                ids.push(row.values["id"].clone());
                assert_eq!(
                    row.values["result"],
                    if typed {
                        Value::Int(7)
                    } else {
                        Value::String("a".repeat(1041))
                    }
                );
            }
            match case {
                NumericBufferCase::Stop => Ok(BatchControl::Stop),
                NumericBufferCase::Error => {
                    Err(HawDBError::Execution("numeric consumer failed".into()))
                }
                _ => Ok(BatchControl::Continue),
            }
        },
    );
    assert_eq!(
        window.allocations(),
        usize::from(matches!(case, NumericBufferCase::Refusal)),
        "numeric source may fit the root, but a refused output must not clone its oversized payload"
    );
    drop(window);
    match case {
        NumericBufferCase::Stop => assert_eq!(result.unwrap(), BatchControl::Stop),
        NumericBufferCase::Error => assert!(
            matches!(result, Err(HawDBError::Execution(message)) if message == "numeric consumer failed")
        ),
        NumericBufferCase::Refusal => {
            let error = result.unwrap_err();
            assert!(error.to_string().contains("batch_payload_bytes"), "{error}");
        }
        _ => assert_eq!(result.unwrap(), BatchControl::Continue),
    }
    let expected_count = if matches!(
        case,
        NumericBufferCase::RetainedRows | NumericBufferCase::TypedScratch
    ) {
        count
    } else {
        1
    };
    assert_eq!(
        ids,
        expected_ids[..expected_count]
            .iter()
            .map(|id| Value::Int(id.0 as i64))
            .collect::<Vec<_>>()
    );
    assert!(callbacks > 0);
    assert_eq!(external.calls.get(), 0);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    // Probe the same persisted source after caller Stop/refusal/error.
    assert_eq!(
        store
            .node_owned(expected_ids[0])
            .unwrap()
            .unwrap()
            .properties["body"],
        Value::String("a".repeat(1041))
    );
}

#[test]
fn owned_numeric_buffer_stops_before_next_large_payload() {
    run_persisted_numeric_buffer(NumericBufferCase::Stop);
}
#[test]
fn owned_numeric_buffer_preserves_consumer_error_without_next_payload() {
    run_persisted_numeric_buffer(NumericBufferCase::Error);
}
#[test]
fn owned_numeric_buffer_refuses_next_output_payload_and_releases_admission() {
    run_persisted_numeric_buffer(NumericBufferCase::Refusal);
}
#[test]
fn owned_numeric_buffer_retains_row_permits_during_consumption() {
    run_persisted_numeric_buffer(NumericBufferCase::RetainedRows);
}
#[test]
fn owned_numeric_buffer_retains_typed_scratch_during_consumption() {
    run_persisted_numeric_buffer(NumericBufferCase::TypedScratch);
}

#[test]
fn owned_numeric_buffer_refuses_typed_capacity_before_allocation() {
    use hawdb_executor::numeric::{
        stream_owned_typed_numeric_nodes, LendingNumericScan, NumericExecutionContext,
        NumericFragment,
    };
    let mut catalog = Catalog::default();
    let mut store = GraphStore::default();
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([("score".into(), Value::Int(1))]),
        )
        .unwrap();
    let plan = PhysicalPlan::SeqNodeScan {
        variable: "n".into(),
        label: "Memory".into(),
    };
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(128),
        batch_payload_bytes: nz(4096),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let items = vec![Projection {
        name: "result".into(),
        expression: ProjectionExpression::Literal(Value::Int(7)),
    }];
    let window = AllocationWindow::start(32 * std::mem::size_of::<i64>());
    let result = stream_owned_typed_numeric_nodes(
        NumericFragment {
            label: "Memory",
            property: "score",
            property_type: PropertyType::Int,
            predicate: hawdb_executor::NumericPredicate::Eq,
            expected: hawdb_executor::NumericLiteral::Int(1),
            fused_operators: None,
        },
        &items,
        catalog.label_id("Memory").unwrap(),
        LendingNumericScan {
            batch_rows: 32,
            needs_node_ids: true,
        },
        NumericExecutionContext {
            catalog: &catalog,
            store: &store,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
        },
        ExecutionLimit::unlimited(),
        &mut |_| panic!("refused numeric capacity reached consumer"),
    );
    assert_eq!(
        window.allocations(),
        0,
        "typed numeric columns allocated before scratch admission"
    );
    drop(window);
    assert!(matches!(
        result,
        Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
    ));
    assert_eq!(ledger.snapshot().used_bytes, 0);
}

#[test]
fn owned_numeric_buffer_refuses_small_validity_capacity_before_allocation() {
    let directory = StoredPropertyDirectory::new();
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = GraphStore::open_with_durability_and_replay_config(
        &directory.0,
        &mut catalog,
        Default::default(),
        replay,
    )
    .unwrap();
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "score",
            PropertyType::Int,
            true,
        )
        .unwrap();
    store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([("score".into(), Value::Null)]),
        )
        .unwrap();
    store.checkpoint(&catalog).unwrap();
    drop(store);
    catalog = Catalog::default();
    let store = GraphStore::open_with_durability_and_replay_config(
        &directory.0,
        &mut catalog,
        Default::default(),
        replay,
    )
    .unwrap();
    assert!(store.is_out_of_core());
    let plan = PhysicalPlan::ProjectExec {
        items: vec![Projection {
            name: "result".into(),
            expression: ProjectionExpression::Literal(Value::Int(7)),
        }],
        input: Box::new(PhysicalPlan::FilterExec {
            predicate: Predicate::PropertyEq {
                variable: "n".into(),
                property: "score".into(),
                value: Value::Int(1),
            },
            input: Box::new(PhysicalPlan::SeqNodeScan {
                variable: "n".into(),
                label: "Memory".into(),
            }),
        }),
    };
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(64 * 1024),
        blocking_operator_bytes: nz(8192),
        batch_payload_bytes: nz(32),
        batch_rows: nz(1),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    // The first lazy bitmap reserves four u64 slots. A 32-byte payload budget
    // cannot also retain its scalar column and selection storage.
    let window = AllocationWindow::start(4 * std::mem::size_of::<u64>());
    let result = execute_binding_batches(
        &plan,
        BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
            host_scorer: None,
        },
        ExecutionLimit::unlimited(),
        &mut |_| panic!("NULL numeric predicate returned a row"),
    );
    assert_eq!(
        window.allocations(),
        0,
        "numeric validity bitmap allocated outside its scratch budget"
    );
    drop(window);
    assert!(matches!(
        result,
        Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
    ));
    assert_eq!(ledger.snapshot().used_bytes, 0);
}

#[derive(Clone, Copy)]
enum FullNodeSource {
    Scan,
    Index,
    Lookup(bool),
    IndexKey,
    LookupKey,
}

fn run_full_node_scan_admission(persisted: bool, materialized: bool) {
    run_full_node_source_admission(persisted, materialized, FullNodeSource::Scan)
}

fn run_full_node_source_admission(persisted: bool, materialized: bool, source: FullNodeSource) {
    let size = 1024 * 1024 + 137;
    let key = if matches!(source, FullNodeSource::IndexKey | FullNodeSource::LookupKey) {
        "K".repeat(size)
    } else {
        "target".into()
    };
    let directory = persisted.then(StoredPropertyDirectory::new);
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = if let Some(directory) = &directory {
        GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap()
    } else {
        GraphStore::default()
    };
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "body",
            PropertyType::String,
            false,
        )
        .unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "id",
            PropertyType::String,
            false,
        )
        .unwrap();
    store
        .create_property_index(&mut catalog, "Memory", "id")
        .unwrap();
    let id = store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([
                ("body".into(), Value::String("X".repeat(size))),
                ("id".into(), Value::String(key.clone())),
            ]),
        )
        .unwrap();
    if let Some(directory) = &directory {
        store.checkpoint(&catalog).unwrap();
        drop(store);
        catalog = Catalog::default();
        store = GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap();
    }
    assert_eq!(store.is_out_of_core(), persisted);
    let plan = if matches!(source, FullNodeSource::Index | FullNodeSource::IndexKey) {
        PhysicalPlan::IndexNodeSeek {
            variable: "n".into(),
            label: "Memory".into(),
            property: "id".into(),
            value: Value::String(key.clone()),
        }
    } else {
        PhysicalPlan::SeqNodeScan {
            variable: "n".into(),
            label: "Memory".into(),
        }
    };
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(64 * 1024),
        blocking_operator_bytes: nz(4096),
        batch_payload_bytes: nz(4096),
        batch_rows: nz(8192),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let rows = Cell::new(0);
    let predicate_calls = Cell::new(0);
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    let input_account = ledger.account(
        hawdb_executor::QueryMemoryClass::BlockingState,
        "full node input guard",
        memory.blocking_operator_bytes,
    );
    let batch_account = ledger.account(
        hawdb_executor::QueryMemoryClass::PipelineBatch,
        "full node batch guard",
        memory.batch_payload_bytes,
    );
    let lookup_input = vec![hawdb_executor::binding::Binding::values(BTreeMap::from([
        ("seed".into(), Value::String(key.clone())),
    ]))];
    let window = AllocationWindow::start(size);
    let result = if matches!(
        source,
        FullNodeSource::Lookup(_) | FullNodeSource::LookupKey
    ) {
        let indexed = !matches!(source, FullNodeSource::Lookup(false));
        hawdb_executor::scan::execute_node_column_lookup(
            hawdb_executor::scan::NodeColumnLookupSpec {
                variable: "n",
                label: if indexed { "Memory" } else { "" },
                property: "id",
                column: "seed",
                optional: false,
                node_visibility_predicate: None,
            },
            lookup_input,
            hawdb_executor::scan::NodeScanContext {
                catalog: &catalog,
                store: &store,
                execution_limit: ExecutionLimit::unlimited(),
                memory_budget: memory.blocking_operator_bytes,
                memory_account: &input_account,
                batch_memory_budget: memory.batch_payload_bytes,
                batch_memory_account: &batch_account,
                batch_rows: memory.batch_rows.get(),
                task_context: None,
            },
            &observer,
        )
        .map(|output| {
            rows.set(output.len());
            BatchControl::Continue
        })
    } else if materialized {
        hawdb_executor::scan::execute_node_scan(
            hawdb_executor::scan::NodeScanSpec {
                variable: "n",
                label: "Memory",
                property_filter: None,
            },
            hawdb_executor::scan::NodeScanContext {
                catalog: &catalog,
                store: &store,
                execution_limit: ExecutionLimit::unlimited(),
                memory_budget: memory.blocking_operator_bytes,
                memory_account: &input_account,
                batch_memory_budget: memory.batch_payload_bytes,
                batch_memory_account: &batch_account,
                batch_rows: memory.batch_rows.get(),
                task_context: None,
            },
            &mut |_| {
                predicate_calls.set(predicate_calls.get() + 1);
                Ok(true)
            },
            &observer,
        )
        .map(|output| {
            rows.set(output.len());
            BatchControl::Continue
        })
    } else {
        execute_binding_batches(
            &plan,
            BatchReadContext {
                catalog: &catalog,
                store: &store,
                parameters: &parameters,
                external: &external,
                memory: &memory,
                memory_ledger: &ledger,
                task_context: None,
                observer: &observer,
                host_scorer: None,
            },
            ExecutionLimit::unlimited(),
            &mut |batch| {
                rows.set(rows.get() + batch.len());
                Ok(BatchControl::Continue)
            },
        )
    };
    assert_eq!(
        window.allocations(),
        0,
        "full node payload was copied before query admission"
    );
    drop(window);
    assert!(matches!(
        result,
        Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
    ));
    assert_eq!(rows.get(), 0);
    assert_eq!(predicate_calls.get(), 0);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert_eq!(external.calls.get(), 0);
    // Caller refusal must not mark the pinned canonical reader corrupt.
    assert_eq!(store.node_owned(id).unwrap().unwrap().id, id);
}

#[test]
fn full_node_scan_admits_resident_streaming_payload_before_clone() {
    run_full_node_scan_admission(false, false);
}

#[test]
fn full_node_scan_admits_persisted_streaming_payload_before_decode() {
    run_full_node_scan_admission(true, false);
}

#[test]
fn full_node_scan_admits_resident_materialized_payload_before_clone() {
    run_full_node_scan_admission(false, true);
}

#[test]
fn full_node_scan_admits_persisted_materialized_payload_before_decode() {
    run_full_node_scan_admission(true, true);
}

#[derive(Clone, Copy)]
enum FullNodeBufferCase {
    Stop,
    Error,
    Refusal,
    Retained,
}

fn run_full_node_scan_buffer(persisted: bool, case: FullNodeBufferCase) {
    let directory = persisted.then(StoredPropertyDirectory::new);
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = if let Some(directory) = &directory {
        GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap()
    } else {
        GraphStore::default()
    };
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "body",
            PropertyType::String,
            false,
        )
        .unwrap();
    let retained = matches!(case, FullNodeBufferCase::Retained);
    let size = 1024 * 1024 + 137;
    let mut expected = Vec::new();
    for index in 0..if retained { 40 } else { 2 } {
        let body = if !retained && index == 1 {
            "X".repeat(size)
        } else {
            "a".repeat(1041)
        };
        expected.push(
            store
                .create_node(
                    &mut catalog,
                    "Memory",
                    BTreeMap::from([("body".into(), Value::String(body))]),
                )
                .unwrap(),
        );
    }
    if let Some(directory) = &directory {
        store.checkpoint(&catalog).unwrap();
        drop(store);
        catalog = Catalog::default();
        store = GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap();
    }
    assert_eq!(store.is_out_of_core(), persisted);
    let plan = PhysicalPlan::SeqNodeScan {
        variable: "n".into(),
        label: "Memory".into(),
    };
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(64 * 1024),
        blocking_operator_bytes: nz(8192),
        batch_payload_bytes: nz(4096),
        batch_rows: nz(if retained { 1 } else { 8192 }),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    let mut delivered = Vec::new();
    let window = AllocationWindow::start(size);
    let result = execute_binding_batches(
        &plan,
        BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
            host_scorer: None,
        },
        ExecutionLimit::unlimited(),
        &mut |batch| {
            let retained_bytes = ledger
                .snapshot()
                .classes
                .iter()
                .find(|entry| entry.class == hawdb_executor::QueryMemoryClass::ExternalRead)
                .map_or(0, |class| class.used_bytes);
            for row in batch {
                assert_eq!(
                    row.nodes["n"].properties["body"],
                    Value::String("a".repeat(1041))
                );
                delivered.push(row.nodes["n"].id);
                if retained {
                    assert!(
                        retained_bytes >= 1041,
                        "buffered node lost its payload permit: {retained_bytes}"
                    );
                }
            }
            match case {
                FullNodeBufferCase::Stop => Ok(BatchControl::Stop),
                FullNodeBufferCase::Error => {
                    Err(HawDBError::Execution("full node consumer sentinel".into()))
                }
                FullNodeBufferCase::Refusal | FullNodeBufferCase::Retained => {
                    Ok(BatchControl::Continue)
                }
            }
        },
    );
    assert_eq!(
        window.allocations(),
        0,
        "full-node buffer owned the next large payload before terminal Stop/error/refusal"
    );
    drop(window);
    match case {
        FullNodeBufferCase::Stop => assert_eq!(result.unwrap(), BatchControl::Stop),
        FullNodeBufferCase::Error => assert!(
            matches!(result, Err(HawDBError::Execution(message)) if message == "full node consumer sentinel")
        ),
        FullNodeBufferCase::Refusal => assert!(matches!(
            result,
            Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
        )),
        FullNodeBufferCase::Retained => assert_eq!(result.unwrap(), BatchControl::Continue),
    }
    assert_eq!(
        delivered.as_slice(),
        if retained {
            expected.as_slice()
        } else {
            &expected[..1]
        }
    );
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert_eq!(external.calls.get(), 0);
    assert_eq!(
        store.node_owned(expected[0]).unwrap().unwrap().id,
        expected[0]
    );
}

#[test]
fn full_node_scan_resident_buffer_stops_before_next_payload() {
    run_full_node_scan_buffer(false, FullNodeBufferCase::Stop);
}

#[test]
fn full_node_scan_persisted_buffer_stops_before_next_payload() {
    run_full_node_scan_buffer(true, FullNodeBufferCase::Stop);
}

#[test]
fn full_node_scan_resident_buffer_preserves_consumer_error() {
    run_full_node_scan_buffer(false, FullNodeBufferCase::Error);
}

#[test]
fn full_node_scan_persisted_buffer_preserves_consumer_error() {
    run_full_node_scan_buffer(true, FullNodeBufferCase::Error);
}

#[test]
fn full_node_scan_resident_buffer_refuses_without_next_payload() {
    run_full_node_scan_buffer(false, FullNodeBufferCase::Refusal);
}

#[test]
fn full_node_scan_persisted_buffer_refuses_without_next_payload() {
    run_full_node_scan_buffer(true, FullNodeBufferCase::Refusal);
}

#[test]
fn full_node_scan_resident_buffer_retains_payload_permit() {
    run_full_node_scan_buffer(false, FullNodeBufferCase::Retained);
}

#[test]
fn full_node_scan_persisted_buffer_retains_payload_permit() {
    run_full_node_scan_buffer(true, FullNodeBufferCase::Retained);
}

#[derive(Clone, Copy)]
enum PointReadCase {
    GraphImport,
    OneHop,
    Bounded,
}

fn run_point_read_admission(persisted: bool, case: PointReadCase) {
    let size = 1024 * 1024 + 137;
    let directory = persisted.then(StoredPropertyDirectory::new);
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = if let Some(directory) = &directory {
        GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap()
    } else {
        GraphStore::default()
    };
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "body",
            PropertyType::String,
            false,
        )
        .unwrap();
    let source = store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([("body".into(), Value::String(String::new()))]),
        )
        .unwrap();
    let target = store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([("body".into(), Value::String("X".repeat(size)))]),
        )
        .unwrap();
    store
        .create_relationship_table(&mut catalog, "LINKS")
        .unwrap();
    store
        .create_relationship(&mut catalog, source, target, "LINKS", BTreeMap::new())
        .unwrap();
    if let Some(directory) = &directory {
        store.checkpoint(&catalog).unwrap();
        drop(store);
        catalog = Catalog::default();
        store = GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap();
    }
    // The real text producer admits its small binding with a 4096-byte
    // envelope. Let that input reach MATCH without admitting the 1 MiB node.
    let point_budget = if matches!(case, PointReadCase::GraphImport) {
        16 * 1024
    } else {
        4096
    };
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(64 * 1024),
        blocking_operator_bytes: nz(point_budget),
        batch_payload_bytes: nz(point_budget),
        batch_rows: nz(1),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let mut input = seed(1);
    if let PhysicalPlan::TextSeedScan {
        resource_profile, ..
    } = &mut input
    {
        resource_profile.max_working_memory_bytes = Some(1024);
    }
    let plan = PhysicalPlan::GraphMatchExec {
        program: hawdb_plan_cypher::GraphMatchProgram {
            imports: vec![hawdb_plan_cypher::GraphBindingImport {
                variable: "n".into(),
                column: "entity".into(),
                kind: hawdb_plan_cypher::GraphEntityKind::Node,
            }],
            introduced: Vec::new(),
            steps: Vec::new(),
            predicate: None,
            optional: false,
        },
        input: Some(Box::new(PhysicalPlan::ProjectExec {
            items: vec![Projection {
                name: "entity".into(),
                expression: ProjectionExpression::Literal(Value::Map(BTreeMap::from([(
                    "_id".into(),
                    Value::Int(target.0 as i64),
                )]))),
            }],
            input: Box::new(input),
        })),
    };
    let observer = QueryExecutionObserver::new(&plan);
    let parameters = BTreeMap::from([("text".into(), Value::String("graph".into()))]);
    let external = TextSource {
        external_id: "small".into(),
        calls: Cell::new(0),
    };
    let point_account = ledger.account(
        hawdb_executor::QueryMemoryClass::BlockingState,
        "point read guard",
        memory.blocking_operator_bytes,
    );
    let rows = Cell::new(0);
    let label_ids = [catalog.label_id("Memory").unwrap()];
    let rel_type_id = catalog.rel_type_id("LINKS").unwrap();
    let rel_properties = BTreeMap::new();
    let window = AllocationWindow::start(size);
    let result = match case {
        PointReadCase::GraphImport => execute_binding_batches(
            &plan,
            BatchReadContext {
                catalog: &catalog,
                store: &store,
                parameters: &parameters,
                external: &external,
                memory: &memory,
                memory_ledger: &ledger,
                task_context: None,
                observer: &observer,
                host_scorer: None,
            },
            ExecutionLimit::unlimited(),
            &mut |batch| {
                rows.set(rows.get() + batch.len());
                Ok(BatchControl::Continue)
            },
        ),
        PointReadCase::OneHop => {
            hawdb_executor::traversal::visit_one_hop_relationships_with_budget(
                &store,
                hawdb_executor::traversal::OneHopRelationshipSpec {
                    source,
                    rel_type_id: Some(rel_type_id),
                    target_label_ids: Some(&label_ids),
                    rel_properties: &rel_properties,
                    relationship_scan_filter: None,
                    direction: hawdb_core::RelationshipDirection::Outgoing,
                },
                hawdb_executor::store::AdjacencyReadMemory {
                    budget_bytes: memory.blocking_operator_bytes.get(),
                    account: Some(&point_account),
                },
                &observer,
                &mut |_, _| {
                    rows.set(rows.get() + 1);
                    Ok(hawdb_executor::store::ScanControl::Continue)
                },
            )
            .map(|_| BatchControl::Continue)
        }
        PointReadCase::Bounded => hawdb_executor::traversal::visit_bounded_expand_targets(
            &store,
            hawdb_executor::traversal::BoundedExpandSpec {
                source,
                rel_type_id,
                target_label_ids: Some(&label_ids),
                min_hops: 1,
                max_hops: 1,
            },
            hawdb_executor::store::AdjacencyReadMemory {
                budget_bytes: memory.blocking_operator_bytes.get(),
                account: Some(&point_account),
            },
            None,
            &mut |_, _| {
                rows.set(rows.get() + 1);
                Ok(hawdb_executor::store::ScanControl::Continue)
            },
        )
        .map(|_| BatchControl::Continue),
    };
    assert_eq!(
        window.allocations(),
        0,
        "graph point hydration copied its payload before admission"
    );
    drop(window);
    assert!(matches!(
        result,
        Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
    ));
    assert_eq!(rows.get(), 0);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert_eq!(store.node_owned(target).unwrap().unwrap().id, target);
}

#[test]
fn point_read_admits_resident_graph_import_before_clone() {
    run_point_read_admission(false, PointReadCase::GraphImport);
}

#[test]
fn point_read_admits_persisted_graph_import_before_decode() {
    run_point_read_admission(true, PointReadCase::GraphImport);
}

#[test]
fn point_read_admits_resident_one_hop_before_clone() {
    run_point_read_admission(false, PointReadCase::OneHop);
}

#[test]
fn point_read_admits_persisted_one_hop_before_decode() {
    run_point_read_admission(true, PointReadCase::OneHop);
}

#[test]
fn point_read_admits_resident_bounded_expand_before_clone() {
    run_point_read_admission(false, PointReadCase::Bounded);
}

#[test]
fn point_read_admits_persisted_bounded_expand_before_decode() {
    run_point_read_admission(true, PointReadCase::Bounded);
}

#[test]
fn numeric_graph_features_do_not_allocate_property_text() {
    use hawdb_executor::graph_seed::numeric_property;
    use hawdb_storage::{NodeId, NodeRecord};
    let numeric_text = format!("{}1.25", "0".repeat(4096 + 137));
    let node = NodeRecord {
        id: NodeId(42),
        labels: Default::default(),
        properties: BTreeMap::from([
            ("id".into(), Value::String(numeric_text.clone())),
            ("source_id".into(), Value::Null),
            (
                "thread_id".into(),
                Value::List(vec![Value::String(numeric_text.clone())]),
            ),
            ("space_id".into(), Value::String(numeric_text)),
            (
                "nested".into(),
                Value::List(vec![Value::List(vec![Value::Int(17)])]),
            ),
            ("binary".into(), Value::Binary(vec![17; 4096])),
            (
                "map".into(),
                Value::Map(BTreeMap::from([("x".repeat(4096), Value::Int(17))])),
            ),
            (
                "list".into(),
                Value::List(vec![Value::String("0".repeat(4096)), Value::Int(17)]),
            ),
        ]),
    };
    for (property, expected) in [
        ("external_id", Some(1.25)),
        ("source_id", Some(1.25)),
        ("space_id", Some(1.25)),
        ("nested", Some(17.0)),
        ("binary", None),
        ("map", None),
        ("list", None),
        ("missing", None),
    ] {
        // usize::MAX observes every allocation on this thread, excluding the
        // fixture and assertion machinery outside the feature read itself.
        let window = AllocationWindow::start(usize::MAX);
        let actual = numeric_property(&node, property);
        let allocations = window.allocations();
        drop(window);
        assert_eq!(actual, expected, "numeric semantics for {property}");
        assert_eq!(
            allocations, 0,
            "numeric feature {property} allocated property text"
        );
    }
}

#[derive(Clone, Copy)]
enum RelationshipReadCase {
    PointRefusal,
    OrderedRefusal,
    Retained,
    PointStop,
    OrderedStop,
}

fn run_relationship_owned_read(persisted: bool, case: RelationshipReadCase) {
    use hawdb_executor::store::{
        admit_graph_read, AdjacencyReadMemory, GraphExecutionRead, ScanControl,
    };
    use hawdb_storage::adjacency::AdjacencyDirection;
    use hawdb_storage::read_view::AdmittedRelationshipRead;
    let retained = matches!(case, RelationshipReadCase::Retained);
    let size = if retained {
        12 * 1024 + 137
    } else {
        64 * 1024 + 137
    };
    let directory = persisted.then(StoredPropertyDirectory::new);
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = if let Some(directory) = &directory {
        GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap()
    } else {
        GraphStore::default()
    };
    store.create_node_table(&mut catalog, "Memory").unwrap();
    let source = store
        .create_node(&mut catalog, "Memory", BTreeMap::new())
        .unwrap();
    let target = store
        .create_node(&mut catalog, "Memory", BTreeMap::new())
        .unwrap();
    store
        .create_relationship_table(&mut catalog, "LINKS")
        .unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Relationship,
            "LINKS",
            "body",
            PropertyType::String,
            false,
        )
        .unwrap();
    let relationship_id = store
        .create_relationship(
            &mut catalog,
            source,
            target,
            "LINKS",
            BTreeMap::from([("body".into(), Value::String("X".repeat(size)))]),
        )
        .unwrap();
    if retained {
        store
            .create_relationship(
                &mut catalog,
                source,
                target,
                "LINKS",
                BTreeMap::from([("body".into(), Value::String("Y".repeat(size)))]),
            )
            .unwrap();
    }
    if let Some(directory) = &directory {
        store.checkpoint(&catalog).unwrap();
        drop(store);
        catalog = Catalog::default();
        store = GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap();
    }
    assert_eq!(store.is_out_of_core(), persisted);
    let limit = nz(if retained { 24 * 1024 } else { 8 * 1024 });
    let ledger = QueryMemoryLedger::new(limit);
    let account = ledger.account(
        hawdb_executor::QueryMemoryClass::BlockingState,
        "relationship source allocation regression",
        limit,
    );
    let stop = matches!(
        case,
        RelationshipReadCase::PointStop | RelationshipReadCase::OrderedStop
    );
    let mut admit = |bytes| {
        if stop && bytes > 0 {
            Ok(None)
        } else {
            admit_graph_read(&account, None, bytes).map(Some)
        }
    };
    let mut buffered = Vec::new();
    let window = AllocationWindow::start(size);
    let result: Result<()> = match case {
        RelationshipReadCase::PointRefusal | RelationshipReadCase::PointStop => {
            GraphExecutionRead::relationship_with_allocation(
                &store,
                relationship_id,
                &mut hawdb_storage::read_view::ControlledGraphReadAllocator::new(&mut admit, None),
            )
            .map(|read| {
                if stop {
                    assert!(matches!(read, AdmittedRelationshipRead::Stopped));
                }
            })
        }
        _ => GraphExecutionRead::visit_ordered_adjacent_relationships_with_allocation(
            &store,
            source,
            None,
            AdjacencyDirection::Outgoing,
            AdjacencyReadMemory {
                budget_bytes: limit.get(),
                account: Some(&account),
            },
            &mut hawdb_storage::read_view::ControlledGraphReadAllocator::new(&mut admit, None),
            &mut |relationship| {
                buffered.push(relationship);
                Ok(ScanControl::Continue)
            },
        )
        .map(|control| {
            if stop {
                assert_eq!(control, ScanControl::Stop);
            }
        }),
    };
    let allocations = window.allocations();
    drop(window);
    assert_eq!(allocations, usize::from(retained), "relationship payload was copied before admission or after a buffered source lost its permit");
    if stop {
        assert!(result.is_ok());
    } else {
        assert!(matches!(
            result,
            Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
        ));
    }
    assert_eq!(buffered.len(), usize::from(retained));
    if retained {
        assert!(
            ledger.snapshot().used_bytes >= size,
            "buffered owned source lost its admission lease"
        );
    }
    drop(buffered);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert!(ledger.snapshot().peak_bytes <= limit.get());
    let recovered = store.relationship_owned(relationship_id).unwrap().unwrap();
    assert!(
        matches!(recovered.properties.get("body"), Some(Value::String(body)) if body.len() == size)
    );
}

macro_rules! relationship_read_guard {
    ($name:ident, $persisted:expr, $case:ident) => {
        #[test]
        fn $name() {
            run_relationship_owned_read($persisted, RelationshipReadCase::$case);
        }
    };
}
relationship_read_guard!(
    relationship_owned_read_live_point_refuses_before_copy,
    false,
    PointRefusal
);
relationship_read_guard!(
    relationship_owned_read_canonical_point_refuses_before_copy,
    true,
    PointRefusal
);
relationship_read_guard!(
    relationship_owned_read_live_ordered_refuses_before_copy,
    false,
    OrderedRefusal
);
relationship_read_guard!(
    relationship_owned_read_canonical_ordered_refuses_before_copy,
    true,
    OrderedRefusal
);
relationship_read_guard!(
    relationship_owned_read_live_retains_buffered_source_permits,
    false,
    Retained
);
relationship_read_guard!(
    relationship_owned_read_canonical_retains_buffered_source_permits,
    true,
    Retained
);
relationship_read_guard!(
    relationship_owned_read_live_point_stops_before_copy,
    false,
    PointStop
);
relationship_read_guard!(
    relationship_owned_read_canonical_point_stops_before_copy,
    true,
    PointStop
);
relationship_read_guard!(
    relationship_owned_read_live_ordered_stops_before_copy,
    false,
    OrderedStop
);
relationship_read_guard!(
    relationship_owned_read_canonical_ordered_stops_before_copy,
    true,
    OrderedStop
);

#[test]
fn relationship_owned_read_typed_window_streams_without_degree_sized_key_buffer() {
    use hawdb_executor::store::{
        admit_graph_read, AdjacencyReadMemory, GraphExecutionRead, ScanControl,
    };
    use hawdb_storage::adjacency::AdjacencyDirection;
    for persisted in [false, true] {
        let directory = persisted.then(StoredPropertyDirectory::new);
        let replay = hawdb_storage::store::WalReplayConfig {
            residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
            ..Default::default()
        };
        let mut catalog = Catalog::default();
        let mut store = match &directory {
            Some(directory) => GraphStore::open_with_durability_and_replay_config(
                &directory.0,
                &mut catalog,
                Default::default(),
                replay,
            )
            .unwrap(),
            None => GraphStore::default(),
        };
        store.create_node_table(&mut catalog, "Memory").unwrap();
        store
            .create_relationship_table(&mut catalog, "LINKS")
            .unwrap();
        let source = store
            .create_node(&mut catalog, "Memory", BTreeMap::new())
            .unwrap();
        let target = store
            .create_node(&mut catalog, "Memory", BTreeMap::new())
            .unwrap();
        let mut expected = Vec::new();
        for index in 0..1000 {
            let id = store
                .create_relationship(&mut catalog, source, target, "LINKS", BTreeMap::new())
                .unwrap();
            if index < 2 {
                expected.push(id);
            }
        }
        if let Some(directory) = &directory {
            store.checkpoint(&catalog).unwrap();
            drop(store);
            catalog = Catalog::default();
            store = GraphStore::open_with_durability_and_replay_config(
                &directory.0,
                &mut catalog,
                Default::default(),
                replay,
            )
            .unwrap();
            // A live tail must merge after the pinned canonical posting list.
            store
                .create_relationship(&mut catalog, source, target, "LINKS", BTreeMap::new())
                .unwrap();
        }
        let ledger = QueryMemoryLedger::new(nz(4096));
        let account = ledger.account(
            hawdb_executor::QueryMemoryClass::BlockingState,
            "typed adjacency window",
            nz(4096),
        );
        let mut selected = Vec::new();
        let control = GraphExecutionRead::visit_ordered_adjacent_relationships_with_allocation(
            &store,
            source,
            Some(catalog.rel_type_id("LINKS").unwrap()),
            AdjacencyDirection::Outgoing,
            AdjacencyReadMemory {
                budget_bytes: 0,
                account: Some(&account),
            },
            &mut hawdb_storage::read_view::ControlledGraphReadAllocator::new(
                &mut |bytes| admit_graph_read(&account, None, bytes).map(Some),
                None,
            ),
            &mut |relationship| {
                selected.push(relationship);
                Ok(if selected.len() == 2 {
                    ScanControl::Stop
                } else {
                    ScanControl::Continue
                })
            },
        )
        .unwrap();
        assert_eq!(control, ScanControl::Stop);
        assert!(ledger.snapshot().used_bytes > 0);
        let actual: Vec<_> = selected
            .into_iter()
            .map(|row| row.into_parts().0.id)
            .collect();
        assert_eq!(actual, expected);
        assert_eq!(ledger.snapshot().used_bytes, 0);
    }
}

#[test]
fn relationship_owned_read_filtered_ordering_admits_before_native_or_indexed_copy() {
    use hawdb_executor::store::{AdjacencyReadMemory, GraphExecutionRead, ScanControl};
    use hawdb_storage::adjacency::AdjacencyDirection;
    use hawdb_storage::mutation::PropertyFilter;
    let size = 64 * 1024 + 151;
    for persisted in [false, true] {
        let directory = persisted.then(StoredPropertyDirectory::new);
        let replay = hawdb_storage::store::WalReplayConfig {
            residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
            ..Default::default()
        };
        let mut catalog = Catalog::default();
        let mut store = match &directory {
            Some(directory) => GraphStore::open_with_durability_and_replay_config(
                &directory.0,
                &mut catalog,
                Default::default(),
                replay,
            )
            .unwrap(),
            None => GraphStore::default(),
        };
        store.create_node_table(&mut catalog, "Memory").unwrap();
        store
            .create_relationship_table(&mut catalog, "LINKS")
            .unwrap();
        for property in ["body", "status"] {
            store
                .create_property_descriptor(
                    &mut catalog,
                    TableKind::Relationship,
                    "LINKS",
                    property,
                    PropertyType::String,
                    false,
                )
                .unwrap();
        }
        let source = store
            .create_node(&mut catalog, "Memory", BTreeMap::new())
            .unwrap();
        let target = store
            .create_node(&mut catalog, "Memory", BTreeMap::new())
            .unwrap();
        let selected = store
            .create_relationship(
                &mut catalog,
                source,
                target,
                "LINKS",
                BTreeMap::from([
                    ("body".into(), Value::String("X".repeat(size))),
                    ("status".into(), Value::String("active".into())),
                ]),
            )
            .unwrap();
        store
            .create_relationship(
                &mut catalog,
                source,
                target,
                "LINKS",
                BTreeMap::from([
                    ("body".into(), Value::String(String::new())),
                    ("status".into(), Value::String("inactive".into())),
                ]),
            )
            .unwrap();
        if let Some(directory) = &directory {
            store.checkpoint(&catalog).unwrap();
            drop(store);
            catalog = Catalog::default();
            store = GraphStore::open_with_durability_and_replay_config(
                &directory.0,
                &mut catalog,
                Default::default(),
                replay,
            )
            .unwrap();
        }
        let limit = nz(8 * 1024);
        let ledger = QueryMemoryLedger::new(limit);
        let account = ledger.account(
            hawdb_executor::QueryMemoryClass::BlockingState,
            "filtered relationship source",
            limit,
        );
        let filter = PropertyFilter::Eq {
            property: "status".into(),
            value: Value::String("active".into()),
        };
        let window = AllocationWindow::start(size);
        let result = GraphExecutionRead::visit_ordered_adjacent_relationships_with_filter_owned(
            &store,
            source,
            Some(catalog.rel_type_id("LINKS").unwrap()),
            AdjacencyDirection::Outgoing,
            &filter,
            AdjacencyReadMemory {
                budget_bytes: limit.get(),
                account: Some(&account),
            },
            &mut |_| panic!("oversized relationship must be refused before consumer"),
        );
        let copies = window.allocations();
        drop(window);
        assert_eq!(copies, 0, "filtered source copied before query admission");
        assert!(matches!(
            result,
            Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
        ));
        assert_eq!(ledger.snapshot().used_bytes, 0);
        // Refusal must preserve the canonical reader's health and index mapping.
        assert_eq!(
            store.relationship_owned(selected).unwrap().unwrap().id,
            selected
        );
        let generous = QueryMemoryLedger::new(nz(256 * 1024));
        let account = generous.account(
            hawdb_executor::QueryMemoryClass::BlockingState,
            "filtered recovery",
            nz(256 * 1024),
        );
        let mut ids = Vec::new();
        let (control, report) =
            GraphExecutionRead::visit_ordered_adjacent_relationships_with_filter_owned(
                &store,
                source,
                Some(catalog.rel_type_id("LINKS").unwrap()),
                AdjacencyDirection::Outgoing,
                &filter,
                AdjacencyReadMemory {
                    budget_bytes: 256 * 1024,
                    account: Some(&account),
                },
                &mut |row| {
                    ids.push(row.id);
                    Ok(ScanControl::Continue)
                },
            )
            .unwrap();
        assert_eq!(control, ScanControl::Continue);
        assert_eq!(ids, vec![selected]);
        if persisted {
            assert!(report.is_some_and(|report| report.pruned));
        }
        assert_eq!(generous.snapshot().used_bytes, 0);
    }
}

#[test]
fn full_node_index_source_admits_native_and_cold_payload_before_clone() {
    for persisted in [false, true] {
        run_full_node_source_admission(persisted, false, FullNodeSource::Index);
    }
}

#[test]
fn full_node_lookup_sources_admit_native_and_cold_payload_before_clone() {
    for persisted in [false, true] {
        for indexed in [false, true] {
            run_full_node_source_admission(persisted, true, FullNodeSource::Lookup(indexed));
        }
    }
}

#[test]
fn full_node_index_key_copy_admitted_before_parameter_clone() {
    for persisted in [false, true] {
        run_full_node_source_admission(persisted, false, FullNodeSource::IndexKey);
    }
}
#[test]
fn full_node_lookup_key_copy_admitted_before_parameter_clone() {
    for persisted in [false, true] {
        run_full_node_source_admission(persisted, true, FullNodeSource::LookupKey);
    }
}

fn run_count_sum_source_admission(persisted: bool) {
    run_blocking_source_admission(persisted, BlockingSource::CountSum);
}

#[derive(Clone, Copy, Debug)]
enum BlockingSource {
    CountSum,
    ThreadRepair,
    ThreadCountTarget,
    ShortestPath,
    GraphAlgorithm,
    ThreadKey,
    ThreadState,
    ShortestOutput,
    GraphRelationship,
}

fn run_blocking_source_admission(persisted: bool, source: BlockingSource) {
    let size = 1024 * 1024 + 137;
    let directory = persisted.then(StoredPropertyDirectory::new);
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = if let Some(directory) = &directory {
        GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap()
    } else {
        GraphStore::default()
    };
    for label in ["Memory", "Other", "Identity", "Message"] {
        store.create_node_table(&mut catalog, label).unwrap();
        store
            .create_property_descriptor(
                &mut catalog,
                TableKind::Node,
                label,
                "body",
                PropertyType::String,
                false,
            )
            .unwrap();
    }
    let (label, property) = match source {
        BlockingSource::ThreadCountTarget => ("Message", "body"),
        BlockingSource::ThreadKey => ("Identity", "ref"),
        BlockingSource::ThreadState => ("Memory", "thread_id"),
        _ => ("Other", "body"),
    };
    if property != "body" {
        store
            .create_property_descriptor(
                &mut catalog,
                TableKind::Node,
                label,
                property,
                PropertyType::String,
                false,
            )
            .unwrap();
    }
    let id = store
        .create_node(
            &mut catalog,
            label,
            BTreeMap::from([
                ("body".into(), Value::String(String::new())),
                (
                    property.into(),
                    Value::String(if matches!(source, BlockingSource::GraphRelationship) {
                        String::new()
                    } else {
                        "X".repeat(size)
                    }),
                ),
            ]),
        )
        .unwrap();
    if matches!(source, BlockingSource::ThreadCountTarget) {
        for property in ["id", "thread_id"] {
            store
                .create_property_descriptor(
                    &mut catalog,
                    TableKind::Node,
                    "Memory",
                    property,
                    PropertyType::Int,
                    false,
                )
                .unwrap();
        }
        let thread = store
            .create_node(
                &mut catalog,
                "Memory",
                BTreeMap::from([
                    ("body".into(), Value::String(String::new())),
                    ("id".into(), Value::Int(7)),
                    ("thread_id".into(), Value::Int(42)),
                ]),
            )
            .unwrap();
        store
            .create_relationship(&mut catalog, thread, id, "MESSAGE", BTreeMap::new())
            .unwrap();
    }
    if matches!(
        source,
        BlockingSource::ShortestOutput | BlockingSource::GraphRelationship
    ) {
        store
            .create_property_descriptor(
                &mut catalog,
                TableKind::Node,
                "Memory",
                "id",
                PropertyType::Int,
                false,
            )
            .unwrap();
        let first = store
            .create_node(
                &mut catalog,
                "Memory",
                BTreeMap::from([
                    ("body".into(), Value::String(String::new())),
                    ("id".into(), Value::Int(1)),
                ]),
            )
            .unwrap();
        let last = store
            .create_node(
                &mut catalog,
                "Memory",
                BTreeMap::from([
                    ("body".into(), Value::String(String::new())),
                    ("id".into(), Value::Int(2)),
                ]),
            )
            .unwrap();
        store
            .create_relationship_table(&mut catalog, "LINK")
            .unwrap();
        if matches!(source, BlockingSource::GraphRelationship) {
            store
                .create_property_descriptor(
                    &mut catalog,
                    TableKind::Relationship,
                    "LINK",
                    "body",
                    PropertyType::String,
                    false,
                )
                .unwrap();
            store
                .create_relationship(
                    &mut catalog,
                    first,
                    last,
                    "LINK",
                    BTreeMap::from([("body".into(), Value::String("X".repeat(size)))]),
                )
                .unwrap();
        } else {
            store
                .create_relationship(&mut catalog, first, id, "LINK", BTreeMap::new())
                .unwrap();
            store
                .create_relationship(&mut catalog, id, last, "LINK", BTreeMap::new())
                .unwrap();
        }
    }
    if let Some(directory) = &directory {
        store.checkpoint(&catalog).unwrap();
        drop(store);
        catalog = Catalog::default();
        store = GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap();
    }
    assert_eq!(store.is_out_of_core(), persisted);
    let plan = match source {
        BlockingSource::CountSum => PhysicalPlan::OptionalRelationshipCountSumExec {
            variable: "n".into(),
            label: "Memory".into(),
            properties: BTreeMap::new(),
            legs: vec![hawdb_plan_cypher::RelationshipCountLeg {
                rel_type: "RELATES_TO".into(),
                direction: hawdb_core::RelationshipDirection::Outgoing,
                distinct: true,
                filter: None,
            }],
            output: "count".into(),
        },
        BlockingSource::ThreadRepair
        | BlockingSource::ThreadCountTarget
        | BlockingSource::ThreadKey
        | BlockingSource::ThreadState => PhysicalPlan::ThreadRepairStatsExec {
            label: "Memory".into(),
            identity_label: "Identity".into(),
            identity_ref_property: "ref".into(),
            thread_id_property: "thread_id".into(),
            message_rel_type: "MESSAGE".into(),
            message_label: "Message".into(),
            memory_rel_type: "MEMORY".into(),
            memory_label: "Memory".into(),
        },
        BlockingSource::ShortestPath | BlockingSource::ShortestOutput => {
            PhysicalPlan::ShortestPathExec {
                source_variable: "s".into(),
                source_label: if matches!(source, BlockingSource::ShortestOutput) {
                    "Memory".into()
                } else {
                    String::new()
                },
                source_id: Value::Int(1),
                source_visibility_predicate: None,
                rel_type: String::new(),
                direction: hawdb_core::RelationshipDirection::Outgoing,
                target_variable: "t".into(),
                target_label: if matches!(source, BlockingSource::ShortestOutput) {
                    "Memory".into()
                } else {
                    String::new()
                },
                target_id: Value::Int(2),
                target_visibility_predicate: None,
                min_hops: 1,
                max_hops: 3,
                returns: if matches!(source, BlockingSource::ShortestOutput) {
                    vec![hawdb_plan_cypher::ShortestPathProjection {
                        name: "bodies".into(),
                        expression:
                            hawdb_plan_cypher::ShortestPathProjectionExpression::NodePropertyList {
                                property: "body".into(),
                            },
                    }]
                } else {
                    vec![]
                },
            }
        }
        BlockingSource::GraphAlgorithm | BlockingSource::GraphRelationship => {
            store
                .register_projected_graph(
                    "graph",
                    hawdb_storage::projection::ProjectedGraphDefinition {
                        node_labels: vec!["Memory".into()],
                        rel_types: vec![],
                        relationship_predicates: BTreeMap::new(),
                    },
                )
                .unwrap();
            PhysicalPlan::GraphAlgorithm {
                algorithm: hawdb_plan_cypher::GraphAlgorithmKind::PageRank,
                graph_name: "graph".into(),
                options: Default::default(),
                score_column: "score".into(),
                return_node_identity: false,
                node_visibility_predicate: Some(Predicate::PropertyEq {
                    variable: "n".into(),
                    property: "visible".into(),
                    value: Value::Bool(true),
                }),
            }
        }
    };
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(
            if matches!(
                source,
                BlockingSource::ThreadKey | BlockingSource::ThreadState
            ) {
                16 * 1024 * 1024
            } else {
                64 * 1024
            },
        ),
        blocking_operator_bytes: nz(
            if matches!(
                source,
                BlockingSource::ThreadKey | BlockingSource::ThreadState
            ) {
                size - 1
            } else {
                4096
            },
        ),
        batch_payload_bytes: nz(4096),
        batch_rows: nz(8192),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    let rows = Cell::new(0);
    let window = AllocationWindow::start(size);
    let result = execute_binding_batches(
        &plan,
        BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
            host_scorer: None,
        },
        ExecutionLimit::unlimited(),
        &mut |batch| {
            if matches!(source, BlockingSource::ThreadCountTarget) {
                assert_eq!(batch.len(), 1);
                assert_eq!(batch[0].values["legacy_messages"], Value::Int(1));
            }
            rows.set(rows.get() + batch.len());
            Ok(BatchControl::Continue)
        },
    );
    assert_eq!(
        window.allocations(),
        usize::from(matches!(
            source,
            BlockingSource::ThreadKey | BlockingSource::ThreadState
        )),
        "{source:?} copied an unused node payload before source admission"
    );
    drop(window);
    if matches!(
        source,
        BlockingSource::ThreadRepair | BlockingSource::ThreadCountTarget
    ) {
        assert!(
            result.is_ok(),
            "unused unrelated payload must not be admitted: {result:?}"
        );
    } else {
        assert!(matches!(
            result,
            Err(HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_))
        ));
    }
    assert_eq!(
        rows.get(),
        usize::from(matches!(source, BlockingSource::ThreadCountTarget))
    );
    assert_eq!(external.calls.get(), 0);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert_eq!(store.node_owned(id).unwrap().unwrap().id, id);
}

#[test]
fn count_sum_admits_native_unused_payload_before_copy() {
    run_count_sum_source_admission(false);
}

#[test]
fn count_sum_admits_cold_unused_payload_before_decode() {
    run_count_sum_source_admission(true);
}

#[test]
fn thread_repair_ignores_native_unused_payload_before_copy() {
    run_blocking_source_admission(false, BlockingSource::ThreadRepair);
}

#[test]
fn thread_repair_ignores_cold_unused_payload_before_copy() {
    run_blocking_source_admission(true, BlockingSource::ThreadRepair);
}

#[test]
fn shortest_path_admits_native_endpoint_payload_before_copy() {
    run_blocking_source_admission(false, BlockingSource::ShortestPath);
}

#[test]
fn shortest_path_admits_cold_endpoint_payload_before_copy() {
    run_blocking_source_admission(true, BlockingSource::ShortestPath);
}

#[test]
fn graph_algorithm_admits_native_visibility_payload_before_copy() {
    run_blocking_source_admission(false, BlockingSource::GraphAlgorithm);
}

#[test]
fn graph_algorithm_admits_cold_visibility_payload_before_copy() {
    run_blocking_source_admission(true, BlockingSource::GraphAlgorithm);
}

#[test]
fn thread_identity_key_admits_native_retained_clone_before_copy() {
    run_blocking_source_admission(false, BlockingSource::ThreadKey);
}

#[test]
fn thread_identity_key_admits_cold_retained_clone_before_copy() {
    run_blocking_source_admission(true, BlockingSource::ThreadKey);
}

#[test]
fn thread_retained_fields_admits_native_retained_clone_before_copy() {
    run_blocking_source_admission(false, BlockingSource::ThreadState);
}

#[test]
fn thread_retained_fields_admits_cold_retained_clone_before_copy() {
    run_blocking_source_admission(true, BlockingSource::ThreadState);
}

#[test]
fn shortest_output_property_admits_native_payload_before_copy() {
    run_blocking_source_admission(false, BlockingSource::ShortestOutput);
}

#[test]
fn shortest_output_property_admits_cold_payload_before_copy() {
    run_blocking_source_admission(true, BlockingSource::ShortestOutput);
}

#[test]
fn graph_relationship_scan_admits_native_payload_before_copy() {
    run_blocking_source_admission(false, BlockingSource::GraphRelationship);
}

#[test]
fn graph_relationship_scan_admits_cold_payload_before_copy() {
    run_blocking_source_admission(true, BlockingSource::GraphRelationship);
}

#[derive(Clone, Copy, Debug)]
enum ExpandOutputOwner {
    OneHop,
    Typed,
    ZeroHop,
    Optional,
}

fn run_expand_output_admission(persisted: bool, owner: ExpandOutputOwner) {
    let size = 1024 * 1024 + 173;
    let directory = persisted.then(StoredPropertyDirectory::new);
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = if let Some(directory) = &directory {
        GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap()
    } else {
        GraphStore::default()
    };
    for label in ["Source", "Target"] {
        store.create_node_table(&mut catalog, label).unwrap();
        store
            .create_property_descriptor(
                &mut catalog,
                TableKind::Node,
                label,
                "body",
                PropertyType::String,
                false,
            )
            .unwrap();
    }
    let source = store
        .create_node(
            &mut catalog,
            "Source",
            BTreeMap::from([(
                "body".into(),
                Value::String(if matches!(owner, ExpandOutputOwner::OneHop) {
                    String::new()
                } else {
                    "S".repeat(size)
                }),
            )]),
        )
        .unwrap();
    let target = store
        .create_node(
            &mut catalog,
            "Target",
            BTreeMap::from([(
                "body".into(),
                Value::String(if matches!(owner, ExpandOutputOwner::OneHop) {
                    "T".repeat(size)
                } else if matches!(owner, ExpandOutputOwner::Typed) {
                    "t".repeat(32768)
                } else {
                    String::new()
                }),
            )]),
        )
        .unwrap();
    store
        .create_relationship_table(&mut catalog, "LINK")
        .unwrap();
    store
        .create_relationship(&mut catalog, source, target, "LINK", BTreeMap::new())
        .unwrap();
    if let Some(directory) = &directory {
        store.checkpoint(&catalog).unwrap();
        drop(store);
        catalog = Catalog::default();
        store = GraphStore::open_with_durability_and_replay_config(
            &directory.0,
            &mut catalog,
            Default::default(),
            replay,
        )
        .unwrap();
    }
    assert_eq!(store.is_out_of_core(), persisted);
    let input = PhysicalPlan::SeqNodeScan {
        variable: "n".into(),
        label: "Source".into(),
    };
    let parameters = BTreeMap::new();
    let external = TextSource {
        external_id: String::new(),
        calls: Cell::new(0),
    };
    let mut memory = ExecutionMemoryConfig {
        query_memory_bytes: nz(32 * 1024 * 1024),
        blocking_operator_bytes: nz(8 * 1024 * 1024),
        batch_payload_bytes: nz(8 * 1024 * 1024),
        batch_rows: nz(1),
        ..Default::default()
    };
    // Observe the real source-only peak outside the allocation window. The
    // typed/optional root permits the complete input, but no larger output.
    let input_ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let input_observer = QueryExecutionObserver::new(&input);
    execute_binding_batches(
        &input,
        BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &input_ledger,
            task_context: None,
            observer: &input_observer,
            host_scorer: None,
        },
        ExecutionLimit::unlimited(),
        &mut |_| Ok(BatchControl::Continue),
    )
    .unwrap();
    let input_peak = input_ledger.snapshot().peak_bytes;
    assert_eq!(input_ledger.snapshot().used_bytes, 0);
    match owner {
        ExpandOutputOwner::OneHop => memory.batch_payload_bytes = nz(4096),
        ExpandOutputOwner::Typed => memory.query_memory_bytes = nz(input_peak + 65536),
        ExpandOutputOwner::Optional => memory.query_memory_bytes = nz(input_peak),
        ExpandOutputOwner::ZeroHop => memory.query_memory_bytes = nz(input_peak + size + 8192),
    }
    let plan = PhysicalPlan::AdjacencyExpandExec {
        source_variable: "n".into(),
        source_label: "Source".into(),
        rel_variable: matches!(owner, ExpandOutputOwner::OneHop).then(|| "r".into()),
        rel_type: if matches!(
            owner,
            ExpandOutputOwner::ZeroHop | ExpandOutputOwner::Optional
        ) {
            "Unknown".into()
        } else {
            "LINK".into()
        },
        rel_properties: BTreeMap::new(),
        direction: hawdb_core::RelationshipDirection::Outgoing,
        target_variable: "m".into(),
        target_label: if matches!(owner, ExpandOutputOwner::ZeroHop) {
            "Source".into()
        } else {
            "Target".into()
        },
        min_hops: usize::from(!matches!(owner, ExpandOutputOwner::ZeroHop)),
        max_hops: if matches!(owner, ExpandOutputOwner::Typed) {
            2
        } else {
            1
        },
        optional: matches!(owner, ExpandOutputOwner::Optional),
        graph_budget: None,
        input: Box::new(input),
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let observer = QueryExecutionObserver::new(&plan);
    let emitted = Cell::new(0);
    let window = AllocationWindow::start(size);
    let result = execute_binding_batches(
        &plan,
        BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
            host_scorer: None,
        },
        ExecutionLimit::unlimited(),
        &mut |rows| {
            emitted.set(emitted.get() + rows.len());
            Ok(BatchControl::Continue)
        },
    );
    let copies = window.allocations();
    drop(window);
    let error = result.expect_err("complete derived row must be refused");
    assert!(
        matches!(
            error,
            HawDBError::Execution(_) | HawDBError::ResourceBudgetExceeded(_)
        ),
        "{error}"
    );
    assert!(
        error
            .to_string()
            .contains(if matches!(owner, ExpandOutputOwner::OneHop) {
                "batch_payload_bytes"
            } else {
                "charging AdjacencyExpandExec output"
            }),
        "wrong refusal boundary: {error}"
    );
    assert_eq!(emitted.get(), 0);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert_eq!(copies, if matches!(owner, ExpandOutputOwner::ZeroHop) { 2 } else { 1 },
        "{owner:?}, cold={persisted}: only admitted input/source copies may occur; no retained output copy before refusal");
    assert_eq!(store.node_owned(source).unwrap().unwrap().id, source);

    // Recover through the same actual query owner with sufficient budgets.
    // The lawful point record is moved; only retained input aliases are cloned.
    memory.query_memory_bytes = nz(32 * 1024 * 1024);
    memory.batch_payload_bytes = nz(8 * 1024 * 1024);
    let admitted_ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let admitted_observer = QueryExecutionObserver::new(&plan);
    let admitted_rows = Cell::new(0);
    let window = AllocationWindow::start(size);
    let result = execute_binding_batches(
        &plan,
        BatchReadContext {
            catalog: &catalog,
            store: &store,
            parameters: &parameters,
            external: &external,
            memory: &memory,
            memory_ledger: &admitted_ledger,
            task_context: None,
            observer: &admitted_observer,
            host_scorer: None,
        },
        ExecutionLimit::unlimited(),
        &mut |rows| {
            admitted_rows.set(admitted_rows.get() + rows.len());
            for row in rows {
                assert_eq!(row.nodes["n"].id, source);
                if matches!(owner, ExpandOutputOwner::Optional) {
                    assert_eq!(row.values["m"], Value::Null);
                } else {
                    assert_eq!(
                        row.nodes["m"].id,
                        if matches!(owner, ExpandOutputOwner::ZeroHop) {
                            source
                        } else {
                            target
                        }
                    );
                }
            }
            Ok(BatchControl::Continue)
        },
    );
    let copies = window.allocations();
    drop(window);
    result.unwrap();
    assert_eq!(admitted_rows.get(), 1);
    assert_eq!(admitted_ledger.snapshot().used_bytes, 0);
    assert_eq!(
        copies,
        match owner {
            ExpandOutputOwner::OneHop => 1,
            ExpandOutputOwner::ZeroHop => 3,
            ExpandOutputOwner::Typed | ExpandOutputOwner::Optional => 2,
        },
        "{owner:?}, cold={persisted}: recovery copied an overwritten or moved record"
    );
}

#[test]
fn expand_one_hop_native_output_admitted_before_copy() {
    run_expand_output_admission(false, ExpandOutputOwner::OneHop);
}
#[test]
fn expand_one_hop_cold_output_admitted_before_copy() {
    run_expand_output_admission(true, ExpandOutputOwner::OneHop);
}
#[test]
fn expand_typed_native_output_admitted_before_copy() {
    run_expand_output_admission(false, ExpandOutputOwner::Typed);
}
#[test]
fn expand_typed_cold_output_admitted_before_copy() {
    run_expand_output_admission(true, ExpandOutputOwner::Typed);
}
#[test]
fn expand_zero_hop_native_output_admitted_before_copy() {
    run_expand_output_admission(false, ExpandOutputOwner::ZeroHop);
}
#[test]
fn expand_zero_hop_cold_output_admitted_before_copy() {
    run_expand_output_admission(true, ExpandOutputOwner::ZeroHop);
}
#[test]
fn expand_optional_native_output_admitted_before_copy() {
    run_expand_output_admission(false, ExpandOutputOwner::Optional);
}
#[test]
fn expand_optional_cold_output_admitted_before_copy() {
    run_expand_output_admission(true, ExpandOutputOwner::Optional);
}

#[test]
fn owned_record_bounds_cover_actual_container_and_nested_value_allocations() {
    use hawdb_core::ids::{
        node_allocation_bytes, project_node_record_ref, projected_node_allocation_bytes,
        relationship_allocation_bytes,
    };
    for count in [0, 1, 2, 5, 11, 12, 23, 32, 128] {
        let properties: BTreeMap<_, _> = (0..count)
            .map(|id| {
                (
                    format!("property-{id}"),
                    Value::Map(BTreeMap::from([(
                        "nested".into(),
                        Value::List(vec![
                            Value::String("payload".repeat(3)),
                            Value::Binary(vec![7; 19]),
                            Value::Vector(hawdb_core::Vector::new(&[1.0; 1024]).unwrap()),
                            Value::Int(id),
                        ]),
                    )])),
                )
            })
            .collect();
        let node = hawdb_core::ids::NodeRecord {
            id: hawdb_core::ids::NodeId(7),
            labels: BTreeSet::from([hawdb_core::LabelId(2)]),
            properties,
        };
        let bound = node_allocation_bytes(&node);
        let window = AllocationWindow::start(usize::MAX);
        let copy = node.clone();
        let allocated = WATCH_BYTES.with(Cell::get);
        drop(window);
        assert!(
            allocated <= bound,
            "full {count}: allocated {allocated} > bound {bound}"
        );
        assert_eq!(copy, node);
        for selected in [count / 2, count] {
            let properties = node
                .properties
                .keys()
                .take(selected as usize)
                .cloned()
                .collect();
            let bound = projected_node_allocation_bytes(&node, &properties);
            let window = AllocationWindow::start(usize::MAX);
            let copy = project_node_record_ref(&node, &properties);
            let allocated = WATCH_BYTES.with(Cell::get);
            drop(window);
            assert!(
                allocated <= bound,
                "projected {selected}/{count}: allocated {allocated} > bound {bound}"
            );
            assert_eq!(copy.properties.len(), selected as usize);
        }
        let relationship = hawdb_core::ids::RelRecord {
            id: hawdb_core::ids::RelId(1),
            source: node.id,
            target: node.id,
            rel_type: hawdb_core::RelTypeId(1),
            properties: node.properties.clone(),
        };
        let bound = relationship_allocation_bytes(&relationship);
        let window = AllocationWindow::start(usize::MAX);
        let copy = relationship.clone();
        let allocated = WATCH_BYTES.with(Cell::get);
        drop(window);
        assert!(
            allocated <= bound,
            "relationship {count}: allocated {allocated} > bound {bound}"
        );
        assert_eq!(copy, relationship);
    }
}

#[test]
fn thread_repair_counts_native_targets_without_copying_unused_payloads() {
    run_blocking_source_admission(false, BlockingSource::ThreadCountTarget);
}

#[test]
fn thread_repair_counts_cold_targets_without_copying_unused_payloads() {
    run_blocking_source_admission(true, BlockingSource::ThreadCountTarget);
}

#[test]
fn property_union_key_admission_keeps_duplicate_charge_and_refuses_second_id() {
    use hawdb_executor::store::admit_graph_read;
    use hawdb_storage::read_view::{AdmittedKeySet, GraphReadAdmission};

    let limit = nz(960);
    let ledger = QueryMemoryLedger::new(limit);
    let source_account = ledger.account(
        hawdb_executor::QueryMemoryClass::ExternalRead,
        "property union source and descriptor",
        limit,
    );
    let account = ledger.account(
        hawdb_executor::QueryMemoryClass::BlockingState,
        "property union key admission",
        limit,
    );
    // Source and access-descriptor grants remain live while union keys grow.
    // The remaining root budget admits one key, but cannot admit a second.
    let source = admit_graph_read(&source_account, None, 512).unwrap();
    let descriptor = admit_graph_read(&source_account, None, 128).unwrap();
    let mut allocate = |bytes| admit_graph_read(&account, None, bytes);
    let admission = GraphReadAdmission::new(&mut allocate);
    let mut seen = AdmittedKeySet::new(&admission).unwrap();
    let first = hawdb_core::ids::NodeId(7);
    let second = hawdb_core::ids::NodeId(8);
    assert!(seen.try_insert(first).unwrap());
    let charged = ledger.snapshot();
    assert!(charged.used_bytes > source.bytes() + descriptor.bytes());
    assert!(charged.used_bytes <= limit.get());
    assert!(!seen.try_insert(first).unwrap());
    assert_eq!(
        ledger.snapshot(),
        charged,
        "duplicates must not grow the grant"
    );
    let error = seen.try_insert(second).unwrap_err();
    assert!(error.to_string().contains("query memory ledger"), "{error}");
    assert_eq!(seen.len(), 1);
    assert!(seen.contains(&first));
    assert!(!seen.contains(&second), "refusal must precede insertion");
    assert_eq!(ledger.snapshot(), charged, "refusal must not leak a grant");
    drop(seen);
    drop((source, descriptor));
    assert_eq!(ledger.snapshot().used_bytes, 0);
}

fn run_union_retained_cap(projected: bool) {
    use hawdb_executor::scan::{NodeProjectionScanSpec, NodeScanContext};
    use hawdb_plan_cypher::ExactPropertySeekBranch;
    let mut catalog = Catalog::default();
    let mut store = GraphStore::default();
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "id",
            PropertyType::Int,
            false,
        )
        .unwrap();
    store
        .create_property_index(&mut catalog, "Memory", "id")
        .unwrap();
    let first = store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([("id".into(), Value::Int(1))]),
        )
        .unwrap();
    let second = store
        .create_node(
            &mut catalog,
            "Memory",
            BTreeMap::from([("id".into(), Value::Int(2))]),
        )
        .unwrap();
    let branches = [1, 1, 2].map(|id| ExactPropertySeekBranch {
        property: "id".into(),
        values: vec![Value::Int(id)],
    });
    for cap in [360, 4096] {
        let ledger = QueryMemoryLedger::new(nz(128 * 1024));
        let source = ledger.account(
            hawdb_executor::QueryMemoryClass::ExternalRead,
            "union source",
            nz(32 * 1024),
        );
        let output = ledger.account(
            hawdb_executor::QueryMemoryClass::PipelineBatch,
            "union output",
            nz(16 * 1024),
        );
        let context = NodeScanContext {
            catalog: &catalog,
            store: &store,
            execution_limit: ExecutionLimit::unlimited(),
            memory_budget: nz(cap),
            memory_account: &source,
            batch_memory_budget: nz(16 * 1024),
            batch_memory_account: &output,
            batch_rows: 1,
            task_context: None,
        };
        let mut ids = Vec::new();
        let mut emit = |batch: BindingBatch| {
            ids.extend(batch.into_iter().map(|binding| binding.nodes["n"].id));
            Ok(BatchControl::Continue)
        };
        let result = if projected {
            hawdb_executor::scan::stream_node_projection_scan_batches(
                NodeProjectionScanSpec {
                    variable: "n",
                    label: "Memory",
                    access: &NodeProjectionAccess::PropertyUnion {
                        branches: branches.to_vec(),
                    },
                    required_properties: &["id".into()],
                    predicate: None,
                    items: &[],
                },
                context,
                &QueryExecutionObserver::default(),
                &mut emit,
            )
        } else {
            hawdb_executor::scan::stream_index_node_union_seek_batches(
                "n",
                "Memory",
                &branches,
                context,
                &QueryExecutionObserver::default(),
                &mut emit,
            )
        };
        if cap == 360 {
            let error = result.unwrap_err();
            assert!(error.to_string().contains("360-byte budget"), "{error}");
            assert_eq!(
                ids,
                vec![first],
                "duplicate must be free and second unique key refused before output"
            );
        } else {
            result.unwrap();
            assert_eq!(ids, vec![first, second]);
        }
        assert_eq!(ledger.snapshot().used_bytes, 0);
    }
}

#[test]
fn source_memory_full_union_retains_its_own_dedup_cap() {
    run_union_retained_cap(false);
}

#[test]
fn source_memory_projected_union_retains_its_own_dedup_cap() {
    run_union_retained_cap(true);
}

fn run_cold_source_retained_keys(projected: bool) {
    use hawdb_executor::scan::{NodeProjectionScanSpec, NodeScanContext};
    let directory = StoredPropertyDirectory::new();
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = GraphStore::open_with_durability_and_replay_config(
        &directory.0,
        &mut catalog,
        Default::default(),
        replay,
    )
    .unwrap();
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "id",
            PropertyType::Int,
            false,
        )
        .unwrap();
    store
        .create_property_index(&mut catalog, "Memory", "id")
        .unwrap();
    let mut expected = Vec::new();
    for id in [1, 2] {
        expected.push(
            store
                .create_node(
                    &mut catalog,
                    "Memory",
                    BTreeMap::from([("id".into(), Value::Int(id))]),
                )
                .unwrap(),
        );
    }
    store.checkpoint(&catalog).unwrap();
    drop(store);
    catalog = Catalog::default();
    store = GraphStore::open_with_durability_and_replay_config(
        &directory.0,
        &mut catalog,
        Default::default(),
        replay,
    )
    .unwrap();
    assert!(store.is_out_of_core());
    let access = NodeProjectionAccess::PropertyValues {
        property: "id".into(),
        values: vec![Value::Int(1), Value::Int(2), Value::Int(1)],
    };
    for cap in [360, 4096] {
        let ledger = QueryMemoryLedger::new(nz(128 * 1024));
        // Direct public callers need not create internal retained metadata.
        let source = ledger.account(
            hawdb_executor::QueryMemoryClass::ExternalRead,
            "cold property source",
            nz(32 * 1024),
        );
        let output = ledger.account(
            hawdb_executor::QueryMemoryClass::PipelineBatch,
            "cold property output",
            nz(16 * 1024),
        );
        let context = NodeScanContext {
            catalog: &catalog,
            store: &store,
            execution_limit: ExecutionLimit::unlimited(),
            memory_budget: nz(cap),
            memory_account: &source,
            batch_memory_budget: nz(16 * 1024),
            batch_memory_account: &output,
            batch_rows: 1,
            task_context: None,
        };
        let observer = QueryExecutionObserver::default();
        let mut ids = Vec::new();
        let result = if projected {
            hawdb_executor::scan::stream_node_projection_scan_batches(
                NodeProjectionScanSpec {
                    variable: "n",
                    label: "Memory",
                    access: &access,
                    required_properties: &[],
                    predicate: None,
                    items: &[Projection {
                        name: "id".into(),
                        expression: ProjectionExpression::Id {
                            variable: "n".into(),
                        },
                    }],
                },
                context,
                &observer,
                &mut |batch| {
                    ids.extend(batch.into_iter().map(|row| match row.values["id"] {
                        Value::Int(id) => hawdb_storage::NodeId(id as u64),
                        _ => panic!("projected ID has wrong type"),
                    }));
                    Ok(BatchControl::Continue)
                },
            )
        } else {
            hawdb_executor::scan::stream_node_access_batches(
                "n",
                "Memory",
                &access,
                context,
                &mut |batch| {
                    ids.extend(batch.into_iter().map(|row| row.nodes["n"].id));
                    Ok(BatchControl::Continue)
                },
            )
        };
        if cap == 360 {
            let error = result.unwrap_err();
            assert!(error.to_string().contains("360-byte budget"), "{error}");
            assert_eq!(ids, expected[..1]);
        } else {
            assert_eq!(result.unwrap(), BatchControl::Continue);
            assert_eq!(ids, expected);
        }
        assert_eq!(ledger.snapshot().used_bytes, 0);
    }
}

#[test]
fn source_memory_cold_full_seek_keeps_storage_dedup_cap() {
    run_cold_source_retained_keys(false);
}
#[test]
fn source_memory_cold_projected_seek_keeps_storage_dedup_cap() {
    run_cold_source_retained_keys(true);
}

#[test]
fn source_memory_cold_union_shares_one_retained_allowance() {
    use hawdb_executor::scan::{NodeProjectionScanSpec, NodeScanContext};
    use hawdb_executor::QueryMemoryClass;
    use hawdb_plan_cypher::ExactPropertySeekBranch;
    let directory = StoredPropertyDirectory::new();
    let replay = hawdb_storage::store::WalReplayConfig {
        residency_mode: hawdb_storage::store::StorageResidencyMode::OutOfCore,
        ..Default::default()
    };
    let mut catalog = Catalog::default();
    let mut store = GraphStore::open_with_durability_and_replay_config(
        &directory.0,
        &mut catalog,
        Default::default(),
        replay,
    )
    .unwrap();
    store.create_node_table(&mut catalog, "Memory").unwrap();
    store
        .create_property_descriptor(
            &mut catalog,
            TableKind::Node,
            "Memory",
            "id",
            PropertyType::Int,
            false,
        )
        .unwrap();
    store
        .create_property_index(&mut catalog, "Memory", "id")
        .unwrap();
    let expected: Vec<_> = [1, 2]
        .into_iter()
        .map(|id| {
            store
                .create_node(
                    &mut catalog,
                    "Memory",
                    BTreeMap::from([("id".into(), Value::Int(id))]),
                )
                .unwrap()
        })
        .collect();
    store.checkpoint(&catalog).unwrap();
    drop(store);
    catalog = Catalog::default();
    store = GraphStore::open_with_durability_and_replay_config(
        &directory.0,
        &mut catalog,
        Default::default(),
        replay,
    )
    .unwrap();
    assert!(store.is_out_of_core());
    let branches = [1, 1, 2].map(|id| ExactPropertySeekBranch {
        property: "id".into(),
        values: vec![Value::Int(id)],
    });
    // Direct public full/projected callers need not attach retained metadata.
    for projected in [false, true] {
        for cap in [512, 4096] {
            let ledger = QueryMemoryLedger::new(nz(128 * 1024));
            let source = ledger.account(
                QueryMemoryClass::ExternalRead,
                "cold union source",
                nz(32 * 1024),
            );
            let output = ledger.account(
                QueryMemoryClass::PipelineBatch,
                "cold union output",
                nz(16 * 1024),
            );
            let context = NodeScanContext {
                catalog: &catalog,
                store: &store,
                execution_limit: ExecutionLimit::unlimited(),
                memory_budget: nz(cap),
                memory_account: &source,
                batch_memory_budget: nz(16 * 1024),
                batch_memory_account: &output,
                batch_rows: 1,
                task_context: None,
            };
            let mut ids = Vec::new();
            let mut emit = |batch: BindingBatch| {
                let snapshot = ledger.snapshot();
                let retained = snapshot
                    .classes
                    .iter()
                    .find(|item| item.class == QueryMemoryClass::BlockingState)
                    .unwrap();
                assert!(
                    retained.used_bytes <= cap,
                    "public union split its retained allowance: {snapshot:?}"
                );
                ids.extend(batch.into_iter().map(|row| row.nodes["n"].id));
                Ok(BatchControl::Continue)
            };
            let observer = QueryExecutionObserver::default();
            let result = if projected {
                hawdb_executor::scan::stream_node_projection_scan_batches(
                    NodeProjectionScanSpec {
                        variable: "n",
                        label: "Memory",
                        access: &NodeProjectionAccess::PropertyUnion {
                            branches: branches.to_vec(),
                        },
                        required_properties: &[],
                        predicate: None,
                        items: &[],
                    },
                    context,
                    &observer,
                    &mut emit,
                )
            } else {
                hawdb_executor::scan::stream_index_node_union_seek_batches(
                    "n", "Memory", &branches, context, &observer, &mut emit,
                )
            };
            if cap == 512 {
                let error = result.unwrap_err();
                assert!(error.to_string().contains("512-byte budget"), "{error}");
                assert!(
                    ids.is_empty(),
                    "cold inner and outer sets cannot fit together"
                );
            } else {
                assert_eq!(result.unwrap(), BatchControl::Continue);
                assert_eq!(
                    ids, expected,
                    "duplicate keys must not grow the retained set"
                );
            }
            assert_eq!(ledger.snapshot().used_bytes, 0);
        }
    }
}

#[test]
fn complete_text_growth_refuses_before_allocating_replacement_slots() {
    use hawdb_executor::{ExternalReadResultBudget, QueryMemoryClass, TextSeedExecutionRow};
    let row_bytes = std::mem::size_of::<TextSeedExecutionRow>();
    for succeeds in [false, true] {
        let bytes = NonZeroUsize::new(6 * row_bytes + if succeeds { 3 } else { 2 }).unwrap();
        let ledger = QueryMemoryLedger::new(bytes);
        let account = ledger.account(
            QueryMemoryClass::ExternalRead,
            "complete growth oracle",
            bytes,
        );
        let mut output = TextSeedExecutionOutput::new_complete(
            &account,
            ExternalReadResultBudget {
                max_rows: 1_000_000,
                max_memory_bytes: bytes,
            },
        )
        .unwrap();
        output.push("a", None, 3.0).unwrap();
        output.push("b", None, 2.0).unwrap();
        let window = AllocationWindow::start(4 * row_bytes);
        let result = output.push("c", None, 1.0);
        let replacements = window.allocations();
        drop(window);
        assert_eq!(replacements, usize::from(succeeds));
        if succeeds {
            result.unwrap();
            assert_eq!(output.finish_complete().unwrap().rows().len(), 3);
        } else {
            assert!(matches!(result, Err(HawDBError::ResourceBudgetExceeded(_))));
            assert!(output.finish_complete().is_err());
        }
        assert_eq!(ledger.snapshot().used_bytes, 0);
    }
}

#[test]
fn grouped_evidence_admits_id_copies_before_allocation_and_keeps_output_owned() {
    use hawdb_plan_cypher::{AggregateFunction, AggregateTarget, Aggregation};
    for succeeds in [false, true] {
        let id_bytes = (1 << 20) + 137;
        let mut source = DirectProjectionSource(Some(vec![Binding::values(BTreeMap::from([
            ("g".into(), Value::String("group".into())),
            ("doc".into(), Value::String("x".repeat(id_bytes))),
            ("score".into(), Value::Float(1.0)),
        ]))]));
        let memory = ExecutionMemoryConfig {
            query_memory_bytes: NonZeroUsize::new(16 << 20).unwrap(),
            blocking_operator_bytes: NonZeroUsize::new(if succeeds { 4 << 20 } else { 8192 })
                .unwrap(),
            batch_payload_bytes: NonZeroUsize::new(32 << 10).unwrap(),
            ..Default::default()
        };
        let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
        let catalog = Catalog::default();
        let observer = QueryExecutionObserver::default();
        let group_keys = [Projection {
            expression: ProjectionExpression::Column("g".into()),
            name: "g".into(),
        }];
        let items = [Aggregation {
            function: AggregateFunction::Collect,
            target: AggregateTarget::Ranked {
                document_id: "doc".into(),
                score: "score".into(),
                limit: 1,
            },
            distinct: false,
            name: "evidence".into(),
        }];
        let window = AllocationWindow::start(id_bytes);
        let mut delivered = 0;
        let result = hawdb_executor::blocking::stream_aggregate_batches(
            &PhysicalPlan::EmptyExec,
            &group_keys,
            &items,
            &mut source,
            BatchExecutionContext {
                catalog: &catalog,
                memory: &memory,
                memory_ledger: &ledger,
                task_context: None,
                observer: &observer,
            },
            ExecutionLimit::unlimited(),
            &mut |rows| {
                delivered += rows.len();
                let Value::List(values) = &rows[0].values["evidence"] else {
                    panic!("missing evidence");
                };
                let Value::String(id) = &values[0] else {
                    panic!("missing ID");
                };
                assert_eq!(id.len(), id_bytes);
                assert!(
                    ledger.snapshot().used_bytes >= 2 * id_bytes,
                    "output lost its original admission"
                );
                Ok(BatchControl::Continue)
            },
        );
        let copies = window.allocations();
        drop(window);
        assert_eq!(copies, if succeeds { 2 } else { 0 });
        assert_eq!(delivered, usize::from(succeeds));
        if succeeds {
            assert_eq!(result.unwrap(), BatchControl::Continue);
        } else {
            assert!(matches!(result, Err(HawDBError::ResourceBudgetExceeded(_))));
        }
        assert_eq!(ledger.snapshot().used_bytes, 0);
    }
}

#[test]
fn grouped_evidence_checks_final_callback_cancellation_and_preserves_consumer_errors() {
    use hawdb_plan_cypher::{AggregateFunction, AggregateTarget, Aggregation};
    for consumer_error in [false, true] {
        let mut source = DirectProjectionSource(Some(vec![Binding::values(BTreeMap::from([
            ("doc".into(), Value::String("a".into())),
            ("score".into(), Value::Float(1.0)),
        ]))]));
        let memory = ExecutionMemoryConfig::default();
        let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
        let catalog = Catalog::default();
        let observer = QueryExecutionObserver::default();
        let task = RuntimeTaskContext::default();
        let items = [Aggregation {
            function: AggregateFunction::Collect,
            target: AggregateTarget::Ranked {
                document_id: "doc".into(),
                score: "score".into(),
                limit: 1,
            },
            distinct: false,
            name: "evidence".into(),
        }];
        let mut delivered = 0;
        let result = hawdb_executor::blocking::stream_aggregate_batches(
            &PhysicalPlan::EmptyExec,
            &[],
            &items,
            &mut source,
            BatchExecutionContext {
                catalog: &catalog,
                memory: &memory,
                memory_ledger: &ledger,
                task_context: Some(&task),
                observer: &observer,
            },
            ExecutionLimit::unlimited(),
            &mut |rows| {
                delivered += rows.len();
                task.cancellation().cancel();
                if consumer_error {
                    Err(HawDBError::Execution("consumer sentinel".into()))
                } else {
                    Ok(BatchControl::Continue)
                }
            },
        );
        assert_eq!(delivered, 1);
        assert!(
            matches!(result, Err(HawDBError::Execution(message)) if if consumer_error { message == "consumer sentinel" } else { message.contains("cancel") })
        );
        assert_eq!(ledger.snapshot().used_bytes, 0);
    }
}

#[test]
fn grouped_evidence_reuses_a_wide_scalar_group_key_without_recopying_it() {
    use hawdb_plan_cypher::{AggregateFunction, AggregateTarget, Aggregation};
    let key_bytes = (1 << 20) + 137;
    let mut source = DirectProjectionSource(Some(
        (0..8)
            .map(|score| {
                Binding::values(BTreeMap::from([
                    ("g".into(), Value::String("g".repeat(key_bytes))),
                    ("doc".into(), Value::String(format!("document-{score}"))),
                    ("score".into(), Value::Float(f64::from(score))),
                ]))
            })
            .collect(),
    ));
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: NonZeroUsize::new(16 << 20).unwrap(),
        blocking_operator_bytes: NonZeroUsize::new(4 << 20).unwrap(),
        batch_payload_bytes: NonZeroUsize::new(32 << 10).unwrap(),
        ..Default::default()
    };
    let ledger = QueryMemoryLedger::new(memory.query_memory_bytes);
    let catalog = Catalog::default();
    let observer = QueryExecutionObserver::default();
    let group_keys = [Projection {
        expression: ProjectionExpression::Column("g".into()),
        name: "g".into(),
    }];
    let items = [Aggregation {
        function: AggregateFunction::Collect,
        target: AggregateTarget::Ranked {
            document_id: "doc".into(),
            score: "score".into(),
            limit: 2,
        },
        distinct: false,
        name: "evidence".into(),
    }];
    let window = AllocationWindow::start(key_bytes);
    let mut delivered = 0;
    hawdb_executor::blocking::stream_aggregate_batches(
        &PhysicalPlan::EmptyExec,
        &group_keys,
        &items,
        &mut source,
        BatchExecutionContext {
            catalog: &catalog,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: None,
            observer: &observer,
        },
        ExecutionLimit::unlimited(),
        &mut |rows| {
            delivered += rows.len();
            let Value::String(key) = &rows[0].values["g"] else {
                panic!("missing key")
            };
            assert_eq!(key.len(), key_bytes);
            assert_eq!(
                rows[0].values["evidence"],
                Value::List(vec![
                    Value::String("document-7".into()),
                    Value::String("document-6".into())
                ])
            );
            assert!(ledger.snapshot().used_bytes >= key_bytes);
            Ok(BatchControl::Continue)
        },
    )
    .unwrap();
    let copies = window.allocations();
    drop(window);
    assert_eq!(copies, 1, "only the first group key needs an owned copy");
    assert_eq!(delivered, 1);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    let reports = observer.into_reports();
    assert_eq!(reports.grouped_evidence.len(), 1);
    assert_eq!(reports.grouped_evidence[0].borrowed_group_key_rows, 7);
    assert_eq!(reports.grouped_evidence[0].input_rows, 8);
}