17045759575

Committed 18 Aug 2025 03:57PM UTC coverage: 86.067% (-1.8%) from 87.913%

Build # 17045759575

Build Type

Pull #4215

github

Committed by

web-flow

Commit Message

Merge 90bf5f689 into 7eb8ac9fa

Pull Request Pull Request #4215: Ji/vectors

Run Details

136 of 1803 new or added lines in 42 files covered. (7.54%)

125 existing lines in 26 files now uncovered.

56919 of 66133 relevant lines covered (86.07%)

612059.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/vortex-vector/src/operators/primitive.rs

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

use std::any::Any;
use std::hash::Hash;
use std::rc::Rc;
use std::task::{Poll, ready};

use vortex_buffer::{Buffer, ByteBuffer};
use vortex_dtype::{NativePType, PType, match_each_native_ptype};
use vortex_error::VortexResult;

use crate::bits::BitView;
use crate::buffers::BufferHandle;
use crate::operators::{BindContext, Operator};
use crate::types::{Element, VType};
use crate::view::ViewMut;
use crate::{Kernel, KernelContext, SC};

#[derive(Debug, Clone, Hash)]
pub struct PrimitiveOperator {
    ptype: PType,
    byte_buffer: ByteBuffer,
}

impl PrimitiveOperator {
    pub fn new(ptype: PType, byte_buffer: ByteBuffer) -> Self {
        Self { ptype, byte_buffer }
    }
}

impl Operator for PrimitiveOperator {
    fn as_any(&self) -> &dyn Any {
        self
    }

    fn vtype(&self) -> VType {
        VType::Primitive(self.ptype)
    }

    fn children(&self) -> &[Rc<dyn Operator>] {
        &[]
    }

    fn with_children(&self, children: Vec<Rc<dyn Operator>>) -> Rc<dyn Operator> {
        Rc::new(self.clone())
    }

    fn bind(&self, ctx: &dyn BindContext) -> VortexResult<Box<dyn Kernel>> {
        match_each_native_ptype!(self.ptype, |T| {
            Ok(Box::new(PrimitiveKernel::<T> {
                buffer: BufferHandle::new(Buffer::from_byte_buffer(self.byte_buffer.clone())),
                offset: 0,
            }) as Box<dyn Kernel>)
        })
    }
}

/// A kernel that produces primitive values from a byte buffer.
pub struct PrimitiveKernel<T: NativePType> {
    buffer: BufferHandle<T>,
    offset: usize,
}

impl<T: Element + NativePType> Kernel for PrimitiveKernel<T> {
    fn seek(&mut self, chunk_idx: usize) -> VortexResult<()> {
        self.offset = chunk_idx * SC;
        Ok(())
    }

    fn step(
        &mut self,
        ctx: &dyn KernelContext,
        mask: BitView,
        out: &mut ViewMut,
    ) -> Poll<VortexResult<()>> {
        // FIXME(ngates): support mask.
        // assert_eq!(mask.true_count(), N, "Mask must have exactly N true bits");

        let buffer = ready!(self.buffer.get_or_load(ctx))?;
        let remaining = buffer.len() - self.offset;

        let out_slice = out.as_slice_mut::<T>();

        if remaining > SC {
            out_slice.copy_from_slice(&buffer[self.offset..][..SC]);
            self.offset += SC;
        } else {
            out_slice[..remaining].copy_from_slice(&buffer[self.offset..]);
            self.offset += remaining;
        }

        // TODO(joe): use mask in copy_from_slice, if faster.
        out.select_mask::<T>(&mask);

        Poll::Ready(Ok(()))
    }
}

#[cfg(test)]
mod tests {
    use std::task::Poll;

    use vortex_array::{IntoArray, ToCanonical};
    use vortex_buffer::{BufferMut, ByteBuffer};

    use super::*;
    use crate::bits::BitView;
    use crate::{BufferId, VectorId, VectorRef};

    struct MockContext;

    impl KernelContext for MockContext {
        fn vector(&self, _vector_id: VectorId) -> VectorRef<'_> {
            unimplemented!("not needed for these tests")
        }

