@@ -3039,16 +3039,11 @@ async fn collect_left_input(
 
             let scratch_reservation = reservation.new_empty();
             if prepared {
-                // Allow one logical null mask per key plus the combined mask.
-                let masks = if null_equality == NullEquality::NullEqualsNothing {
-                    on_left.len() + 1
-                } else {
-                    0
-                };
-                scratch_reservation.try_grow(
-                    max_batch_rows * size_of::<u64>()
-                        + (max_batch_rows.div_ceil(8) + 64) * masks,
-                )?;
+                scratch_reservation.try_grow(prepared::prepared_scratch_bytes(
+                    max_batch_rows,
+                    on_left.len(),
+                    null_equality,
+                )?)?;
             }
             // The maximum is known: avoid geometric growth and its excess capacity.
             let mut hashes_buffer = Vec::with_capacity(max_batch_rows);
@@ -3213,10 +3208,15 @@ async fn collect_left_input(
 
     if prepared {
         drop(batches);
+        // Prepared keys are direct columns. IN-list arrays share these batch
+        // buffers, including through multi-key StructArray children.
         let retained = RecordBatchMemoryCounter::new().count_batch(&batch);
-        let allowance = input_bytes + copy_bytes;
-        debug_assert!(retained <= allowance);
-        reservation.shrink(allowance - retained);
+        prepared::reconcile_prepared_copy_reservation(
+            &reservation,
+            input_bytes,
+            copy_bytes,
+            retained,
+        )?;
     }
 
     let data = JoinLeftData {
 
@@ -19,6 +19,7 @@
 
 use super::*;
 use crate::spill::spill_manager::GetSlicedSize;
+use datafusion_common::exec_datafusion_err;
 use datafusion_execution::memory_pool::MemoryPool;
 
 /// An immutable, fully prepared broadcast build, independent of any probe task.
@@ -210,7 +211,8 @@ impl HashJoinExec {
 /// Bound copy allocations, including validity, offsets and alignment. Aliased
 /// columns count separately because concatenation materializes each column.
 pub(super) fn prepared_copy_bytes(batches: &[RecordBatch]) -> Result<usize> {
-    let mut bytes = 0;
+    let overflow = || exec_datafusion_err!("Prepared hash-join copy size overflow");
+    let mut bytes = 0usize;
     for batch in batches {
         // Concat can materialize validity for an all-valid input when another input
         // contains nulls. Arrow's slice measurement only counts existing bitmaps.
@@ -219,17 +221,73 @@ pub(super) fn prepared_copy_bytes(batches: &[RecordBatch]) -> Result<usize> {
             .iter()
             .filter(|array| array.nulls().is_none())
             .count();
-        bytes +=
-            batch.get_sliced_size()? + batch.num_rows().div_ceil(8) * missing_validity;
+        let validity_bytes = batch
+            .num_rows()
+            .div_ceil(8)
+            .checked_mul(missing_validity)
+            .ok_or_else(overflow)?;
+        bytes = bytes
+            .checked_add(batch.get_sliced_size()?)
+            .and_then(|bytes| bytes.checked_add(validity_bytes))
+            .ok_or_else(overflow)?;
     }
     // These are flat arrays: allow one alignment unit per output buffer,
     // including potential validity. Concat allocates each buffer only once.
     let buffers = batches[0]
         .columns()
         .iter()
-        .map(|array| array.to_data().buffers().len() + 1)
-        .sum::<usize>();
-    Ok(bytes + buffers * 64)
+        .try_fold(0usize, |buffers, array| {
+            buffers
+                .checked_add(array.to_data().buffers().len())
+                .and_then(|buffers| buffers.checked_add(1))
+                .ok_or_else(overflow)
+        })?;
+    buffers
+        .checked_mul(64)
+        .and_then(|padding| bytes.checked_add(padding))
+        .ok_or_else(overflow)
+}
+
+/// Bound hashes and flat-key null masks for the largest input batch.
+pub(super) fn prepared_scratch_bytes(
+    max_batch_rows: usize,
+    num_keys: usize,
+    null_equality: NullEquality,
+) -> Result<usize> {
+    let overflow = || exec_datafusion_err!("Prepared hash-join scratch size overflow");
+    let hash_bytes = max_batch_rows
+        .checked_mul(size_of::<u64>())
+        .ok_or_else(overflow)?;
+    if null_equality == NullEquality::NullEqualsNull {
+        return Ok(hash_bytes);
+    }
+    // Allow one logical null mask per key plus the combined mask.
+    let masks = num_keys.checked_add(1).ok_or_else(overflow)?;
+    let mask_bytes = max_batch_rows
+        .div_ceil(8)
+        .checked_add(64)
+        .and_then(|bytes| bytes.checked_mul(masks))
+        .ok_or_else(overflow)?;
+    hash_bytes.checked_add(mask_bytes).ok_or_else(overflow)
+}
+
+/// Release the unused copy allowance only after validating the retained size.
+pub(super) fn reconcile_prepared_copy_reservation(
+    reservation: &MemoryReservation,
+    input_bytes: usize,
+    copy_bytes: usize,
+    retained: usize,
+) -> Result<()> {
+    let allowance = input_bytes.checked_add(copy_bytes).ok_or_else(|| {
+        exec_datafusion_err!("Prepared hash-join copy allowance size overflow")
+    })?;
+    if retained > allowance {
+        return internal_err!(
+            "Prepared hash-join concat exceeded its admitted copy bound"
+        );
+    }
+    reservation.shrink(allowance - retained);
+    Ok(())
 }
 
 impl HashJoinExecBuilder {
 
@@ -30,7 +30,8 @@ use crate::{
     test::TestMemoryExec,
 };
 use arrow::array::{
-    BooleanArray, FixedSizeBinaryArray, Int64Array, StringArray, StringViewArray,
+    BooleanArray, FixedSizeBinaryArray, Int64Array, NullArray, StringArray,
+    StringViewArray, StructArray,
 };
 use arrow::buffer::Buffer;
 use arrow::compute::kernels::sort::SortOptions;
@@ -698,6 +699,125 @@ fn prepared_concat_admits_validity_for_non_nullable_input_arrays() -> Result<()>
     Ok(())
 }
 
+#[tokio::test]
+async fn prepared_inlist_shares_accounted_build_buffers() -> Result<()> {
+    let build = batch(vec![Some(1), Some(1_000_000), None, Some(1)]);
+    for keys in [1, 2] {
+        let on = (0..keys)
+            .map(|index| {
+                let name = build.schema().field(index).name().clone();
+                let key: PhysicalExprRef = Arc::new(Column::new(&name, index));
+                (Arc::clone(&key), key)
+            })
+            .collect();
+        let base = join(build.schema(), build.clone())?
+            .builder()
+            .with_on(on)
+            .build()?;
+        for batches in [
+            vec![build.clone()],
+            vec![build.slice(0, 2), build.slice(2, 2)],
+        ] {
+            let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 20));
+            let prepared = prepare(&base, batches.clone(), Arc::clone(&pool)).await?;
+            let PushdownStrategy::InList(values) = &prepared.build.membership else {
+                panic!("expected IN-list membership");
+            };
+            let columns = prepared.build.batch.columns();
+            if keys == 1 {
+                assert!(Arc::ptr_eq(values, &columns[0]));
+            } else {
+                let values = values.as_any().downcast_ref::<StructArray>().unwrap();
+                for (value, column) in values.columns().iter().zip(columns) {
+                    assert!(Arc::ptr_eq(value, column));
+                }
+            }
+            let bytes = prepared.reserved_bytes();
+            let mut counter = RecordBatchMemoryCounter::new();
+            let batch_bytes = counter.count_batch(&prepared.build.batch);
+            assert!(bytes > batch_bytes, "hash storage must also be reserved");
+            let membership_batch =
+                RecordBatch::try_from_iter([("membership", Arc::clone(values))])?;
+            assert_eq!(counter.count_batch(&membership_batch), 0);
+            drop(membership_batch);
+
+            // Disabling IN-list membership changes no retained key buffers.
+            let mut config = ConfigOptions::default();
+            config.optimizer.hash_join_inlist_pushdown_max_size = 0;
+            let map = base
+                .prepare_build(
+                    input(batches, build.schema()),
+                    Arc::clone(&pool),
+                    Arc::new(config),
+                )
+                .await?;
+            assert!(matches!(map.build.membership, PushdownStrategy::Map(_)));
+            assert_eq!(map.reserved_bytes(), bytes);
+            assert_eq!(pool.reserved(), 2 * bytes);
+            drop(map);
+            let lease = Arc::clone(&prepared);
+            drop(prepared);
+            assert_eq!(pool.reserved(), bytes);
+            drop(lease);
+            assert_eq!(pool.reserved(), 0);
+        }
+    }
+    Ok(())
+}
+
+#[test]
+fn prepared_concat_rejects_under_admission_without_shrinking() -> Result<()> {
+    let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1024));
+    let reservation = MemoryConsumer::new("prepared concat test").register(&pool);
+    // 128 bytes of map storage plus 128 bytes of input/copy allowance.
+    reservation.try_grow(256)?;
+    let error =
+        reconcile_prepared_copy_reservation(&reservation, 64, 64, 129).unwrap_err();
+    assert!(matches!(error, DataFusionError::Internal(_)));
+    assert!(
+        error
+            .to_string()
+            .contains("exceeded its admitted copy bound")
+    );
+    assert_eq!(reservation.size(), 256);
+    assert!(reconcile_prepared_copy_reservation(&reservation, usize::MAX, 1, 0).is_err());
+    assert_eq!(reservation.size(), 256);
+    reconcile_prepared_copy_reservation(&reservation, 64, 64, 128)?;
+    assert_eq!(reservation.size(), 256);
+    reconcile_prepared_copy_reservation(&reservation, 64, 64, 80)?;
+    assert_eq!(reservation.size(), 208);
+    drop(reservation);
+    assert_eq!(pool.reserved(), 0);
+    Ok(())
+}
+
+#[test]
+fn prepared_sizing_rejects_overflow() -> Result<()> {
+    use NullEquality::{NullEqualsNothing, NullEqualsNull};
+    for (rows, keys, equality) in [
+        (usize::MAX / 8 + 1, 1, NullEqualsNull),
+        (0, usize::MAX, NullEqualsNothing),
+        (0, usize::MAX / 64, NullEqualsNothing),
+        (usize::MAX / 8, 0, NullEqualsNothing),
+    ] {
+        let error = prepared_scratch_bytes(rows, keys, equality).unwrap_err();
+        assert!(error.to_string().contains("scratch size overflow"));
+    }
+    assert!(prepared_scratch_bytes(usize::MAX / 8, 1, NullEqualsNull).is_ok());
+
+    // Null arrays can describe huge row counts without allocating data buffers.
+    // Exercise both per-batch validity multiplication and cumulative addition.
+    let array: ArrayRef = Arc::new(NullArray::new(usize::MAX));
+    for (columns, batches) in [(8, 1), (4, 2)] {
+        let batch = RecordBatch::try_from_iter(
+            (0..columns).map(|i| (format!("key{i}"), Arc::clone(&array))),
+        )?;
+        let error = prepared_copy_bytes(&vec![batch; batches]).unwrap_err();
+        assert!(error.to_string().contains("copy size overflow"));
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn prepared_byte_keys_use_hash_membership() -> Result<()> {
     for array in [