1use 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
63pub 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 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 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 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 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
334pub(crate) const UNCOMPRESSED_LENGTHS_SLOT: usize = 0;
336pub(crate) const CODES_OFFSETS_SLOT: usize = 1;
338pub(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#[derive(Clone)]
353pub struct FSSTData {
354 symbol_table: Arc<FSSTSymbolTable>,
355 codes_bytes: BufferHandle,
357 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 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 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 #[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 pub fn try_new(
565 symbols: Buffer<Symbol>,
566 symbol_lengths: Buffer<u8>,
567 codes_bytes: BufferHandle,
568 len: usize,
569 ) -> VortexResult<Self> {
570 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 #[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 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 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 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 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 pub fn len(&self) -> usize {
746 self.len
747 }
748
749 pub fn is_empty(&self) -> bool {
751 self.len == 0
752 }
753
754 pub fn symbols(&self) -> &Buffer<Symbol> {
756 &self.symbol_table.symbols
757 }
758
759 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 pub fn codes_bytes_handle(&self) -> &BufferHandle {
770 &self.codes_bytes
771 }
772
773 pub fn codes_bytes(&self) -> &ByteBuffer {
775 self.codes_bytes.as_host()
776 }
777
778 pub fn decompressor(&self) -> Decompressor<'_> {
781 Decompressor::new(self.symbols().as_slice(), self.symbol_lengths().as_slice())
782 }
783
784 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 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 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 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 #[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 let buffers = [
949 BufferHandle::new_host(symbols.into_byte_buffer()),
950 BufferHandle::new_host(symbol_lengths.into_byte_buffer()),
951 ];
952
953 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 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}