        fn buffer(&self, _buffer_id: BufferId) -> Poll<VortexResult<ByteBuffer>> {
            unimplemented!("not needed for these tests")
        }
    }

    #[test]
    fn test_primitive_kernel_basic_operation() {
        // Create a primitive array with values 0..16
        let size = 16;
        let values = (0..i32::try_from(size).unwrap()).collect::<BufferMut<_>>();
        let primitive_array = values.into_array().to_primitive().unwrap();

        // Create the kernel
        let mut kernel = PrimitiveKernel::<i32> {
            buffer: BufferHandle::new(primitive_array.buffer()),
            offset: 0,
        };

        // Create an all-true mask for simplicity
        let mask_data = [u64::MAX; SC / 64];
        let mask_view = BitView::new(&mask_data);

        // Create output buffer
        let mut output = BufferMut::<i32>::with_capacity(SC);
        unsafe { output.set_len(SC) };
        let mut output_view = ViewMut::new(&mut output[..], None);

        // Create a mock context
        let ctx = MockContext;

        // Execute the step
        let result = kernel.step(&ctx, mask_view, &mut output_view);
        assert!(matches!(result, Poll::Ready(Ok(()))));

        // Verify the first elements contain our values
        for i in 0..size {
            assert_eq!(
                output[i],
                i32::try_from(i).unwrap(),
                "Mismatch at position {}: expected {}, got {}",
                i,
                i,
                output[i]
            );
        }
    }

    #[test]
    fn test_primitive_kernel_with_mask() {
        // Create a primitive array with values 0..16
        let size = 16;
        let values = (0..size).collect::<BufferMut<_>>();
        let primitive_array = values.into_array().to_primitive().unwrap();

        // Create the kernel
        let mut kernel = PrimitiveKernel::<i32> {
            buffer: BufferHandle::new(primitive_array.buffer()),
            offset: 0,
        };

        // Create a mask with alternating bits (every other element selected)
        let mut mask_data = [0u64; SC / 64];
        // Set bits 0, 2, 4, 6, 8, 10, 12, 14 (first 8 even positions)
        for i in 0..8 {
            let bit_pos = i * 2;
            let word_idx = bit_pos / 64;
            let bit_idx = bit_pos % 64;
            mask_data[word_idx] |= 1u64 << bit_idx; // MSB ordering
        }
        let true_count = 8;
        let mask_view = BitView::new(&mask_data);

        // Create output buffer
        let mut output = BufferMut::<i32>::with_capacity(SC);
        unsafe { output.set_len(SC) };
        let mut output_view = ViewMut::new(&mut output[..], None);

        // Create a mock context
        let ctx = MockContext;

        // Execute the step
        let result = kernel.step(&ctx, mask_view, &mut output_view);
        assert!(matches!(result, Poll::Ready(Ok(()))));
        unsafe { output.set_len(mask_view.true_count()) };

        // Verify that the mask was applied successfully
        // The select_mask operation filters elements based on the mask

        // Count elements that have been affected by mask selection
        // Note: element 0 is a valid selected value, so we need to count differently
        let non_zero_count = output.iter().filter(|&&x| x != 0).count();

        // Verify that element 0 was selected (first bit in mask is 1)
        assert_eq!(output[0], 0, "First element should be 0 since bit 0 is set");

        // Since element 0 is valid but counts as zero, the actual selection count is non_zero_count + 1
        let actual_selected = non_zero_count + 1; // +1 for the zero at position 0

        // The exact number of selected elements should match our true_count
        assert_eq!(
            actual_selected, true_count,
            "Selected element count should match true_count"
        )
    }

    #[test]
    fn test_primitive_kernel_offset_tracking() {
        // Create a primitive array with more than N values
        let total_size = SC + 100;
        let values = (0..i32::try_from(total_size).unwrap()).collect::<BufferMut<_>>();
        let primitive_array = values.into_array().to_primitive().unwrap();

        // Create the kernel
        let mut kernel = PrimitiveKernel::<i32> {
            buffer: BufferHandle::new(primitive_array.buffer()),
            offset: 0,
        };

        // All-true mask
        let mask_data = [u64::MAX; SC / 64];
        let mask_view = BitView::new(&mask_data);
        let ctx = MockContext;

        // First step should process first N elements
        {
            let mut output = BufferMut::<i32>::with_capacity(SC);
            unsafe { output.set_len(SC) };
            let mut output_view = ViewMut::new(&mut output[..], None);

            let result = kernel.step(&ctx, mask_view, &mut output_view);
            assert!(matches!(result, Poll::Ready(Ok(()))));
            assert_eq!(kernel.offset, SC);

            // Verify first chunk
            for i in 0..SC {
                assert_eq!(output[i], i32::try_from(i).unwrap(), "{i}");
            }
        }

        // Second step should process remaining elements (partial chunk)
        {
            let mut output = BufferMut::<i32>::with_capacity(SC);
            unsafe { output.set_len(SC) };
            let mut output_view = ViewMut::new(&mut output[..], None);

            let result = kernel.step(&ctx, mask_view, &mut output_view);
            assert!(matches!(result, Poll::Ready(Ok(()))));
            assert_eq!(kernel.offset, total_size);

            // Verify remaining elements (first 100 should be valid)
            for i in 0..100 {
                assert_eq!(output[i], i32::try_from(SC + i).unwrap());
            }
        }
    }
}

