1#[cfg(feature = "fast_hash")]
21use ahash::AHashSet as HashSet;
22#[cfg(not(feature = "fast_hash"))]
23use std::collections::HashSet;
24use std::simd::{Mask, MaskElement};
25use std::{fmt::Display, sync::Arc};
26
27use crate::enums::error::KernelError;
28#[cfg(feature = "chunked")]
29use crate::enums::error::MinarrowError;
30#[cfg(feature = "chunked")]
31use crate::structs::field_array::create_field_for_array;
32use crate::traits::masked_array::MaskedArray;
33#[cfg(feature = "chunked")]
34use crate::{Array, FieldArray, SuperArray};
35use crate::{
36 Bitmask, CategoricalArray, Float, FloatArray, Integer, IntegerArray, StringArray, TextArray,
37};
38
39#[inline(always)]
40pub fn validate_null_mask_len(data_len: usize, null_mask: &Option<Bitmask>) {
41 if let Some(mask) = null_mask {
42 assert_eq!(
43 mask.len(),
44 data_len,
45 "Validation Error: Null mask length ({}) does not match data length ({})",
46 mask.len(),
47 data_len
48 );
49 }
50}
51
52pub fn parse_datetime_str(s: &str) -> Option<i64> {
58 if s.is_empty() {
60 return None;
61 }
62
63 #[cfg(feature = "datetime_ops")]
64 {
65 use time::{
66 Date, OffsetDateTime, PrimitiveDateTime, Time,
67 format_description::well_known::{Iso8601, Rfc3339},
68 macros::format_description,
69 };
70
71 if let Ok(dt) = OffsetDateTime::parse(s, &Rfc3339) {
73 return Some(dt.unix_timestamp() * 1_000 + (dt.nanosecond() / 1_000_000) as i64);
74 }
75
76 if let Ok(dt) = OffsetDateTime::parse(s, &Iso8601::DEFAULT) {
78 return Some(dt.unix_timestamp() * 1_000 + (dt.nanosecond() / 1_000_000) as i64);
79 }
80
81 let format = format_description!("[year]-[month]-[day] [hour]:[minute]:[second]");
83 if let Ok(dt) = PrimitiveDateTime::parse(s, format) {
84 let dt_utc = dt.assume_utc();
85 return Some(
86 dt_utc.unix_timestamp() * 1_000 + (dt_utc.nanosecond() / 1_000_000) as i64,
87 );
88 }
89
90 let date_format = format_description!("[year]-[month]-[day]");
92 if let Ok(date) = Date::parse(s, date_format) {
93 if let Ok(dt) = date.with_hms(0, 0, 0) {
94 let dt_utc = dt.assume_utc();
95 return Some(dt_utc.unix_timestamp() * 1_000);
96 }
97 }
98
99 let time_format = format_description!("[hour]:[minute]:[second]");
101 if let Ok(time) = Time::parse(s, time_format) {
102 let today = OffsetDateTime::now_utc().date();
103 let dt_primitive = today.with_time(time);
104 let dt_utc = dt_primitive.assume_utc();
105 return Some(
106 dt_utc.unix_timestamp() * 1_000 + (dt_utc.nanosecond() / 1_000_000) as i64,
107 );
108 }
109 }
110
111 if let Ok(ms) = s.parse::<i64>() {
113 return Some(ms);
114 }
115
116 None
117}
118
119pub fn int_to_text_array<T: Display + Integer>(arr: &Arc<IntegerArray<T>>) -> TextArray {
121 let mut strings: Vec<String> = Vec::with_capacity(arr.len());
122 for i in 0..arr.len() {
123 if arr.is_null(i) {
124 strings.push(String::new()); } else {
126 strings.push(format!("{}", arr.data[i]));
127 }
128 }
129 let refs: Vec<&str> = strings.iter().map(String::as_str).collect();
130 let string_array = StringArray::<u32>::from_vec(refs, arr.null_mask.clone());
131 TextArray::String32(Arc::new(string_array))
132}
133
134pub fn float_to_text_array<T: Display + Float>(arr: &Arc<FloatArray<T>>) -> TextArray {
136 let mut strings: Vec<String> = Vec::with_capacity(arr.len());
137 for i in 0..arr.len() {
138 if arr.is_null(i) {
139 strings.push(String::new()); } else {
141 strings.push(format!("{}", arr.data[i]));
142 }
143 }
144 let refs: Vec<&str> = strings.iter().map(String::as_str).collect();
145 let string_array = StringArray::<u32>::from_vec(refs, arr.null_mask.clone());
146 TextArray::String32(Arc::new(string_array))
147}
148
149#[inline(always)]
163pub fn confirm_equal_len(label: &str, a: usize, b: usize) -> Result<(), KernelError> {
164 if a != b {
165 return Err(KernelError::LengthMismatch(format!(
166 "{}: length mismatch (lhs: {}, rhs: {})",
167 label, a, b
168 )));
169 }
170 Ok(())
171}
172
173#[inline(always)]
178pub fn align64(byte_count: usize) -> usize {
179 (byte_count + 63) & !63
180}
181
182#[inline(always)]
185pub fn is_simd_aligned<T>(slice: &[T]) -> bool {
186 if slice.is_empty() {
187 true
188 } else {
189 (slice.as_ptr() as usize) % 64 == 0
190 }
191}
192
193#[inline(always)]
221pub fn simd_mask<T: MaskElement, const N: usize>(
222 mask: &Bitmask,
223 offset: usize,
224 len: usize,
225) -> Mask<T, N>
226where
227{
228 let word_idx = offset / 64;
231 let bit_shift = offset % 64;
232 let raw = unsafe { mask.word_unchecked(word_idx) } >> bit_shift;
233
234 let raw = if bit_shift > 0 && word_idx + 1 < (mask.len + 63) / 64 {
236 raw | (unsafe { mask.word_unchecked(word_idx + 1) } << (64 - bit_shift))
237 } else {
238 raw
239 };
240
241 let remaining = if offset < len { len - offset } else { 0 };
243 let raw = if remaining < N && remaining < 64 {
244 raw & ((1u64 << remaining) - 1)
245 } else {
246 raw
247 };
248
249 Mask::from_bitmask(raw)
250}
251
252#[inline(always)]
255pub fn write_simd_mask_bits<T: MaskElement, const N: usize>(
256 out_mask: &mut Bitmask,
257 offset: usize,
258 m: Mask<T, N>,
259)
260where
261{
262 let mbits = m.to_bitmask();
263 let word_idx = offset / 64;
264 let bit_shift = offset % 64;
265
266 unsafe {
267 let existing = out_mask.word_unchecked(word_idx);
269 let lane_mask = if N >= 64 { !0u64 } else { (1u64 << N) - 1 };
271 let cleared = existing & !(lane_mask << bit_shift);
272 out_mask.set_word_unchecked(word_idx, cleared | (mbits << bit_shift));
273
274 if bit_shift > 0 && bit_shift + N > 64 {
276 let overflow_bits = N - (64 - bit_shift);
277 let next_existing = out_mask.word_unchecked(word_idx + 1);
278 let overflow_mask = (1u64 << overflow_bits) - 1;
279 let cleared_next = next_existing & !overflow_mask;
280 out_mask.set_word_unchecked(word_idx + 1, cleared_next | (mbits >> (64 - bit_shift)));
281 }
282 }
283}
284
285#[inline(always)]
288pub fn confirm_mask_capacity(cmp_len: usize, mask: Option<&Bitmask>) -> Result<(), KernelError> {
289 if let Some(m) = mask {
290 confirm_capacity("mask (Bitmask)", m.capacity(), cmp_len)?;
291 }
292 Ok(())
293}
294
295#[inline(always)]
313pub fn confirm_capacity(label: &str, actual: usize, expected: usize) -> Result<(), KernelError> {
314 if actual != expected {
315 return Err(KernelError::InvalidArguments(format!(
316 "{}: capacity mismatch (expected {}, got {})",
317 label, expected, actual
318 )));
319 }
320 Ok(())
321}
322
323#[inline(always)]
327pub fn estimate_categorical_cardinality(cat: &CategoricalArray<u32>, sample_size: usize) -> f64 {
328 let len = cat.data.len();
329 if len == 0 {
330 return 0.0;
331 }
332 let mut seen = HashSet::with_capacity(sample_size.min(len));
333 let step = (len / sample_size.max(1)).max(1);
334 for i in (0..len).step_by(step) {
335 let s = unsafe { cat.get_str_unchecked(i) };
336 seen.insert(s);
337 if seen.len() >= sample_size {
338 break;
339 }
340 }
341 (seen.len() as f64) / (sample_size.min(len) as f64)
342}
343
344#[inline(always)]
348pub fn estimate_string_cardinality<T: Integer>(arr: &StringArray<T>, sample_size: usize) -> f64 {
349 let len = arr.len();
350 if len == 0 {
351 return 0.0;
352 }
353 let mut seen = HashSet::with_capacity(sample_size.min(len));
354 let step = (len / sample_size.max(1)).max(1);
355 for i in (0..len).step_by(step) {
356 let s = unsafe { arr.get_str_unchecked(i) };
357 seen.insert(s);
358 if seen.len() >= sample_size {
359 break;
360 }
361 }
362 (seen.len() as f64) / (sample_size.min(len) as f64)
363}
364
365#[cfg(feature = "chunked")]
366fn union_array_superarray_masks(
368 array: &Array,
369 super_array: &SuperArray,
370) -> Result<Option<Bitmask>, MinarrowError> {
371 let array_mask = array.null_mask();
372 let super_array_masks: Vec<Option<&Bitmask>> = super_array
373 .chunks()
374 .iter()
375 .map(|chunk| chunk.null_mask())
376 .collect();
377
378 let super_array_concatenated_mask = if super_array_masks
379 .iter()
380 .any(|m: &Option<&Bitmask>| m.is_some())
381 {
382 let mut concatenated_bits = Vec::new();
383 for (chunk, mask_opt) in super_array.chunks().iter().zip(super_array_masks.iter()) {
384 if let Some(mask) = mask_opt {
385 concatenated_bits.extend((0..mask.len()).map(|i| mask.get(i)));
386 } else {
387 concatenated_bits.extend(std::iter::repeat(true).take(chunk.len()));
388 }
389 }
390 Some(Bitmask::from_bools(&concatenated_bits))
391 } else {
392 None
393 };
394
395 match (array_mask, super_array_concatenated_mask) {
396 (Some(arr_mask), Some(super_mask)) => {
397 if arr_mask.len() == super_mask.len() {
398 Ok(Some(arr_mask.union(&super_mask)))
399 } else {
400 Err(MinarrowError::ShapeError {
401 message: format!(
402 "Mask lengths must match for union: {} vs {}",
403 arr_mask.len(),
404 super_mask.len()
405 ),
406 })
407 }
408 }
409 (Some(mask), None) => Ok(Some(mask.clone())),
410 (None, Some(mask)) => Ok(Some(mask)),
411 (None, None) => Ok(None),
412 }
413}
414
415#[cfg(feature = "chunked")]
416pub fn create_aligned_chunks_from_array(
418 array: Array,
419 super_array: &SuperArray,
420 field_name: &str,
421) -> Result<SuperArray, MinarrowError> {
422 if array.len() != super_array.len() {
424 return Err(MinarrowError::ShapeError {
425 message: format!(
426 "Array and SuperArray must have same total length for broadcasting: {} vs {}",
427 array.len(),
428 super_array.len()
429 ),
430 });
431 }
432
433 let full_mask = union_array_superarray_masks(&array, super_array)?;
435
436 let chunk_lengths: Vec<usize> = super_array
438 .chunks()
439 .iter()
440 .map(|chunk| chunk.len())
441 .collect();
442
443 let mut start = 0;
445 let mut mask_start = 0;
446 let chunks: Result<Vec<_>, _> = chunk_lengths
447 .iter()
448 .map(|&chunk_len| {
449 let end = start + chunk_len;
450 if end > array.len() {
451 return Err(MinarrowError::ShapeError {
452 message: format!(
453 "Chunk alignment failed: index {} out of bounds for length {}",
454 end,
455 array.len()
456 ),
457 });
458 }
459 let view = array.view(start, chunk_len);
460 let mut array_chunk = view.array.slice_clone(view.offset, view.len());
461
462 if let Some(ref mask) = full_mask {
464 let mask_end = mask_start + chunk_len;
465 let chunk_mask_bits: Vec<bool> =
466 (mask_start..mask_end).map(|i| mask.get(i)).collect();
467 let chunk_mask = Bitmask::from_bools(&chunk_mask_bits);
468 array_chunk.set_null_mask(chunk_mask);
469 mask_start = mask_end;
470 }
471
472 start = end;
473 let first_super_chunk = &super_array.chunks()[0];
474 let field =
475 create_field_for_array(field_name, &array_chunk, Some(first_super_chunk), None);
476 Ok(FieldArray::new(field, array_chunk))
477 })
478 .collect();
479
480 Ok(SuperArray::from_chunks(chunks?))
481}