Skip to main content

vortex_fsst/
array.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::fmt::Debug;
5use std::fmt::Display;
6use std::fmt::Formatter;
7use std::hash::Hasher;
8use std::sync::Arc;
9use std::sync::OnceLock;
10
11use fsst::Compressor;
12use fsst::Decompressor;
13use fsst::Symbol;
14use prost::Message as _;
15use vortex_array::Array;
16use vortex_array::ArrayEq;
17use vortex_array::ArrayHash;
18use vortex_array::ArrayId;
19use vortex_array::ArrayParts;
20use vortex_array::ArrayRef;
21use vortex_array::ArraySlots;
22use vortex_array::ArrayView;
23use vortex_array::Canonical;
24use vortex_array::EqMode;
25use vortex_array::ExecutionCtx;
26use vortex_array::ExecutionResult;
27use vortex_array::IntoArray;
28use vortex_array::LEGACY_SESSION;
29use vortex_array::TypedArrayRef;
30use vortex_array::VortexSessionExecute;
31use vortex_array::arrays::VarBin;
32use vortex_array::arrays::VarBinArray;
33use vortex_array::arrays::varbin::VarBinArrayExt;
34use vortex_array::buffer::BufferHandle;
35use vortex_array::builders::ArrayBuilder;
36use vortex_array::builders::VarBinViewBuilder;
37use vortex_array::dtype::DType;
38use vortex_array::dtype::Nullability;
39use vortex_array::dtype::PType;
40use vortex_array::serde::ArrayChildren;
41use vortex_array::smallvec::smallvec;
42use vortex_array::validity::Validity;
43use vortex_array::vtable::VTable;
44use vortex_array::vtable::ValidityVTable;
45use vortex_array::vtable::child_to_validity;
46use vortex_array::vtable::validity_to_child;
47use vortex_buffer::Buffer;
48use vortex_buffer::ByteBuffer;
49use vortex_error::VortexExpect;
50use vortex_error::VortexResult;
51use vortex_error::vortex_bail;
52use vortex_error::vortex_ensure;
53use vortex_error::vortex_err;
54use vortex_error::vortex_panic;
55use vortex_session::VortexSession;
56use vortex_session::registry::CachedId;
57
58use crate::canonical::canonicalize_fsst;
59use crate::canonical::fsst_decode_views;
60use crate::kernel::PARENT_KERNELS;
61use crate::rules::RULES;
62
63/// A [`FSST`]-encoded Vortex array.
64pub type FSSTArray = Array<FSST>;
65
66#[derive(Clone, prost::Message)]
67pub struct FSSTMetadata {
68    #[prost(enumeration = "PType", tag = "1")]
69    uncompressed_lengths_ptype: i32,
70
71    #[prost(enumeration = "PType", tag = "2")]
72    codes_offsets_ptype: i32,
73}
74
75impl FSSTMetadata {
76    pub fn get_uncompressed_lengths_ptype(&self) -> VortexResult<PType> {
77        PType::try_from(self.uncompressed_lengths_ptype)
78            .map_err(|_| vortex_err!("Invalid PType {}", self.uncompressed_lengths_ptype))
79    }
80}
81
82impl ArrayHash for FSSTData {
83    fn array_hash<H: Hasher>(&self, state: &mut H, precision: EqMode) {
84        self.symbol_table.symbols.array_hash(state, precision);
85        self.symbol_table
86            .symbol_lengths
87            .array_hash(state, precision);
88        self.codes_bytes.as_host().array_hash(state, precision);
89    }
90}
91
92impl ArrayEq for FSSTData {
93    fn array_eq(&self, other: &Self, precision: EqMode) -> bool {
94        self.symbol_table
95            .symbols
96            .array_eq(&other.symbol_table.symbols, precision)
97            && self
98                .symbol_table
99                .symbol_lengths
100                .array_eq(&other.symbol_table.symbol_lengths, precision)
101            && self
102                .codes_bytes
103                .as_host()
104                .array_eq(other.codes_bytes.as_host(), precision)
105    }
106}
107
108impl VTable for FSST {
109    type TypedArrayData = FSSTData;
110    type OperationsVTable = Self;
111    type ValidityVTable = Self;
112
113    fn id(&self) -> ArrayId {
114        static ID: CachedId = CachedId::new("vortex.fsst");
115        *ID
116    }
117
118    fn validate(
119        &self,
120        data: &Self::TypedArrayData,
121        dtype: &DType,
122        len: usize,
123        slots: &[Option<ArrayRef>],
124    ) -> VortexResult<()> {
125        // TODO(ctx): trait fixes - VTable::validate has a fixed signature.
126        let mut ctx = LEGACY_SESSION.create_execution_ctx();
127        data.validate(dtype, len, slots, &mut ctx)
128    }
129
130    fn nbuffers(_array: ArrayView<'_, Self>) -> usize {
131        3
132    }
133
134    fn buffer(array: ArrayView<'_, Self>, idx: usize) -> BufferHandle {
135        match idx {
136            0 => BufferHandle::new_host(array.symbols().clone().into_byte_buffer()),
137            1 => BufferHandle::new_host(array.symbol_lengths().clone().into_byte_buffer()),
138            2 => array.codes_bytes_handle().clone(),
139            _ => vortex_panic!("FSSTArray buffer index {idx} out of bounds"),
140        }
141    }
142
143    fn buffer_name(_array: ArrayView<'_, Self>, idx: usize) -> Option<String> {
144        match idx {
145            0 => Some("symbols".to_string()),
146            1 => Some("symbol_lengths".to_string()),
147            2 => Some("compressed_codes".to_string()),
148            _ => vortex_panic!("FSSTArray buffer_name index {idx} out of bounds"),
149        }
150    }
151
152    fn serialize(
153        array: ArrayView<'_, Self>,
154        _session: &VortexSession,
155    ) -> VortexResult<Option<Vec<u8>>> {
156        let codes_offsets = array.as_ref().slots()[CODES_OFFSETS_SLOT]
157            .as_ref()
158            .vortex_expect("FSSTArray codes_offsets slot");
159        Ok(Some(
160            FSSTMetadata {
161                uncompressed_lengths_ptype: array.uncompressed_lengths().dtype().as_ptype().into(),
162                codes_offsets_ptype: codes_offsets.dtype().as_ptype().into(),
163            }
164            .encode_to_vec(),
165        ))
166    }
167
168    /// Deserializes an FSST array from its serialized components.
169    ///
170    /// Supports two serialization formats:
171    ///
172    /// ## Legacy format (2 buffers, 2 children)
173    ///
174    /// The original FSST layout stored the compressed codes as a full `VarBinArray` child.
175    /// - **Buffers**: `[symbols, symbol_lengths]`
176    /// - **Children**: `[codes (VarBinArray), uncompressed_lengths (Primitive)]`
177    ///
178    /// The codes VarBinArray child is decomposed: its bytes become the `codes_bytes` buffer,
179    /// and its offsets/validity are extracted into slots.
180    /// See `FSST::deserialize_legacy`.
181    ///
182    /// ## Current format (3 buffers, 2-3 children)
183    ///
184    /// The current layout stores the compressed bytes as a raw buffer alongside the symbol
185    /// table, with offsets and validity as separate children.
186    /// - **Buffers**: `[symbols, symbol_lengths, compressed_codes_bytes]`
187    /// - **Children**: `[uncompressed_lengths, codes_offsets, (optional) codes_validity]`
188    ///
189    /// The `codes_bytes` buffer is stored directly in `FSSTData`. A `VarBinArray` for the
190    /// codes can be reconstructed on demand via [`FSSTArrayExt::codes()`] using the bytes
191    /// from `FSSTData` combined with offsets and validity from the array's slots.
192    fn deserialize(
193        &self,
194        dtype: &DType,
195        len: usize,
196        metadata: &[u8],
197        buffers: &[BufferHandle],
198        children: &dyn ArrayChildren,
199        session: &VortexSession,
200    ) -> VortexResult<ArrayParts<Self>> {
201        let metadata = FSSTMetadata::decode(metadata)?;
202        let symbols = Buffer::<Symbol>::from_byte_buffer(buffers[0].clone().try_to_host_sync()?);
203        let symbol_lengths = Buffer::<u8>::from_byte_buffer(buffers[1].clone().try_to_host_sync()?);
204
205        let mut ctx = session.create_execution_ctx();
206        if buffers.len() == 2 {
207            return Self::deserialize_legacy(
208                self,
209                dtype,
210                len,
211                &metadata,
212                &symbols,
213                &symbol_lengths,
214                children,
215                &mut ctx,
216            );
217        }
218
219        if buffers.len() == 3 {
220            let uncompressed_lengths = children.get(
221                0,
222                &DType::Primitive(
223                    metadata.get_uncompressed_lengths_ptype()?,
224                    Nullability::NonNullable,
225                ),
226                len,
227            )?;
228
229            let codes_bytes = buffers[2].clone();
230            let codes_offsets = children.get(
231                1,
232                &DType::Primitive(
233                    PType::try_from(metadata.codes_offsets_ptype)?,
234                    Nullability::NonNullable,
235                ),
236                // VarBin offsets are len + 1
237                len + 1,
238            )?;
239
240            let codes_validity = if children.len() == 2 {
241                Validity::from(dtype.nullability())
242            } else if children.len() == 3 {
243                let validity = children.get(2, &Validity::DTYPE, len)?;
244                Validity::Array(validity)
245            } else {
246                vortex_bail!("Expected 2 or 3 children, got {}", children.len());
247            };
248
249            FSSTData::validate_parts(
250                &symbols,
251                &symbol_lengths,
252                &codes_bytes,
253                &codes_offsets,
254                dtype.nullability(),
255                &uncompressed_lengths,
256                dtype,
257                len,
258                &mut ctx,
259            )?;
260            let slots = smallvec![
261                Some(uncompressed_lengths),
262                Some(codes_offsets),
263                validity_to_child(&codes_validity, len),
264            ];
265            let data = FSSTData::try_new(symbols, symbol_lengths, codes_bytes, len)?;
266            return Ok(ArrayParts::new(self.clone(), dtype.clone(), len, data).with_slots(slots));
267        }
268
269        vortex_bail!(
270            "InvalidArgument: Expected 2 or 3 buffers, got {}",
271            buffers.len()
272        );
273    }
274
275    fn slot_name(_array: ArrayView<'_, Self>, idx: usize) -> String {
276        SLOT_NAMES[idx].to_string()
277    }
278
279    fn execute(array: Array<Self>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
280        canonicalize_fsst(array.as_view(), ctx).map(ExecutionResult::done)
281    }
282
283    fn append_to_builder(
284        array: ArrayView<'_, Self>,
285        builder: &mut dyn ArrayBuilder,
286        ctx: &mut ExecutionCtx,
287    ) -> VortexResult<()> {
288        let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() else {
289            builder.extend_from_array(
290                &array
291                    .array()
292                    .clone()
293                    .execute::<Canonical>(ctx)?
294                    .into_array(),
295            );
296            return Ok(());
297        };
298
299        // Decompress the whole block of data into a new buffer, and create some views
300        // from it instead. The new buffer lands after any pending in-progress
301        // buffer that push_buffer_and_adjusted_views will flush first.
302        let next_buffer_index = builder.completed_block_count() + u32::from(builder.in_progress());
303        let (buffers, views) = fsst_decode_views(array, next_buffer_index, ctx)?;
304
305        builder.push_buffer_and_adjusted_views(
306            &buffers,
307            &views,
308            array
309                .array()
310                .validity()?
311                .execute_mask(array.array().len(), ctx)?,
312        );
313        Ok(())
314    }
315
316    fn execute_parent(
317        array: ArrayView<'_, Self>,
318        parent: &ArrayRef,
319        child_idx: usize,
320        ctx: &mut ExecutionCtx,
321    ) -> VortexResult<Option<ArrayRef>> {
322        PARENT_KERNELS.execute(array, parent, child_idx, ctx)
323    }
324
325    fn reduce_parent(
326        array: ArrayView<'_, Self>,
327        parent: &ArrayRef,
328        child_idx: usize,
329    ) -> VortexResult<Option<ArrayRef>> {
330        RULES.evaluate(array, parent, child_idx)
331    }
332}
333
334/// Lengths of the original values before compression, can be compressed.
335pub(crate) const UNCOMPRESSED_LENGTHS_SLOT: usize = 0;
336/// The offsets array for the FSST-compressed codes.
337pub(crate) const CODES_OFFSETS_SLOT: usize = 1;
338/// The validity bitmap for the compressed codes.
339pub(crate) const CODES_VALIDITY_SLOT: usize = 2;
340pub(crate) const NUM_SLOTS: usize = 3;
341pub(crate) const SLOT_NAMES: [&str; NUM_SLOTS] =
342    ["uncompressed_lengths", "codes_offsets", "codes_validity"];
343
344/// The inner data for an FSST-compressed array.
345///
346/// Holds the FSST symbol table (`symbols` + `symbol_lengths`) and the raw compressed
347/// codes bytes buffer. The codes offsets and validity live in the outer array's slots
348/// (slots 1 and 2 respectively).
349///
350/// A full [`VarBinArray`] representing the codes can be reconstructed on demand via
351/// [`FSSTArrayExt::codes()`], combining this buffer with the offsets/validity from slots.
352#[derive(Clone)]
353pub struct FSSTData {
354    symbol_table: Arc<FSSTSymbolTable>,
355    /// The raw compressed codes bytes, equivalent to `VarBinData::bytes`.
356    codes_bytes: BufferHandle,
357    /// Cached length (number of elements).
358    len: usize,
359}
360
361impl Display for FSSTData {
362    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
363        write!(
364            f,
365            "len: {}, nsymbols: {}",
366            self.len,
367            self.symbol_table.symbols.len()
368        )
369    }
370}
371
372impl Debug for FSSTData {
373    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
374        f.debug_struct("FSSTArray")
375            .field("symbols", &self.symbol_table.symbols)
376            .field("symbol_lengths", &self.symbol_table.symbol_lengths)
377            .field("codes_bytes_len", &self.codes_bytes.len())
378            .field("len", &self.len)
379            .field("uncompressed_lengths", &"<outer slot>")
380            .field("codes_offsets", &"<outer slot>")
381            .field("codes_validity", &"<outer slot>")
382            .finish()
383    }
384}
385
386pub(crate) struct FSSTSymbolTable {
387    symbols: Buffer<Symbol>,
388    symbol_lengths: Buffer<u8>,
389    /// Memoized compressor used for push-down of compute by compressing the RHS.
390    compressor: OnceLock<Compressor>,
391}
392
393impl FSSTSymbolTable {
394    fn new(symbols: Buffer<Symbol>, symbol_lengths: Buffer<u8>) -> Self {
395        Self {
396            symbols,
397            symbol_lengths,
398            compressor: OnceLock::new(),
399        }
400    }
401
402    fn compressor(&self) -> &Compressor {
403        self.compressor.get_or_init(|| {
404            Compressor::rebuild_from(self.symbols.as_slice(), self.symbol_lengths.as_slice())
405        })
406    }
407}
408
409#[derive(Clone, Debug)]
410pub struct FSST;
411
412impl FSST {
413    /// Build an FSST array from a set of `symbols` and `codes`.
414    ///
415    /// The `codes` VarBinArray is decomposed: its bytes are stored in [`FSSTData`], while
416    /// its offsets and validity become array slots. The codes VarBinArray can be
417    /// reconstructed on demand via [`FSSTArrayExt::codes()`].
418    pub fn try_new(
419        dtype: DType,
420        symbols: Buffer<Symbol>,
421        symbol_lengths: Buffer<u8>,
422        codes: VarBinArray,
423        uncompressed_lengths: ArrayRef,
424        ctx: &mut ExecutionCtx,
425    ) -> VortexResult<FSSTArray> {
426        let len = codes.len();
427        FSSTData::validate_parts_from_codes(
428            &symbols,
429            &symbol_lengths,
430            &codes,
431            &uncompressed_lengths,
432            &dtype,
433            len,
434            ctx,
435        )?;
436        let slots = FSSTData::make_slots(&codes, &uncompressed_lengths);
437        let codes_bytes = codes.bytes_handle().clone();
438        let data = FSSTData::try_new(symbols, symbol_lengths, codes_bytes, len)?;
439        Ok(unsafe {
440            Array::from_parts_unchecked(ArrayParts::new(FSST, dtype, len, data).with_slots(slots))
441        })
442    }
443
444    pub(crate) fn try_new_with_symbol_table(
445        dtype: DType,
446        symbol_table: Arc<FSSTSymbolTable>,
447        codes: VarBinArray,
448        uncompressed_lengths: ArrayRef,
449        ctx: &mut ExecutionCtx,
450    ) -> VortexResult<FSSTArray> {
451        let len = codes.len();
452        FSSTData::validate_parts_from_codes(
453            &symbol_table.symbols,
454            &symbol_table.symbol_lengths,
455            &codes,
456            &uncompressed_lengths,
457            &dtype,
458            len,
459            ctx,
460        )?;
461        let slots = FSSTData::make_slots(&codes, &uncompressed_lengths);
462        let codes_bytes = codes.bytes_handle().clone();
463        let data =
464            unsafe { FSSTData::new_unchecked_with_symbol_table(symbol_table, codes_bytes, len) };
465        Ok(unsafe {
466            Array::from_parts_unchecked(ArrayParts::new(FSST, dtype, len, data).with_slots(slots))
467        })
468    }
469
470    /// Legacy deserialization path (2 buffers): the codes were stored as a full
471    /// `VarBinArray` child. We decompose the VarBinArray into its bytes (stored in
472    /// FSSTData) and offsets/validity (stored in slots).
473    #[allow(clippy::too_many_arguments)]
474    fn deserialize_legacy(
475        &self,
476        dtype: &DType,
477        len: usize,
478        metadata: &FSSTMetadata,
479        symbols: &Buffer<Symbol>,
480        symbol_lengths: &Buffer<u8>,
481        children: &dyn ArrayChildren,
482        ctx: &mut ExecutionCtx,
483    ) -> VortexResult<ArrayParts<Self>> {
484        if children.len() != 2 {
485            vortex_bail!(InvalidArgument: "Expected 2 children, got {}", children.len());
486        }
487        let codes = children.get(0, &DType::Binary(dtype.nullability()), len)?;
488        let codes: VarBinArray = codes
489            .as_opt::<VarBin>()
490            .ok_or_else(|| {
491                vortex_err!(
492                    "Expected VarBinArray for codes, got {}",
493                    codes.encoding_id()
494                )
495            })?
496            .into_owned();
497        let uncompressed_lengths = children.get(
498            1,
499            &DType::Primitive(
500                metadata.get_uncompressed_lengths_ptype()?,
501                Nullability::NonNullable,
502            ),
503            len,
504        )?;
505
506        FSSTData::validate_parts_from_codes(
507            symbols,
508            symbol_lengths,
509            &codes,
510            &uncompressed_lengths,
511            dtype,
512            len,
513            ctx,
514        )?;
515        let slots = FSSTData::make_slots(&codes, &uncompressed_lengths);
516        let codes_bytes = codes.bytes_handle().clone();
517        let data = FSSTData::try_new(symbols.clone(), symbol_lengths.clone(), codes_bytes, len)?;
518        Ok(ArrayParts::new(self.clone(), dtype.clone(), len, data).with_slots(slots))
519    }
520
521    pub(crate) unsafe fn new_unchecked_with_symbol_table(
522        dtype: DType,
523        symbol_table: Arc<FSSTSymbolTable>,
524        codes: VarBinArray,
525        uncompressed_lengths: ArrayRef,
526    ) -> FSSTArray {
527        let len = codes.len();
528        let slots = FSSTData::make_slots(&codes, &uncompressed_lengths);
529        let codes_bytes = codes.bytes_handle().clone();
530        let data =
531            unsafe { FSSTData::new_unchecked_with_symbol_table(symbol_table, codes_bytes, len) };
532        unsafe {
533            Array::from_parts_unchecked(ArrayParts::new(FSST, dtype, len, data).with_slots(slots))
534        }
535    }
536}
537
538impl FSSTData {
539    fn make_slots(codes: &VarBinArray, uncompressed_lengths: &ArrayRef) -> ArraySlots {
540        smallvec![
541            Some(uncompressed_lengths.clone()),
542            Some(codes.offsets().clone()),
543            validity_to_child(
544                &codes
545                    .validity()
546                    .vortex_expect("FSST codes validity should be derivable"),
547                codes.len(),
548            ),
549        ]
550    }
551
552    /// Build FSST data from a set of `symbols`, `symbol_lengths`, and compressed codes bytes.
553    ///
554    /// Symbols are 8-bytes and can represent short strings, each of which is assigned
555    /// a code.
556    ///
557    /// The `codes_bytes` buffer contains the concatenated compressed bytecodes for all elements.
558    /// Each element's compressed bytecodes are a sequence of 8-bit codes, where each code
559    /// corresponds either to a symbol or to the "escape code" (which tells the decoder to
560    /// emit the following byte without doing a table lookup).
561    ///
562    /// The offsets and validity for the codes are stored in the array's slots, not here.
563    /// Use [`FSSTArrayExt::codes()`] to reconstruct a full `VarBinArray`.
564    pub fn try_new(
565        symbols: Buffer<Symbol>,
566        symbol_lengths: Buffer<u8>,
567        codes_bytes: BufferHandle,
568        len: usize,
569    ) -> VortexResult<Self> {
570        // SAFETY: all components validated above
571        unsafe {
572            Ok(Self::new_unchecked(
573                symbols,
574                symbol_lengths,
575                codes_bytes,
576                len,
577            ))
578        }
579    }
580
581    pub fn validate(
582        &self,
583        dtype: &DType,
584        len: usize,
585        slots: &[Option<ArrayRef>],
586        ctx: &mut ExecutionCtx,
587    ) -> VortexResult<()> {
588        let codes_offsets = slots[CODES_OFFSETS_SLOT]
589            .as_ref()
590            .vortex_expect("FSSTArray codes_offsets slot");
591        Self::validate_parts(
592            &self.symbol_table.symbols,
593            &self.symbol_table.symbol_lengths,
594            &self.codes_bytes,
595            codes_offsets,
596            dtype.nullability(),
597            uncompressed_lengths_from_slots(slots),
598            dtype,
599            len,
600            ctx,
601        )
602    }
603
604    /// Validate using the decomposed components (codes bytes + offsets + nullability).
605    #[expect(clippy::too_many_arguments)]
606    fn validate_parts(
607        symbols: &Buffer<Symbol>,
608        symbol_lengths: &Buffer<u8>,
609        codes_bytes: &BufferHandle,
610        codes_offsets: &ArrayRef,
611        codes_nullability: Nullability,
612        uncompressed_lengths: &ArrayRef,
613        dtype: &DType,
614        len: usize,
615        ctx: &mut ExecutionCtx,
616    ) -> VortexResult<()> {
617        vortex_ensure!(
618            matches!(dtype, DType::Binary(_) | DType::Utf8(_)),
619            "FSST arrays must be Binary or Utf8, found {dtype}"
620        );
621
622        if symbols.len() > 255 {
623            vortex_bail!(InvalidArgument: "symbols array must have length <= 255");
624        }
625
626        if symbols.len() != symbol_lengths.len() {
627            vortex_bail!(InvalidArgument: "symbols and symbol_lengths arrays must have same length");
628        }
629
630        Self::validate_symbol_lengths(symbol_lengths.as_slice())?;
631
632        // codes_offsets.len() - 1 == number of elements
633        let codes_len = codes_offsets.len().saturating_sub(1);
634        if codes_len != len {
635            vortex_bail!(InvalidArgument: "codes must have same len as outer array");
636        }
637
638        if uncompressed_lengths.len() != len {
639            vortex_bail!(InvalidArgument: "uncompressed_lengths must be same len as codes");
640        }
641
642        if !uncompressed_lengths.dtype().is_int() || uncompressed_lengths.dtype().is_nullable() {
643            vortex_bail!(InvalidArgument: "uncompressed_lengths must have integer type and cannot be nullable, found {}", uncompressed_lengths.dtype());
644        }
645
646        // Offsets must be non-nullable integer.
647        if !codes_offsets.dtype().is_int() || codes_offsets.dtype().is_nullable() {
648            vortex_bail!(InvalidArgument: "codes offsets must be non-nullable integer type, found {}", codes_offsets.dtype());
649        }
650
651        if codes_nullability != dtype.nullability() {
652            vortex_bail!(InvalidArgument: "codes nullability must match outer dtype nullability");
653        }
654
655        // Validate that last offset doesn't exceed bytes length (when host-resident).
656        if codes_bytes.is_on_host() && codes_offsets.is_host() && !codes_offsets.is_empty() {
657            let last_offset: usize = (&codes_offsets
658                .execute_scalar(codes_offsets.len() - 1, ctx)
659                .vortex_expect("offsets must support scalar_at"))
660                .try_into()
661                .vortex_expect("Failed to convert offset to usize");
662            vortex_ensure!(
663                last_offset <= codes_bytes.len(),
664                InvalidArgument: "Last codes offset {} exceeds codes bytes length {}",
665                last_offset,
666                codes_bytes.len()
667            );
668        }
669
670        Ok(())
671    }
672
673    fn validate_symbol_lengths(symbol_lengths: &[u8]) -> VortexResult<()> {
674        let mut expected = 2;
675        for (idx, &len) in symbol_lengths.iter().enumerate() {
676            if len > 8 || len == 0 {
677                vortex_bail!(InvalidArgument: "symbol length at index {idx} must be between 1 and 8, found {len}");
678            }
679
680            if expected == 1 {
681                if len != 1 {
682                    vortex_bail!(InvalidArgument: "symbol length at index {idx} must be 1 after one-byte symbols begin, found {len}");
683                }
684            } else {
685                if len == 1 {
686                    expected = 1;
687                }
688
689                if len < expected {
690                    vortex_bail!(InvalidArgument: "symbol length at index {idx} violates FSST symbol table ordering");
691                }
692                expected = len;
693            }
694        }
695
696        Ok(())
697    }
698
699    /// Validate using a VarBinArray for the codes (convenience for construction paths).
700    fn validate_parts_from_codes(
701        symbols: &Buffer<Symbol>,
702        symbol_lengths: &Buffer<u8>,
703        codes: &VarBinArray,
704        uncompressed_lengths: &ArrayRef,
705        dtype: &DType,
706        len: usize,
707        ctx: &mut ExecutionCtx,
708    ) -> VortexResult<()> {
709        Self::validate_parts(
710            symbols,
711            symbol_lengths,
712            codes.bytes_handle(),
713            codes.offsets(),
714            codes.dtype().nullability(),
715            uncompressed_lengths,
716            dtype,
717            len,
718            ctx,
719        )
720    }
721
722    pub(crate) unsafe fn new_unchecked(
723        symbols: Buffer<Symbol>,
724        symbol_lengths: Buffer<u8>,
725        codes_bytes: BufferHandle,
726        len: usize,
727    ) -> Self {
728        let symbol_table = Arc::new(FSSTSymbolTable::new(symbols, symbol_lengths));
729        unsafe { Self::new_unchecked_with_symbol_table(symbol_table, codes_bytes, len) }
730    }
731
732    pub(crate) unsafe fn new_unchecked_with_symbol_table(
733        symbol_table: Arc<FSSTSymbolTable>,
734        codes_bytes: BufferHandle,
735        len: usize,
736    ) -> Self {
737        Self {
738            symbol_table,
739            codes_bytes,
740            len,
741        }
742    }
743
744    /// Returns the number of elements in the array.
745    pub fn len(&self) -> usize {
746        self.len
747    }
748
749    /// Returns `true` if the array contains no elements.
750    pub fn is_empty(&self) -> bool {
751        self.len == 0
752    }
753
754    /// Access the symbol table array.
755    pub fn symbols(&self) -> &Buffer<Symbol> {
756        &self.symbol_table.symbols
757    }
758
759    /// Access the symbol lengths array.
760    pub fn symbol_lengths(&self) -> &Buffer<u8> {
761        &self.symbol_table.symbol_lengths
762    }
763
764    pub(crate) fn symbol_table(&self) -> Arc<FSSTSymbolTable> {
765        Arc::clone(&self.symbol_table)
766    }
767
768    /// Access the compressed codes bytes buffer handle (may be on host or device).
769    pub fn codes_bytes_handle(&self) -> &BufferHandle {
770        &self.codes_bytes
771    }
772
773    /// Access the compressed codes bytes on the host.
774    pub fn codes_bytes(&self) -> &ByteBuffer {
775        self.codes_bytes.as_host()
776    }
777
778    /// Build a [`Decompressor`] that can be used to decompress values from
779    /// this array.
780    pub fn decompressor(&self) -> Decompressor<'_> {
781        Decompressor::new(self.symbols().as_slice(), self.symbol_lengths().as_slice())
782    }
783
784    /// Retrieves the FSST compressor.
785    pub fn compressor(&self) -> &Compressor {
786        self.symbol_table.compressor()
787    }
788}
789
790fn uncompressed_lengths_from_slots(slots: &[Option<ArrayRef>]) -> &ArrayRef {
791    slots[UNCOMPRESSED_LENGTHS_SLOT]
792        .as_ref()
793        .vortex_expect("FSSTArray uncompressed_lengths slot")
794}
795
796pub trait FSSTArrayExt: TypedArrayRef<FSST> {
797    fn uncompressed_lengths(&self) -> &ArrayRef {
798        uncompressed_lengths_from_slots(self.as_ref().slots())
799    }
800
801    fn uncompressed_lengths_dtype(&self) -> &DType {
802        self.uncompressed_lengths().dtype()
803    }
804
805    /// Reconstruct a [`VarBinArray`] for the compressed codes by combining the bytes
806    /// from [`FSSTData`] with the offsets and validity stored in the array's slots.
807    fn codes(&self) -> VarBinArray {
808        let offsets = self.as_ref().slots()[CODES_OFFSETS_SLOT]
809            .as_ref()
810            .vortex_expect("FSSTArray codes_offsets slot")
811            .clone();
812        let validity = child_to_validity(
813            self.as_ref().slots()[CODES_VALIDITY_SLOT].as_ref(),
814            self.as_ref().dtype().nullability(),
815        );
816        let codes_bytes = self.codes_bytes_handle().clone();
817        // SAFETY: components were validated at construction time.
818        unsafe {
819            VarBinArray::new_unchecked_from_handle(
820                offsets,
821                codes_bytes,
822                DType::Binary(self.as_ref().dtype().nullability()),
823                validity,
824            )
825        }
826    }
827
828    /// Get the DType of the codes array.
829    fn codes_dtype(&self) -> DType {
830        DType::Binary(self.as_ref().dtype().nullability())
831    }
832}
833
834impl<T: TypedArrayRef<FSST>> FSSTArrayExt for T {}
835
836impl ValidityVTable<FSST> for FSST {
837    fn validity(array: ArrayView<'_, FSST>) -> VortexResult<Validity> {
838        Ok(child_to_validity(
839            array.slots()[CODES_VALIDITY_SLOT].as_ref(),
840            array.dtype().nullability(),
841        ))
842    }
843}
844
845#[cfg(test)]
846mod test {
847    use fsst::Compressor;
848    use fsst::Symbol;
849    use prost::Message;
850    use vortex_array::ArrayPlugin;
851    use vortex_array::IntoArray;
852    use vortex_array::LEGACY_SESSION;
853    use vortex_array::VortexSessionExecute;
854    use vortex_array::accessor::ArrayAccessor;
855    use vortex_array::arrays::VarBinViewArray;
856    use vortex_array::buffer::BufferHandle;
857    use vortex_array::dtype::DType;
858    use vortex_array::dtype::Nullability;
859    use vortex_array::dtype::PType;
860    use vortex_array::test_harness::check_metadata;
861    use vortex_buffer::Buffer;
862    use vortex_error::VortexError;
863    use vortex_error::VortexResult;
864    use vortex_error::vortex_err;
865
866    use crate::FSST;
867    use crate::array::FSSTArrayExt;
868    use crate::array::FSSTMetadata;
869    use crate::fsst_compress_iter;
870
871    #[test]
872    fn slice_reuses_initialized_compressor() -> VortexResult<()> {
873        let symbols = Buffer::<Symbol>::copy_from([
874            Symbol::from_slice(b"abc00000"),
875            Symbol::from_slice(b"defghijk"),
876        ]);
877        let symbol_lengths = Buffer::<u8>::copy_from([3, 8]);
878
879        let compressor = Compressor::rebuild_from(symbols.as_slice(), symbol_lengths.as_slice());
880        let mut ctx = LEGACY_SESSION.create_execution_ctx();
881        let fsst_array = fsst_compress_iter(
882            [
883                Some(b"abcabcab".as_ref()),
884                Some(b"defghijk".as_ref()),
885                Some(b"abcxyz".as_ref()),
886            ]
887            .into_iter(),
888            3,
889            DType::Utf8(Nullability::NonNullable),
890            &compressor,
891            &mut ctx,
892        );
893
894        let compressor_ptr = fsst_array.compressor() as *const Compressor;
895        let sliced = fsst_array
896            .slice(1..3)?
897            .try_downcast::<FSST>()
898            .map_err(|_| vortex_err!("slice must return an FSST array"))?;
899        let sliced_compressor_ptr = sliced.compressor() as *const Compressor;
900
901        assert_eq!(compressor_ptr, sliced_compressor_ptr);
902        Ok(())
903    }
904
905    #[cfg_attr(miri, ignore)]
906    #[test]
907    fn test_fsst_metadata() {
908        check_metadata(
909            "fsst.metadata",
910            &FSSTMetadata {
911                uncompressed_lengths_ptype: PType::U64 as i32,
912                codes_offsets_ptype: PType::I32 as i32,
913            }
914            .encode_to_vec(),
915        );
916    }
917
918    /// The original FSST array stored codes as a VarBinArray child and required that the child
919    /// have this encoding. Vortex forbids this kind of introspection, therefore we had to fix
920    /// the array to store the compressed offsets and compressed data buffer separately, and only
921    /// use VarBinArray to delegate behavior.
922    ///
923    /// This test manually constructs an old-style FSST array and ensures that it can still be
924    /// deserialized.
925    #[test]
926    fn test_back_compat() {
927        let symbols = Buffer::<Symbol>::copy_from([
928            Symbol::from_slice(b"abc00000"),
929            Symbol::from_slice(b"defghijk"),
930        ]);
931        let symbol_lengths = Buffer::<u8>::copy_from([3, 8]);
932
933        let compressor = Compressor::rebuild_from(symbols.as_slice(), symbol_lengths.as_slice());
934        let mut ctx = LEGACY_SESSION.create_execution_ctx();
935        let fsst_array = fsst_compress_iter(
936            [Some(b"abcabcab".as_ref()), Some(b"defghijk".as_ref())].into_iter(),
937            2,
938            DType::Utf8(Nullability::NonNullable),
939            &compressor,
940            &mut ctx,
941        );
942
943        let compressed_codes = fsst_array.codes();
944
945        // There were two buffers:
946        // 1. The 8 byte symbols
947        // 2. The symbol lengths as u8.
948        let buffers = [
949            BufferHandle::new_host(symbols.into_byte_buffer()),
950            BufferHandle::new_host(symbol_lengths.into_byte_buffer()),
951        ];
952
953        // There were 2 children:
954        // 1. The compressed codes, stored as a VarBinArray.
955        // 2. The uncompressed lengths, stored as a Primitive array.
956        let children = vec![
957            compressed_codes.into_array(),
958            fsst_array.uncompressed_lengths().clone(),
959        ];
960
961        let fsst = ArrayPlugin::deserialize(
962            &FSST,
963            &DType::Utf8(Nullability::NonNullable),
964            2,
965            &FSSTMetadata {
966                uncompressed_lengths_ptype: fsst_array
967                    .uncompressed_lengths()
968                    .dtype()
969                    .as_ptype()
970                    .into(),
971                // Legacy array did not store this field, use Protobuf default of 0.
972                codes_offsets_ptype: 0,
973            }
974            .encode_to_vec(),
975            &buffers,
976            &children.as_slice(),
977            &LEGACY_SESSION,
978        )
979        .unwrap();
980
981        let decompressed = fsst
982            .execute::<VarBinViewArray>(&mut LEGACY_SESSION.create_execution_ctx())
983            .unwrap();
984        decompressed
985            .with_iterator(|it| {
986                assert_eq!(it.next().unwrap(), Some(b"abcabcab".as_ref()));
987                assert_eq!(it.next().unwrap(), Some(b"defghijk".as_ref()));
988                Ok::<_, VortexError>(())
989            })
990            .unwrap()
991    }
992}