1	// SPDX-License-Identifier: Apache-2.0
2	// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4	use std::any::Any;
5	use std::hash::Hash;
6	use std::rc::Rc;
7	use std::task::{Poll, ready};
8
9	use vortex_buffer::{Buffer, ByteBuffer};
10	use vortex_dtype::{NativePType, PType, match_each_native_ptype};
11	use vortex_error::VortexResult;
12
13	use crate::bits::BitView;
14	use crate::buffers::BufferHandle;
15	use crate::operators::{BindContext, Operator};
16	use crate::types::{Element, VType};
17	use crate::view::ViewMut;
18	use crate::{Kernel, KernelContext, SC};
19
20	#[derive(Debug, Clone, Hash)]
21	pub struct PrimitiveOperator {
22	ptype: PType,
23	byte_buffer: ByteBuffer,
24	}
25
26	impl PrimitiveOperator {
NEW 27	pub fn new(ptype: PType, byte_buffer: ByteBuffer) -> Self {	×
NEW 28	Self { ptype, byte_buffer }	×
NEW 29	}	×
30	}
31
32	impl Operator for PrimitiveOperator {
NEW 33	fn as_any(&self) -> &dyn Any {	×
NEW 34	self	×
NEW 35	}	×
36
NEW 37	fn vtype(&self) -> VType {	×
NEW 38	VType::Primitive(self.ptype)	×
NEW 39	}	×
40
NEW 41	fn children(&self) -> &[Rc<dyn Operator>] {	×
NEW 42	&[]	×
NEW 43	}	×
44
NEW 45	fn with_children(&self, children: Vec<Rc<dyn Operator>>) -> Rc<dyn Operator> {	×
NEW 46	Rc::new(self.clone())	×
NEW 47	}	×
48
NEW 49	fn bind(&self, ctx: &dyn BindContext) -> VortexResult<Box<dyn Kernel>> {	×
NEW 50	match_each_native_ptype!(self.ptype, \|T\| {	×
NEW 51	Ok(Box::new(PrimitiveKernel::<T> {	×
NEW 52	buffer: BufferHandle::new(Buffer::from_byte_buffer(self.byte_buffer.clone())),	×
NEW 53	offset: 0,	×
NEW 54	}) as Box<dyn Kernel>)	×
55	})
NEW 56	}	×
57	}
58
59	/// A kernel that produces primitive values from a byte buffer.
60	pub struct PrimitiveKernel<T: NativePType> {
61	buffer: BufferHandle<T>,
62	offset: usize,
63	}
64
65	impl<T: Element + NativePType> Kernel for PrimitiveKernel<T> {
NEW 66	fn seek(&mut self, chunk_idx: usize) -> VortexResult<()> {	×
NEW 67	self.offset = chunk_idx * SC;	×
NEW 68	Ok(())	×
NEW 69	}	×
70
NEW 71	fn step(	×
NEW 72	&mut self,	×
NEW 73	ctx: &dyn KernelContext,	×
NEW 74	mask: BitView,	×
NEW 75	out: &mut ViewMut,	×
NEW 76	) -> Poll<VortexResult<()>> {	×
77	// FIXME(ngates): support mask.
78	// assert_eq!(mask.true_count(), N, "Mask must have exactly N true bits");
79
NEW 80	let buffer = ready!(self.buffer.get_or_load(ctx))?;	×
NEW 81	let remaining = buffer.len() - self.offset;	×
82
NEW 83	let out_slice = out.as_slice_mut::<T>();	×
84
NEW 85	if remaining > SC {	×
NEW 86	out_slice.copy_from_slice(&buffer[self.offset..][..SC]);	×
NEW 87	self.offset += SC;	×
NEW 88	} else {	×
NEW 89	out_slice[..remaining].copy_from_slice(&buffer[self.offset..]);	×
NEW 90	self.offset += remaining;	×
NEW 91	}	×
92
93	// TODO(joe): use mask in copy_from_slice, if faster.
NEW 94	out.select_mask::<T>(&mask);	×
95
NEW 96	Poll::Ready(Ok(()))	×
NEW 97	}	×
98	}
99
100	#[cfg(test)]
101	mod tests {
102	use std::task::Poll;
103
104	use vortex_array::{IntoArray, ToCanonical};
105	use vortex_buffer::{BufferMut, ByteBuffer};
106
107	use super::*;
108	use crate::bits::BitView;
109	use crate::{BufferId, VectorId, VectorRef};
110
111	struct MockContext;
112
113	impl KernelContext for MockContext {
114	fn vector(&self, _vector_id: VectorId) -> VectorRef<'_> {
115	unimplemented!("not needed for these tests")
116	}
117
118	fn buffer(&self, _buffer_id: BufferId) -> Poll<VortexResult<ByteBuffer>> {
119	unimplemented!("not needed for these tests")
120	}
121	}
122
123	#[test]
124	fn test_primitive_kernel_basic_operation() {
125	// Create a primitive array with values 0..16
126	let size = 16;
127	let values = (0..i32::try_from(size).unwrap()).collect::<BufferMut<_>>();
128	let primitive_array = values.into_array().to_primitive().unwrap();
129
130	// Create the kernel
131	let mut kernel = PrimitiveKernel::<i32> {
132	buffer: BufferHandle::new(primitive_array.buffer()),
133	offset: 0,
134	};
135
136	// Create an all-true mask for simplicity
137	let mask_data = [u64::MAX; SC / 64];
138	let mask_view = BitView::new(&mask_data);
139
140	// Create output buffer
141	let mut output = BufferMut::<i32>::with_capacity(SC);
142	unsafe { output.set_len(SC) };
143	let mut output_view = ViewMut::new(&mut output[..], None);
144
145	// Create a mock context
146	let ctx = MockContext;
147
148	// Execute the step
149	let result = kernel.step(&ctx, mask_view, &mut output_view);
150	assert!(matches!(result, Poll::Ready(Ok(()))));
151
152	// Verify the first elements contain our values
153	for i in 0..size {
154	assert_eq!(
155	output[i],
156	i32::try_from(i).unwrap(),
157	"Mismatch at position {}: expected {}, got {}",
158	i,
159	i,
160	output[i]
161	);
162	}
163	}
164
165	#[test]
166	fn test_primitive_kernel_with_mask() {
167	// Create a primitive array with values 0..16
168	let size = 16;
169	let values = (0..size).collect::<BufferMut<_>>();
170	let primitive_array = values.into_array().to_primitive().unwrap();
171
172	// Create the kernel
173	let mut kernel = PrimitiveKernel::<i32> {
174	buffer: BufferHandle::new(primitive_array.buffer()),
175	offset: 0,
176	};
177
178	// Create a mask with alternating bits (every other element selected)
179	let mut mask_data = [0u64; SC / 64];
180	// Set bits 0, 2, 4, 6, 8, 10, 12, 14 (first 8 even positions)
181	for i in 0..8 {
182	let bit_pos = i * 2;
183	let word_idx = bit_pos / 64;
184	let bit_idx = bit_pos % 64;
185	mask_data[word_idx] \|= 1u64 << bit_idx; // MSB ordering
186	}
187	let true_count = 8;
188	let mask_view = BitView::new(&mask_data);
189
190	// Create output buffer
191	let mut output = BufferMut::<i32>::with_capacity(SC);
192	unsafe { output.set_len(SC) };
193	let mut output_view = ViewMut::new(&mut output[..], None);
194
195	// Create a mock context
196	let ctx = MockContext;
197
198	// Execute the step
199	let result = kernel.step(&ctx, mask_view, &mut output_view);
200	assert!(matches!(result, Poll::Ready(Ok(()))));
201	unsafe { output.set_len(mask_view.true_count()) };
202
203	// Verify that the mask was applied successfully
204	// The select_mask operation filters elements based on the mask
205
206	// Count elements that have been affected by mask selection
207	// Note: element 0 is a valid selected value, so we need to count differently
208	let non_zero_count = output.iter().filter(\|&&x\| x != 0).count();
209
210	// Verify that element 0 was selected (first bit in mask is 1)
211	assert_eq!(output[0], 0, "First element should be 0 since bit 0 is set");
212
213	// Since element 0 is valid but counts as zero, the actual selection count is non_zero_count + 1
214	let actual_selected = non_zero_count + 1; // +1 for the zero at position 0
215
216	// The exact number of selected elements should match our true_count
217	assert_eq!(
218	actual_selected, true_count,
219	"Selected element count should match true_count"
220	)
221	}
222
223	#[test]
224	fn test_primitive_kernel_offset_tracking() {
225	// Create a primitive array with more than N values
226	let total_size = SC + 100;
227	let values = (0..i32::try_from(total_size).unwrap()).collect::<BufferMut<_>>();
228	let primitive_array = values.into_array().to_primitive().unwrap();
229
230	// Create the kernel
231	let mut kernel = PrimitiveKernel::<i32> {
232	buffer: BufferHandle::new(primitive_array.buffer()),
233	offset: 0,
234	};
235
236	// All-true mask
237	let mask_data = [u64::MAX; SC / 64];
238	let mask_view = BitView::new(&mask_data);
239	let ctx = MockContext;
240
241	// First step should process first N elements
242	{
243	let mut output = BufferMut::<i32>::with_capacity(SC);
244	unsafe { output.set_len(SC) };
245	let mut output_view = ViewMut::new(&mut output[..], None);
246
247	let result = kernel.step(&ctx, mask_view, &mut output_view);
248	assert!(matches!(result, Poll::Ready(Ok(()))));
249	assert_eq!(kernel.offset, SC);
250
251	// Verify first chunk
252	for i in 0..SC {
253	assert_eq!(output[i], i32::try_from(i).unwrap(), "{i}");
254	}
255	}
256
257	// Second step should process remaining elements (partial chunk)
258	{
259	let mut output = BufferMut::<i32>::with_capacity(SC);
260	unsafe { output.set_len(SC) };
261	let mut output_view = ViewMut::new(&mut output[..], None);
262
263	let result = kernel.step(&ctx, mask_view, &mut output_view);
264	assert!(matches!(result, Poll::Ready(Ok(()))));
265	assert_eq!(kernel.offset, total_size);
266
267	// Verify remaining elements (first 100 should be valid)
268	for i in 0..100 {
269	assert_eq!(output[i], i32::try_from(SC + i).unwrap());
270	}
271	}
272	}
273	}

vortex-data / vortex / 17045759575

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous