vortex_array/arrays/varbinview/vtable/
mod.rs1use std::hash::Hasher;
5use std::mem::size_of;
6use std::sync::Arc;
7
8use vortex_buffer::Buffer;
9use vortex_error::VortexResult;
10use vortex_error::vortex_bail;
11use vortex_error::vortex_ensure;
12use vortex_error::vortex_ensure_eq;
13use vortex_error::vortex_err;
14use vortex_error::vortex_panic;
15use vortex_session::VortexSession;
16use vortex_session::registry::CachedId;
17
18use crate::ArrayParts;
19use crate::ArrayRef;
20use crate::EqMode;
21use crate::ExecutionCtx;
22use crate::ExecutionResult;
23use crate::VortexSessionExecute;
24use crate::array::Array;
25use crate::array::ArrayId;
26use crate::array::ArrayView;
27use crate::array::VTable;
28use crate::arrays::varbinview::BinaryView;
29use crate::arrays::varbinview::VarBinViewData;
30use crate::arrays::varbinview::array::VarBinViewSlots;
31use crate::arrays::varbinview::compute::rules::PARENT_RULES;
32use crate::buffer::BufferHandle;
33use crate::builders::ArrayBuilder;
34use crate::builders::VarBinViewBuilder;
35use crate::dtype::DType;
36use crate::hash::ArrayEq;
37use crate::hash::ArrayHash;
38use crate::match_each_varbin_builder;
39use crate::serde::ArrayChildren;
40use crate::validity::Validity;
41mod kernel;
42mod operations;
43mod validity;
44pub type VarBinViewArray = Array<VarBinView>;
46
47pub(crate) fn initialize(session: &VortexSession) {
48 kernel::initialize(session);
49}
50
51#[derive(Clone, Debug)]
52pub struct VarBinView;
53
54impl ArrayHash for VarBinViewData {
55 fn array_hash<H: Hasher>(&self, state: &mut H, accuracy: EqMode) {
56 for buffer in self.buffers.iter() {
57 buffer.array_hash(state, accuracy);
58 }
59 self.views.array_hash(state, accuracy);
60 }
61}
62
63impl ArrayEq for VarBinViewData {
64 fn array_eq(&self, other: &Self, accuracy: EqMode) -> bool {
65 self.buffers.len() == other.buffers.len()
66 && self
67 .buffers
68 .iter()
69 .zip(other.buffers.iter())
70 .all(|(a, b)| a.array_eq(b, accuracy))
71 && self.views.array_eq(&other.views, accuracy)
72 }
73}
74
75impl VTable for VarBinView {
76 type TypedArrayData = VarBinViewData;
77
78 type OperationsVTable = Self;
79 type ValidityVTable = Self;
80
81 fn id(&self) -> ArrayId {
82 static ID: CachedId = CachedId::new("vortex.varbinview");
83 *ID
84 }
85
86 fn nbuffers(array: ArrayView<'_, Self>) -> usize {
87 array.data_buffers().len() + 1
88 }
89
90 fn validate(
91 &self,
92 data: &VarBinViewData,
93 dtype: &DType,
94 len: usize,
95 slots: &[Option<ArrayRef>],
96 ) -> VortexResult<()> {
97 vortex_ensure_eq!(slots.len(), VarBinViewSlots::COUNT);
98 vortex_ensure_eq!(
99 data.len(),
100 len,
101 "VarBinViewArray length does not match outer length",
102 );
103 vortex_ensure!(
104 matches!(dtype, DType::Binary(_) | DType::Utf8(_)),
105 "VarBinViewArray dtype must be binary or utf8, got {dtype}"
106 );
107 Ok(())
108 }
109
110 fn buffer(array: ArrayView<'_, Self>, idx: usize) -> BufferHandle {
111 let ndata = array.data_buffers().len();
112 if idx < ndata {
113 array.data_buffers()[idx].clone()
114 } else if idx == ndata {
115 array.views_handle().clone()
116 } else {
117 vortex_panic!("VarBinViewArray buffer index {idx} out of bounds")
118 }
119 }
120
121 fn buffer_name(array: ArrayView<'_, Self>, idx: usize) -> Option<String> {
122 let ndata = array.data_buffers().len();
123 if idx < ndata {
124 Some(format!("buffer_{idx}"))
125 } else if idx == ndata {
126 Some("views".to_string())
127 } else {
128 vortex_panic!("VarBinViewArray buffer_name index {idx} out of bounds")
129 }
130 }
131
132 fn with_buffers(
133 &self,
134 array: ArrayView<'_, Self>,
135 buffers: &[BufferHandle],
136 ) -> VortexResult<ArrayParts<Self>> {
137 let Some((views, data_buffers)) = buffers.split_last() else {
138 vortex_bail!("Expected at least 1 buffer, got 0");
139 };
140 let data = VarBinViewData::try_new_handle(
141 views.clone(),
142 Arc::from(data_buffers.to_vec()),
143 array.dtype().clone(),
144 array.validity()?,
145 )?;
146 Ok(ArrayParts::new(
147 self.clone(),
148 array.dtype().clone(),
149 array.len(),
150 data,
151 array.slots().iter().cloned().collect(),
152 ))
153 }
154
155 fn serialize(
156 _array: ArrayView<'_, Self>,
157 _session: &VortexSession,
158 ) -> VortexResult<Option<Vec<u8>>> {
159 Ok(Some(vec![]))
160 }
161
162 fn deserialize(
163 &self,
164 dtype: &DType,
165 len: usize,
166 metadata: &[u8],
167
168 buffers: &[BufferHandle],
169 children: &dyn ArrayChildren,
170 session: &VortexSession,
171 ) -> VortexResult<ArrayParts<Self>> {
172 if !metadata.is_empty() {
173 vortex_bail!(
174 "VarBinViewArray expects empty metadata, got {} bytes",
175 metadata.len()
176 );
177 }
178 let Some((views_handle, data_handles)) = buffers.split_last() else {
179 vortex_bail!("Expected at least 1 buffer, got 0");
180 };
181
182 let validity = if children.is_empty() {
183 Validity::from(dtype.nullability())
184 } else if children.len() == 1 {
185 let validity = children.get(0, &Validity::DTYPE, len)?;
186 Validity::Array(validity)
187 } else {
188 vortex_bail!("Expected 0 or 1 children, got {}", children.len());
189 };
190
191 let views_nbytes = views_handle.len();
192 let expected_views_nbytes = len
193 .checked_mul(size_of::<BinaryView>())
194 .ok_or_else(|| vortex_err!("views byte length overflow for len={len}"))?;
195 if views_nbytes != expected_views_nbytes {
196 vortex_bail!(
197 "Expected views buffer length {} bytes, got {} bytes",
198 expected_views_nbytes,
199 views_nbytes
200 );
201 }
202
203 if buffers.iter().any(|b| b.is_on_device()) {
205 let data = VarBinViewData::try_new_handle(
206 views_handle.clone(),
207 Arc::from(data_handles.to_vec()),
208 dtype.clone(),
209 validity.clone(),
210 )?;
211 let slots = VarBinViewData::make_slots(&validity, len);
212 return Ok(ArrayParts::new(
213 self.clone(),
214 dtype.clone(),
215 len,
216 data,
217 slots,
218 ));
219 }
220
221 let data_buffers = data_handles
222 .iter()
223 .map(|b| b.as_host().clone())
224 .collect::<Vec<_>>();
225 let views = Buffer::<BinaryView>::from_byte_buffer(views_handle.clone().as_host().clone());
226
227 let data = VarBinViewData::try_new(
228 views,
229 Arc::from(data_buffers),
230 dtype.clone(),
231 validity.clone(),
232 &mut session.create_execution_ctx(),
233 )?;
234 let slots = VarBinViewData::make_slots(&validity, len);
235 Ok(ArrayParts::new(
236 self.clone(),
237 dtype.clone(),
238 len,
239 data,
240 slots,
241 ))
242 }
243
244 fn slot_name(_array: ArrayView<'_, Self>, idx: usize) -> String {
245 VarBinViewSlots::NAMES[idx].to_string()
246 }
247
248 fn reduce_parent(
249 array: ArrayView<'_, Self>,
250 parent: &ArrayRef,
251 child_idx: usize,
252 ) -> VortexResult<Option<ArrayRef>> {
253 PARENT_RULES.evaluate(array, parent, child_idx)
254 }
255
256 fn append_to_builder(
257 array: ArrayView<'_, Self>,
258 builder: &mut dyn ArrayBuilder,
259 ctx: &mut ExecutionCtx,
260 ) -> VortexResult<()> {
261 if let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() {
262 return builder.append_varbinview_array(&array.into_owned(), ctx);
263 }
264 if let Some(result) =
265 match_each_varbin_builder!(builder, |builder| builder.append_varbinview(array, ctx))
266 {
267 return result;
268 }
269 vortex_bail!("append_to_builder for VarBinView requires a variable-binary builder")
270 }
271
272 fn execute(array: Array<Self>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
273 Ok(ExecutionResult::done(array))
274 }
275}
276
277#[cfg(test)]
278mod tests {
279 use vortex_buffer::ByteBufferMut;
280 use vortex_session::registry::ReadContext;
281
282 use super::*;
283 use crate::ArrayContext;
284 use crate::IntoArray;
285 use crate::VortexSessionExecute;
286 use crate::array_session;
287 use crate::assert_arrays_eq;
288 use crate::serde::SerializeOptions;
289 use crate::serde::SerializedArray;
290
291 #[test]
292 fn test_nullable_varbinview_serde_roundtrip() {
293 let array = VarBinViewArray::from_iter_nullable_str([
294 Some("hello"),
295 None,
296 Some("world"),
297 None,
298 Some("a moderately long string for testing"),
299 ]);
300 let dtype = array.dtype().clone();
301 let len = array.len();
302
303 let session = array_session();
304 let mut ctx = session.create_execution_ctx();
305 let array_ctx = ArrayContext::empty();
306 let serialized = array
307 .clone()
308 .into_array()
309 .serialize(&array_ctx, &session, &SerializeOptions::default())
310 .unwrap();
311
312 let mut concat = ByteBufferMut::empty();
313 for buf in serialized {
314 concat.extend_from_slice(buf.as_ref());
315 }
316 let parts = SerializedArray::try_from(concat.freeze()).unwrap();
317 let decoded = parts
318 .decode(&dtype, len, &ReadContext::new(array_ctx.to_ids()), &session)
319 .unwrap();
320
321 assert_arrays_eq!(decoded, array, &mut ctx);
322 }
323}