1use std::env;
16use std::sync::Arc;
17use std::sync::LazyLock;
18
19use async_trait::async_trait;
20use vortex_array::ArrayContext;
21use vortex_array::dtype::Field;
22use vortex_array::dtype::FieldName;
23use vortex_array::dtype::FieldPath;
24use vortex_error::VortexResult;
25use vortex_session::VortexSession;
26use vortex_utils::aliases::hash_map::HashMap;
27use vortex_utils::aliases::hash_set::HashSet;
28
29use crate::LayoutRef;
30use crate::LayoutStrategy;
31use crate::layouts::list::writer::ListLayoutStrategy;
32use crate::layouts::struct_::StructStrategy;
33use crate::segments::SegmentSinkRef;
34use crate::sequence::SendableSequentialStream;
35use crate::sequence::SequencePointer;
36
37pub fn use_experimental_list_layout() -> bool {
43 static USE_EXPERIMENTAL_LIST_LAYOUT: LazyLock<bool> =
44 LazyLock::new(|| env::var("VORTEX_EXPERIMENTAL_LIST_LAYOUT").is_ok_and(|v| v == "1"));
45 *USE_EXPERIMENTAL_LIST_LAYOUT
46}
47
48type ListLayoutFactory = Arc<dyn Fn(ListLayoutStrategy) -> Arc<dyn LayoutStrategy> + Send + Sync>;
49
50pub struct TableStrategy {
65 leaf_writers: HashMap<FieldPath, Arc<dyn LayoutStrategy>>,
68 validity: Arc<dyn LayoutStrategy>,
70 leaf: Arc<dyn LayoutStrategy>,
72 list_layout_factory: Option<ListLayoutFactory>,
77}
78
79impl TableStrategy {
80 pub fn new(validity: Arc<dyn LayoutStrategy>, fallback: Arc<dyn LayoutStrategy>) -> Self {
99 Self {
100 leaf_writers: Default::default(),
101 validity,
102 leaf: fallback,
103 list_layout_factory: None,
104 }
105 }
106
107 pub fn with_field_writer(
136 mut self,
137 field_path: impl Into<FieldPath>,
138 writer: Arc<dyn LayoutStrategy>,
139 ) -> Self {
140 self.leaf_writers
141 .insert(self.validate_path(field_path.into()), writer);
142 self
143 }
144
145 pub fn with_field_writers(
149 mut self,
150 writers: impl IntoIterator<Item = (FieldPath, Arc<dyn LayoutStrategy>)>,
151 ) -> Self {
152 for (field_path, strategy) in writers {
153 self.leaf_writers
154 .insert(self.validate_path(field_path), strategy);
155 }
156 self
157 }
158
159 pub fn with_default_strategy(mut self, default: Arc<dyn LayoutStrategy>) -> Self {
161 self.leaf = default;
162 self
163 }
164
165 pub fn with_validity_strategy(mut self, validity: Arc<dyn LayoutStrategy>) -> Self {
167 self.validity = validity;
168 self
169 }
170
171 pub fn with_list_layout(self) -> Self {
176 self.with_list_layout_factory(|strategy| Arc::new(strategy))
177 }
178
179 pub fn with_list_layout_factory(
185 mut self,
186 factory: impl Fn(ListLayoutStrategy) -> Arc<dyn LayoutStrategy> + Send + Sync + 'static,
187 ) -> Self {
188 self.list_layout_factory = Some(Arc::new(factory));
189 self
190 }
191}
192
193impl TableStrategy {
194 fn struct_strategy(&self) -> StructStrategy {
199 let mut field_writers: HashMap<FieldName, Arc<dyn LayoutStrategy>> = HashMap::default();
200
201 let mut named_first: HashSet<FieldName> = HashSet::default();
204 for path in self.leaf_writers.keys() {
205 if let Some(Field::Name(name)) = path.parts().first() {
206 named_first.insert(name.clone());
207 }
208 }
209
210 for name in named_first {
211 let writer = match self.leaf_writers.get(&FieldPath::from_name(name.clone())) {
214 Some(exact) => Arc::clone(exact),
215 None => {
216 Arc::new(self.descend(&Field::Name(name.clone()))) as Arc<dyn LayoutStrategy>
217 }
218 };
219 field_writers.insert(name, writer);
220 }
221
222 StructStrategy::new(Arc::clone(&self.validity), Arc::new(self.descend_clean()))
223 .with_field_writers(field_writers)
224 }
225
226 fn list_strategy(&self) -> Option<Arc<dyn LayoutStrategy>> {
232 let factory = self.list_layout_factory.as_ref()?;
233 let list_layout = ListLayoutStrategy::default()
234 .with_elements(Arc::new(self.descend_clean()))
235 .with_offsets(Arc::clone(&self.leaf))
236 .with_validity(Arc::clone(&self.validity))
237 .with_fallback(Arc::clone(&self.leaf));
238 Some(factory(list_layout))
239 }
240
241 fn descend(&self, field: &Field) -> Self {
244 let mut new_writers = HashMap::with_capacity(self.leaf_writers.len());
245
246 for (field_path, strategy) in &self.leaf_writers {
247 if field_path.parts().first() == Some(field)
248 && let Some(subpath) = field_path.clone().step_into()
249 && !subpath.is_root()
250 {
251 new_writers.insert(subpath, Arc::clone(strategy));
252 }
253 }
254
255 Self {
256 leaf_writers: new_writers,
257 validity: Arc::clone(&self.validity),
258 leaf: Arc::clone(&self.leaf),
259 list_layout_factory: self.list_layout_factory.clone(),
260 }
261 }
262
263 fn descend_clean(&self) -> Self {
266 Self {
267 leaf_writers: HashMap::default(),
268 validity: Arc::clone(&self.validity),
269 leaf: Arc::clone(&self.leaf),
270 list_layout_factory: self.list_layout_factory.clone(),
271 }
272 }
273
274 fn validate_path(&self, path: FieldPath) -> FieldPath {
275 assert!(
276 !path.is_root(),
277 "Do not set override as a root strategy, instead set the default strategy"
278 );
279
280 for field_path in self.leaf_writers.keys() {
283 assert!(
284 !path.overlap(field_path),
285 "Override for field_path {path} conflicts with existing override for {field_path}"
286 );
287 }
288
289 path
290 }
291}
292
293#[async_trait]
295impl LayoutStrategy for TableStrategy {
296 async fn write_stream(
297 &self,
298 ctx: ArrayContext,
299 segment_sink: SegmentSinkRef,
300 stream: SendableSequentialStream,
301 eof: SequencePointer,
302 session: &VortexSession,
303 ) -> VortexResult<LayoutRef> {
304 let dtype = stream.dtype().clone();
305
306 if dtype.is_struct() {
307 return self
308 .struct_strategy()
309 .write_stream(ctx, segment_sink, stream, eof, session)
310 .await;
311 }
312
313 if dtype.is_list()
314 && let Some(list_strategy) = self.list_strategy()
315 {
316 return list_strategy
317 .write_stream(ctx, segment_sink, stream, eof, session)
318 .await;
319 }
320
321 self.leaf
323 .write_stream(ctx, segment_sink, stream, eof, session)
324 .await
325 }
326}
327
328#[cfg(test)]
329mod tests {
330 use std::num::NonZeroUsize;
331 use std::sync::Arc;
332 use std::task::Poll;
333
334 use vortex_array::ArrayContext;
335 use vortex_array::ArrayRef;
336 use vortex_array::IntoArray;
337 use vortex_array::arrays::BoolArray;
338 use vortex_array::arrays::ChunkedArray;
339 use vortex_array::arrays::ListArray;
340 use vortex_array::arrays::PrimitiveArray;
341 use vortex_array::arrays::StructArray;
342 use vortex_array::dtype::DType;
343 use vortex_array::dtype::FieldPath;
344 use vortex_array::dtype::Nullability;
345 use vortex_array::dtype::PType;
346 use vortex_array::dtype::StructFields;
347 use vortex_array::field_path;
348 use vortex_array::validity::Validity;
349 use vortex_buffer::buffer;
350 use vortex_error::VortexExpect;
351 use vortex_error::VortexResult;
352 use vortex_io::runtime::single::block_on;
353 use vortex_io::session::RuntimeSessionExt;
354
355 use crate::LayoutRef;
356 use crate::LayoutStrategy;
357 use crate::layouts::chunked::writer::ChunkedLayoutStrategy;
358 use crate::layouts::flat::writer::FlatLayoutStrategy;
359 use crate::layouts::list::List;
360 use crate::layouts::repartition::RepartitionStrategy;
361 use crate::layouts::repartition::RepartitionWriterOptions;
362 use crate::layouts::table::TableStrategy;
363 use crate::layouts::zoned::Zoned;
364 use crate::layouts::zoned::writer::ZonedLayoutOptions;
365 use crate::layouts::zoned::writer::ZonedStrategy;
366 use crate::segments::TestSegments;
367 use crate::sequence::SequenceId;
368 use crate::sequence::SequentialArrayStreamExt;
369 use crate::sequence::SequentialStreamAdapter;
370 use crate::sequence::SequentialStreamExt;
371 use crate::test::SESSION;
372
373 async fn write<S: LayoutStrategy>(strategy: &S, array: ArrayRef) -> VortexResult<LayoutRef> {
374 let segments = Arc::new(TestSegments::default());
375 let (ptr, eof) = SequenceId::root().split();
376 let stream = array.to_array_stream().sequenced(ptr);
377 strategy
378 .write_stream(ArrayContext::empty(), segments, stream, eof, &SESSION)
379 .await
380 }
381
382 fn flat_table() -> TableStrategy {
385 let flat: Arc<dyn LayoutStrategy> = Arc::new(FlatLayoutStrategy::default());
386 TableStrategy::new(Arc::clone(&flat), flat)
387 }
388
389 #[tokio::test]
391 async fn dispatches_struct() -> VortexResult<()> {
392 let struct_array = StructArray::from_fields(
393 [
394 ("a", buffer![1i32, 2, 3].into_array()),
395 ("b", buffer![10i32, 20, 30].into_array()),
396 ]
397 .as_slice(),
398 )?
399 .into_array();
400
401 let layout = write(&flat_table(), struct_array).await?;
402 insta::assert_snapshot!(layout.display_tree(), @r"
403 vortex.struct, dtype: {a=i32, b=i32}, children: 2
404 ├── a: vortex.flat, dtype: i32, segment: 0
405 └── b: vortex.flat, dtype: i32, segment: 1
406 ");
407 Ok(())
408 }
409
410 #[tokio::test]
413 async fn dispatches_nested_list() -> VortexResult<()> {
414 let inner = ListArray::try_new(
415 buffer![1i32, 2, 3, 4, 5, 6].into_array(),
416 buffer![0u32, 2, 5, 5, 6].into_array(),
417 Validity::NonNullable,
418 )?
419 .into_array();
420 let outer = ListArray::try_new(
421 inner,
422 buffer![0u32, 2, 4].into_array(),
423 Validity::NonNullable,
424 )?
425 .into_array();
426
427 let layout = write(&flat_table().with_list_layout(), outer).await?;
428 insta::assert_snapshot!(layout.display_tree(), @r"
429 vortex.list, dtype: list(list(i32)), children: 2
430 ├── elements: vortex.list, dtype: list(i32), children: 2
431 │ ├── elements: vortex.flat, dtype: i32, segment: 1
432 │ └── offsets: vortex.flat, dtype: u64, segment: 2
433 └── offsets: vortex.flat, dtype: u64, segment: 0
434 ");
435 Ok(())
436 }
437
438 #[tokio::test]
441 async fn dispatches_struct_list_struct() -> VortexResult<()> {
442 let inner_struct = StructArray::from_fields(
443 [
444 ("a", buffer![1i32, 2, 3, 4, 5].into_array()),
445 ("b", buffer![10i32, 20, 30, 40, 50].into_array()),
446 ]
447 .as_slice(),
448 )?
449 .into_array();
450 let items = ListArray::try_new(
451 inner_struct,
452 buffer![0u32, 2, 5, 5].into_array(),
453 Validity::Array(BoolArray::from_iter([true, false, true]).into_array()),
454 )?
455 .into_array();
456 let st = StructArray::from_fields([("items", items)].as_slice())?.into_array();
457
458 let layout = write(&flat_table().with_list_layout(), st).await?;
459 insta::assert_snapshot!(layout.display_tree(), @r"
460 vortex.struct, dtype: {items=list({a=i32, b=i32})?}, children: 1
461 └── items: vortex.list, dtype: list({a=i32, b=i32})?, children: 3
462 ├── elements: vortex.struct, dtype: {a=i32, b=i32}, children: 2
463 │ ├── a: vortex.flat, dtype: i32, segment: 2
464 │ └── b: vortex.flat, dtype: i32, segment: 3
465 ├── offsets: vortex.flat, dtype: u64, segment: 0
466 └── validity: vortex.flat, dtype: bool, segment: 1
467 ");
468 Ok(())
469 }
470
471 #[tokio::test]
475 async fn dispatches_chunked_list() -> VortexResult<()> {
476 let chunk0 = ListArray::try_new(
477 buffer![1i32, 2, 3].into_array(),
478 buffer![0u32, 2, 3].into_array(),
479 Validity::NonNullable,
480 )?
481 .into_array();
482 let chunk1 = ListArray::try_new(
483 buffer![4i32, 5, 6, 7].into_array(),
484 buffer![0u32, 1, 4].into_array(),
485 Validity::NonNullable,
486 )?
487 .into_array();
488 let dtype = chunk0.dtype().clone();
489 let chunked = ChunkedArray::try_new(vec![chunk0, chunk1], dtype)?.into_array();
490
491 let flat: Arc<dyn LayoutStrategy> = Arc::new(FlatLayoutStrategy::default());
492 let dispatcher = TableStrategy::new(
493 Arc::clone(&flat),
494 Arc::new(ChunkedLayoutStrategy::new(FlatLayoutStrategy::default())),
495 )
496 .with_list_layout();
497 let layout = write(&dispatcher, chunked).await?;
498 insta::assert_snapshot!(layout.display_tree(), @r"
499 vortex.list, dtype: list(i32), children: 2
500 ├── elements: vortex.chunked, dtype: i32, children: 2
501 │ ├── [0]: vortex.flat, dtype: i32, segment: 0
502 │ └── [1]: vortex.flat, dtype: i32, segment: 1
503 └── offsets: vortex.chunked, dtype: u64, children: 2
504 ├── [0]: vortex.flat, dtype: u64, segment: 2
505 └── [1]: vortex.flat, dtype: u64, segment: 3
506 ");
507 Ok(())
508 }
509
510 #[tokio::test]
512 async fn wraps_list_strategy_before_decomposition() -> VortexResult<()> {
513 let list = ListArray::try_new(
514 PrimitiveArray::from_iter(0..9_i32).into_array(),
515 PrimitiveArray::from_iter(0..=9_u32).into_array(),
516 Validity::NonNullable,
517 )?
518 .into_array();
519
520 let row_block_size = NonZeroUsize::new(4).vortex_expect("4 is non-zero");
521 let flat: Arc<dyn LayoutStrategy> = Arc::new(FlatLayoutStrategy::default());
522 let stats = Arc::clone(&flat);
523 let chunked: Arc<dyn LayoutStrategy> =
524 Arc::new(ChunkedLayoutStrategy::new(FlatLayoutStrategy::default()));
525 let dispatcher = TableStrategy::new(Arc::clone(&flat), chunked).with_list_layout_factory(
526 move |list_layout| {
527 let zoned = ZonedStrategy::new(
528 list_layout,
529 Arc::clone(&stats),
530 ZonedLayoutOptions {
531 block_size: row_block_size,
532 ..Default::default()
533 },
534 );
535 Arc::new(RepartitionStrategy::new(
536 zoned,
537 RepartitionWriterOptions {
538 block_size_minimum: 0,
539 block_len_multiple: row_block_size.get(),
540 block_size_target: None,
541 canonicalize: false,
542 },
543 )) as Arc<dyn LayoutStrategy>
544 },
545 );
546
547 let layout = write(&dispatcher, list).await?;
548 let zoned = layout.as_::<Zoned>();
549 assert_eq!(zoned.zone_len(), 4);
550 assert_eq!(zoned.nzones(), 3);
551
552 let data = layout.child(0)?;
553 assert!(data.is::<List>());
554 assert_eq!(data.row_count(), 9);
555 Ok(())
556 }
557
558 #[tokio::test]
560 async fn non_struct_input_uses_leaf() -> VortexResult<()> {
561 let primitive = PrimitiveArray::from_iter([1i32, 2, 3]).into_array();
562 let layout = write(&flat_table(), primitive).await?;
563 insta::assert_snapshot!(layout.display_tree(), @"vortex.flat, dtype: i32, segment: 0");
564 Ok(())
565 }
566
567 #[tokio::test]
569 async fn chunked_struct() -> VortexResult<()> {
570 let validity: Arc<dyn LayoutStrategy> = Arc::new(FlatLayoutStrategy::default());
571 let chunked_flat: Arc<dyn LayoutStrategy> =
572 Arc::new(ChunkedLayoutStrategy::new(FlatLayoutStrategy::default()));
573 let dispatcher = TableStrategy::new(validity, chunked_flat);
574
575 let c0 = StructArray::from_fields(
576 [
577 ("a", buffer![1i32, 2].into_array()),
578 ("b", buffer![10i32, 20].into_array()),
579 ]
580 .as_slice(),
581 )?
582 .into_array();
583 let c1 = StructArray::from_fields(
584 [
585 ("a", buffer![3i32].into_array()),
586 ("b", buffer![30i32].into_array()),
587 ]
588 .as_slice(),
589 )?
590 .into_array();
591 let dtype = c0.dtype().clone();
592 let chunked = ChunkedArray::try_new(vec![c0, c1], dtype)?.into_array();
593
594 let layout = write(&dispatcher, chunked).await?;
595 insta::assert_snapshot!(layout.display_tree(), @r"
596 vortex.struct, dtype: {a=i32, b=i32}, children: 2
597 ├── a: vortex.chunked, dtype: i32, children: 2
598 │ ├── [0]: vortex.flat, dtype: i32, segment: 0
599 │ └── [1]: vortex.flat, dtype: i32, segment: 1
600 └── b: vortex.chunked, dtype: i32, children: 2
601 ├── [0]: vortex.flat, dtype: i32, segment: 2
602 └── [1]: vortex.flat, dtype: i32, segment: 3
603 ");
604 Ok(())
605 }
606
607 #[tokio::test]
609 async fn field_override_is_used() -> VortexResult<()> {
610 let struct_array = StructArray::from_fields(
611 [
612 ("a", buffer![1i32, 2, 3].into_array()),
613 ("b", buffer![10i32, 20, 30].into_array()),
614 ]
615 .as_slice(),
616 )?
617 .into_array();
618
619 let strategy =
620 flat_table().with_field_writer(field_path!(a), Arc::new(FlatLayoutStrategy::default()));
621 let layout = write(&strategy, struct_array).await?;
622 insta::assert_snapshot!(layout.display_tree(), @r"
623 vortex.struct, dtype: {a=i32, b=i32}, children: 2
624 ├── a: vortex.flat, dtype: i32, segment: 0
625 └── b: vortex.flat, dtype: i32, segment: 1
626 ");
627 Ok(())
628 }
629
630 #[test]
631 #[should_panic(
632 expected = "Override for field_path $a.$b conflicts with existing override for $a.$b.$c"
633 )]
634 fn test_overlapping_paths_fail() {
635 let flat = Arc::new(FlatLayoutStrategy::default());
636
637 let path = TableStrategy::new(
639 Arc::<FlatLayoutStrategy>::clone(&flat),
640 Arc::<FlatLayoutStrategy>::clone(&flat),
641 )
642 .with_field_writer(field_path!(a.b.c), Arc::<FlatLayoutStrategy>::clone(&flat));
643
644 let _path = path.with_field_writer(field_path!(a.b), flat);
646 }
647
648 #[test]
649 #[should_panic(
650 expected = "Do not set override as a root strategy, instead set the default strategy"
651 )]
652 fn test_root_override() {
653 let flat = Arc::new(FlatLayoutStrategy::default());
654 let _strategy = TableStrategy::new(
655 Arc::<FlatLayoutStrategy>::clone(&flat),
656 Arc::<FlatLayoutStrategy>::clone(&flat),
657 )
658 .with_field_writer(FieldPath::root(), flat);
659 }
660
661 #[test]
662 #[should_panic(expected = "panic while transposing table stream")]
663 fn table_fanout_panic_propagates() {
664 let ctx = ArrayContext::empty();
665 let segments = Arc::new(TestSegments::default());
666 let (_, eof) = SequenceId::root().split();
667 let dtype = DType::Struct(
668 StructFields::from_iter([(
669 "a",
670 DType::Primitive(PType::I32, Nullability::NonNullable),
671 )]),
672 Nullability::NonNullable,
673 );
674 let stream =
675 futures::stream::poll_fn(|_| -> Poll<Option<VortexResult<(SequenceId, ArrayRef)>>> {
676 panic!("panic while transposing table stream");
677 });
678 let strategy = TableStrategy::new(
679 Arc::new(FlatLayoutStrategy::default()),
680 Arc::new(FlatLayoutStrategy::default()),
681 );
682
683 block_on(|handle| async move {
684 let session = SESSION.clone().with_handle(handle);
685 strategy
686 .write_stream(
687 ctx,
688 segments,
689 SequentialStreamAdapter::new(dtype, stream).sendable(),
690 eof,
691 &session,
692 )
693 .await
694 .unwrap();
695 });
696 }
697}