algeon-cudf 0.1.3

//! Integration tests for execution-context propagation.
//!
//! These tests verify that non-default CUDA streams and memory resources are
//! correctly propagated through the FFI shim layer and through the current
//! cudf safe-API slices.
//!
//! # Test categories
//!
//! 1. **Existing `_ex` FFI**: `cudf_column_copy_ex`, `cudf_column_view_to_host_ex`,
//!    `cudf_column_create_from_device_data_ex` use explicit streams.
//! 2. **High-level execution-context contract**: hidden safe-layer hooks
//!    route explicit-lineage operations through `_ex` entry points and preserve
//!    explicit retained provenance under borrowed overrides.
//! 3. **Retained-domain default behavior**: APIs outside the scoped override path
//!    derive explicit execution from the retained source domain.
//! 4. **Allocator context**: explicit memory resource is passed through.
//! 5. **Borrowed handle identity**: `as_borrowed_sys` preserves the sys-handle identity.
//! 6. **Strict-context rejection**: production `_ex` APIs reject missing or
//!    partial execution options.
use std::{
    ffi::{c_char, c_void},
    ptr,
};

use cudf::{
    DeviceId,
    core::memory::DeviceMemoryResource,
    runtime::{CudaEvent, CudaStream},
    testing::{prelude::*, run_named_cases},
};
use cudf_sys::test_support::{
    cudf_test_current_device_resource_raw, cudf_test_default_stream_value, cudf_test_exec_last_device,
    cudf_test_exec_last_mr_raw, cudf_test_exec_last_stream_value, cudf_test_exec_last_sync_stream_value,
    cudf_test_exec_reset_observation, cudf_test_memory_resource_raw, cudf_test_stream_handle_value,
};

use crate::common::{column_from_slice, series};

/// Reset observation state so the next `_ex` call's context is cleanly visible.
fn obs_reset() {
    unsafe { cudf_test_exec_reset_observation() };
}

fn obs_last_stream() -> usize {
    let mut out: usize = 0;
    let status = unsafe { cudf_test_exec_last_stream_value(&mut out) };
    assert_eq!(status, 0, "cudf_test_exec_last_stream_value failed");
    out
}

/// Last synchronized stream value (for host-materializing ops).
fn obs_last_sync_stream() -> usize {
    let mut out: usize = 0;
    let status = unsafe { cudf_test_exec_last_sync_stream_value(&mut out) };
    assert_eq!(status, 0, "cudf_test_exec_last_sync_stream_value failed");
    out
}

fn obs_last_device() -> i32 {
    let mut out: i32 = -1;
    let status = unsafe { cudf_test_exec_last_device(&mut out) };
    assert_eq!(status, 0, "cudf_test_exec_last_device failed");
    out
}

fn obs_last_mr() -> *mut c_void {
    let mut out: *mut c_void = ptr::null_mut();
    let status = unsafe { cudf_test_exec_last_mr_raw(&mut out) };
    assert_eq!(status, 0, "cudf_test_exec_last_mr_raw failed");
    out
}

/// Process-global default stream value.
fn obs_default_stream() -> usize {
    let mut out: usize = 0;
    let status = unsafe { cudf_test_default_stream_value(&mut out) };
    assert_eq!(status, 0, "cudf_test_default_stream_value failed");
    out
}

/// Convert a stream handle to its raw uintptr value (for comparison).
fn obs_stream_handle_value(handle: cudf_sys::cudf_stream_handle) -> usize {
    let mut out: usize = 0;
    let status = unsafe { cudf_test_stream_handle_value(handle, &mut out) };
    assert_eq!(status, 0, "cudf_test_stream_handle_value failed");
    out
}

/// Process-global current device memory resource raw pointer.
fn obs_current_device_resource() -> *mut c_void {
    let mut out: *mut c_void = ptr::null_mut();
    let status = unsafe { cudf_test_current_device_resource_raw(&mut out) };
    assert_eq!(status, 0, "cudf_test_current_device_resource_raw failed");
    out
}

fn required_exec_options(
    stream: &cudf_sys::BorrowedStream<'_>,
    mr: &cudf_sys::BorrowedMemoryResource<'_>,
) -> cudf_sys::cudf_exec_options_t {
    let mut opts = cudf_sys::cudf_exec_options_t::defaults();
    opts.device = DeviceId::current().expect("current device must be available").ordinal() as i32;
    opts.flags = cudf_sys::CUDF_EXEC_FLAG_DEVICE_EXPLICIT;
    opts.stream = stream.as_raw();
    opts.mr = mr.as_raw();
    opts
}

fn strict_exec_owners() -> (CudaStream, DeviceMemoryResource, cudf_sys::cudf_exec_options_t) {
    let stream = CudaStream::new().expect("stream must be created");
    let mr = DeviceMemoryResource::cuda().expect("memory resource must be created");
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let exec = required_exec_options(&borrowed_stream, &borrowed_mr);
    (stream, mr, exec)
}

/// Create an i32 column and return its raw cudf_column_handle.
///
/// The caller is responsible for destroying the handle.
fn create_i32_column_raw(data: &[i32]) -> cudf_sys::cudf_column_handle {
    let mut handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let (_stream, _mr, exec) = strict_exec_owners();
    let status = unsafe {
        cudf_sys::cudf_column_create_numeric_ex(
            cudf_sys::CUDF_TYPE_INT32,
            data.as_ptr() as *const c_void,
            data.len() as i32,
            ptr::null(), // no validity mask
            0,           // null_count = 0
            &exec,
            &mut handle,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create i32 column");
    assert!(!handle.is_null());
    handle
}

fn create_i64_column_raw(data: &[i64]) -> cudf_sys::cudf_column_handle {
    let mut handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let (_stream, _mr, exec) = strict_exec_owners();
    let status = unsafe {
        cudf_sys::cudf_column_create_numeric_ex(
            cudf_sys::CUDF_TYPE_INT64,
            data.as_ptr() as *const c_void,
            data.len() as i32,
            ptr::null(),
            0,
            &exec,
            &mut handle,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create i64 column");
    assert!(!handle.is_null());
    handle
}

fn create_bool_column_raw(data: &[bool]) -> cudf_sys::cudf_column_handle {
    let mut handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let (_stream, _mr, exec) = strict_exec_owners();
    let status = unsafe {
        cudf_sys::cudf_column_create_numeric_ex(
            cudf_sys::CUDF_TYPE_BOOL8,
            data.as_ptr() as *const c_void,
            data.len() as i32,
            ptr::null(),
            0,
            &exec,
            &mut handle,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create bool column");
    assert!(!handle.is_null());
    handle
}

fn create_i32_nullable_column_raw(data: &[i32], validity_bits: u8) -> cudf_sys::cudf_column_handle {
    let mut handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let mask = [validity_bits];
    let null_count = data.len() as i32 - validity_bits.count_ones() as i32;
    let (_stream, _mr, exec) = strict_exec_owners();
    let status = unsafe {
        cudf_sys::cudf_column_create_numeric_ex(
            cudf_sys::CUDF_TYPE_INT32,
            data.as_ptr() as *const c_void,
            data.len() as i32,
            mask.as_ptr(),
            null_count,
            &exec,
            &mut handle,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create nullable i32 column");
    assert!(!handle.is_null());
    handle
}

fn create_i32_scalar_raw(value: i32) -> cudf_sys::cudf_scalar_handle {
    let mut handle: cudf_sys::cudf_scalar_handle = ptr::null_mut();
    let (stream, _mr, exec) = strict_exec_owners();
    let status = unsafe {
        cudf_sys::cudf_scalar_create_numeric_ex(
            cudf_sys::CUDF_TYPE_INT32,
            &value as *const i32 as *const c_void,
            1,
            &exec,
            &mut handle,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create i32 scalar");
    assert!(!handle.is_null());
    // This helper returns the scalar for use on unrelated test streams.
    stream.synchronize().expect("scalar producer stream sync failed");
    handle
}

/// Get a view handle from a column handle. Caller must destroy the view.
fn column_view_raw(col: cudf_sys::cudf_column_handle) -> cudf_sys::cudf_column_view_handle {
    let mut view: cudf_sys::cudf_column_view_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_view(col, &mut view) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create column view");
    assert!(!view.is_null());
    view
}

/// Get a table view handle from a table handle. Caller must destroy the view.
fn table_view_raw(table: cudf_sys::cudf_table_handle) -> cudf_sys::cudf_table_view_handle {
    let mut view: cudf_sys::cudf_table_view_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_table_view(table, &mut view) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create table view");
    assert!(!view.is_null());
    view
}

/// Get a column view handle from a table view by index. Caller must destroy the view.
fn table_view_column_raw(table: cudf_sys::cudf_table_view_handle, index: i32) -> cudf_sys::cudf_column_view_handle {
    let mut view: cudf_sys::cudf_column_view_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_table_view_column(table, index, &mut view) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create table column view");
    assert!(!view.is_null());
    view
}

fn column_view_device_ptr(view: cudf_sys::cudf_column_view_handle) -> *const c_void {
    let mut ptr: *const c_void = ptr::null();
    let status = unsafe { cudf_sys::cudf_column_view_data_ptr(view, &mut ptr) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to get device data ptr");
    ptr
}

fn read_i32_column_raw(col: cudf_sys::cudf_column_handle, len: usize) -> Vec<i32> {
    let view = column_view_raw(col);
    let mut host = vec![0i32; len];
    let (_stream, _mr, exec) = strict_exec_owners();
    let status = unsafe {
        cudf_sys::cudf_column_view_to_host_ex(
            view,
            host.as_mut_ptr() as *mut c_void,
            std::mem::size_of_val(host.as_slice()),
            &exec,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to read i32 column to host");
    unsafe { cudf_sys::cudf_column_view_destroy(view) };
    host
}

fn read_first_i32_table_column_raw(table: cudf_sys::cudf_table_handle, len: usize) -> Vec<i32> {
    let table_view = table_view_raw(table);
    let column_view = table_view_column_raw(table_view, 0);
    let mut host = vec![0i32; len];
    let (_stream, _mr, exec) = strict_exec_owners();
    let status = unsafe {
        cudf_sys::cudf_column_view_to_host_ex(
            column_view,
            host.as_mut_ptr() as *mut c_void,
            std::mem::size_of_val(host.as_slice()),
            &exec,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to read first table column to host");
    unsafe {
        cudf_sys::cudf_column_view_destroy(column_view);
        cudf_sys::cudf_table_view_destroy(table_view);
    }
    host
}

fn destroy_column(handle: cudf_sys::cudf_column_handle) {
    unsafe { cudf_sys::cudf_column_destroy(handle) };
}

fn destroy_column_view(handle: cudf_sys::cudf_column_view_handle) {
    unsafe { cudf_sys::cudf_column_view_destroy(handle) };
}

fn destroy_table(handle: cudf_sys::cudf_table_handle) {
    unsafe { cudf_sys::cudf_table_destroy(handle) };
}

fn destroy_table_view(handle: cudf_sys::cudf_table_view_handle) {
    unsafe { cudf_sys::cudf_table_view_destroy(handle) };
}

fn destroy_scalar(handle: cudf_sys::cudf_scalar_handle) {
    unsafe { cudf_sys::cudf_scalar_destroy(handle) };
}

fn create_single_column_table_raw(data: &[i32]) -> cudf_sys::cudf_table_handle {
    let col = create_i32_column_raw(data);
    let mut columns = [col];
    let mut table: cudf_sys::cudf_table_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_table_create(columns.as_mut_ptr(), columns.len() as i32, &mut table) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "failed to create table");
    assert!(!table.is_null());
    table
}

fn explicit_i32_series(_name: &str, data: &[i32], identity: u64) -> cudf::Series {
    cudf::Series::from_column(
        column_from_slice(data).unwrap().testing_rebind_execution_domain_explicit(identity).unwrap(),
    )
    .unwrap()
}

fn explicit_i64_series(_name: &str, data: &[i64], identity: u64) -> cudf::Series {
    cudf::Series::from_column(
        column_from_slice(data).unwrap().testing_rebind_execution_domain_explicit(identity).unwrap(),
    )
    .unwrap()
}

fn explicit_bool_series(_name: &str, data: &[bool], identity: u64) -> cudf::Series {
    cudf::Series::from_column(
        column_from_slice(data).unwrap().testing_rebind_execution_domain_explicit(identity).unwrap(),
    )
    .unwrap()
}

const TEST_POOL_MR_BYTES: usize = 16 * 1024 * 1024;

fn pool_mr() -> DeviceMemoryResource {
    DeviceMemoryResource::pool(TEST_POOL_MR_BYTES, None).expect("failed to create pool MR")
}

fn debug_ctx_with_stream(stream: &CudaStream) -> cudf::Exec<'_> {
    cudf::Exec::RETAINED.with_stream(stream)
}

fn debug_ctx_with_mr(mr: &DeviceMemoryResource) -> cudf::Exec<'_> {
    cudf::Exec::RETAINED.with_memory_resource(mr)
}

fn assert_observed_stream(explicit_stream: usize) {
    assert_eq!(obs_last_stream(), explicit_stream, "observation hook should record the explicit stream");
}

fn assert_observed_sync_stream(explicit_stream: usize) {
    assert_eq!(
        obs_last_sync_stream(),
        explicit_stream,
        "host materialization should synchronize on the explicit stream"
    );
}

fn assert_observed_non_default_mr(default_mr: *mut c_void) {
    let observed_mr = obs_last_mr();
    assert_ne!(observed_mr, ptr::null_mut(), "observed mr should be non-null");
    assert_ne!(observed_mr, default_mr, "observation hook should record the explicit memory resource");
}

fn assert_dataframe_stream_override(df: &cudf::DataFrame, expected_domain: u64, expected_stream: usize) {
    assert_eq!(df.testing_execution_domain_id(), expected_domain);
    assert_eq!(df.testing_execution_domain_stream_value(), expected_stream);
}

fn assert_dataframe_mr_override(df: &cudf::DataFrame, expected_domain: u64, expected_mr: usize) {
    assert_eq!(df.testing_execution_domain_id(), expected_domain);
    assert_eq!(df.testing_execution_domain_mr_value(), expected_mr);
}

fn assert_dataframe_i32_values(df: &cudf::DataFrame, column: &str, expected: &[i32]) {
    let values: Vec<i32> = df.series(column).unwrap().testing_view().unwrap().to_vec().unwrap();
    assert_eq!(values.as_slice(), expected);
}

fn assert_exec_options_copy_validation_error(
    opts: cudf_sys::cudf_exec_options_t,
    expected_status: cudf_sys::cudf_status_t,
    expectation: &str,
) {
    let col = create_i32_column_raw(&[1, 2, 3]);
    let mut out: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe {
        cudf_sys::cudf_column_copy_ex(col, &opts as *const cudf_sys::cudf_exec_options_t as *const _, &mut out)
    };

    assert_eq!(status, expected_status, "{expectation}");
    assert!(out.is_null(), "output handle should remain null on validation error");

    destroy_column(col);
}

// These tests use the existing `_ex` FFI variants to verify that explicit
// stream handles reach the C++ shim's observation hooks.
//
// All column creation and manipulation uses cudf_sys FFI directly (not the
// cudf safe API) because Column::raw_handle() is pub(crate).

/// `cudf_column_copy_ex` with an explicit stream records that stream in the
/// observation hook. The explicit stream differs from the process-global default.
fn column_copy_ex_uses_explicit_stream() {
    let col = create_i32_column_raw(&[10, 20, 30]);

    let stream = CudaStream::new().expect("failed to create stream");
    let mr = DeviceMemoryResource::cuda().expect("failed to create mr");
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);

    let explicit_stream = stream.testing_raw_handle_value();
    assert_ne!(
        explicit_stream,
        obs_default_stream(),
        "explicit stream must differ from default for this test to be meaningful"
    );

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_copy_ex(col, &ctx, &mut out_handle) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_column_copy_ex failed");
    assert!(!out_handle.is_null());

    assert_observed_stream(explicit_stream);

    stream.synchronize().expect("stream sync failed");
    destroy_column(out_handle);
    destroy_column(col);
}

/// `cudf_column_view_to_host_ex` with an explicit stream records that stream
/// in the observation hook.
fn column_to_host_ex_uses_explicit_stream() {
    let data: [i32; 3] = [100, 200, 300];
    let col = create_i32_column_raw(&data);
    let view = column_view_raw(col);

    let stream = CudaStream::new().expect("failed to create stream");
    let mr = DeviceMemoryResource::cuda().expect("failed to create mr");
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);

    let explicit_stream = stream.testing_raw_handle_value();

    obs_reset();
    let mut host_buf = [0i32; 3];
    let buf_size = std::mem::size_of_val(&host_buf);
    let status =
        unsafe { cudf_sys::cudf_column_view_to_host_ex(view, host_buf.as_mut_ptr() as *mut c_void, buf_size, &ctx) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_column_view_to_host_ex failed");

    assert_observed_stream(explicit_stream);

    assert_eq!(host_buf, data, "host data should match original");

    destroy_column_view(view);
    destroy_column(col);
}

/// `cudf_column_view_to_host_ex` on a sliced view must copy from the sliced
/// element offset, scaled by the element width; not by single bytes.
fn column_to_host_ex_reads_sliced_view_from_element_offset() {
    let data: [i32; 6] = [10, 20, 30, 40, 50, 60];
    let col = create_i32_column_raw(&data);
    let view = column_view_raw(col);

    let (_stream, _mr, exec) = strict_exec_owners();
    let mut sliced: cudf_sys::cudf_column_view_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_view_slice_ex(view, 2, 3, &exec, &mut sliced) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_column_view_slice_ex failed");
    assert!(!sliced.is_null());

    let mut host_buf = [0i32; 3];
    let status = unsafe {
        cudf_sys::cudf_column_view_to_host_ex(
            sliced,
            host_buf.as_mut_ptr() as *mut c_void,
            std::mem::size_of_val(&host_buf),
            &exec,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_column_view_to_host_ex failed");
    assert_eq!(host_buf, [30, 40, 50], "sliced view must copy from offset * elem_size");

    destroy_column_view(sliced);
    destroy_column_view(view);
    destroy_column(col);
}

/// `cudf_column_view_strings_to_host_ex` on a sliced STRING view must read
/// the offsets child starting at the parent view's offset.
fn strings_to_host_ex_accounts_for_sliced_parent_view_offset() {
    let strings = [c"aa", c"bbb", c"c", c"dddd"];
    let mut ptrs: Vec<*const c_char> = strings.iter().map(|s| s.as_ptr()).collect();
    let lengths: Vec<i32> = strings.iter().map(|s| s.to_bytes().len() as i32).collect();

    let (_stream, _mr, exec) = strict_exec_owners();
    let mut col: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe {
        cudf_sys::cudf_column_create_strings_ex(
            ptrs.as_mut_ptr(),
            lengths.as_ptr(),
            strings.len() as i32,
            ptr::null(),
            0,
            &exec,
            &mut col,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_column_create_strings_ex failed");
    assert!(!col.is_null());
    let view = column_view_raw(col);

    let mut sliced: cudf_sys::cudf_column_view_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_view_slice_ex(view, 1, 2, &exec, &mut sliced) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_column_view_slice_ex failed");
    assert!(!sliced.is_null());

    let mut offsets = [0i32; 3];
    let mut chars = [0u8; 16];
    let mut total_chars_bytes = 0i32;
    let status = unsafe {
        cudf_sys::cudf_column_view_strings_to_host_ex(
            sliced,
            offsets.as_mut_ptr(),
            std::mem::size_of_val(&offsets),
            chars.as_mut_ptr() as *mut c_char,
            chars.len(),
            &mut total_chars_bytes,
            &exec,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_column_view_strings_to_host_ex failed");
    assert_eq!(total_chars_bytes, 4, "sliced rows are \"bbb\" and \"c\"");
    assert_eq!(offsets, [0, 3, 4], "offsets must be rebased to the slice");
    assert_eq!(&chars[..4], b"bbbc", "chars must come from the sliced rows");

    destroy_column_view(sliced);
    destroy_column_view(view);
    destroy_column(col);
}

/// `cudf_column_create_from_device_data_ex` with an explicit stream records
/// that stream in the observation hook.
fn column_from_device_data_ex_uses_explicit_stream() {
    let data: [i64; 4] = [1, 2, 3, 4];
    let src_col = create_i64_column_raw(&data);
    let src_view = column_view_raw(src_col);
    let device_ptr = column_view_device_ptr(src_view);

    let stream = CudaStream::new().expect("failed to create stream");
    let mr = DeviceMemoryResource::cuda().expect("failed to create mr");
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);

    let explicit_stream = stream.testing_raw_handle_value();

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe {
        cudf_sys::cudf_column_create_from_device_data_ex(
            cudf_sys::CUDF_TYPE_INT64,
            device_ptr,
            4,
            &ctx,
            &mut out_handle,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_column_create_from_device_data_ex failed");
    assert!(!out_handle.is_null());

    assert_observed_stream(explicit_stream);

    // The _ex variant does NOT synchronize; the caller must keep src alive
    // until the stream completes.
    stream.synchronize().expect("stream sync failed");

    let out_view = column_view_raw(out_handle);
    let mut host_buf = [0i64; 4];
    let (_read_stream, _read_mr, read_exec) = strict_exec_owners();
    let s3 = unsafe {
        cudf_sys::cudf_column_view_to_host_ex(
            out_view,
            host_buf.as_mut_ptr() as *mut c_void,
            std::mem::size_of_val(&host_buf),
            &read_exec,
        )
    };
    assert_eq!(s3, cudf_sys::CUDF_SUCCESS);
    assert_eq!(host_buf, data, "round-trip data should match");

    destroy_column_view(out_view);
    destroy_column_view(src_view);
    destroy_column(out_handle);
    destroy_column(src_col);
}

fn replace_nulls_scalar_ex_uses_explicit_stream() {
    let input = create_i32_nullable_column_raw(&[1, 0, 3], 0b0000_0101);
    let input_view = column_view_raw(input);
    let replacement = create_i32_scalar_raw(99);

    let stream = CudaStream::new().expect("failed to create stream");
    let mr = DeviceMemoryResource::cuda().expect("failed to create mr");
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);
    let explicit_stream = stream.testing_raw_handle_value();

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_replace_nulls_scalar_ex(input_view, replacement, &mut out_handle, &ctx) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_replace_nulls_scalar_ex failed");
    assert!(!out_handle.is_null());

    assert_observed_stream(explicit_stream);

    stream.synchronize().expect("stream sync failed");
    assert_eq!(read_i32_column_raw(out_handle, 3), vec![1, 99, 3]);

    destroy_column(out_handle);
    destroy_scalar(replacement);
    destroy_column_view(input_view);
    destroy_column(input);
}

fn replace_nulls_column_ex_uses_explicit_stream() {
    let input = create_i32_nullable_column_raw(&[1, 0, 3], 0b0000_0101);
    let input_view = column_view_raw(input);
    let replacement = create_i32_column_raw(&[10, 20, 30]);
    let replacement_view = column_view_raw(replacement);

    let stream = CudaStream::new().expect("failed to create stream");
    let mr = DeviceMemoryResource::cuda().expect("failed to create mr");
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);
    let explicit_stream = stream.testing_raw_handle_value();

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_replace_nulls_column_ex(input_view, replacement_view, &mut out_handle, &ctx) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_replace_nulls_column_ex failed");
    assert!(!out_handle.is_null());

    assert_observed_stream(explicit_stream);

    stream.synchronize().expect("stream sync failed");
    assert_eq!(read_i32_column_raw(out_handle, 3), vec![1, 20, 3]);

    destroy_column(out_handle);
    destroy_column_view(replacement_view);
    destroy_column(replacement);
    destroy_column_view(input_view);
    destroy_column(input);
}

fn gather_ex_uses_explicit_stream_and_mr() {
    let table = create_single_column_table_raw(&[10, 20, 30]);
    let table_view = table_view_raw(table);
    let gather_map = create_i32_column_raw(&[2, 0]);
    let gather_map_view = column_view_raw(gather_map);

    let stream = CudaStream::new().expect("failed to create stream");
    let mr = pool_mr();
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);
    let explicit_stream = stream.testing_raw_handle_value();
    let default_mr = obs_current_device_resource();

    obs_reset();
    let mut out_table: cudf_sys::cudf_table_handle = ptr::null_mut();
    let status = unsafe {
        cudf_sys::cudf_gather_ex(
            table_view,
            gather_map_view,
            cudf_sys::CUDF_OUT_OF_BOUNDS_POLICY_NULLIFY,
            &mut out_table,
            &ctx,
        )
    };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_gather_ex failed");
    assert!(!out_table.is_null());

    assert_observed_stream(explicit_stream);
    assert_observed_non_default_mr(default_mr);

    stream.synchronize().expect("stream sync failed");
    assert_eq!(read_first_i32_table_column_raw(out_table, 2), vec![30, 10]);

    destroy_table(out_table);
    destroy_column_view(gather_map_view);
    destroy_column(gather_map);
    destroy_table_view(table_view);
    destroy_table(table);
}

fn apply_boolean_mask_ex_uses_explicit_stream_and_mr() {
    let table = create_single_column_table_raw(&[10, 20, 30]);
    let table_view = table_view_raw(table);
    let mask = create_bool_column_raw(&[true, false, true]);
    let mask_view = column_view_raw(mask);

    let stream = CudaStream::new().expect("failed to create stream");
    let mr = pool_mr();
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);
    let explicit_stream = stream.testing_raw_handle_value();
    let default_mr = obs_current_device_resource();

    obs_reset();
    let mut out_table: cudf_sys::cudf_table_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_apply_boolean_mask_ex(table_view, mask_view, &mut out_table, &ctx) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS, "cudf_apply_boolean_mask_ex failed");
    assert!(!out_table.is_null());

    assert_observed_stream(explicit_stream);
    assert_observed_non_default_mr(default_mr);

    stream.synchronize().expect("stream sync failed");
    assert_eq!(read_first_i32_table_column_raw(out_table, 2), vec![10, 30]);

    destroy_table(out_table);
    destroy_column_view(mask_view);
    destroy_column(mask);
    destroy_table_view(table_view);
    destroy_table(table);
}

/// Production `_ex` ops reject null options; they do not fall back to ambient
/// defaults.
fn column_copy_ex_null_options_are_rejected() {
    let col = create_i32_column_raw(&[10, 20, 30]);

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_copy_ex(col, ptr::null(), &mut out_handle) };
    assert_eq!(status, cudf_sys::CUDF_ERROR_INVALID_OPERATION);
    assert!(out_handle.is_null());

    assert_eq!(obs_last_device(), -1, "rejected null options should not record execution");

    destroy_column(col);
}

/// Production `_ex` ops reject default `cudf_sys::ExecutionContext` values
/// because they contain no explicit device, stream, or memory resource.
fn column_copy_ex_default_context_is_rejected() {
    let col = create_i32_column_raw(&[1, 2, 3]);

    let ctx = cudf_sys::ExecutionContext::default();

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_copy_ex(col, ctx.as_raw(), &mut out_handle) };
    assert_eq!(status, cudf_sys::CUDF_ERROR_INVALID_OPERATION);
    assert!(out_handle.is_null());

    assert_eq!(obs_last_device(), -1, "rejected default context should not record execution");

    destroy_column(col);
}

/// Explicit memory resource is recorded by the observation hook.
fn column_copy_ex_explicit_mr_is_observed() {
    let col = create_i32_column_raw(&[5, 10, 15]);

    let stream = CudaStream::new().expect("failed to create stream");
    let mr = DeviceMemoryResource::cuda().expect("failed to create mr");
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_copy_ex(col, &ctx, &mut out_handle) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS);

    let observed_mr = obs_last_mr();
    assert!(!observed_mr.is_null(), "observed mr should be non-null");

    stream.synchronize().expect("stream sync failed");
    destroy_column(out_handle);
    destroy_column(col);
}

fn column_copy_from_view_ex_uses_explicit_stream_and_mr() {
    let col = create_i32_column_raw(&[5, 10, 15]);
    let view = column_view_raw(col);
    let stream = CudaStream::new().expect("failed to create stream");
    let mr = DeviceMemoryResource::cuda().expect("failed to create mr");
    let default_mr = obs_current_device_resource();
    let explicit_stream = stream.testing_raw_handle_value();
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_copy_from_view_ex(view, &ctx, &mut out_handle) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS);
    assert!(!out_handle.is_null());

    assert_observed_stream(explicit_stream);
    assert_observed_non_default_mr(default_mr);

    destroy_column(out_handle);
    destroy_column_view(view);
    destroy_column(col);
}

fn table_copy_ex_uses_explicit_stream_and_mr() {
    let table = create_single_column_table_raw(&[5, 10, 15]);
    let stream = CudaStream::new().expect("failed to create stream");
    let mr = DeviceMemoryResource::cuda().expect("failed to create mr");
    let default_mr = obs_current_device_resource();
    let explicit_stream = stream.testing_raw_handle_value();
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = mr.as_borrowed_sys();
    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);

    obs_reset();
    let mut out_handle: cudf_sys::cudf_table_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_table_copy_ex(table, &ctx, &mut out_handle) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS);
    assert!(!out_handle.is_null());

    assert_observed_stream(explicit_stream);
    assert_observed_non_default_mr(default_mr);

    destroy_table(out_handle);
    destroy_table(table);
}

// These tests exercise hidden safe-layer hooks that validate execution-context
// routing below the current public `Runtime v1` surface.

fn sort_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override() {
    let df = cudf::DataFrame::from_series([("id", explicit_i32_series("id", &[3, 1, 2], 501))]).unwrap();

    let stream = CudaStream::new().unwrap();
    let explicit_stream = stream.testing_raw_handle_value();
    let ctx = debug_ctx_with_stream(&stream);

    obs_reset();
    let sorted = df.testing_sort_values_in_context(["id"], cudf::SortOptions::default(), ctx).unwrap();

    assert_observed_stream(explicit_stream);
    assert_dataframe_stream_override(&sorted, 501, explicit_stream);

    obs_reset();
    assert_dataframe_i32_values(&sorted, "id", &[1, 2, 3]);
    assert_observed_sync_stream(explicit_stream);
}

fn sort_with_execution_context_accepts_runtime_seeded_inputs() {
    let ids = series::<i32>("id", &[3, 1, 2]).unwrap();
    let df = cudf::DataFrame::from_series([("id", ids)]).unwrap();

    let stream = CudaStream::new().unwrap();
    let explicit_stream = stream.testing_raw_handle_value();
    let ctx = debug_ctx_with_stream(&stream);

    obs_reset();
    let sorted = df.testing_sort_values_in_context(["id"], cudf::SortOptions::default(), ctx).unwrap();
    assert_observed_stream(explicit_stream);
    assert_eq!(sorted.testing_execution_domain_kind(), "explicit");
    assert_dataframe_i32_values(&sorted, "id", &[1, 2, 3]);
}

fn sort_with_execution_context_on_explicit_domain_uses_explicit_memory_resource() {
    let df = cudf::DataFrame::from_series([("id", explicit_i32_series("id", &[3, 1, 2], 502))]).unwrap();

    let mr = pool_mr();
    let explicit_mr = mr.testing_raw_handle_value();
    let ctx = debug_ctx_with_mr(&mr);
    let default_mr = obs_current_device_resource() as usize;

    obs_reset();
    let sorted = df.testing_sort_values_in_context(["id"], cudf::SortOptions::default(), ctx).unwrap();

    assert_ne!(obs_last_mr() as usize, default_mr);
    assert_dataframe_mr_override(&sorted, 502, explicit_mr);
}

fn merge_with_execution_context_on_explicit_domains_uses_explicit_stream() {
    let left = cudf::DataFrame::from_series([
        ("id", explicit_i32_series("id", &[1, 2, 3], 503)),
        ("val", explicit_i64_series("val", &[10, 20, 30], 503)),
    ])
    .unwrap();
    let right = cudf::DataFrame::from_series([
        ("id", explicit_i32_series("id", &[2, 3, 4], 503)),
        ("score", explicit_i64_series("score", &[200, 300, 400], 503)),
    ])
    .unwrap();

    let stream = CudaStream::new().unwrap();
    let explicit_stream = stream.testing_raw_handle_value();
    let ctx = debug_ctx_with_stream(&stream);

    obs_reset();
    let joined = left.testing_merge_in_context(&right, cudf::MergeOptions::new().how_inner().on(["id"]), ctx).unwrap();

    assert_observed_stream(explicit_stream);
    assert_eq!(joined.num_rows(), 2);
    assert_dataframe_stream_override(&joined, 503, explicit_stream);
}

fn groupby_with_execution_context_on_explicit_domain_uses_explicit_stream() {
    let df = cudf::DataFrame::from_series([
        ("key", explicit_i32_series("key", &[1, 1, 2, 2], 504)),
        ("val", explicit_i64_series("val", &[10, 20, 30, 40], 504)),
    ])
    .unwrap();

    let stream = CudaStream::new().unwrap();
    let explicit_stream = stream.testing_raw_handle_value();
    let ctx = debug_ctx_with_stream(&stream);

    obs_reset();
    let result =
        df.groupby(["key"]).testing_agg_in_context(&[cudf::AggSpec::new("val", cudf::GroupByAgg::Sum)], ctx).unwrap();

    assert_observed_stream(explicit_stream);
    assert_eq!(result.num_rows(), 2);
    assert_dataframe_stream_override(&result, 504, explicit_stream);
}

fn gather_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override() {
    let df = cudf::DataFrame::from_series([("id", explicit_i32_series("id", &[3, 1, 2], 505))]).unwrap();
    let indices = explicit_i32_series("gather", &[2, 0], 505);

    let stream = CudaStream::new().unwrap();
    let explicit_stream = stream.testing_raw_handle_value();
    let ctx = debug_ctx_with_stream(&stream);

    obs_reset();
    let gathered = df.testing_gather_in_context(&indices, ctx).unwrap();

    assert_observed_stream(explicit_stream);
    assert_dataframe_stream_override(&gathered, 505, explicit_stream);

    obs_reset();
    assert_dataframe_i32_values(&gathered, "id", &[2, 3]);
    assert_observed_sync_stream(explicit_stream);
}

fn gather_with_execution_context_accepts_runtime_seeded_inputs() {
    let ids = series::<i32>("id", &[3, 1, 2]).unwrap();
    let df = cudf::DataFrame::from_series([("id", ids)]).unwrap();
    let indices = series::<i32>("gather", &[2, 0]).unwrap();

    let stream = CudaStream::new().unwrap();
    let explicit_stream = stream.testing_raw_handle_value();
    let ctx = debug_ctx_with_stream(&stream);

    obs_reset();
    let gathered = df.testing_gather_in_context(&indices, ctx).unwrap();
    assert_observed_stream(explicit_stream);
    assert_eq!(gathered.testing_execution_domain_kind(), "explicit");
    assert_dataframe_i32_values(&gathered, "id", &[2, 3]);
}

fn filter_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override() {
    let df = cudf::DataFrame::from_series([("id", explicit_i32_series("id", &[3, 1, 2], 506))]).unwrap();
    let mask = explicit_bool_series("mask", &[true, false, true], 506);

    let stream = CudaStream::new().unwrap();
    let explicit_stream = stream.testing_raw_handle_value();
    let ctx = debug_ctx_with_stream(&stream);

    obs_reset();
    let filtered = df.testing_filter_in_context(&mask, ctx).unwrap();

    assert_observed_stream(explicit_stream);
    assert_dataframe_stream_override(&filtered, 506, explicit_stream);

    obs_reset();
    assert_dataframe_i32_values(&filtered, "id", &[3, 2]);
    assert_observed_sync_stream(explicit_stream);
}

fn filter_with_execution_context_accepts_runtime_seeded_inputs() {
    let ids = series::<i32>("id", &[3, 1, 2]).unwrap();
    let df = cudf::DataFrame::from_series([("id", ids)]).unwrap();
    let mask = series::<bool>("mask", &[true, false, true]).unwrap();

    let stream = CudaStream::new().unwrap();
    let explicit_stream = stream.testing_raw_handle_value();
    let ctx = debug_ctx_with_stream(&stream);

    obs_reset();
    let filtered = df.testing_filter_in_context(&mask, ctx).unwrap();
    assert_observed_stream(explicit_stream);
    assert_eq!(filtered.testing_execution_domain_kind(), "explicit");
    assert_dataframe_i32_values(&filtered, "id", &[3, 2]);
}

// Existing high-level APIs outside the scoped override path still derive
// explicit execution from the retained source domain.

/// Calling `Column::deep_copy()` without an override keeps using the source
/// column's retained runtime stream.
fn column_deep_copy_uses_source_runtime_stream_without_override() {
    let col = column_from_slice(&[1i32, 2, 3]).unwrap();
    let source_stream = col.testing_execution_domain_stream_value();

    obs_reset();
    let _copy = col.deep_copy().unwrap();

    let observed = obs_last_stream();
    assert_eq!(observed, source_stream, "Column::deep_copy() should use the source runtime stream");
}

/// Column copy with an explicit pool memory resource records that resource.
fn column_copy_ex_pool_mr_is_observed() {
    let col = create_i32_column_raw(&[42, 43, 44]);

    let stream = CudaStream::new().expect("failed to create stream");
    let pool_mr = pool_mr();
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = pool_mr.as_borrowed_sys();

    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);

    let default_mr = obs_current_device_resource();

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_copy_ex(col, &ctx, &mut out_handle) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS);

    let observed_mr = obs_last_mr();
    assert!(!observed_mr.is_null(), "observed mr should be non-null after _ex call");

    assert_ne!(observed_mr, default_mr, "pool MR should differ from the process-global default MR");

    stream.synchronize().expect("stream sync failed");
    destroy_column(out_handle);
    destroy_column(col);
}

fn column_copy_ex_stream_and_mr_combined() {
    let col = create_i32_column_raw(&[7, 8, 9]);

    let stream = CudaStream::new().expect("failed to create stream");
    let pool_mr = pool_mr();
    let borrowed_stream = stream.as_borrowed_sys();
    let borrowed_mr = pool_mr.as_borrowed_sys();

    let ctx = required_exec_options(&borrowed_stream, &borrowed_mr);

    let explicit_stream = stream.testing_raw_handle_value();
    let default_mr = obs_current_device_resource();

    obs_reset();
    let mut out_handle: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe { cudf_sys::cudf_column_copy_ex(col, &ctx, &mut out_handle) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS);

    assert_observed_stream(explicit_stream);
    assert_observed_non_default_mr(default_mr);

    stream.synchronize().expect("stream sync failed");
    destroy_column(out_handle);
    destroy_column(col);
}

// These verify that the cudf wrappers (CudaStream, DeviceMemoryResource)
// have the properties needed for Exec construction.

/// `CudaStream::as_borrowed_sys` preserves the sys-handle identity.
fn cuda_stream_as_borrowed_sys_preserves_handle() {
    let stream = CudaStream::new().unwrap();
    let borrowed = stream.as_borrowed_sys();
    assert_eq!(stream.testing_raw_handle_value(), obs_stream_handle_value(borrowed.as_raw()));
}

/// `CudaEvent::as_borrowed_sys` preserves the sys-handle identity.
fn cuda_event_as_borrowed_sys_preserves_handle() {
    let stream = CudaStream::new().unwrap();
    let event = CudaEvent::new().unwrap();
    event.record(&stream).unwrap();

    let borrowed = event.as_borrowed_sys();
    assert_eq!(event.testing_raw_handle_value(), borrowed.as_raw() as usize);
}

/// `DeviceMemoryResource::as_borrowed_sys` preserves the sys-handle identity.
fn device_memory_resource_as_borrowed_sys_preserves_handle() {
    let mr = DeviceMemoryResource::cuda().unwrap();
    let borrowed = mr.as_borrowed_sys();
    let mut underlying = ptr::null_mut();
    let status = unsafe { cudf_test_memory_resource_raw(borrowed.as_raw(), &mut underlying) };
    assert_eq!(status, cudf_sys::CUDF_SUCCESS);
    assert_eq!(mr.testing_raw_handle_value(), underlying as usize);
}

// These tests construct raw `cudf_exec_options_t` structs with deliberately
// invalid fields and verify that the C++ shim rejects them before performing
// any side-effects (such as device switching).

/// A `cudf_exec_options_t` with `struct_size` smaller than the actual struct
/// is rejected with `CUDF_ERROR_INVALID_OPERATION`.
fn exec_options_struct_size_too_small_returns_error() {
    let mut opts = cudf_sys::cudf_exec_options_t::defaults();
    opts.struct_size = 8; // too small

    assert_exec_options_copy_validation_error(
        opts,
        cudf_sys::CUDF_ERROR_INVALID_OPERATION,
        "struct_size too small should return CUDF_ERROR_INVALID_OPERATION",
    );
}

/// A `cudf_exec_options_t` with unknown flag bits is rejected with
/// `CUDF_ERROR_INVALID_OPERATION`.
fn exec_options_unknown_flags_returns_error() {
    let mut opts = cudf_sys::cudf_exec_options_t::defaults();
    opts.flags = 0x8000_0000; // unknown bit

    assert_exec_options_copy_validation_error(
        opts,
        cudf_sys::CUDF_ERROR_INVALID_OPERATION,
        "unknown flag bits should return CUDF_ERROR_INVALID_OPERATION",
    );
}

/// A `cudf_exec_options_t` with a non-zero reserved field is rejected with
/// `CUDF_ERROR_INVALID_OPERATION`.
fn exec_options_nonzero_reserved_returns_error() {
    let mut opts = cudf_sys::cudf_exec_options_t::defaults();
    opts.reserved[2] = 0xDEAD_BEEF;

    assert_exec_options_copy_validation_error(
        opts,
        cudf_sys::CUDF_ERROR_INVALID_OPERATION,
        "non-zero reserved field should return CUDF_ERROR_INVALID_OPERATION",
    );
}

/// A `cudf_exec_options_t` with an explicit device ordinal that does not exist
/// is rejected as invalid before any execution observation occurs.
fn exec_options_invalid_explicit_device_returns_error_without_side_effects() {
    let col = create_i32_column_raw(&[1, 2, 3]);
    let before = DeviceId::current().unwrap();

    let (_stream, _mr, mut opts) = strict_exec_owners();
    opts.device = DeviceId::count().unwrap() as i32 + 1000;

    obs_reset();
    let mut out: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe {
        cudf_sys::cudf_column_copy_ex(col, &opts as *const cudf_sys::cudf_exec_options_t as *const _, &mut out)
    };

    assert_eq!(
        status,
        cudf_sys::CUDF_ERROR_INVALID_OPERATION,
        "invalid explicit device ordinal should be rejected during strict validation"
    );
    assert!(out.is_null(), "output handle should remain null on device-selection error");
    assert_eq!(obs_last_device(), -1, "failed device selection should not record execution observation");
    assert_eq!(
        DeviceId::current().unwrap(),
        before,
        "failed device selection should not change the calling thread's current device"
    );

    destroy_column(col);
}

/// A `cudf_exec_options_t` constructed via `defaults()` is rejected for
/// production `_ex` calls because it has no explicit device, stream, or memory
/// resource.
fn exec_options_defaults_are_rejected() {
    let col = create_i32_column_raw(&[1, 2, 3]);

    let opts = cudf_sys::cudf_exec_options_t::defaults();

    let mut out: cudf_sys::cudf_column_handle = ptr::null_mut();
    let status = unsafe {
        cudf_sys::cudf_column_copy_ex(col, &opts as *const cudf_sys::cudf_exec_options_t as *const _, &mut out)
    };

    assert_eq!(status, cudf_sys::CUDF_ERROR_INVALID_OPERATION, "defaults should not satisfy strict execution");
    assert!(out.is_null(), "output handle should remain null on validation error");

    destroy_column(col);
}

#[test]
fn execution_context_tests_cases() {
    run_named_cases(&[
        ("column_copy_ex_uses_explicit_stream", column_copy_ex_uses_explicit_stream),
        ("column_to_host_ex_uses_explicit_stream", column_to_host_ex_uses_explicit_stream),
        (
            "column_to_host_ex_reads_sliced_view_from_element_offset",
            column_to_host_ex_reads_sliced_view_from_element_offset,
        ),
        (
            "strings_to_host_ex_accounts_for_sliced_parent_view_offset",
            strings_to_host_ex_accounts_for_sliced_parent_view_offset,
        ),
        ("column_from_device_data_ex_uses_explicit_stream", column_from_device_data_ex_uses_explicit_stream),
        ("replace_nulls_scalar_ex_uses_explicit_stream", replace_nulls_scalar_ex_uses_explicit_stream),
        ("replace_nulls_column_ex_uses_explicit_stream", replace_nulls_column_ex_uses_explicit_stream),
        ("gather_ex_uses_explicit_stream_and_mr", gather_ex_uses_explicit_stream_and_mr),
        ("apply_boolean_mask_ex_uses_explicit_stream_and_mr", apply_boolean_mask_ex_uses_explicit_stream_and_mr),
        ("column_copy_ex_null_options_are_rejected", column_copy_ex_null_options_are_rejected),
        ("column_copy_ex_default_context_is_rejected", column_copy_ex_default_context_is_rejected),
        ("column_copy_ex_explicit_mr_is_observed", column_copy_ex_explicit_mr_is_observed),
        ("column_copy_from_view_ex_uses_explicit_stream_and_mr", column_copy_from_view_ex_uses_explicit_stream_and_mr),
        ("table_copy_ex_uses_explicit_stream_and_mr", table_copy_ex_uses_explicit_stream_and_mr),
        (
            "sort_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override",
            sort_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override,
        ),
        (
            "sort_with_execution_context_accepts_runtime_seeded_inputs",
            sort_with_execution_context_accepts_runtime_seeded_inputs,
        ),
        (
            "sort_with_execution_context_on_explicit_domain_uses_explicit_memory_resource",
            sort_with_execution_context_on_explicit_domain_uses_explicit_memory_resource,
        ),
        (
            "merge_with_execution_context_on_explicit_domains_uses_explicit_stream",
            merge_with_execution_context_on_explicit_domains_uses_explicit_stream,
        ),
        (
            "groupby_with_execution_context_on_explicit_domain_uses_explicit_stream",
            groupby_with_execution_context_on_explicit_domain_uses_explicit_stream,
        ),
        (
            "gather_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override",
            gather_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override,
        ),
        (
            "gather_with_execution_context_accepts_runtime_seeded_inputs",
            gather_with_execution_context_accepts_runtime_seeded_inputs,
        ),
        (
            "filter_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override",
            filter_with_execution_context_on_explicit_domain_uses_explicit_stream_and_retains_override,
        ),
        (
            "filter_with_execution_context_accepts_runtime_seeded_inputs",
            filter_with_execution_context_accepts_runtime_seeded_inputs,
        ),
        (
            "column_deep_copy_uses_source_runtime_stream_without_override",
            column_deep_copy_uses_source_runtime_stream_without_override,
        ),
        ("column_copy_ex_pool_mr_is_observed", column_copy_ex_pool_mr_is_observed),
        ("column_copy_ex_stream_and_mr_combined", column_copy_ex_stream_and_mr_combined),
        ("cuda_stream_as_borrowed_sys_preserves_handle", cuda_stream_as_borrowed_sys_preserves_handle),
        ("cuda_event_as_borrowed_sys_preserves_handle", cuda_event_as_borrowed_sys_preserves_handle),
        (
            "device_memory_resource_as_borrowed_sys_preserves_handle",
            device_memory_resource_as_borrowed_sys_preserves_handle,
        ),
        ("exec_options_struct_size_too_small_returns_error", exec_options_struct_size_too_small_returns_error),
        ("exec_options_unknown_flags_returns_error", exec_options_unknown_flags_returns_error),
        ("exec_options_nonzero_reserved_returns_error", exec_options_nonzero_reserved_returns_error),
        (
            "exec_options_invalid_explicit_device_returns_error_without_side_effects",
            exec_options_invalid_explicit_device_returns_error_without_side_effects,
        ),
        ("exec_options_defaults_are_rejected", exec_options_defaults_are_rejected),
    ]);
}