Skip to main content

vortex_layout/layouts/
table.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4//! A configurable writer strategy for tabular data.
5//!
6//! [`TableStrategy`] is a *dispatcher*: it inspects the dtype of the stream it is handed and
7//! routes struct columns to [`StructStrategy`], list columns to [`ListLayoutStrategy`], and
8//! everything else to the configured leaf strategy. Because it hands *itself* (suitably descended)
9//! to those structural writers as the strategy for their children, arbitrarily nested struct/list
10//! trees are written with no manual wiring.
11//!
12//! The dispatcher also owns field-path overrides, letting callers force a specific leaf field —
13//! at any depth — onto a custom strategy.
14
15use 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
37/// Whether [`TableStrategy`] writes list fields using a [`ListLayoutStrategy`] by
38/// default. Disabled unless the environment variable `VORTEX_EXPERIMENTAL_LIST_LAYOUT`
39/// is set to `1`.
40///
41/// [`ListLayoutStrategy`]: crate::layouts::list::writer::ListLayoutStrategy
42pub 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
50/// A configurable strategy for writing nested tabular data, dispatching each (sub)stream to the
51/// structural writer for its dtype.
52///
53/// Dispatch rules, applied to the dtype of the stream handed to [`write_stream`]:
54/// - **struct** → [`StructStrategy`], with each field written by its override (if any) or by a
55///   descended copy of this dispatcher.
56/// - **list** → [`ListLayoutStrategy`], with `elements` written by a descended copy of this
57///   dispatcher (so nested structs/lists recurse) and `offsets`/`validity` by the leaf/validity
58///   strategies. Gated: only when list decomposition is enabled via
59///   [`with_list_layout`][Self::with_list_layout] (off by default); otherwise a list falls through
60///   to the leaf strategy.
61/// - **anything else** → the leaf strategy.
62///
63/// [`write_stream`]: LayoutStrategy::write_stream
64pub struct TableStrategy {
65    /// A set of field-path overrides, e.g. to force one column to be compact-compressed. Keys are
66    /// paths relative to the level this dispatcher sits at.
67    leaf_writers: HashMap<FieldPath, Arc<dyn LayoutStrategy>>,
68    /// The writer for any validity arrays that may be present, at any level of the tree.
69    validity: Arc<dyn LayoutStrategy>,
70    /// The writer for leaf fields, i.e. anything that is not a struct.
71    leaf: Arc<dyn LayoutStrategy>,
72    /// Optional factory applied to each dynamically constructed [`ListLayoutStrategy`].
73    /// Its presence also enables list decomposition.
74    ///
75    /// [`ListLayoutStrategy`]: crate::layouts::list::writer::ListLayoutStrategy
76    list_layout_factory: Option<ListLayoutFactory>,
77}
78
79impl TableStrategy {
80    /// Create a new dispatcher with the given `validity` strategy and `fallback` leaf strategy and
81    /// no overrides.
82    ///
83    /// Additional per-field overrides can be configured with
84    /// [`with_field_writer`][Self::with_field_writer].
85    ///
86    /// ## Example
87    ///
88    /// ```ignore
89    /// # use std::sync::Arc;
90    /// # use vortex_layout::layouts::flat::writer::FlatLayoutStrategy;
91    /// # use vortex_layout::layouts::table::TableStrategy;
92    ///
93    /// // Build a write strategy that does not compress validity or any leaf fields.
94    /// let flat = Arc::new(FlatLayoutStrategy::default());
95    ///
96    /// let strategy = TableStrategy::new(Arc::<FlatLayoutStrategy>::clone(&flat), Arc::<FlatLayoutStrategy>::clone(&flat));
97    /// ```
98    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    /// Add a custom write strategy for the given leaf field.
108    ///
109    /// ## Example
110    ///
111    /// ```ignore
112    /// # use std::sync::Arc;
113    /// # use vortex_array::dtype::{field_path, Field, FieldPath};
114    /// # use vortex_btrblocks::BtrBlocksCompressor;
115    /// # use vortex_layout::layouts::compressed::CompressingStrategy;
116    /// # use vortex_layout::layouts::flat::writer::FlatLayoutStrategy;
117    /// # use vortex_layout::layouts::table::TableStrategy;
118    ///
119    /// // A strategy for compressing data using the balanced BtrBlocks compressor.
120    /// let compress =
121    ///     CompressingStrategy::new(FlatLayoutStrategy::default(), BtrBlocksCompressor::default());
122    ///
123    /// // Our combined strategy uses no compression for validity buffers, BtrBlocks compression
124    /// // for most columns, and stores a nested binary column uncompressed (flat) because it
125    /// // is pre-compressed or never filtered on.
126    /// let strategy = TableStrategy::new(
127    ///         Arc::new(FlatLayoutStrategy::default()),
128    ///         Arc::new(compress),
129    ///     )
130    ///     .with_field_writer(
131    ///         field_path!(request.body.bytes),
132    ///         Arc::new(FlatLayoutStrategy::default()),
133    ///     );
134    /// ```
135    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    /// Set writers for several fields at once.
146    ///
147    /// See also: [`with_field_writer`][Self::with_field_writer].
148    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    /// Override the default strategy for leaf columns that don't have overrides.
160    pub fn with_default_strategy(mut self, default: Arc<dyn LayoutStrategy>) -> Self {
161        self.leaf = default;
162        self
163    }
164
165    /// Override the strategy for compressing struct validity at all levels of the schema tree.
166    pub fn with_validity_strategy(mut self, validity: Arc<dyn LayoutStrategy>) -> Self {
167        self.validity = validity;
168        self
169    }
170
171    /// Enable writing list fields with [`ListLayoutStrategy`].
172    ///
173    /// **Note**: this is an unstable and experimental layout that is expected to change.
174    /// Using it may lead to unreadable files in the future.
175    pub fn with_list_layout(self) -> Self {
176        self.with_list_layout_factory(|strategy| Arc::new(strategy))
177    }
178
179    /// Enable writing list fields with [`ListLayoutStrategy`] and wrap each list writer. This
180    /// allows repartitioning or zoning to operate in the list's outer-row space before shredding.
181    ///
182    /// **Note**: this is an unstable and experimental layout that is expected to change.
183    /// Using it may lead to unreadable files in the future.
184    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    /// Build the [`StructStrategy`] used to write a struct-typed stream at this level.
195    ///
196    /// Each field that has an override (or a deeper override beneath it) is resolved up front;
197    /// every other field falls through to a clean descended dispatcher.
198    fn struct_strategy(&self) -> StructStrategy {
199        let mut field_writers: HashMap<FieldName, Arc<dyn LayoutStrategy>> = HashMap::default();
200
201        // The distinct named first-segments of our override paths are the only fields that need
202        // anything other than the default dispatcher.
203        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            // `validate_path` forbids overlapping overrides, so a name has *either* an exact
212            // single-segment override *or* deeper overrides, never both.
213            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    /// Build the [`ListLayoutStrategy`] used to write a list field stream at this level.
227    ///
228    /// The `elements` sub-column is routed back through a clean descended dispatcher so nested
229    /// structs/lists recurse; `offsets` go straight to the leaf (they are always a primitive
230    /// column); and `validity` uses the shared validity strategy.
231    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    /// Descend into a subfield, retaining only the overrides that apply beneath it (rebased to be
242    /// relative to the child).
243    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    /// A copy of this dispatcher with no overrides, used as the default child strategy for fields
264    /// that carry no override.
265    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        // Validate that the field path does not conflict with any overrides
281        // that we've added by overlapping.
282        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/// Dispatches each stream to the structural writer for its dtype.
294#[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        // Leaf: hand off to the leaf strategy.
322        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    /// A plain table dispatcher with no overrides. `flat` here is both the validity and leaf
383    /// strategy.
384    fn flat_table() -> TableStrategy {
385        let flat: Arc<dyn LayoutStrategy> = Arc::new(FlatLayoutStrategy::default());
386        TableStrategy::new(Arc::clone(&flat), flat)
387    }
388
389    /// The dispatcher shreds a top-level struct into one child per field.
390    #[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    /// A `list<list<i32>>` column: the dispatcher recurses into itself so the outer list's
411    /// `elements` are decomposed as a nested `ListLayout`.
412    #[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    /// A `struct<{ items: list<struct<{a,b}>>? }>` column: list decomposition recurses into struct
439    /// decomposition for the elements, and a nullable list writes a validity child.
440    #[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    /// A multi-chunk `list<i32>` written with a chunked leaf: each sub-column (`elements`,
472    /// `offsets`) becomes its own `ChunkedLayout`, so elements are chunked independently of rows.
473    /// This is the "list-of-chunkeds" topology top-level decomposition unlocks.
474    #[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    /// A wrapper can repartition and zone lists in outer-row space before decomposition.
511    #[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    /// A non-struct stream is not shredded; it is handed straight to the leaf strategy.
559    #[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    /// A multi-chunk struct is transposed per field; each column is written by a chunked leaf.
568    #[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    /// A field override on a struct field is honored ahead of the default leaf strategy.
608    #[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        // Success
638        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        // Should panic right here.
645        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}