1use anyhow::{Result, anyhow};
11use arrow_array::builder::{
12 BinaryBuilder, BooleanBufferBuilder, BooleanBuilder, Date32Builder, DurationMicrosecondBuilder,
13 FixedSizeBinaryBuilder, FixedSizeListBuilder, Float32Builder, Float64Builder, Int32Builder,
14 Int64Builder, IntervalMonthDayNanoBuilder, LargeBinaryBuilder, ListBuilder, StringBuilder,
15 StructBuilder, Time64MicrosecondBuilder, Time64NanosecondBuilder, TimestampNanosecondBuilder,
16 UInt8Builder, UInt32Builder, UInt64Builder,
17};
18use arrow_array::{
19 Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, FixedSizeBinaryArray,
20 FixedSizeListArray, Float32Array, Float64Array, Int32Array, Int64Array,
21 IntervalMonthDayNanoArray, LargeBinaryArray, ListArray, StringArray, StructArray,
22 Time64NanosecondArray, TimestampNanosecondArray, UInt8Array, UInt32Array, UInt64Array,
23};
24use arrow_schema::{DataType as ArrowDataType, Field};
25use std::collections::HashMap;
26use std::sync::Arc;
27use uni_common::DataType;
28use uni_common::Value;
29use uni_common::core::id::{Eid, Vid};
30use uni_common::core::schema;
31use uni_common::core::schema::PointType;
32use uni_crdt::Crdt;
33
34fn build_timestamp_column_from_id_map<K, I>(
39 ids: I,
40 timestamps: Option<&HashMap<K, i64>>,
41) -> ArrayRef
42where
43 K: Eq + std::hash::Hash,
44 I: IntoIterator<Item = K>,
45{
46 let mut builder = TimestampNanosecondBuilder::new().with_timezone("UTC");
47 for id in ids {
48 match timestamps.and_then(|m| m.get(&id)) {
49 Some(&ts) => builder.append_value(ts),
50 None => builder.append_null(),
51 }
52 }
53 Arc::new(builder.finish())
54}
55
56pub fn build_timestamp_column_from_vid_map<I>(
57 ids: I,
58 timestamps: Option<&HashMap<Vid, i64>>,
59) -> ArrayRef
60where
61 I: IntoIterator<Item = Vid>,
62{
63 build_timestamp_column_from_id_map(ids, timestamps)
64}
65
66pub fn build_timestamp_column_from_eid_map<I>(
67 ids: I,
68 timestamps: Option<&HashMap<Eid, i64>>,
69) -> ArrayRef
70where
71 I: IntoIterator<Item = Eid>,
72{
73 build_timestamp_column_from_id_map(ids, timestamps)
74}
75
76pub fn build_timestamp_column<I>(timestamps: I) -> ArrayRef
80where
81 I: IntoIterator<Item = Option<i64>>,
82{
83 let mut builder = TimestampNanosecondBuilder::new().with_timezone("UTC");
84 for ts in timestamps {
85 builder.append_option(ts);
86 }
87 Arc::new(builder.finish())
88}
89
90pub fn labels_from_list_array(list_arr: &ListArray, row: usize) -> Vec<String> {
96 if list_arr.is_null(row) {
97 return Vec::new();
98 }
99 let values = list_arr.value(row);
100 let Some(str_arr) = values.as_any().downcast_ref::<StringArray>() else {
101 return Vec::new();
102 };
103 (0..str_arr.len())
104 .filter(|&j| !str_arr.is_null(j))
105 .map(|j| str_arr.value(j).to_string())
106 .collect()
107}
108
109fn parse_datetime_to_nanos(s: &str) -> Option<i64> {
114 chrono::DateTime::parse_from_rfc3339(s)
115 .map(|dt| {
116 dt.with_timezone(&chrono::Utc)
117 .timestamp_nanos_opt()
118 .unwrap_or(0)
119 })
120 .or_else(|_| {
121 chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S")
122 .map(|ndt| ndt.and_utc().timestamp_nanos_opt().unwrap_or(0))
123 })
124 .or_else(|_| {
125 chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%SZ")
126 .map(|ndt| ndt.and_utc().timestamp_nanos_opt().unwrap_or(0))
127 })
128 .or_else(|_| {
129 chrono::DateTime::parse_from_str(s, "%Y-%m-%dT%H:%M%:z").map(|dt| {
130 dt.with_timezone(&chrono::Utc)
131 .timestamp_nanos_opt()
132 .unwrap_or(0)
133 })
134 })
135 .ok()
136 .or_else(|| {
137 s.strip_suffix('Z')
138 .and_then(|base| chrono::NaiveDateTime::parse_from_str(base, "%Y-%m-%dT%H:%M").ok())
139 .map(|ndt| ndt.and_utc().timestamp_nanos_opt().unwrap_or(0))
140 })
141}
142
143pub(crate) fn try_reconstruct_map(arr: &ArrayRef) -> Option<HashMap<String, Value>> {
149 let structs = arr.as_any().downcast_ref::<StructArray>()?;
150 let fields = structs.fields();
151 if fields.len() != 2 || fields[0].name() != "key" || fields[1].name() != "value" {
152 return None;
153 }
154 let value_hint = raw_bytes_hint(fields[1].metadata());
157 let key_col = structs.column(0);
158 let val_col = structs.column(1);
159 let mut map = HashMap::new();
160 for i in 0..structs.len() {
161 if let Value::String(k) = arrow_to_value(key_col.as_ref(), i, None) {
162 map.insert(k, arrow_to_value(val_col.as_ref(), i, value_hint));
163 }
164 }
165 Some(map)
166}
167
168fn array_to_value_list(arr: &ArrayRef, elem_type: Option<&DataType>) -> Vec<Value> {
173 (0..arr.len())
174 .map(|i| arrow_to_value(arr.as_ref(), i, elem_type))
175 .collect()
176}
177
178fn raw_bytes_hint(metadata: &HashMap<String, String>) -> Option<&'static DataType> {
182 if metadata.get("uni_raw_bytes").is_some_and(|v| v == "true") {
183 Some(&DataType::Bytes)
184 } else {
185 None
186 }
187}
188
189fn list_child_bytes_hint(dt: &ArrowDataType) -> Option<&'static DataType> {
191 match dt {
192 ArrowDataType::List(f)
193 | ArrowDataType::LargeList(f)
194 | ArrowDataType::FixedSizeList(f, _) => raw_bytes_hint(f.metadata()),
195 _ => None,
196 }
197}
198
199pub fn arrow_to_value(col: &dyn Array, row: usize, data_type: Option<&DataType>) -> Value {
206 if col.is_null(row) {
207 return Value::Null;
208 }
209
210 if let Some(dt) = data_type {
212 match dt {
213 DataType::DateTime => {
214 if let Some(struct_arr) = col.as_any().downcast_ref::<StructArray>()
216 && let (Some(nanos_col), Some(offset_col), Some(tz_col)) = (
217 struct_arr.column_by_name("nanos_since_epoch"),
218 struct_arr.column_by_name("offset_seconds"),
219 struct_arr.column_by_name("timezone_name"),
220 )
221 && let (Some(nanos_arr), Some(offset_arr), Some(tz_arr)) = (
222 nanos_col
223 .as_any()
224 .downcast_ref::<TimestampNanosecondArray>(),
225 offset_col.as_any().downcast_ref::<Int32Array>(),
226 tz_col.as_any().downcast_ref::<StringArray>(),
227 )
228 {
229 if nanos_arr.is_null(row) {
230 return Value::Null;
231 }
232 let nanos = nanos_arr.value(row);
233 if offset_arr.is_null(row) {
234 return Value::Temporal(uni_common::TemporalValue::LocalDateTime {
236 nanos_since_epoch: nanos,
237 });
238 }
239 let offset = offset_arr.value(row);
240 let tz_name = (!tz_arr.is_null(row)).then(|| tz_arr.value(row).to_string());
241 return Value::Temporal(uni_common::TemporalValue::DateTime {
242 nanos_since_epoch: nanos,
243 offset_seconds: offset,
244 timezone_name: tz_name,
245 });
246 }
247 if let Some(ts) = col.as_any().downcast_ref::<TimestampNanosecondArray>() {
249 let nanos = ts.value(row);
250 let tz_name = ts.timezone().map(|s| s.to_string());
251 return Value::Temporal(uni_common::TemporalValue::DateTime {
252 nanos_since_epoch: nanos,
253 offset_seconds: 0,
254 timezone_name: tz_name,
255 });
256 }
257 }
258 DataType::Time => {
259 if let Some(struct_arr) = col.as_any().downcast_ref::<StructArray>()
261 && let (Some(nanos_col), Some(offset_col)) = (
262 struct_arr.column_by_name("nanos_since_midnight"),
263 struct_arr.column_by_name("offset_seconds"),
264 )
265 && let (Some(nanos_arr), Some(offset_arr)) = (
266 nanos_col.as_any().downcast_ref::<Time64NanosecondArray>(),
267 offset_col.as_any().downcast_ref::<Int32Array>(),
268 )
269 {
270 if nanos_arr.is_null(row) || offset_arr.is_null(row) {
272 return Value::Null;
273 }
274 let nanos = nanos_arr.value(row);
275 let offset = offset_arr.value(row);
276 return Value::Temporal(uni_common::TemporalValue::Time {
277 nanos_since_midnight: nanos,
278 offset_seconds: offset,
279 });
280 }
281 if let Some(t) = col.as_any().downcast_ref::<Time64NanosecondArray>() {
283 let nanos = t.value(row);
284 return Value::Temporal(uni_common::TemporalValue::Time {
285 nanos_since_midnight: nanos,
286 offset_seconds: 0,
287 });
288 }
289 }
290 DataType::Point(pt) => {
291 if let Some(struct_arr) = col.as_any().downcast_ref::<StructArray>() {
298 let f64_at = |name: &str| -> Option<f64> {
299 struct_arr
300 .column_by_name(name)
301 .and_then(|c| c.as_any().downcast_ref::<Float64Array>())
302 .filter(|a| !a.is_null(row))
303 .map(|a| a.value(row))
304 };
305 let crs = struct_arr
306 .column_by_name("crs")
307 .and_then(|c| c.as_any().downcast_ref::<StringArray>())
308 .filter(|a| !a.is_null(row))
309 .map(|a| a.value(row).to_string());
310
311 match pt {
312 PointType::Geographic => {
313 if let (Some(lat), Some(lon)) =
314 (f64_at("latitude"), f64_at("longitude"))
315 {
316 return Value::Map(HashMap::from([
317 ("type".to_string(), Value::String("Point".into())),
318 (
319 "crs".to_string(),
320 Value::String(crs.unwrap_or_else(|| "WGS84".into())),
321 ),
322 ("latitude".to_string(), Value::Float(lat)),
323 ("longitude".to_string(), Value::Float(lon)),
324 ]));
325 }
326 }
327 PointType::Cartesian2D => {
328 if let (Some(x), Some(y)) = (f64_at("x"), f64_at("y")) {
329 return Value::Map(HashMap::from([
330 ("type".to_string(), Value::String("Point".into())),
331 (
332 "crs".to_string(),
333 Value::String(crs.unwrap_or_else(|| "cartesian".into())),
334 ),
335 ("x".to_string(), Value::Float(x)),
336 ("y".to_string(), Value::Float(y)),
337 ]));
338 }
339 }
340 PointType::Cartesian3D => {
341 if let (Some(x), Some(y), Some(z)) =
342 (f64_at("x"), f64_at("y"), f64_at("z"))
343 {
344 return Value::Map(HashMap::from([
345 ("type".to_string(), Value::String("Point".into())),
346 (
347 "crs".to_string(),
348 Value::String(crs.unwrap_or_else(|| "cartesian-3d".into())),
349 ),
350 ("x".to_string(), Value::Float(x)),
351 ("y".to_string(), Value::Float(y)),
352 ("z".to_string(), Value::Float(z)),
353 ]));
354 }
355 }
356 }
357 }
358 return Value::Null;
360 }
361 DataType::Bytes => {
362 let Some(arr) = col.as_any().downcast_ref::<LargeBinaryArray>() else {
363 log::warn!("Bytes column is not LargeBinaryArray");
364 return Value::Null;
365 };
366 if arr.is_null(row) {
367 return Value::Null;
368 }
369 return Value::Bytes(arr.value(row).to_vec());
370 }
371 DataType::Btic => {
372 let Some(fsb) = col.as_any().downcast_ref::<FixedSizeBinaryArray>() else {
373 log::warn!("BTIC column is not FixedSizeBinaryArray");
374 return Value::Null;
375 };
376 let bytes = fsb.value(row);
377 return match uni_btic::encode::decode_slice(bytes) {
378 Ok(btic) => Value::Temporal(uni_common::TemporalValue::Btic {
379 lo: btic.lo(),
380 hi: btic.hi(),
381 meta: btic.meta(),
382 }),
383 Err(e) => {
384 log::warn!("BTIC decode error: {}", e);
385 Value::Null
386 }
387 };
388 }
389 DataType::SparseVector { .. } => {
390 let Some(struct_arr) = col.as_any().downcast_ref::<StructArray>() else {
391 log::warn!("SparseVector column is not StructArray");
392 return Value::Null;
393 };
394 if struct_arr.is_null(row) {
395 return Value::Null;
396 }
397 let (Some(indices_list), Some(values_list)) = (
398 struct_arr
399 .column_by_name("indices")
400 .and_then(|c| c.as_any().downcast_ref::<ListArray>()),
401 struct_arr
402 .column_by_name("values")
403 .and_then(|c| c.as_any().downcast_ref::<ListArray>()),
404 ) else {
405 log::warn!("SparseVector struct missing indices/values list columns");
406 return Value::Null;
407 };
408 let idx_vals = indices_list.value(row);
409 let Some(idx_arr) = idx_vals.as_any().downcast_ref::<UInt32Array>() else {
410 log::warn!("SparseVector 'indices' inner not UInt32");
411 return Value::Null;
412 };
413 let w_vals = values_list.value(row);
414 let Some(w_arr) = w_vals.as_any().downcast_ref::<Float32Array>() else {
415 log::warn!("SparseVector 'values' inner not Float32");
416 return Value::Null;
417 };
418 let indices: Vec<u32> = (0..idx_arr.len()).map(|i| idx_arr.value(i)).collect();
419 let values: Vec<f32> = (0..w_arr.len()).map(|i| w_arr.value(i)).collect();
420 return Value::SparseVector { indices, values };
421 }
422 _ => {}
423 }
424 }
425
426 if let Some(s) = col.as_any().downcast_ref::<StringArray>() {
428 return Value::String(s.value(row).to_string());
429 }
430
431 if let Some(u) = col.as_any().downcast_ref::<UInt64Array>() {
433 return Value::Int(u.value(row) as i64);
434 }
435 if let Some(i) = col.as_any().downcast_ref::<Int64Array>() {
436 return Value::Int(i.value(row));
437 }
438 if let Some(i) = col.as_any().downcast_ref::<Int32Array>() {
439 return Value::Int(i.value(row) as i64);
440 }
441
442 if let Some(f) = col.as_any().downcast_ref::<Float64Array>() {
444 return Value::Float(f.value(row));
445 }
446 if let Some(f) = col.as_any().downcast_ref::<Float32Array>() {
447 return Value::Float(f.value(row) as f64);
448 }
449
450 if let Some(b) = col.as_any().downcast_ref::<BooleanArray>() {
452 return Value::Bool(b.value(row));
453 }
454
455 if let Some(list) = col.as_any().downcast_ref::<FixedSizeListArray>() {
462 let inner = list.value(row);
463 if let Some(floats) = inner.as_any().downcast_ref::<Float32Array>() {
464 return Value::Vector((0..floats.len()).map(|i| floats.value(i)).collect());
465 }
466 if let Some(bytes) = inner.as_any().downcast_ref::<UInt8Array>() {
469 return Value::BinaryVector((0..bytes.len()).map(|i| bytes.value(i)).collect());
470 }
471 let elem_hint = list_child_bytes_hint(list.data_type());
472 return Value::List(array_to_value_list(&inner, elem_hint));
473 }
474
475 if let Some(list) = col.as_any().downcast_ref::<ListArray>() {
477 let arr = list.value(row);
478
479 if let Some(obj) = try_reconstruct_map(&arr) {
481 return Value::Map(obj);
482 }
483
484 let elem_hint = list_child_bytes_hint(list.data_type());
485 return Value::List(array_to_value_list(&arr, elem_hint));
486 }
487
488 if let Some(list) = col.as_any().downcast_ref::<arrow_array::LargeListArray>() {
490 let elem_hint = list_child_bytes_hint(list.data_type());
491 return Value::List(array_to_value_list(&list.value(row), elem_hint));
492 }
493
494 if let Some(s) = col.as_any().downcast_ref::<StructArray>() {
496 if schema::is_sparse_vector_struct(col.data_type()) {
500 if s.is_null(row) {
501 return Value::Null;
502 }
503 if let (Some(idx_list), Some(val_list)) = (
504 s.column_by_name("indices")
505 .and_then(|c| c.as_any().downcast_ref::<ListArray>()),
506 s.column_by_name("values")
507 .and_then(|c| c.as_any().downcast_ref::<ListArray>()),
508 ) {
509 let idx_vals = idx_list.value(row);
510 let val_vals = val_list.value(row);
511 if let (Some(ia), Some(va)) = (
512 idx_vals.as_any().downcast_ref::<UInt32Array>(),
513 val_vals.as_any().downcast_ref::<Float32Array>(),
514 ) {
515 let indices = (0..ia.len()).map(|i| ia.value(i)).collect();
516 let values = (0..va.len()).map(|i| va.value(i)).collect();
517 return Value::SparseVector { indices, values };
518 }
519 }
520 }
521
522 let field_names: Vec<&str> = s.fields().iter().map(|f| f.name().as_str()).collect();
523
524 if field_names.contains(&"nanos_since_epoch")
526 && field_names.contains(&"offset_seconds")
527 && field_names.contains(&"timezone_name")
528 && let (Some(nanos_col), Some(offset_col), Some(tz_col)) = (
529 s.column_by_name("nanos_since_epoch"),
530 s.column_by_name("offset_seconds"),
531 s.column_by_name("timezone_name"),
532 )
533 {
534 let nanos_opt = nanos_col
536 .as_any()
537 .downcast_ref::<TimestampNanosecondArray>()
538 .map(|a| {
539 if a.is_null(row) {
540 None
541 } else {
542 Some(a.value(row))
543 }
544 })
545 .or_else(|| {
546 nanos_col.as_any().downcast_ref::<Int64Array>().map(|a| {
547 if a.is_null(row) {
548 None
549 } else {
550 Some(a.value(row))
551 }
552 })
553 });
554 let offset_opt = offset_col.as_any().downcast_ref::<Int32Array>().map(|a| {
555 if a.is_null(row) {
556 None
557 } else {
558 Some(a.value(row))
559 }
560 });
561
562 if let Some(Some(nanos)) = nanos_opt {
563 match offset_opt {
564 Some(Some(offset)) => {
565 let tz_name = tz_col.as_any().downcast_ref::<StringArray>().and_then(|a| {
566 if a.is_null(row) {
567 None
568 } else {
569 Some(a.value(row).to_string())
570 }
571 });
572 return Value::Temporal(uni_common::TemporalValue::DateTime {
573 nanos_since_epoch: nanos,
574 offset_seconds: offset,
575 timezone_name: tz_name,
576 });
577 }
578 _ => {
579 return Value::Temporal(uni_common::TemporalValue::LocalDateTime {
581 nanos_since_epoch: nanos,
582 });
583 }
584 }
585 }
586 }
587
588 if field_names.contains(&"nanos_since_midnight")
590 && field_names.contains(&"offset_seconds")
591 && let (Some(nanos_col), Some(offset_col)) = (
592 s.column_by_name("nanos_since_midnight"),
593 s.column_by_name("offset_seconds"),
594 )
595 {
596 let nanos_opt = nanos_col
598 .as_any()
599 .downcast_ref::<Time64NanosecondArray>()
600 .map(|a| {
601 if a.is_null(row) {
602 None
603 } else {
604 Some(a.value(row))
605 }
606 })
607 .or_else(|| {
608 nanos_col.as_any().downcast_ref::<Int64Array>().map(|a| {
609 if a.is_null(row) {
610 None
611 } else {
612 Some(a.value(row))
613 }
614 })
615 });
616 let offset_opt = offset_col.as_any().downcast_ref::<Int32Array>().map(|a| {
617 if a.is_null(row) {
618 None
619 } else {
620 Some(a.value(row))
621 }
622 });
623
624 if let (Some(Some(nanos)), Some(Some(offset))) = (nanos_opt, offset_opt) {
625 return Value::Temporal(uni_common::TemporalValue::Time {
626 nanos_since_midnight: nanos,
627 offset_seconds: offset,
628 });
629 }
630 }
631
632 let mut map = HashMap::new();
634 for (field, child) in s.fields().iter().zip(s.columns()) {
635 map.insert(
636 field.name().clone(),
637 arrow_to_value(child.as_ref(), row, None),
638 );
639 }
640 return Value::Map(map);
641 }
642
643 if let Some(d) = col.as_any().downcast_ref::<Date32Array>() {
645 let days = d.value(row);
646 return Value::Temporal(uni_common::TemporalValue::Date {
647 days_since_epoch: days,
648 });
649 }
650
651 if let Some(ts) = col.as_any().downcast_ref::<TimestampNanosecondArray>() {
653 let nanos = ts.value(row);
654 return match ts.timezone() {
655 Some(tz) => Value::Temporal(uni_common::TemporalValue::DateTime {
656 nanos_since_epoch: nanos,
657 offset_seconds: 0,
658 timezone_name: Some(tz.to_string()),
659 }),
660 None => Value::Temporal(uni_common::TemporalValue::LocalDateTime {
661 nanos_since_epoch: nanos,
662 }),
663 };
664 }
665
666 if let Some(t) = col.as_any().downcast_ref::<Time64NanosecondArray>() {
668 let nanos = t.value(row);
669 return Value::Temporal(uni_common::TemporalValue::LocalTime {
670 nanos_since_midnight: nanos,
671 });
672 }
673
674 if let Some(t) = col
676 .as_any()
677 .downcast_ref::<arrow_array::Time64MicrosecondArray>()
678 {
679 let micros = t.value(row);
680 return Value::Temporal(uni_common::TemporalValue::LocalTime {
681 nanos_since_midnight: micros * 1000,
682 });
683 }
684
685 if let Some(d) = col
687 .as_any()
688 .downcast_ref::<arrow_array::DurationMicrosecondArray>()
689 {
690 let micros = d.value(row);
691 let total_nanos = micros * 1000;
692 let seconds = total_nanos / 1_000_000_000;
693 let remaining_nanos = total_nanos % 1_000_000_000;
694 return Value::Temporal(uni_common::TemporalValue::Duration {
695 months: 0,
696 days: 0,
697 nanos: seconds * 1_000_000_000 + remaining_nanos,
698 });
699 }
700
701 if let Some(interval) = col.as_any().downcast_ref::<IntervalMonthDayNanoArray>() {
703 let val = interval.value(row);
704 return Value::Temporal(uni_common::TemporalValue::Duration {
705 months: val.months as i64,
706 days: val.days as i64,
707 nanos: val.nanoseconds,
708 });
709 }
710
711 if let Some(b) = col.as_any().downcast_ref::<LargeBinaryArray>() {
713 let bytes = b.value(row);
714 if bytes.is_empty() {
715 return Value::Null;
716 }
717 return uni_common::cypher_value_codec::decode(bytes).unwrap_or_else(|e| {
718 eprintln!("CypherValue decode error: {}", e);
719 Value::Null
720 });
721 }
722
723 if let Some(fsb) = col.as_any().downcast_ref::<FixedSizeBinaryArray>()
725 && fsb.value_length() == 24
726 {
727 let bytes = fsb.value(row);
728 return match uni_btic::encode::decode_slice(bytes) {
729 Ok(btic) => Value::Temporal(uni_common::TemporalValue::Btic {
730 lo: btic.lo(),
731 hi: btic.hi(),
732 meta: btic.meta(),
733 }),
734 Err(e) => {
735 log::warn!("BTIC decode error: {}", e);
736 Value::Null
737 }
738 };
739 }
740
741 if let Some(b) = col.as_any().downcast_ref::<BinaryArray>() {
743 let bytes = b.value(row);
744 return Crdt::from_msgpack(bytes)
745 .ok()
746 .and_then(|crdt| serde_json::to_value(&crdt).ok())
747 .map(Value::from)
748 .unwrap_or(Value::Null);
749 }
750
751 Value::Null
753}
754
755fn values_to_uint64_array(values: &[Value]) -> ArrayRef {
756 let mut builder = UInt64Builder::with_capacity(values.len());
757 for v in values {
758 if let Some(n) = v.as_u64() {
759 builder.append_value(n);
760 } else {
761 builder.append_null();
762 }
763 }
764 Arc::new(builder.finish())
765}
766
767fn values_to_int64_array(values: &[Value]) -> ArrayRef {
768 let mut builder = Int64Builder::with_capacity(values.len());
769 for v in values {
770 if let Some(n) = v.as_i64() {
771 builder.append_value(n);
772 } else {
773 builder.append_null();
774 }
775 }
776 Arc::new(builder.finish())
777}
778
779fn values_to_int32_array(values: &[Value]) -> ArrayRef {
780 let mut builder = Int32Builder::with_capacity(values.len());
781 for v in values {
782 if let Some(n) = v.as_i64() {
783 builder.append_value(n as i32);
784 } else {
785 builder.append_null();
786 }
787 }
788 Arc::new(builder.finish())
789}
790
791fn values_to_string_array(values: &[Value]) -> ArrayRef {
792 let mut builder = StringBuilder::with_capacity(values.len(), values.len() * 10);
793 for v in values {
794 if let Some(s) = v.as_str() {
795 builder.append_value(s);
796 } else if v.is_null() {
797 builder.append_null();
798 } else {
799 builder.append_value(v.to_string());
800 }
801 }
802 Arc::new(builder.finish())
803}
804
805fn values_to_bool_array(values: &[Value]) -> ArrayRef {
806 let mut builder = BooleanBuilder::with_capacity(values.len());
807 for v in values {
808 if let Some(b) = v.as_bool() {
809 builder.append_value(b);
810 } else {
811 builder.append_null();
812 }
813 }
814 Arc::new(builder.finish())
815}
816
817fn values_to_float32_array(values: &[Value]) -> ArrayRef {
818 let mut builder = Float32Builder::with_capacity(values.len());
819 for v in values {
820 if let Some(n) = v.as_f64() {
821 builder.append_value(n as f32);
822 } else {
823 builder.append_null();
824 }
825 }
826 Arc::new(builder.finish())
827}
828
829fn values_to_float64_array(values: &[Value]) -> ArrayRef {
830 let mut builder = Float64Builder::with_capacity(values.len());
831 for v in values {
832 if let Some(n) = v.as_f64() {
833 builder.append_value(n);
834 } else {
835 builder.append_null();
836 }
837 }
838 Arc::new(builder.finish())
839}
840
841fn values_to_fixed_size_binary_array(values: &[Value], size: i32) -> Result<ArrayRef> {
842 let mut builder = FixedSizeBinaryBuilder::with_capacity(values.len(), size);
843 for v in values {
844 match v {
845 Value::Temporal(uni_common::TemporalValue::Btic { lo, hi, meta }) if size == 24 => {
846 let btic = uni_btic::Btic::new(*lo, *hi, *meta)
847 .map_err(|e| anyhow!("invalid BTIC value: {}", e))?;
848 builder.append_value(uni_btic::encode::encode(&btic))?;
849 }
850 Value::String(s) if size == 24 => match uni_btic::parse::parse_btic_literal(s) {
851 Ok(b) => builder.append_value(uni_btic::encode::encode(&b))?,
852 Err(_) => builder.append_null(),
853 },
854 Value::List(bytes) => {
855 let b: Vec<u8> = bytes
856 .iter()
857 .map(|bv| bv.as_u64().unwrap_or(0) as u8)
858 .collect();
859 if b.len() as i32 == size {
860 builder.append_value(&b)?;
861 } else {
862 builder.append_null();
863 }
864 }
865 _ => builder.append_null(),
866 }
867 }
868 Ok(Arc::new(builder.finish()))
869}
870
871pub fn extract_vector_f32_values(
886 val: Option<&Value>,
887 is_deleted: bool,
888 dimensions: usize,
889) -> (Vec<f32>, bool) {
890 let zeros = || vec![0.0_f32; dimensions];
891
892 if is_deleted {
894 return (zeros(), true);
895 }
896
897 match val {
898 Some(Value::Vector(v)) if v.len() == dimensions => (v.clone(), true),
900 Some(Value::Vector(_)) => (zeros(), false), Some(Value::List(arr)) if arr.len() == dimensions => {
903 let values: Vec<f32> = arr
904 .iter()
905 .map(|v| v.as_f64().unwrap_or(0.0) as f32)
906 .collect();
907 (values, true)
908 }
909 Some(Value::List(_)) => (zeros(), false), _ => (zeros(), false), }
912}
913
914pub fn extract_vector_f32_values_strict(
928 val: Option<&Value>,
929 is_deleted: bool,
930 dimensions: usize,
931) -> Result<(Vec<f32>, bool)> {
932 if !is_deleted && let Some(v) = val {
933 uni_common::core::schema::check_dense_vector_value(v, dimensions)?;
934 }
935 Ok(extract_vector_f32_values(val, is_deleted, dimensions))
936}
937
938pub fn extract_binary_vector_values(
949 val: Option<&Value>,
950 is_deleted: bool,
951 dimensions: usize,
952) -> (Vec<u8>, bool) {
953 let zeros = || vec![0_u8; dimensions];
954
955 if is_deleted {
956 return (zeros(), true);
957 }
958
959 match val {
960 Some(Value::BinaryVector(b)) if b.len() == dimensions => (b.clone(), true),
961 Some(Value::BinaryVector(_)) => (zeros(), false), Some(Value::List(arr)) if arr.len() == dimensions => {
963 let mut bytes = Vec::with_capacity(dimensions);
964 for v in arr {
965 match v.as_i64() {
966 Some(n @ 0..=255) => bytes.push(n as u8),
967 _ => return (zeros(), false), }
969 }
970 (bytes, true)
971 }
972 Some(Value::List(_)) => (zeros(), false), _ => (zeros(), false), }
975}
976
977pub fn extract_binary_vector_values_strict(
985 val: Option<&Value>,
986 is_deleted: bool,
987 dimensions: usize,
988) -> Result<(Vec<u8>, bool)> {
989 if !is_deleted && let Some(v) = val {
990 uni_common::core::schema::check_binary_vector_value(v, dimensions)?;
991 }
992 Ok(extract_binary_vector_values(val, is_deleted, dimensions))
993}
994
995fn values_to_fixed_size_list_f32_array(values: &[Value], size: i32) -> ArrayRef {
996 let mut builder = FixedSizeListBuilder::new(Float32Builder::new(), size);
997 for v in values {
998 let (vals, valid) = extract_vector_f32_values(Some(v), false, size as usize);
999 for val in vals {
1000 builder.values().append_value(val);
1001 }
1002 builder.append(valid);
1003 }
1004 Arc::new(builder.finish())
1005}
1006
1007fn values_to_timestamp_array(values: &[Value], tz: Option<&Arc<str>>) -> ArrayRef {
1008 let mut builder = TimestampNanosecondBuilder::with_capacity(values.len());
1009 for v in values {
1010 if v.is_null() {
1011 builder.append_null();
1012 } else if let Value::Temporal(tv) = v {
1013 match tv {
1014 uni_common::TemporalValue::DateTime {
1015 nanos_since_epoch, ..
1016 }
1017 | uni_common::TemporalValue::LocalDateTime {
1018 nanos_since_epoch, ..
1019 } => builder.append_value(*nanos_since_epoch),
1020 _ => builder.append_null(),
1021 }
1022 } else if let Some(n) = v.as_i64() {
1023 builder.append_value(n);
1024 } else if let Some(s) = v.as_str() {
1025 match parse_datetime_to_nanos(s) {
1026 Some(nanos) => builder.append_value(nanos),
1027 None => builder.append_null(),
1028 }
1029 } else {
1030 builder.append_null();
1031 }
1032 }
1033
1034 let arr = builder.finish();
1035 if let Some(tz) = tz {
1036 Arc::new(arr.with_timezone(tz.as_ref()))
1037 } else {
1038 Arc::new(arr)
1039 }
1040}
1041
1042fn values_to_datetime_struct_array(values: &[Value]) -> ArrayRef {
1047 let mut nanos_builder = TimestampNanosecondBuilder::with_capacity(values.len());
1048 let mut offset_builder = Int32Builder::with_capacity(values.len());
1049 let mut tz_builder = StringBuilder::with_capacity(values.len(), values.len() * 20);
1050 let mut null_buffer = BooleanBufferBuilder::new(values.len());
1051
1052 for v in values {
1053 match v {
1054 Value::Temporal(uni_common::TemporalValue::DateTime {
1055 nanos_since_epoch,
1056 offset_seconds,
1057 timezone_name,
1058 }) => {
1059 nanos_builder.append_value(*nanos_since_epoch);
1060 offset_builder.append_value(*offset_seconds);
1061 tz_builder.append_option(timezone_name.as_deref());
1062 null_buffer.append(true);
1063 }
1064 Value::Temporal(uni_common::TemporalValue::LocalDateTime { nanos_since_epoch }) => {
1065 nanos_builder.append_value(*nanos_since_epoch);
1066 offset_builder.append_null();
1067 tz_builder.append_null();
1068 null_buffer.append(true);
1069 }
1070 _ => {
1071 nanos_builder.append_null();
1072 offset_builder.append_null();
1073 tz_builder.append_null();
1074 null_buffer.append(false);
1075 }
1076 }
1077 }
1078
1079 let struct_arr = StructArray::new(
1080 schema::datetime_struct_fields(),
1081 vec![
1082 Arc::new(nanos_builder.finish()) as ArrayRef,
1083 Arc::new(offset_builder.finish()) as ArrayRef,
1084 Arc::new(tz_builder.finish()) as ArrayRef,
1085 ],
1086 Some(null_buffer.finish().into()),
1087 );
1088 Arc::new(struct_arr)
1089}
1090
1091fn values_to_time_struct_array(values: &[Value]) -> ArrayRef {
1096 let mut nanos_builder = Time64NanosecondBuilder::with_capacity(values.len());
1097 let mut offset_builder = Int32Builder::with_capacity(values.len());
1098 let mut null_buffer = BooleanBufferBuilder::new(values.len());
1099
1100 for v in values {
1101 match v {
1102 Value::Temporal(uni_common::TemporalValue::Time {
1103 nanos_since_midnight,
1104 offset_seconds,
1105 }) => {
1106 nanos_builder.append_value(*nanos_since_midnight);
1107 offset_builder.append_value(*offset_seconds);
1108 null_buffer.append(true);
1109 }
1110 Value::Temporal(uni_common::TemporalValue::LocalTime {
1111 nanos_since_midnight,
1112 }) => {
1113 nanos_builder.append_value(*nanos_since_midnight);
1114 offset_builder.append_null();
1115 null_buffer.append(true);
1116 }
1117 _ => {
1118 nanos_builder.append_null();
1119 offset_builder.append_null();
1120 null_buffer.append(false);
1121 }
1122 }
1123 }
1124
1125 let struct_arr = StructArray::new(
1126 schema::time_struct_fields(),
1127 vec![
1128 Arc::new(nanos_builder.finish()) as ArrayRef,
1129 Arc::new(offset_builder.finish()) as ArrayRef,
1130 ],
1131 Some(null_buffer.finish().into()),
1132 );
1133 Arc::new(struct_arr)
1134}
1135
1136pub(crate) fn values_to_point_struct_array(values: &[Value], point_type: PointType) -> ArrayRef {
1145 use arrow_array::builder::Float64Builder;
1146
1147 let n = values.len();
1148 let fields = match schema::DataType::Point(point_type).to_arrow() {
1149 ArrowDataType::Struct(f) => f,
1150 _ => unreachable!("Point maps to an Arrow struct"),
1152 };
1153
1154 let coord = |v: &Value, key: &str| -> Option<f64> {
1156 v.as_object()
1157 .and_then(|m| m.get(key))
1158 .and_then(Value::as_f64)
1159 };
1160 let crs_of = |v: &Value, default: &str| -> String {
1161 v.as_object()
1162 .and_then(|m| m.get("crs"))
1163 .and_then(|c| c.as_str())
1164 .unwrap_or(default)
1165 .to_string()
1166 };
1167
1168 let mut null_buffer = BooleanBufferBuilder::new(n);
1169 let mut crs_b = StringBuilder::with_capacity(n, n * 8);
1170
1171 let (keys, default_crs): (&[&str], &str) = match point_type {
1173 PointType::Geographic => (&["latitude", "longitude"], "WGS84"),
1174 PointType::Cartesian2D => (&["x", "y"], "cartesian"),
1175 PointType::Cartesian3D => (&["x", "y", "z"], "cartesian-3d"),
1176 };
1177 let mut coord_builders: Vec<Float64Builder> = keys
1178 .iter()
1179 .map(|_| Float64Builder::with_capacity(n))
1180 .collect();
1181
1182 for v in values {
1183 let coords: Option<Vec<f64>> = keys.iter().map(|k| coord(v, k)).collect();
1184 match coords {
1185 Some(cs) => {
1186 for (b, c) in coord_builders.iter_mut().zip(cs) {
1187 b.append_value(c);
1188 }
1189 crs_b.append_value(crs_of(v, default_crs));
1190 null_buffer.append(true);
1191 }
1192 None => {
1193 for b in &mut coord_builders {
1194 b.append_value(0.0);
1195 }
1196 crs_b.append_value(default_crs);
1197 null_buffer.append(false);
1198 }
1199 }
1200 }
1201
1202 let mut columns: Vec<ArrayRef> = coord_builders
1203 .into_iter()
1204 .map(|mut b| Arc::new(b.finish()) as ArrayRef)
1205 .collect();
1206 columns.push(Arc::new(crs_b.finish()) as ArrayRef);
1207
1208 Arc::new(StructArray::new(
1209 fields,
1210 columns,
1211 Some(null_buffer.finish().into()),
1212 ))
1213}
1214
1215fn values_to_large_binary_array(values: &[Value]) -> ArrayRef {
1216 let mut builder =
1217 arrow_array::builder::LargeBinaryBuilder::with_capacity(values.len(), values.len() * 64);
1218 for v in values {
1219 if v.is_null() {
1220 builder.append_null();
1221 } else {
1222 let cv_bytes = uni_common::cypher_value_codec::encode(v);
1224 builder.append_value(&cv_bytes);
1225 }
1226 }
1227 Arc::new(builder.finish())
1228}
1229
1230pub fn values_to_array(values: &[Value], dt: &ArrowDataType) -> Result<ArrayRef> {
1232 match dt {
1233 ArrowDataType::UInt64 => Ok(values_to_uint64_array(values)),
1234 ArrowDataType::Int64 => Ok(values_to_int64_array(values)),
1235 ArrowDataType::Int32 => Ok(values_to_int32_array(values)),
1236 ArrowDataType::Utf8 => Ok(values_to_string_array(values)),
1237 ArrowDataType::Boolean => Ok(values_to_bool_array(values)),
1238 ArrowDataType::Float32 => Ok(values_to_float32_array(values)),
1239 ArrowDataType::Float64 => Ok(values_to_float64_array(values)),
1240 ArrowDataType::FixedSizeBinary(size) => values_to_fixed_size_binary_array(values, *size),
1241 ArrowDataType::FixedSizeList(inner, size) => {
1242 if inner.data_type() == &ArrowDataType::Float32 {
1243 Ok(values_to_fixed_size_list_f32_array(values, *size))
1244 } else {
1245 Err(anyhow!("Unsupported FixedSizeList inner type"))
1246 }
1247 }
1248 ArrowDataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
1249 Ok(values_to_timestamp_array(values, tz.as_ref()))
1250 }
1251 ArrowDataType::Timestamp(arrow_schema::TimeUnit::Microsecond, tz) => {
1252 Ok(values_to_timestamp_array(values, tz.as_ref()))
1253 }
1254 ArrowDataType::Date32 => {
1255 let mut builder = Date32Builder::with_capacity(values.len());
1256 for v in values {
1257 if v.is_null() {
1258 builder.append_null();
1259 } else if let Value::Temporal(uni_common::TemporalValue::Date {
1260 days_since_epoch,
1261 }) = v
1262 {
1263 builder.append_value(*days_since_epoch);
1264 } else if let Some(n) = v.as_i64() {
1265 builder.append_value(n as i32);
1266 } else {
1267 builder.append_null();
1268 }
1269 }
1270 Ok(Arc::new(builder.finish()))
1271 }
1272 ArrowDataType::Time64(arrow_schema::TimeUnit::Nanosecond) => {
1273 let mut builder = Time64NanosecondBuilder::with_capacity(values.len());
1274 for v in values {
1275 if v.is_null() {
1276 builder.append_null();
1277 } else if let Value::Temporal(tv) = v {
1278 match tv {
1279 uni_common::TemporalValue::LocalTime {
1280 nanos_since_midnight,
1281 }
1282 | uni_common::TemporalValue::Time {
1283 nanos_since_midnight,
1284 ..
1285 } => builder.append_value(*nanos_since_midnight),
1286 _ => builder.append_null(),
1287 }
1288 } else if let Some(n) = v.as_i64() {
1289 builder.append_value(n);
1290 } else {
1291 builder.append_null();
1292 }
1293 }
1294 Ok(Arc::new(builder.finish()))
1295 }
1296 ArrowDataType::Time64(arrow_schema::TimeUnit::Microsecond) => {
1297 let mut builder = Time64MicrosecondBuilder::with_capacity(values.len());
1298 for v in values {
1299 if v.is_null() {
1300 builder.append_null();
1301 } else if let Value::Temporal(tv) = v {
1302 match tv {
1303 uni_common::TemporalValue::LocalTime {
1304 nanos_since_midnight,
1305 }
1306 | uni_common::TemporalValue::Time {
1307 nanos_since_midnight,
1308 ..
1309 } => builder.append_value(*nanos_since_midnight / 1_000), _ => builder.append_null(),
1311 }
1312 } else if let Some(n) = v.as_i64() {
1313 builder.append_value(n);
1314 } else {
1315 builder.append_null();
1316 }
1317 }
1318 Ok(Arc::new(builder.finish()))
1319 }
1320 ArrowDataType::Interval(arrow_schema::IntervalUnit::MonthDayNano) => {
1321 let mut builder = IntervalMonthDayNanoBuilder::with_capacity(values.len());
1322 for v in values {
1323 if v.is_null() {
1324 builder.append_null();
1325 } else if let Value::Temporal(uni_common::TemporalValue::Duration {
1326 months,
1327 days,
1328 nanos,
1329 }) = v
1330 {
1331 builder.append_value(arrow::datatypes::IntervalMonthDayNano {
1332 months: *months as i32,
1333 days: *days as i32,
1334 nanoseconds: *nanos,
1335 });
1336 } else {
1337 builder.append_null();
1338 }
1339 }
1340 Ok(Arc::new(builder.finish()))
1341 }
1342 ArrowDataType::Duration(arrow_schema::TimeUnit::Microsecond) => {
1343 let mut builder = DurationMicrosecondBuilder::with_capacity(values.len());
1344 for v in values {
1345 if v.is_null() {
1346 builder.append_null();
1347 } else if let Value::Temporal(uni_common::TemporalValue::Duration {
1348 months,
1349 days,
1350 nanos,
1351 }) = v
1352 {
1353 let total_micros =
1354 months * 30 * 86_400_000_000i64 + days * 86_400_000_000i64 + nanos / 1_000;
1355 builder.append_value(total_micros);
1356 } else if let Some(n) = v.as_i64() {
1357 builder.append_value(n);
1358 } else {
1359 builder.append_null();
1360 }
1361 }
1362 Ok(Arc::new(builder.finish()))
1363 }
1364 ArrowDataType::LargeBinary => Ok(values_to_large_binary_array(values)),
1365 ArrowDataType::List(field) => {
1366 if field.data_type() == &ArrowDataType::Utf8 {
1367 let mut builder = ListBuilder::new(StringBuilder::new());
1368 for v in values {
1369 if let Value::List(arr) = v {
1370 for item in arr {
1371 if let Some(s) = item.as_str() {
1372 builder.values().append_value(s);
1373 } else {
1374 builder.values().append_null();
1375 }
1376 }
1377 builder.append(true);
1378 } else {
1379 builder.append_null();
1380 }
1381 }
1382 Ok(Arc::new(builder.finish()))
1383 } else {
1384 Err(anyhow!(
1385 "Unsupported List inner type: {:?}",
1386 field.data_type()
1387 ))
1388 }
1389 }
1390 ArrowDataType::Struct(_) if schema::is_datetime_struct(dt) => {
1391 Ok(values_to_datetime_struct_array(values))
1392 }
1393 ArrowDataType::Struct(_) if schema::is_time_struct(dt) => {
1394 Ok(values_to_time_struct_array(values))
1395 }
1396 _ => Err(anyhow!("Unsupported type for conversion: {:?}", dt)),
1397 }
1398}
1399
1400pub struct PropertyExtractor<'a> {
1402 name: &'a str,
1404 data_type: &'a DataType,
1405}
1406
1407fn sparse_pair_from_value(v: &Value) -> Option<(Vec<u32>, Vec<f32>)> {
1416 match v {
1417 Value::SparseVector { indices, values } => {
1418 if indices.len() != values.len() {
1424 return None;
1425 }
1426 Some((indices.clone(), values.clone()))
1427 }
1428 Value::Map(m) => {
1429 let idx = match m.get("indices") {
1430 Some(Value::List(l)) => l,
1431 _ => return None,
1432 };
1433 let vals = match m.get("values") {
1434 Some(Value::List(l)) => l,
1435 _ => return None,
1436 };
1437 let indices: Vec<u32> = idx
1438 .iter()
1439 .map(|x| x.as_u64().map(|n| n as u32))
1440 .collect::<Option<_>>()?;
1441 let values: Vec<f32> = vals
1442 .iter()
1443 .map(|x| x.as_f64().map(|n| n as f32))
1444 .collect::<Option<_>>()?;
1445 if indices.len() != values.len() {
1446 return None;
1447 }
1448 Some((indices, values))
1449 }
1450 _ => None,
1451 }
1452}
1453
1454pub fn build_sparse_vector_array(values: &[Option<Value>]) -> ArrayRef {
1460 let mut indices_builder = ListBuilder::new(UInt32Builder::new());
1461 let mut values_builder = ListBuilder::new(Float32Builder::new());
1462 let mut null_buffer = BooleanBufferBuilder::new(values.len());
1463 for v in values {
1464 match v.as_ref().and_then(sparse_pair_from_value) {
1465 Some((indices, vals)) => {
1466 for ix in indices {
1467 indices_builder.values().append_value(ix);
1468 }
1469 indices_builder.append(true);
1470 for w in vals {
1471 values_builder.values().append_value(w);
1472 }
1473 values_builder.append(true);
1474 null_buffer.append(true);
1475 }
1476 None => {
1477 indices_builder.append(true);
1478 values_builder.append(true);
1479 null_buffer.append(false);
1480 }
1481 }
1482 }
1483 let struct_arr = StructArray::new(
1484 schema::sparse_vector_struct_fields(),
1485 vec![
1486 Arc::new(indices_builder.finish()) as ArrayRef,
1487 Arc::new(values_builder.finish()) as ArrayRef,
1488 ],
1489 Some(null_buffer.finish().into()),
1490 );
1491 Arc::new(struct_arr)
1492}
1493
1494pub fn build_multivector_array(values: &[Option<Value>], dimensions: usize) -> ArrayRef {
1508 let dim = dimensions as i32;
1509 let mut builder = ListBuilder::new(FixedSizeListBuilder::new(Float32Builder::new(), dim));
1510 for v in values {
1511 match v.as_ref().and_then(|v| v.as_array()) {
1512 Some(arr) => {
1513 for tok in arr {
1517 let (vals, valid) = extract_vector_f32_values(Some(tok), false, dimensions);
1518 for f in vals {
1519 builder.values().values().append_value(f);
1520 }
1521 builder.values().append(valid);
1522 }
1523 builder.append(true);
1524 }
1525 None => builder.append_null(),
1526 }
1527 }
1528 Arc::new(builder.finish())
1529}
1530
1531impl<'a> PropertyExtractor<'a> {
1532 pub fn new(name: &'a str, data_type: &'a DataType) -> Self {
1533 Self { name, data_type }
1534 }
1535
1536 pub fn build_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
1539 where
1540 F: Fn(usize) -> Option<&'a Value>,
1541 {
1542 match self.data_type {
1543 DataType::String => self.build_string_column(len, deleted, get_props),
1544 DataType::Int32 => self.build_int32_column(len, deleted, get_props),
1545 DataType::Int64 => self.build_int64_column(len, deleted, get_props),
1546 DataType::Float32 => self.build_float32_column(len, deleted, get_props),
1547 DataType::Float64 => self.build_float64_column(len, deleted, get_props),
1548 DataType::Bool => self.build_bool_column(len, deleted, get_props),
1549 DataType::Vector { dimensions } => {
1550 self.build_vector_column(len, deleted, get_props, *dimensions)
1551 }
1552 DataType::SparseVector { .. } => {
1553 self.build_sparse_vector_column(len, deleted, get_props)
1554 }
1555 DataType::BinaryVector { dimensions } => {
1556 self.build_binary_vector_column(len, deleted, get_props, *dimensions)
1557 }
1558 DataType::CypherValue => self.build_json_column(len, deleted, get_props),
1559 DataType::Bytes => self.build_bytes_column(len, deleted, get_props),
1560 DataType::List(inner) => self.build_list_column(len, deleted, get_props, inner),
1561 DataType::Map(key, value) => self.build_map_column(len, deleted, get_props, key, value),
1562 DataType::Crdt(_) => self.build_crdt_column(len, deleted, get_props),
1563 DataType::DateTime => self.build_datetime_struct_column(len, deleted, get_props),
1564 DataType::Timestamp => self.build_timestamp_column(len, deleted, get_props),
1565 DataType::Date => self.build_date32_column(len, deleted, get_props),
1566 DataType::Time => self.build_time_struct_column(len, deleted, get_props),
1567 DataType::Point(pt) => self.build_point_struct_column(len, deleted, get_props, *pt),
1568 DataType::Duration => self.build_duration_column(len, deleted, get_props),
1569 DataType::Btic => self.build_btic_column(len, deleted, get_props),
1570 _ => Err(anyhow!(
1571 "Unsupported data type for arrow conversion: {:?}",
1572 self.data_type
1573 )),
1574 }
1575 }
1576
1577 fn build_string_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
1578 where
1579 F: Fn(usize) -> Option<&'a Value>,
1580 {
1581 let mut builder = arrow_array::builder::StringBuilder::with_capacity(len, len * 32);
1582 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1583 let prop = get_props(i);
1584 if let Some(s) = prop.and_then(|v| v.as_str()) {
1585 builder.append_value(s);
1586 } else if let Some(Value::Temporal(tv)) = prop {
1587 builder.append_value(tv.to_string());
1588 } else if is_deleted {
1589 builder.append_value("");
1590 } else {
1591 builder.append_null();
1592 }
1593 }
1594 Ok(Arc::new(builder.finish()))
1595 }
1596
1597 fn build_int32_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
1598 where
1599 F: Fn(usize) -> Option<&'a Value>,
1600 {
1601 let mut values = Vec::with_capacity(len);
1602 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1603 let val = get_props(i)
1606 .and_then(|v| v.as_i64())
1607 .and_then(|v| i32::try_from(v).ok());
1608 if val.is_none() && is_deleted {
1609 values.push(Some(0));
1610 } else {
1611 values.push(val);
1612 }
1613 }
1614 Ok(Arc::new(Int32Array::from(values)))
1615 }
1616
1617 fn build_int64_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
1618 where
1619 F: Fn(usize) -> Option<&'a Value>,
1620 {
1621 let mut values = Vec::with_capacity(len);
1622 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1623 let val = get_props(i).and_then(|v| v.as_i64());
1624 if val.is_none() && is_deleted {
1625 values.push(Some(0));
1626 } else {
1627 values.push(val);
1628 }
1629 }
1630 Ok(Arc::new(Int64Array::from(values)))
1631 }
1632
1633 fn build_timestamp_column<F>(
1634 &self,
1635 len: usize,
1636 deleted: &[bool],
1637 get_props: F,
1638 ) -> Result<ArrayRef>
1639 where
1640 F: Fn(usize) -> Option<&'a Value>,
1641 {
1642 let mut values = Vec::with_capacity(len);
1643 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1644 let val = get_props(i);
1645 let ts = if let Some(Value::Temporal(tv)) = val {
1650 match tv {
1651 uni_common::TemporalValue::DateTime {
1652 nanos_since_epoch, ..
1653 }
1654 | uni_common::TemporalValue::LocalDateTime {
1655 nanos_since_epoch, ..
1656 } => Some(*nanos_since_epoch),
1657 _ => None,
1658 }
1659 } else if let Some(v) = val.and_then(|v| v.as_i64()) {
1660 Some(v)
1661 } else if let Some(s) = val.and_then(|v| v.as_str()) {
1662 parse_datetime_to_nanos(s)
1663 } else {
1664 None
1665 };
1666
1667 if is_deleted {
1668 values.push(Some(0));
1669 } else {
1670 values.push(ts);
1671 }
1672 }
1673 let arr = TimestampNanosecondArray::from(values).with_timezone("UTC");
1674 Ok(Arc::new(arr))
1675 }
1676
1677 fn build_datetime_struct_column<F>(
1678 &self,
1679 len: usize,
1680 deleted: &[bool],
1681 get_props: F,
1682 ) -> Result<ArrayRef>
1683 where
1684 F: Fn(usize) -> Option<&'a Value>,
1685 {
1686 let values = self.collect_values_or_null(len, deleted, &get_props);
1687 Ok(values_to_datetime_struct_array(&values))
1688 }
1689
1690 fn build_time_struct_column<F>(
1691 &self,
1692 len: usize,
1693 deleted: &[bool],
1694 get_props: F,
1695 ) -> Result<ArrayRef>
1696 where
1697 F: Fn(usize) -> Option<&'a Value>,
1698 {
1699 let values = self.collect_values_or_null(len, deleted, &get_props);
1700 Ok(values_to_time_struct_array(&values))
1701 }
1702
1703 fn build_point_struct_column<F>(
1704 &self,
1705 len: usize,
1706 deleted: &[bool],
1707 get_props: F,
1708 point_type: PointType,
1709 ) -> Result<ArrayRef>
1710 where
1711 F: Fn(usize) -> Option<&'a Value>,
1712 {
1713 let values = self.collect_values_or_null(len, deleted, &get_props);
1714 Ok(values_to_point_struct_array(&values, point_type))
1715 }
1716
1717 fn collect_values_or_null<F>(&self, len: usize, deleted: &[bool], get_props: &F) -> Vec<Value>
1719 where
1720 F: Fn(usize) -> Option<&'a Value>,
1721 {
1722 deleted
1723 .iter()
1724 .enumerate()
1725 .take(len)
1726 .map(|(i, &is_deleted)| {
1727 if is_deleted {
1728 Value::Null
1729 } else {
1730 get_props(i).cloned().unwrap_or(Value::Null)
1731 }
1732 })
1733 .collect()
1734 }
1735
1736 fn build_date32_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
1737 where
1738 F: Fn(usize) -> Option<&'a Value>,
1739 {
1740 let mut builder = Date32Builder::with_capacity(len);
1741 let epoch = chrono::NaiveDate::from_ymd_opt(1970, 1, 1).unwrap();
1742
1743 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1744 let val = get_props(i);
1745 let days = if let Some(Value::Temporal(uni_common::TemporalValue::Date {
1748 days_since_epoch,
1749 })) = val
1750 {
1751 Some(*days_since_epoch)
1752 } else if let Some(v) = val.and_then(|v| v.as_i64()) {
1753 i32::try_from(v).ok()
1756 } else if let Some(s) = val.and_then(|v| v.as_str()) {
1757 match chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d") {
1758 Ok(date) => Some(date.signed_duration_since(epoch).num_days() as i32),
1759 Err(_) => None,
1760 }
1761 } else {
1762 None
1763 };
1764
1765 if is_deleted {
1766 builder.append_value(0);
1767 } else if let Some(v) = days {
1768 builder.append_value(v);
1769 } else {
1770 builder.append_null();
1771 }
1772 }
1773 Ok(Arc::new(builder.finish()))
1774 }
1775
1776 fn build_duration_column<F>(
1777 &self,
1778 len: usize,
1779 deleted: &[bool],
1780 get_props: F,
1781 ) -> Result<ArrayRef>
1782 where
1783 F: Fn(usize) -> Option<&'a Value>,
1784 {
1785 let mut builder = LargeBinaryBuilder::with_capacity(len, len * 32);
1787 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1788 let raw_val = get_props(i);
1789 if let Some(val @ Value::Temporal(uni_common::TemporalValue::Duration { .. })) = raw_val
1790 {
1791 let encoded = uni_common::cypher_value_codec::encode(val);
1792 builder.append_value(&encoded);
1793 } else if is_deleted {
1794 let zero = Value::Temporal(uni_common::TemporalValue::Duration {
1795 months: 0,
1796 days: 0,
1797 nanos: 0,
1798 });
1799 let encoded = uni_common::cypher_value_codec::encode(&zero);
1800 builder.append_value(&encoded);
1801 } else {
1802 builder.append_null();
1803 }
1804 }
1805 Ok(Arc::new(builder.finish()))
1806 }
1807
1808 fn build_btic_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
1809 where
1810 F: Fn(usize) -> Option<&'a Value>,
1811 {
1812 const ENCODED_LEN: i32 = 24;
1813 let mut builder = FixedSizeBinaryBuilder::with_capacity(len, ENCODED_LEN);
1814 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1815 let raw_val = get_props(i);
1816 let btic = match raw_val {
1817 Some(Value::Temporal(uni_common::TemporalValue::Btic { lo, hi, meta })) => Some(
1818 uni_btic::Btic::new(*lo, *hi, *meta)
1819 .map_err(|e| anyhow!("invalid BTIC value: {}", e))?,
1820 ),
1821 Some(Value::String(s)) => Some(
1822 uni_btic::parse::parse_btic_literal(s)
1823 .map_err(|e| anyhow!("BTIC parse error for '{}': {}", s, e))?,
1824 ),
1825 _ => None,
1826 };
1827
1828 if let Some(b) = btic {
1829 builder.append_value(uni_btic::encode::encode(&b))?;
1830 } else if is_deleted {
1831 builder.append_value([0u8; ENCODED_LEN as usize])?;
1832 } else {
1833 builder.append_null();
1834 }
1835 }
1836 Ok(Arc::new(builder.finish()))
1837 }
1838
1839 fn build_float32_column<F>(
1840 &self,
1841 len: usize,
1842 deleted: &[bool],
1843 get_props: F,
1844 ) -> Result<ArrayRef>
1845 where
1846 F: Fn(usize) -> Option<&'a Value>,
1847 {
1848 let mut values = Vec::with_capacity(len);
1849 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1850 let val = get_props(i).and_then(|v| v.as_f64()).map(|v| v as f32);
1851 if val.is_none() && is_deleted {
1852 values.push(Some(0.0));
1853 } else {
1854 values.push(val);
1855 }
1856 }
1857 Ok(Arc::new(Float32Array::from(values)))
1858 }
1859
1860 fn build_float64_column<F>(
1861 &self,
1862 len: usize,
1863 deleted: &[bool],
1864 get_props: F,
1865 ) -> Result<ArrayRef>
1866 where
1867 F: Fn(usize) -> Option<&'a Value>,
1868 {
1869 let mut values = Vec::with_capacity(len);
1870 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1871 let val = get_props(i).and_then(|v| v.as_f64());
1872 if val.is_none() && is_deleted {
1873 values.push(Some(0.0));
1874 } else {
1875 values.push(val);
1876 }
1877 }
1878 Ok(Arc::new(Float64Array::from(values)))
1879 }
1880
1881 fn build_bool_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
1882 where
1883 F: Fn(usize) -> Option<&'a Value>,
1884 {
1885 let mut values = Vec::with_capacity(len);
1886 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1887 let val = get_props(i).and_then(|v| v.as_bool());
1888 if val.is_none() && is_deleted {
1889 values.push(Some(false));
1890 } else {
1891 values.push(val);
1892 }
1893 }
1894 Ok(Arc::new(BooleanArray::from(values)))
1895 }
1896
1897 fn build_vector_column<F>(
1898 &self,
1899 len: usize,
1900 deleted: &[bool],
1901 get_props: F,
1902 dimensions: usize,
1903 ) -> Result<ArrayRef>
1904 where
1905 F: Fn(usize) -> Option<&'a Value>,
1906 {
1907 let mut builder = FixedSizeListBuilder::new(Float32Builder::new(), dimensions as i32);
1908
1909 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1910 let val = get_props(i);
1911 let (values, valid) = extract_vector_f32_values_strict(val, is_deleted, dimensions)
1915 .map_err(|e| {
1916 anyhow!(
1917 "flush: property '{}' row {}: {}; wrong-dimension vectors are rejected \
1918 at write time as of issue #137 — this value was likely written by an \
1919 older version",
1920 self.name,
1921 i,
1922 e
1923 )
1924 })?;
1925 for v in values {
1926 builder.values().append_value(v);
1927 }
1928 builder.append(valid);
1929 }
1930 Ok(Arc::new(builder.finish()))
1931 }
1932
1933 fn build_binary_vector_column<F>(
1941 &self,
1942 len: usize,
1943 deleted: &[bool],
1944 get_props: F,
1945 dimensions: usize,
1946 ) -> Result<ArrayRef>
1947 where
1948 F: Fn(usize) -> Option<&'a Value>,
1949 {
1950 let mut builder = FixedSizeListBuilder::new(UInt8Builder::new(), dimensions as i32);
1951
1952 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1953 let val = get_props(i);
1954 let (values, valid) = extract_binary_vector_values_strict(val, is_deleted, dimensions)
1955 .map_err(|e| {
1956 anyhow!(
1957 "flush: property '{}' row {}: {}; wrong-shape binary vectors are \
1958 rejected at write time — this value was likely written by an older \
1959 version",
1960 self.name,
1961 i,
1962 e
1963 )
1964 })?;
1965 for v in values {
1966 builder.values().append_value(v);
1967 }
1968 builder.append(valid);
1969 }
1970 Ok(Arc::new(builder.finish()))
1971 }
1972
1973 fn build_sparse_vector_column<F>(
1979 &self,
1980 len: usize,
1981 deleted: &[bool],
1982 get_props: F,
1983 ) -> Result<ArrayRef>
1984 where
1985 F: Fn(usize) -> Option<&'a Value>,
1986 {
1987 let mut indices_builder = ListBuilder::new(UInt32Builder::new());
1988 let mut values_builder = ListBuilder::new(Float32Builder::new());
1989 let mut null_buffer = BooleanBufferBuilder::new(len);
1990
1991 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
1992 let pair = if is_deleted {
1993 None
1994 } else {
1995 get_props(i).and_then(sparse_pair_from_value)
1996 };
1997 match pair {
1998 Some((indices, values)) => {
1999 for ix in indices {
2000 indices_builder.values().append_value(ix);
2001 }
2002 indices_builder.append(true);
2003 for w in values {
2004 values_builder.values().append_value(w);
2005 }
2006 values_builder.append(true);
2007 null_buffer.append(true);
2008 }
2009 None => {
2010 indices_builder.append(true);
2011 values_builder.append(true);
2012 null_buffer.append(false);
2013 }
2014 }
2015 }
2016
2017 let struct_arr = StructArray::new(
2018 schema::sparse_vector_struct_fields(),
2019 vec![
2020 Arc::new(indices_builder.finish()) as ArrayRef,
2021 Arc::new(values_builder.finish()) as ArrayRef,
2022 ],
2023 Some(null_buffer.finish().into()),
2024 );
2025 Ok(Arc::new(struct_arr))
2026 }
2027
2028 fn build_json_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
2029 where
2030 F: Fn(usize) -> Option<&'a Value>,
2031 {
2032 let null_val = Value::Null;
2033 let mut builder = arrow_array::builder::LargeBinaryBuilder::with_capacity(len, len * 64);
2034 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
2035 let val = get_props(i);
2036 let uni_val = if val.is_none() && is_deleted {
2037 &null_val
2038 } else {
2039 val.unwrap_or(&null_val)
2040 };
2041 let cv_bytes = uni_common::cypher_value_codec::encode(uni_val);
2043 builder.append_value(&cv_bytes);
2044 }
2045 Ok(Arc::new(builder.finish()))
2046 }
2047
2048 fn build_bytes_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
2049 where
2050 F: Fn(usize) -> Option<&'a Value>,
2051 {
2052 let mut builder = LargeBinaryBuilder::with_capacity(len, len * 64);
2053 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
2054 let val = get_props(i);
2055 if let Some(Value::Bytes(b)) = val {
2056 builder.append_value(b);
2057 } else if is_deleted {
2058 builder.append_value(&[][..]);
2059 } else {
2060 builder.append_null();
2061 }
2062 }
2063 Ok(Arc::new(builder.finish()))
2064 }
2065
2066 fn build_list_column<F>(
2067 &self,
2068 len: usize,
2069 deleted: &[bool],
2070 get_props: F,
2071 inner: &DataType,
2072 ) -> Result<ArrayRef>
2073 where
2074 F: Fn(usize) -> Option<&'a Value>,
2075 {
2076 match inner {
2077 DataType::String => {
2078 self.build_typed_list(len, deleted, &get_props, StringBuilder::new(), |v, b| {
2079 if let Some(s) = v.as_str() {
2080 b.append_value(s);
2081 } else {
2082 b.append_null();
2083 }
2084 })
2085 }
2086 DataType::Int64 => {
2087 self.build_typed_list(len, deleted, &get_props, Int64Builder::new(), |v, b| {
2088 if let Some(n) = v.as_i64() {
2089 b.append_value(n);
2090 } else {
2091 b.append_null();
2092 }
2093 })
2094 }
2095 DataType::Float64 => {
2096 self.build_typed_list(len, deleted, &get_props, Float64Builder::new(), |v, b| {
2097 if let Some(f) = v.as_f64() {
2098 b.append_value(f);
2099 } else {
2100 b.append_null();
2101 }
2102 })
2103 }
2104 DataType::Bytes => {
2105 let item_field = Arc::new(
2109 Field::new("item", ArrowDataType::LargeBinary, true)
2110 .with_metadata(schema::raw_bytes_field_metadata()),
2111 );
2112 let mut builder =
2113 ListBuilder::new(LargeBinaryBuilder::new()).with_field(item_field);
2114 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
2115 let val_array = get_props(i).and_then(|v| v.as_array());
2116 if val_array.is_none() && is_deleted {
2117 builder.append_null();
2118 } else if let Some(arr) = val_array {
2119 for v in arr {
2120 if let Value::Bytes(b) = v {
2121 builder.values().append_value(b);
2122 } else {
2123 builder.values().append_null();
2124 }
2125 }
2126 builder.append(true);
2127 } else {
2128 builder.append_null();
2129 }
2130 }
2131 Ok(Arc::new(builder.finish()))
2132 }
2133 DataType::Vector { dimensions } => {
2134 let dim = *dimensions as i32;
2145 let mut builder =
2146 ListBuilder::new(FixedSizeListBuilder::new(Float32Builder::new(), dim));
2147 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
2148 let raw = get_props(i);
2149 if !is_deleted && let Some(v) = raw {
2155 self.data_type.check_vector_dims(v).map_err(|e| {
2156 anyhow!(
2157 "flush: property '{}' row {}: {}; wrong-dimension vectors \
2158 are rejected at write time as of issue #137 — this value \
2159 was likely written by an older version",
2160 self.name,
2161 i,
2162 e
2163 )
2164 })?;
2165 }
2166 let val_array = raw.and_then(|v| v.as_array());
2167 if val_array.is_none() && is_deleted {
2168 builder.append_null();
2169 } else if let Some(arr) = val_array {
2170 for tok in arr {
2175 let (vals, valid) =
2176 extract_vector_f32_values(Some(tok), false, *dimensions);
2177 for v in vals {
2178 builder.values().values().append_value(v);
2179 }
2180 builder.values().append(valid);
2181 }
2182 builder.append(true);
2183 } else {
2184 builder.append_null();
2185 }
2186 }
2187 Ok(Arc::new(builder.finish()))
2188 }
2189 _ => Err(anyhow!("Unsupported inner type for List: {:?}", inner)),
2190 }
2191 }
2192
2193 fn build_typed_list<F, B, A>(
2195 &self,
2196 len: usize,
2197 deleted: &[bool],
2198 get_props: &F,
2199 inner_builder: B,
2200 mut append_value: A,
2201 ) -> Result<ArrayRef>
2202 where
2203 F: Fn(usize) -> Option<&'a Value>,
2204 B: arrow_array::builder::ArrayBuilder,
2205 A: FnMut(&Value, &mut B),
2206 {
2207 let mut builder = ListBuilder::new(inner_builder);
2208 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
2209 let val_array = get_props(i).and_then(|v| v.as_array());
2210 if val_array.is_none() && is_deleted {
2211 builder.append_null();
2212 } else if let Some(arr) = val_array {
2213 for v in arr {
2214 append_value(v, builder.values());
2215 }
2216 builder.append(true);
2217 } else {
2218 builder.append_null();
2219 }
2220 }
2221 Ok(Arc::new(builder.finish()))
2222 }
2223
2224 fn build_map_column<F>(
2225 &self,
2226 len: usize,
2227 deleted: &[bool],
2228 get_props: F,
2229 key: &DataType,
2230 value: &DataType,
2231 ) -> Result<ArrayRef>
2232 where
2233 F: Fn(usize) -> Option<&'a Value>,
2234 {
2235 if !matches!(key, DataType::String) {
2236 return Err(anyhow!("Map keys must be String (JSON limitation)"));
2237 }
2238
2239 match value {
2240 DataType::String => self.build_typed_map(
2241 len,
2242 deleted,
2243 &get_props,
2244 StringBuilder::new(),
2245 arrow_schema::DataType::Utf8,
2246 None,
2247 |v, b: &mut StringBuilder| {
2248 if let Some(s) = v.as_str() {
2249 b.append_value(s);
2250 } else {
2251 b.append_null();
2252 }
2253 },
2254 ),
2255 DataType::Int64 => self.build_typed_map(
2256 len,
2257 deleted,
2258 &get_props,
2259 Int64Builder::new(),
2260 arrow_schema::DataType::Int64,
2261 None,
2262 |v, b: &mut Int64Builder| {
2263 if let Some(n) = v.as_i64() {
2264 b.append_value(n);
2265 } else {
2266 b.append_null();
2267 }
2268 },
2269 ),
2270 DataType::Int32 => self.build_typed_map(
2271 len,
2272 deleted,
2273 &get_props,
2274 Int32Builder::new(),
2275 arrow_schema::DataType::Int32,
2276 None,
2277 |v, b: &mut Int32Builder| match v.as_i64().and_then(|n| i32::try_from(n).ok()) {
2278 Some(n) => b.append_value(n),
2279 None => b.append_null(),
2280 },
2281 ),
2282 DataType::Float64 => self.build_typed_map(
2283 len,
2284 deleted,
2285 &get_props,
2286 Float64Builder::new(),
2287 arrow_schema::DataType::Float64,
2288 None,
2289 |v, b: &mut Float64Builder| match v.as_f64() {
2290 Some(f) => b.append_value(f),
2291 None => b.append_null(),
2292 },
2293 ),
2294 DataType::Float32 => self.build_typed_map(
2295 len,
2296 deleted,
2297 &get_props,
2298 Float32Builder::new(),
2299 arrow_schema::DataType::Float32,
2300 None,
2301 |v, b: &mut Float32Builder| match v.as_f64() {
2302 Some(f) => b.append_value(f as f32),
2303 None => b.append_null(),
2304 },
2305 ),
2306 DataType::Bool => self.build_typed_map(
2307 len,
2308 deleted,
2309 &get_props,
2310 BooleanBuilder::new(),
2311 arrow_schema::DataType::Boolean,
2312 None,
2313 |v, b: &mut BooleanBuilder| match v.as_bool() {
2314 Some(x) => b.append_value(x),
2315 None => b.append_null(),
2316 },
2317 ),
2318 DataType::Bytes => self.build_typed_map(
2319 len,
2320 deleted,
2321 &get_props,
2322 LargeBinaryBuilder::new(),
2323 arrow_schema::DataType::LargeBinary,
2324 Some(schema::raw_bytes_field_metadata()),
2327 |v, b: &mut LargeBinaryBuilder| {
2328 if let Value::Bytes(bytes) = v {
2329 b.append_value(bytes);
2330 } else {
2331 b.append_null();
2332 }
2333 },
2334 ),
2335 _ => self.build_typed_map(
2341 len,
2342 deleted,
2343 &get_props,
2344 LargeBinaryBuilder::new(),
2345 arrow_schema::DataType::LargeBinary,
2346 None,
2347 |v, b: &mut LargeBinaryBuilder| {
2348 if v.is_null() {
2349 b.append_null();
2350 } else {
2351 b.append_value(uni_common::cypher_value_codec::encode(v));
2352 }
2353 },
2354 ),
2355 }
2356 }
2357
2358 #[expect(
2360 clippy::too_many_arguments,
2361 reason = "builder plumbing: value type + optional child metadata are distinct knobs"
2362 )]
2363 fn build_typed_map<F, B, A>(
2364 &self,
2365 len: usize,
2366 deleted: &[bool],
2367 get_props: &F,
2368 value_builder: B,
2369 value_arrow_type: arrow_schema::DataType,
2370 value_metadata: Option<HashMap<String, String>>,
2371 mut append_value: A,
2372 ) -> Result<ArrayRef>
2373 where
2374 F: Fn(usize) -> Option<&'a Value>,
2375 B: arrow_array::builder::ArrayBuilder,
2376 A: FnMut(&Value, &mut B),
2377 {
2378 let key_builder = Box::new(StringBuilder::new());
2379 let value_builder = Box::new(value_builder);
2380 let value_field = match value_metadata {
2381 Some(meta) => Field::new("value", value_arrow_type, true).with_metadata(meta),
2382 None => Field::new("value", value_arrow_type, true),
2383 };
2384 let struct_builder = StructBuilder::new(
2385 vec![
2386 Field::new("key", arrow_schema::DataType::Utf8, false),
2387 value_field,
2388 ],
2389 vec![key_builder, value_builder],
2390 );
2391 let mut builder = ListBuilder::new(struct_builder);
2392
2393 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
2394 self.append_map_entry(&mut builder, get_props(i), is_deleted, &mut append_value);
2395 }
2396 Ok(Arc::new(builder.finish()))
2397 }
2398
2399 fn append_map_entry<B, A>(
2401 &self,
2402 builder: &mut ListBuilder<StructBuilder>,
2403 val: Option<&'a Value>,
2404 is_deleted: bool,
2405 append_value: &mut A,
2406 ) where
2407 B: arrow_array::builder::ArrayBuilder,
2408 A: FnMut(&Value, &mut B),
2409 {
2410 let val_obj = val.and_then(|v| v.as_object());
2411 if val_obj.is_none() && is_deleted {
2412 builder.append(false);
2413 } else if let Some(obj) = val_obj {
2414 let struct_b = builder.values();
2415 for (k, v) in obj {
2416 struct_b
2417 .field_builder::<StringBuilder>(0)
2418 .unwrap()
2419 .append_value(k);
2420 let value_b = struct_b.field_builder::<B>(1).unwrap();
2422 append_value(v, value_b);
2423 struct_b.append(true);
2424 }
2425 builder.append(true);
2426 } else {
2427 builder.append(false);
2428 }
2429 }
2430
2431 fn build_crdt_column<F>(&self, len: usize, deleted: &[bool], get_props: F) -> Result<ArrayRef>
2432 where
2433 F: Fn(usize) -> Option<&'a Value>,
2434 {
2435 let mut builder = BinaryBuilder::new();
2436 for (i, &is_deleted) in deleted.iter().enumerate().take(len) {
2437 if is_deleted {
2438 builder.append_null();
2439 continue;
2440 }
2441 if let Some(val) = get_props(i) {
2442 let crdt_result = if let Some(s) = val.as_str() {
2445 serde_json::from_str::<Crdt>(s)
2446 } else {
2447 let json_val: serde_json::Value = val.clone().into();
2449 serde_json::from_value::<Crdt>(json_val)
2450 };
2451
2452 if let Ok(crdt) = crdt_result {
2453 if let Ok(bytes) = crdt.to_msgpack() {
2454 builder.append_value(&bytes);
2455 } else {
2456 builder.append_null();
2457 }
2458 } else {
2459 builder.append_null();
2460 }
2461 } else {
2462 builder.append_null();
2463 }
2464 }
2465 Ok(Arc::new(builder.finish()))
2466 }
2467}
2468
2469pub fn build_edge_column<'a>(
2471 name: &'a str,
2472 data_type: &'a DataType,
2473 len: usize,
2474 get_props: impl Fn(usize) -> Option<&'a Value>,
2475) -> Result<ArrayRef> {
2476 let deleted = vec![false; len];
2478 let extractor = PropertyExtractor::new(name, data_type);
2479 extractor.build_column(len, &deleted, get_props)
2480}
2481
2482#[cfg(test)]
2483mod tests {
2484 use super::*;
2485 use arrow_array::{
2486 Array, DurationMicrosecondArray,
2487 builder::{BinaryBuilder, Time64MicrosecondBuilder, TimestampNanosecondBuilder},
2488 };
2489 use std::collections::HashMap;
2490 use uni_common::TemporalValue;
2491 use uni_crdt::{Crdt, GCounter};
2492
2493 #[test]
2494 fn test_sparse_vector_columnar_roundtrip_and_no_silent_null() {
2495 use crate::storage::value_codec::{CrdtDecodeMode, decode_column_value, value_from_column};
2496
2497 let dt = DataType::SparseVector { dimensions: 100 };
2498 let v0 = Value::SparseVector {
2499 indices: vec![1, 5, 9],
2500 values: vec![0.5, -1.0, 2.0],
2501 };
2502 let v1 = Value::SparseVector {
2504 indices: vec![],
2505 values: vec![],
2506 };
2507 let props = [Some(v0.clone()), Some(v1.clone()), None];
2509 let deleted = [false, false, true];
2510
2511 let extractor = PropertyExtractor::new("emb", &dt);
2512 let col = extractor
2513 .build_column(3, &deleted, |i| props[i].as_ref())
2514 .unwrap();
2515
2516 assert_eq!(
2518 decode_column_value(&col, &dt, 0, CrdtDecodeMode::Strict).unwrap(),
2519 v0
2520 );
2521 assert_eq!(
2522 decode_column_value(&col, &dt, 1, CrdtDecodeMode::Strict).unwrap(),
2523 v1
2524 );
2525 assert_eq!(
2526 decode_column_value(&col, &dt, 2, CrdtDecodeMode::Strict).unwrap(),
2527 Value::Null
2528 );
2529
2530 let json0 = value_from_column(&col, &dt, 0, CrdtDecodeMode::Strict).unwrap();
2533 assert!(
2534 json0.is_object(),
2535 "sparse column was silently nulled by value_from_column: {json0:?}"
2536 );
2537 assert_eq!(json0["indices"], serde_json::json!([1u32, 5u32, 9u32]));
2538 assert_eq!(
2539 json0["values"],
2540 serde_json::json!([0.5f32, -1.0f32, 2.0f32])
2541 );
2542 }
2543
2544 #[test]
2545 fn test_arrow_to_value_string() {
2546 let arr = StringArray::from(vec![Some("hello"), None, Some("world")]);
2547 assert_eq!(
2548 arrow_to_value(&arr, 0, None),
2549 Value::String("hello".to_string())
2550 );
2551 assert_eq!(arrow_to_value(&arr, 1, None), Value::Null);
2552 assert_eq!(
2553 arrow_to_value(&arr, 2, None),
2554 Value::String("world".to_string())
2555 );
2556 }
2557
2558 #[test]
2559 fn test_arrow_to_value_int64() {
2560 let arr = Int64Array::from(vec![Some(42), None, Some(-10)]);
2561 assert_eq!(arrow_to_value(&arr, 0, None), Value::Int(42));
2562 assert_eq!(arrow_to_value(&arr, 1, None), Value::Null);
2563 assert_eq!(arrow_to_value(&arr, 2, None), Value::Int(-10));
2564 }
2565
2566 #[test]
2567 #[allow(clippy::approx_constant)]
2568 fn test_arrow_to_value_float64() {
2569 let arr = Float64Array::from(vec![Some(3.14), None]);
2570 assert_eq!(arrow_to_value(&arr, 0, None), Value::Float(3.14));
2571 assert_eq!(arrow_to_value(&arr, 1, None), Value::Null);
2572 }
2573
2574 #[test]
2575 fn test_arrow_to_value_bool() {
2576 let arr = BooleanArray::from(vec![Some(true), Some(false), None]);
2577 assert_eq!(arrow_to_value(&arr, 0, None), Value::Bool(true));
2578 assert_eq!(arrow_to_value(&arr, 1, None), Value::Bool(false));
2579 assert_eq!(arrow_to_value(&arr, 2, None), Value::Null);
2580 }
2581
2582 #[test]
2583 fn test_values_to_array_int64() {
2584 let values = vec![Value::Int(1), Value::Int(2), Value::Null, Value::Int(4)];
2585 let arr = values_to_array(&values, &ArrowDataType::Int64).unwrap();
2586 assert_eq!(arr.len(), 4);
2587
2588 let int_arr = arr.as_any().downcast_ref::<Int64Array>().unwrap();
2589 assert_eq!(int_arr.value(0), 1);
2590 assert_eq!(int_arr.value(1), 2);
2591 assert!(int_arr.is_null(2));
2592 assert_eq!(int_arr.value(3), 4);
2593 }
2594
2595 #[test]
2596 fn test_values_to_array_string() {
2597 let values = vec![
2598 Value::String("a".to_string()),
2599 Value::String("b".to_string()),
2600 Value::Null,
2601 ];
2602 let arr = values_to_array(&values, &ArrowDataType::Utf8).unwrap();
2603 assert_eq!(arr.len(), 3);
2604
2605 let str_arr = arr.as_any().downcast_ref::<StringArray>().unwrap();
2606 assert_eq!(str_arr.value(0), "a");
2607 assert_eq!(str_arr.value(1), "b");
2608 assert!(str_arr.is_null(2));
2609 }
2610
2611 #[test]
2612 fn test_property_extractor_string() {
2613 let props: Vec<HashMap<String, Value>> = vec![
2614 [("name".to_string(), Value::String("Alice".to_string()))]
2615 .into_iter()
2616 .collect(),
2617 [("name".to_string(), Value::String("Bob".to_string()))]
2618 .into_iter()
2619 .collect(),
2620 HashMap::new(),
2621 ];
2622 let deleted = vec![false, false, true];
2623
2624 let extractor = PropertyExtractor::new("name", &DataType::String);
2625 let arr = extractor
2626 .build_column(3, &deleted, |i| props[i].get("name"))
2627 .unwrap();
2628
2629 let str_arr = arr.as_any().downcast_ref::<StringArray>().unwrap();
2630 assert_eq!(str_arr.value(0), "Alice");
2631 assert_eq!(str_arr.value(1), "Bob");
2632 assert_eq!(str_arr.value(2), ""); }
2634
2635 #[test]
2636 fn test_property_extractor_int64() {
2637 let props: Vec<HashMap<String, Value>> = vec![
2638 [("age".to_string(), Value::Int(25))].into_iter().collect(),
2639 [("age".to_string(), Value::Int(30))].into_iter().collect(),
2640 HashMap::new(),
2641 ];
2642 let deleted = vec![false, false, true];
2643
2644 let extractor = PropertyExtractor::new("age", &DataType::Int64);
2645 let arr = extractor
2646 .build_column(3, &deleted, |i| props[i].get("age"))
2647 .unwrap();
2648
2649 let int_arr = arr.as_any().downcast_ref::<Int64Array>().unwrap();
2650 assert_eq!(int_arr.value(0), 25);
2651 assert_eq!(int_arr.value(1), 30);
2652 assert_eq!(int_arr.value(2), 0); }
2654
2655 #[test]
2656 fn test_property_extractor_bytes_roundtrip() {
2657 let blob = vec![0u8, 1, 2, 255];
2658 let props: Vec<HashMap<String, Value>> = vec![
2659 [("blob".to_string(), Value::Bytes(blob.clone()))]
2660 .into_iter()
2661 .collect(),
2662 [("blob".to_string(), Value::Bytes(Vec::new()))]
2663 .into_iter()
2664 .collect(),
2665 HashMap::new(),
2666 ];
2667 let deleted = vec![false, false, false];
2668
2669 let extractor = PropertyExtractor::new("blob", &DataType::Bytes);
2670 let arr = extractor
2671 .build_column(3, &deleted, |i| props[i].get("blob"))
2672 .unwrap();
2673
2674 assert_eq!(
2676 arrow_to_value(arr.as_ref(), 0, Some(&DataType::Bytes)),
2677 Value::Bytes(blob)
2678 );
2679 assert_eq!(
2680 arrow_to_value(arr.as_ref(), 1, Some(&DataType::Bytes)),
2681 Value::Bytes(Vec::new())
2682 );
2683 assert_eq!(
2685 arrow_to_value(arr.as_ref(), 2, Some(&DataType::Bytes)),
2686 Value::Null
2687 );
2688 }
2689
2690 #[test]
2691 fn test_bytes_vs_cypher_value_disambiguation() {
2692 let raw = vec![0xDEu8, 0xAD, 0xBE, 0xEF];
2696 let props: Vec<HashMap<String, Value>> = vec![
2697 [("blob".to_string(), Value::Bytes(raw.clone()))]
2698 .into_iter()
2699 .collect(),
2700 ];
2701 let extractor = PropertyExtractor::new("blob", &DataType::Bytes);
2702 let arr = extractor
2703 .build_column(1, &[false], |i| props[i].get("blob"))
2704 .unwrap();
2705 assert_eq!(
2707 arrow_to_value(arr.as_ref(), 0, Some(&DataType::Bytes)),
2708 Value::Bytes(raw)
2709 );
2710 }
2711
2712 #[test]
2713 fn test_data_type_bytes_to_arrow() {
2714 assert_eq!(DataType::Bytes.to_arrow(), ArrowDataType::LargeBinary);
2715 }
2716
2717 #[test]
2718 fn test_arrow_to_value_time64() {
2719 let mut builder = Time64MicrosecondBuilder::new();
2721 builder.append_value(37_845_000_000);
2723 builder.append_value(0);
2725 builder.append_value(86_399_123_456);
2727 builder.append_null();
2728
2729 let arr = builder.finish();
2730 assert_eq!(arrow_to_value(&arr, 0, None).to_string(), "10:30:45");
2732 assert_eq!(arrow_to_value(&arr, 1, None).to_string(), "00:00");
2733 assert_eq!(arrow_to_value(&arr, 2, None).to_string(), "23:59:59.123456");
2734 assert_eq!(arrow_to_value(&arr, 3, None), Value::Null);
2735 }
2736
2737 #[test]
2738 fn test_arrow_to_value_duration() {
2739 let arr = DurationMicrosecondArray::from(vec![
2742 Some(1_000_000), Some(3_600_000_000), Some(86_400_000_000), None,
2746 ]);
2747
2748 assert_eq!(arrow_to_value(&arr, 0, None).to_string(), "PT1S");
2749 assert_eq!(arrow_to_value(&arr, 1, None).to_string(), "PT1H");
2750 assert_eq!(arrow_to_value(&arr, 2, None).to_string(), "PT24H");
2751 assert_eq!(arrow_to_value(&arr, 3, None), Value::Null);
2752 }
2753
2754 #[test]
2755 fn test_arrow_to_value_binary_crdt() {
2756 let mut builder = BinaryBuilder::new();
2758
2759 let mut counter = GCounter::new();
2761 counter.increment("actor1", 5);
2762 let crdt = Crdt::GCounter(counter);
2763 let bytes = crdt.to_msgpack().unwrap();
2764 builder.append_value(&bytes);
2765
2766 builder.append_null();
2768
2769 let arr = builder.finish();
2770
2771 let result = arrow_to_value(&arr, 0, None);
2773 assert!(result.as_object().is_some());
2774 let obj = result.as_object().unwrap();
2775 assert_eq!(obj.get("t"), Some(&Value::String("gc".to_string())));
2777
2778 assert_eq!(arrow_to_value(&arr, 1, None), Value::Null);
2780 }
2781
2782 #[test]
2783 fn test_datetime_struct_encode_decode_roundtrip() {
2784 let values = vec![
2786 Value::Temporal(TemporalValue::DateTime {
2787 nanos_since_epoch: 441763200000000000, offset_seconds: 3600, timezone_name: Some("Europe/Paris".to_string()),
2790 }),
2791 Value::Temporal(TemporalValue::DateTime {
2792 nanos_since_epoch: 1704067200000000000, offset_seconds: -18000, timezone_name: None,
2795 }),
2796 Value::Temporal(TemporalValue::DateTime {
2797 nanos_since_epoch: 0, offset_seconds: 0,
2799 timezone_name: Some("UTC".to_string()),
2800 }),
2801 ];
2802
2803 let arr_ref = values_to_datetime_struct_array(&values);
2805 let arr = arr_ref.as_any().downcast_ref::<StructArray>().unwrap();
2806 assert_eq!(arr.len(), 3);
2807
2808 let decoded_0 = arrow_to_value(arr_ref.as_ref(), 0, Some(&DataType::DateTime));
2810 let decoded_1 = arrow_to_value(arr_ref.as_ref(), 1, Some(&DataType::DateTime));
2811 let decoded_2 = arrow_to_value(arr_ref.as_ref(), 2, Some(&DataType::DateTime));
2812
2813 assert_eq!(decoded_0, values[0]);
2815 assert_eq!(decoded_1, values[1]);
2816 assert_eq!(decoded_2, values[2]);
2817
2818 if let Value::Temporal(TemporalValue::DateTime {
2820 nanos_since_epoch,
2821 offset_seconds,
2822 timezone_name,
2823 }) = decoded_0
2824 {
2825 assert_eq!(nanos_since_epoch, 441763200000000000);
2826 assert_eq!(offset_seconds, 3600);
2827 assert_eq!(timezone_name, Some("Europe/Paris".to_string()));
2828 } else {
2829 panic!("Expected DateTime value");
2830 }
2831 }
2832
2833 #[test]
2834 fn test_datetime_struct_null_handling() {
2835 let values = vec![
2837 Value::Temporal(TemporalValue::DateTime {
2838 nanos_since_epoch: 441763200000000000,
2839 offset_seconds: 3600,
2840 timezone_name: Some("Europe/Paris".to_string()),
2841 }),
2842 Value::Null,
2843 Value::Temporal(TemporalValue::DateTime {
2844 nanos_since_epoch: 0,
2845 offset_seconds: 0,
2846 timezone_name: None,
2847 }),
2848 ];
2849
2850 let arr_ref = values_to_datetime_struct_array(&values);
2851 let arr = arr_ref.as_any().downcast_ref::<StructArray>().unwrap();
2852 assert_eq!(arr.len(), 3);
2853
2854 let decoded_0 = arrow_to_value(arr_ref.as_ref(), 0, Some(&DataType::DateTime));
2856 assert_eq!(decoded_0, values[0]);
2857
2858 assert!(arr.is_null(1));
2860 let decoded_1 = arrow_to_value(arr_ref.as_ref(), 1, Some(&DataType::DateTime));
2861 assert_eq!(decoded_1, Value::Null);
2862
2863 let decoded_2 = arrow_to_value(arr_ref.as_ref(), 2, Some(&DataType::DateTime));
2865 assert_eq!(decoded_2, values[2]);
2866 }
2867
2868 #[test]
2869 fn test_datetime_struct_boundary_values() {
2870 let values = vec![
2872 Value::Temporal(TemporalValue::DateTime {
2873 nanos_since_epoch: 441763200000000000,
2874 offset_seconds: 0, timezone_name: None,
2876 }),
2877 Value::Temporal(TemporalValue::DateTime {
2878 nanos_since_epoch: 441763200000000000,
2879 offset_seconds: 43200, timezone_name: None,
2881 }),
2882 Value::Temporal(TemporalValue::DateTime {
2883 nanos_since_epoch: 441763200000000000,
2884 offset_seconds: -43200, timezone_name: None,
2886 }),
2887 ];
2888
2889 let arr_ref = values_to_datetime_struct_array(&values);
2890 let arr = arr_ref.as_any().downcast_ref::<StructArray>().unwrap();
2891 assert_eq!(arr.len(), 3);
2892
2893 for (i, expected) in values.iter().enumerate() {
2895 let decoded = arrow_to_value(arr_ref.as_ref(), i, Some(&DataType::DateTime));
2896 assert_eq!(&decoded, expected);
2897 }
2898 }
2899
2900 #[test]
2901 fn test_datetime_old_schema_migration() {
2902 let mut builder = TimestampNanosecondBuilder::new().with_timezone("UTC");
2904 builder.append_value(441763200000000000); builder.append_value(1704067200000000000); builder.append_null();
2907
2908 let arr = builder.finish();
2909
2910 let decoded_0 = arrow_to_value(&arr, 0, Some(&DataType::DateTime));
2912 let _decoded_1 = arrow_to_value(&arr, 1, Some(&DataType::DateTime));
2913 let decoded_2 = arrow_to_value(&arr, 2, Some(&DataType::DateTime));
2914
2915 if let Value::Temporal(TemporalValue::DateTime {
2917 nanos_since_epoch,
2918 offset_seconds,
2919 timezone_name,
2920 }) = decoded_0
2921 {
2922 assert_eq!(nanos_since_epoch, 441763200000000000);
2923 assert_eq!(offset_seconds, 0);
2924 assert_eq!(timezone_name, Some("UTC".to_string()));
2925 } else {
2926 panic!("Expected DateTime value");
2927 }
2928
2929 assert_eq!(decoded_2, Value::Null);
2931 }
2932
2933 #[test]
2934 fn test_time_struct_encode_decode_roundtrip() {
2935 let values = vec![
2937 Value::Temporal(TemporalValue::Time {
2938 nanos_since_midnight: 37845000000000, offset_seconds: 3600, }),
2941 Value::Temporal(TemporalValue::Time {
2942 nanos_since_midnight: 0, offset_seconds: 0,
2944 }),
2945 Value::Temporal(TemporalValue::Time {
2946 nanos_since_midnight: 86399999999999, offset_seconds: -18000, }),
2949 ];
2950
2951 let arr_ref = values_to_time_struct_array(&values);
2953 let arr = arr_ref.as_any().downcast_ref::<StructArray>().unwrap();
2954 assert_eq!(arr.len(), 3);
2955
2956 let decoded_0 = arrow_to_value(arr_ref.as_ref(), 0, Some(&DataType::Time));
2958 let decoded_1 = arrow_to_value(arr_ref.as_ref(), 1, Some(&DataType::Time));
2959 let decoded_2 = arrow_to_value(arr_ref.as_ref(), 2, Some(&DataType::Time));
2960
2961 assert_eq!(decoded_0, values[0]);
2963 assert_eq!(decoded_1, values[1]);
2964 assert_eq!(decoded_2, values[2]);
2965
2966 if let Value::Temporal(TemporalValue::Time {
2968 nanos_since_midnight,
2969 offset_seconds,
2970 }) = decoded_0
2971 {
2972 assert_eq!(nanos_since_midnight, 37845000000000);
2973 assert_eq!(offset_seconds, 3600);
2974 } else {
2975 panic!("Expected Time value");
2976 }
2977 }
2978
2979 #[test]
2980 fn test_time_struct_null_handling() {
2981 let values = vec![
2983 Value::Temporal(TemporalValue::Time {
2984 nanos_since_midnight: 37845000000000,
2985 offset_seconds: 3600,
2986 }),
2987 Value::Null,
2988 Value::Temporal(TemporalValue::Time {
2989 nanos_since_midnight: 0,
2990 offset_seconds: 0,
2991 }),
2992 ];
2993
2994 let arr_ref = values_to_time_struct_array(&values);
2995 let arr = arr_ref.as_any().downcast_ref::<StructArray>().unwrap();
2996 assert_eq!(arr.len(), 3);
2997
2998 let decoded_0 = arrow_to_value(arr_ref.as_ref(), 0, Some(&DataType::Time));
3000 assert_eq!(decoded_0, values[0]);
3001
3002 assert!(arr.is_null(1));
3004 let decoded_1 = arrow_to_value(arr_ref.as_ref(), 1, Some(&DataType::Time));
3005 assert_eq!(decoded_1, Value::Null);
3006
3007 let decoded_2 = arrow_to_value(arr_ref.as_ref(), 2, Some(&DataType::Time));
3009 assert_eq!(decoded_2, values[2]);
3010 }
3011
3012 #[test]
3015 fn test_extract_vector_f32_values_valid_vector() {
3016 let v = vec![1.0, 2.0, 3.0];
3017 let val = Value::Vector(v.clone());
3018 let (result, valid) = extract_vector_f32_values(Some(&val), false, 3);
3019 assert_eq!(result, v);
3020 assert!(valid);
3021 }
3022
3023 #[test]
3024 fn test_extract_vector_f32_values_vector_wrong_dims() {
3025 let v = vec![1.0, 2.0];
3029 let val = Value::Vector(v);
3030 let (result, valid) = extract_vector_f32_values(Some(&val), false, 3);
3031 assert_eq!(result, vec![0.0, 0.0, 0.0]);
3032 assert!(!valid);
3033 }
3034
3035 #[test]
3036 fn test_extract_vector_f32_values_valid_list() {
3037 let v = vec![Value::Float(1.0), Value::Float(2.0), Value::Float(3.0)];
3038 let val = Value::List(v);
3039 let (result, valid) = extract_vector_f32_values(Some(&val), false, 3);
3040 assert_eq!(result, vec![1.0, 2.0, 3.0]);
3041 assert!(valid);
3042 }
3043
3044 #[test]
3045 fn test_extract_vector_f32_values_list_wrong_dims() {
3046 let v = vec![Value::Float(1.0), Value::Float(2.0)];
3048 let val = Value::List(v);
3049 let (result, valid) = extract_vector_f32_values(Some(&val), false, 3);
3050 assert_eq!(result, vec![0.0, 0.0, 0.0]);
3051 assert!(!valid);
3052 }
3053
3054 #[test]
3055 fn test_extract_vector_f32_values_list_int_coercion() {
3056 let v = vec![Value::Int(1), Value::Int(2), Value::Int(3)];
3057 let val = Value::List(v);
3058 let (result, valid) = extract_vector_f32_values(Some(&val), false, 3);
3059 assert_eq!(result, vec![1.0, 2.0, 3.0]);
3060 assert!(valid);
3061 }
3062
3063 #[test]
3064 fn test_extract_vector_f32_values_none() {
3065 let (result, valid) = extract_vector_f32_values(None, false, 3);
3066 assert_eq!(result, vec![0.0, 0.0, 0.0]);
3067 assert!(!valid);
3068 }
3069
3070 #[test]
3071 fn test_extract_vector_f32_values_null() {
3072 let val = Value::Null;
3073 let (result, valid) = extract_vector_f32_values(Some(&val), false, 3);
3074 assert_eq!(result, vec![0.0, 0.0, 0.0]);
3075 assert!(!valid);
3076 }
3077
3078 #[test]
3079 fn test_extract_vector_f32_values_unsupported_type() {
3080 let val = Value::String("not a vector".to_string());
3081 let (result, valid) = extract_vector_f32_values(Some(&val), false, 3);
3082 assert_eq!(result, vec![0.0, 0.0, 0.0]);
3083 assert!(!valid);
3084 }
3085
3086 #[test]
3087 fn test_extract_vector_f32_values_deleted_with_none() {
3088 let (result, valid) = extract_vector_f32_values(None, true, 3);
3089 assert_eq!(result, vec![0.0, 0.0, 0.0]);
3090 assert!(valid); }
3092
3093 #[test]
3094 fn test_extract_vector_f32_values_deleted_with_null() {
3095 let val = Value::Null;
3096 let (result, valid) = extract_vector_f32_values(Some(&val), true, 3);
3097 assert_eq!(result, vec![0.0, 0.0, 0.0]);
3098 assert!(valid); }
3100
3101 #[test]
3104 fn test_values_to_fixed_size_list_vector_with_nulls() {
3105 let values = vec![
3106 Value::Vector(vec![1.0, 2.0]),
3107 Value::Null,
3108 Value::Vector(vec![3.0, 4.0]),
3109 Value::String("invalid".to_string()),
3110 ];
3111 let arr_ref = values_to_array(
3112 &values,
3113 &ArrowDataType::FixedSizeList(
3114 Arc::new(Field::new("item", ArrowDataType::Float32, false)),
3115 2,
3116 ),
3117 )
3118 .unwrap();
3119
3120 let arr = arr_ref
3121 .as_any()
3122 .downcast_ref::<FixedSizeListArray>()
3123 .unwrap();
3124
3125 assert_eq!(arr.len(), 4);
3126 assert!(arr.is_valid(0));
3127 assert!(!arr.is_valid(1)); assert!(arr.is_valid(2));
3129 assert!(!arr.is_valid(3)); }
3131
3132 #[test]
3133 fn test_values_to_fixed_size_list_from_list() {
3134 let values = vec![
3135 Value::List(vec![Value::Float(1.0), Value::Float(2.0)]),
3136 Value::List(vec![Value::Int(3), Value::Int(4)]),
3137 ];
3138 let arr_ref = values_to_array(
3139 &values,
3140 &ArrowDataType::FixedSizeList(
3141 Arc::new(Field::new("item", ArrowDataType::Float32, false)),
3142 2,
3143 ),
3144 )
3145 .unwrap();
3146
3147 let arr = arr_ref
3148 .as_any()
3149 .downcast_ref::<FixedSizeListArray>()
3150 .unwrap();
3151
3152 assert_eq!(arr.len(), 2);
3153 assert!(arr.is_valid(0));
3154 assert!(arr.is_valid(1));
3155
3156 let child = arr
3158 .values()
3159 .as_any()
3160 .downcast_ref::<Float32Array>()
3161 .unwrap();
3162 assert_eq!(child.value(0), 1.0);
3163 assert_eq!(child.value(1), 2.0);
3164 assert_eq!(child.value(2), 3.0);
3165 assert_eq!(child.value(3), 4.0);
3166 }
3167
3168 #[test]
3169 fn test_values_to_fixed_size_list_wrong_dimensions() {
3170 let values = vec![
3171 Value::Vector(vec![1.0, 2.0, 3.0]), Value::List(vec![Value::Float(4.0)]), ];
3174 let arr_ref = values_to_array(
3175 &values,
3176 &ArrowDataType::FixedSizeList(
3177 Arc::new(Field::new("item", ArrowDataType::Float32, false)),
3178 2,
3179 ),
3180 )
3181 .unwrap();
3182
3183 let arr = arr_ref
3184 .as_any()
3185 .downcast_ref::<FixedSizeListArray>()
3186 .unwrap();
3187
3188 assert_eq!(arr.len(), 2);
3189 assert!(!arr.is_valid(0)); assert!(!arr.is_valid(1)); let child = arr
3194 .values()
3195 .as_any()
3196 .downcast_ref::<Float32Array>()
3197 .unwrap();
3198 assert_eq!(child.value(0), 0.0);
3199 assert_eq!(child.value(1), 0.0);
3200 assert_eq!(child.value(2), 0.0);
3201 assert_eq!(child.value(3), 0.0);
3202 }
3203
3204 #[test]
3205 fn test_values_to_fixed_size_list_all_nulls() {
3206 let values = vec![Value::Null, Value::Null, Value::Null];
3207 let arr_ref = values_to_array(
3208 &values,
3209 &ArrowDataType::FixedSizeList(
3210 Arc::new(Field::new("item", ArrowDataType::Float32, false)),
3211 3,
3212 ),
3213 )
3214 .unwrap();
3215
3216 let arr = arr_ref
3217 .as_any()
3218 .downcast_ref::<FixedSizeListArray>()
3219 .unwrap();
3220
3221 assert_eq!(arr.len(), 3);
3222 assert!(!arr.is_valid(0));
3223 assert!(!arr.is_valid(1));
3224 assert!(!arr.is_valid(2));
3225
3226 let child = arr
3228 .values()
3229 .as_any()
3230 .downcast_ref::<Float32Array>()
3231 .unwrap();
3232 assert_eq!(child.len(), 9);
3233 }
3234
3235 #[test]
3236 fn test_values_to_fixed_size_list_mixed_types() {
3237 let values = vec![
3238 Value::Vector(vec![1.0, 2.0]),
3239 Value::List(vec![Value::Float(3.0), Value::Float(4.0)]),
3240 Value::Null,
3241 Value::String("invalid".to_string()),
3242 ];
3243 let arr_ref = values_to_array(
3244 &values,
3245 &ArrowDataType::FixedSizeList(
3246 Arc::new(Field::new("item", ArrowDataType::Float32, false)),
3247 2,
3248 ),
3249 )
3250 .unwrap();
3251
3252 let arr = arr_ref
3253 .as_any()
3254 .downcast_ref::<FixedSizeListArray>()
3255 .unwrap();
3256
3257 assert_eq!(arr.len(), 4);
3258 assert!(arr.is_valid(0)); assert!(arr.is_valid(1)); assert!(!arr.is_valid(2)); assert!(!arr.is_valid(3)); let child = arr
3265 .values()
3266 .as_any()
3267 .downcast_ref::<Float32Array>()
3268 .unwrap();
3269 assert_eq!(child.value(0), 1.0);
3270 assert_eq!(child.value(1), 2.0);
3271 assert_eq!(child.value(2), 3.0);
3272 assert_eq!(child.value(3), 4.0);
3273 }
3274
3275 #[test]
3278 fn test_build_vector_column_with_nulls_and_deleted() {
3279 let data_type = DataType::Vector { dimensions: 3 };
3280 let extractor = PropertyExtractor::new("test_vec", &data_type);
3281
3282 let props = [
3283 Some(Value::Vector(vec![1.0, 2.0, 3.0])),
3284 None, Some(Value::Null), Some(Value::Vector(vec![4.0, 5.0, 6.0])),
3287 ];
3288 let deleted = [false, false, false, true]; let arr_ref = extractor
3291 .build_vector_column(4, &deleted, |i| props[i].as_ref(), 3)
3292 .unwrap();
3293
3294 let arr = arr_ref
3295 .as_any()
3296 .downcast_ref::<FixedSizeListArray>()
3297 .unwrap();
3298
3299 assert_eq!(arr.len(), 4);
3300 assert!(arr.is_valid(0)); assert!(!arr.is_valid(1)); assert!(!arr.is_valid(2)); assert!(arr.is_valid(3)); let child = arr
3307 .values()
3308 .as_any()
3309 .downcast_ref::<Float32Array>()
3310 .unwrap();
3311 assert_eq!(child.value(0), 1.0);
3312 assert_eq!(child.value(1), 2.0);
3313 assert_eq!(child.value(2), 3.0);
3314 assert_eq!(child.value(9), 0.0);
3318 assert_eq!(child.value(10), 0.0);
3319 assert_eq!(child.value(11), 0.0);
3320 }
3321
3322 #[test]
3323 fn test_build_vector_column_with_list_input() {
3324 let data_type = DataType::Vector { dimensions: 2 };
3325 let extractor = PropertyExtractor::new("test_vec", &data_type);
3326
3327 let props = [
3328 Some(Value::List(vec![Value::Float(1.0), Value::Float(2.0)])),
3329 Some(Value::List(vec![Value::Int(3), Value::Int(4)])),
3330 Some(Value::Vector(vec![5.0, 6.0])),
3331 ];
3332 let deleted = [false, false, false];
3333
3334 let arr_ref = extractor
3335 .build_vector_column(3, &deleted, |i| props[i].as_ref(), 2)
3336 .unwrap();
3337
3338 let arr = arr_ref
3339 .as_any()
3340 .downcast_ref::<FixedSizeListArray>()
3341 .unwrap();
3342
3343 assert_eq!(arr.len(), 3);
3344 assert!(arr.is_valid(0));
3345 assert!(arr.is_valid(1));
3346 assert!(arr.is_valid(2));
3347
3348 let child = arr
3350 .values()
3351 .as_any()
3352 .downcast_ref::<Float32Array>()
3353 .unwrap();
3354 assert_eq!(child.value(0), 1.0);
3355 assert_eq!(child.value(1), 2.0);
3356 assert_eq!(child.value(2), 3.0);
3357 assert_eq!(child.value(3), 4.0);
3358 assert_eq!(child.value(4), 5.0);
3359 assert_eq!(child.value(5), 6.0);
3360 }
3361
3362 #[test]
3363 fn test_build_binary_vector_column_roundtrip() {
3364 let data_type = DataType::BinaryVector { dimensions: 3 };
3368 let extractor = PropertyExtractor::new("bits", &data_type);
3369
3370 let props = [
3371 Some(Value::BinaryVector(vec![0x00, 0xFF, 0xA5])),
3372 Some(Value::List(vec![
3373 Value::Int(1),
3374 Value::Int(2),
3375 Value::Int(255),
3376 ])),
3377 None, None, ];
3380 let deleted = [false, false, true, false];
3381
3382 let arr_ref = extractor
3383 .build_binary_vector_column(4, &deleted, |i| props[i].as_ref(), 3)
3384 .unwrap();
3385
3386 let arr = arr_ref
3387 .as_any()
3388 .downcast_ref::<FixedSizeListArray>()
3389 .unwrap();
3390 assert_eq!(arr.len(), 4);
3391 assert!(arr.is_valid(0));
3392 assert!(arr.is_valid(1));
3393 assert!(arr.is_valid(2)); assert!(!arr.is_valid(3)); let child = arr.values().as_any().downcast_ref::<UInt8Array>().unwrap();
3397 assert_eq!(child.value(0), 0x00);
3398 assert_eq!(child.value(1), 0xFF);
3399 assert_eq!(child.value(2), 0xA5);
3400 assert_eq!(child.value(3), 1);
3401 assert_eq!(child.value(5), 255);
3402
3403 assert_eq!(
3405 arrow_to_value(&arr_ref, 0, Some(&data_type)),
3406 Value::BinaryVector(vec![0x00, 0xFF, 0xA5])
3407 );
3408 assert_eq!(
3409 arrow_to_value(&arr_ref, 1, Some(&data_type)),
3410 Value::BinaryVector(vec![1, 2, 255])
3411 );
3412 assert_eq!(arrow_to_value(&arr_ref, 3, Some(&data_type)), Value::Null);
3413 }
3414
3415 #[test]
3416 fn test_build_binary_vector_column_wrong_length_fails_closed() {
3417 let data_type = DataType::BinaryVector { dimensions: 3 };
3419 let extractor = PropertyExtractor::new("bits", &data_type);
3420 let props = [Some(Value::BinaryVector(vec![1, 2]))]; let deleted = [false];
3422 let err = extractor
3423 .build_binary_vector_column(1, &deleted, |i| props[i].as_ref(), 3)
3424 .unwrap_err();
3425 assert!(err.to_string().contains("bits"), "got: {err}");
3426 }
3427
3428 #[test]
3431 fn test_build_multivector_list_column_roundtrip() {
3432 let data_type = DataType::List(Box::new(DataType::Vector { dimensions: 3 }));
3438 let extractor = PropertyExtractor::new("tokens", &data_type);
3439
3440 let props = [
3441 Some(Value::List(vec![
3443 Value::Vector(vec![1.0, 2.0, 3.0]),
3444 Value::Vector(vec![4.0, 5.0, 6.0]),
3445 ])),
3446 Some(Value::List(vec![
3448 Value::Vector(vec![7.0, 8.0, 9.0]),
3449 Value::Vector(vec![10.0, 11.0, 12.0]),
3450 Value::Vector(vec![13.0, 14.0, 15.0]),
3451 ])),
3452 Some(Value::List(vec![])),
3454 None,
3456 ];
3457 let deleted = [false, false, false, true];
3458
3459 let arr_ref = extractor
3460 .build_column(4, &deleted, |i| props[i].as_ref())
3461 .unwrap();
3462
3463 let outer = arr_ref.as_any().downcast_ref::<ListArray>().unwrap();
3465 assert_eq!(outer.len(), 4);
3466 assert!(outer.is_valid(0));
3467 assert!(outer.is_valid(1));
3468 assert!(outer.is_valid(2)); assert!(!outer.is_valid(3)); assert_eq!(outer.value(0).len(), 2);
3473 assert_eq!(outer.value(1).len(), 3);
3474 assert_eq!(outer.value(2).len(), 0);
3475
3476 let row0 = arrow_to_value(arr_ref.as_ref(), 0, Some(&data_type));
3479 assert_eq!(
3480 row0,
3481 Value::List(vec![
3482 Value::Vector(vec![1.0, 2.0, 3.0]),
3483 Value::Vector(vec![4.0, 5.0, 6.0]),
3484 ])
3485 );
3486
3487 let row1 = arrow_to_value(arr_ref.as_ref(), 1, Some(&data_type));
3488 let Value::List(tokens) = row1 else {
3489 panic!("row1 should decode to a list of tokens");
3490 };
3491 assert_eq!(tokens.len(), 3);
3492 assert_eq!(tokens[2], Value::Vector(vec![13.0, 14.0, 15.0]));
3493 }
3494
3495 #[test]
3496 fn test_build_multivector_invalid_inner_tokens() {
3497 let data_type = DataType::List(Box::new(DataType::Vector { dimensions: 2 }));
3501 let extractor = PropertyExtractor::new("tokens", &data_type);
3502
3503 let props = [Some(Value::List(vec![
3504 Value::Vector(vec![1.0, 2.0]), Value::Vector(vec![9.0, 9.0, 9.0]), ]))];
3507 let deleted = [false];
3508
3509 let err = extractor
3510 .build_column(1, &deleted, |i| props[i].as_ref())
3511 .unwrap_err()
3512 .to_string();
3513 assert!(err.contains("'tokens'"), "message: {err}");
3514 assert!(err.contains("token 1"), "message: {err}");
3515
3516 let props = [Some(Value::List(vec![Value::String("nope".to_string())]))];
3518 let err = extractor
3519 .build_column(1, &deleted, |i| props[i].as_ref())
3520 .unwrap_err()
3521 .to_string();
3522 assert!(err.contains("token 0"), "message: {err}");
3523 }
3524
3525 #[test]
3526 fn test_build_multivector_valid_tokens_and_null_rows() {
3527 let data_type = DataType::List(Box::new(DataType::Vector { dimensions: 2 }));
3531 let extractor = PropertyExtractor::new("tokens", &data_type);
3532
3533 let props = [
3534 Some(Value::List(vec![
3535 Value::Vector(vec![1.0, 2.0]),
3536 Value::List(vec![Value::Float(3.0), Value::Float(4.0)]),
3537 ])),
3538 None,
3539 Some(Value::Null),
3540 Some(Value::List(vec![])),
3541 ];
3542 let deleted = [false; 4];
3543
3544 let arr_ref = extractor
3545 .build_column(4, &deleted, |i| props[i].as_ref())
3546 .unwrap();
3547 let outer = arr_ref.as_any().downcast_ref::<ListArray>().unwrap();
3548 assert!(outer.is_valid(0));
3549 assert!(!outer.is_valid(1)); assert!(!outer.is_valid(2)); assert!(outer.is_valid(3)); let inner_row = outer.value(0);
3553 let inner = inner_row
3554 .as_any()
3555 .downcast_ref::<FixedSizeListArray>()
3556 .unwrap();
3557 assert_eq!(inner.len(), 2);
3558 assert!(inner.is_valid(0));
3559 assert!(inner.is_valid(1));
3560 assert_eq!(outer.value(3).len(), 0);
3561 }
3562
3563 #[test]
3564 fn test_build_vector_column_wrong_dims_is_error() {
3565 let data_type = DataType::Vector { dimensions: 3 };
3568 let extractor = PropertyExtractor::new("embedding", &data_type);
3569
3570 let props = [
3571 Some(Value::Vector(vec![1.0, 2.0, 3.0])),
3572 Some(Value::Vector(vec![1.0, 2.0])),
3573 ];
3574 let deleted = [false; 2];
3575
3576 let err = extractor
3577 .build_column(2, &deleted, |i| props[i].as_ref())
3578 .unwrap_err()
3579 .to_string();
3580 assert!(err.contains("'embedding'"), "message: {err}");
3581 assert!(err.contains("row 1"), "message: {err}");
3582
3583 let props = [
3585 Some(Value::Vector(vec![1.0, 2.0, 3.0])),
3586 None,
3587 Some(Value::Null),
3588 ];
3589 let deleted = [false, false, true];
3590 let arr_ref = extractor
3591 .build_column(3, &deleted, |i| props[i].as_ref())
3592 .unwrap();
3593 let arr = arr_ref
3594 .as_any()
3595 .downcast_ref::<FixedSizeListArray>()
3596 .unwrap();
3597 assert!(arr.is_valid(0));
3598 assert!(!arr.is_valid(1));
3599 }
3600
3601 #[test]
3602 fn test_extract_vector_f32_values_strict() {
3603 assert!(
3605 extract_vector_f32_values_strict(Some(&Value::Vector(vec![1.0, 2.0])), false, 3)
3606 .is_err()
3607 );
3608 assert!(
3609 extract_vector_f32_values_strict(Some(&Value::String("x".into())), false, 3).is_err()
3610 );
3611 let (vals, valid) = extract_vector_f32_values_strict(None, false, 3).unwrap();
3612 assert_eq!((vals, valid), (vec![0.0, 0.0, 0.0], false));
3613 let (vals, valid) = extract_vector_f32_values_strict(Some(&Value::Null), false, 3).unwrap();
3614 assert_eq!((vals, valid), (vec![0.0, 0.0, 0.0], false));
3615 let (vals, valid) =
3616 extract_vector_f32_values_strict(Some(&Value::Vector(vec![1.0, 2.0])), true, 3)
3617 .unwrap();
3618 assert_eq!((vals, valid), (vec![0.0, 0.0, 0.0], true));
3619 let (vals, valid) =
3620 extract_vector_f32_values_strict(Some(&Value::Vector(vec![1.0, 2.0, 3.0])), false, 3)
3621 .unwrap();
3622 assert_eq!((vals, valid), (vec![1.0, 2.0, 3.0], true));
3623 }
3624
3625 #[test]
3626 fn test_values_to_array_multivector_schemaless_deferred() {
3627 let values = vec![Value::List(vec![Value::Vector(vec![1.0, 2.0])])];
3632 let dt = ArrowDataType::List(Arc::new(Field::new(
3633 "item",
3634 ArrowDataType::FixedSizeList(
3635 Arc::new(Field::new("item", ArrowDataType::Float32, true)),
3636 2,
3637 ),
3638 true,
3639 )));
3640 assert!(values_to_array(&values, &dt).is_err());
3641 }
3642
3643 #[test]
3646 fn test_int32_and_date32_columns_null_out_of_range() {
3647 let dt = DataType::Int64;
3648 let extractor = PropertyExtractor::new("x", &dt);
3649
3650 let over = Value::Int(i64::from(i32::MAX) + 1);
3651 let ok = Value::Int(42);
3652 let vals = [over, ok];
3653 let deleted = [false, false];
3654
3655 let arr = extractor
3656 .build_int32_column(2, &deleted, |i| Some(&vals[i]))
3657 .unwrap();
3658 let arr = arr.as_any().downcast_ref::<Int32Array>().unwrap();
3659 assert!(
3660 arr.is_null(0),
3661 "out-of-range i64 must be NULL in an int32 column, not wrapped"
3662 );
3663 assert_eq!(arr.value(1), 42);
3664
3665 let arr = extractor
3666 .build_date32_column(2, &deleted, |i| Some(&vals[i]))
3667 .unwrap();
3668 let arr = arr.as_any().downcast_ref::<Date32Array>().unwrap();
3669 assert!(
3670 arr.is_null(0),
3671 "out-of-range day count must be NULL in a date32 column, not wrapped"
3672 );
3673 assert_eq!(arr.value(1), 42);
3674 }
3675}