1- //! Write-time assembly for zoned layouts.
2-
31// SPDX-License-Identifier: Apache-2.0
42// SPDX-FileCopyrightText: Copyright the Vortex contributors
53
4+ //! Write-time assembly for zoned layouts.
5+
66use std:: num:: NonZeroUsize ;
77use std:: sync:: Arc ;
88
@@ -112,24 +112,8 @@ impl LayoutStrategy for ZonedStrategy {
112112 if let Some ( skip_indexes) = & self . options . skip_indexes {
113113 for skip_index in skip_indexes. iter ( ) {
114114 match skip_index. aggregate_fn ( stream. dtype ( ) ) {
115- Some ( skip_aggregate_fn) => {
116- if aggregate_fns
117- . iter ( )
118- . any ( |aggregate_fn| aggregate_fn. id ( ) == skip_aggregate_fn. id ( ) )
119- {
120- vortex_bail ! (
121- "skip index aggregate {} is configured more than once" ,
122- skip_aggregate_fn. id( )
123- ) ;
124- }
125-
126- aggregate_fns. push ( skip_aggregate_fn) ;
127- }
128- None => vortex_bail ! (
129- "skip index aggregate {} is unsupported for type {}" ,
130- skip_index. aggregate_id( ) ,
131- stream. dtype( )
132- ) ,
115+ Some ( skip_aggregate_fn) => aggregate_fns. push ( skip_aggregate_fn) ,
116+ None => vortex_bail ! ( "skip index is unsupported for type {}" , stream. dtype( ) ) ,
133117 }
134118 }
135119 }
@@ -230,6 +214,8 @@ mod tests {
230214 use vortex_array:: ArrayContext ;
231215 use vortex_array:: IntoArray ;
232216 use vortex_array:: aggregate_fn:: AggregateFnVTable ;
217+ use vortex_array:: aggregate_fn:: AggregateFnVTableExt ;
218+ use vortex_array:: aggregate_fn:: NumericalAggregateOpts ;
233219 use vortex_array:: aggregate_fn:: fns:: bounded_max:: BoundedMax ;
234220 use vortex_array:: aggregate_fn:: fns:: bounded_min:: BoundedMin ;
235221 use vortex_array:: aggregate_fn:: fns:: max:: Max ;
@@ -256,21 +242,22 @@ mod tests {
256242 use crate :: layouts:: flat:: writer:: FlatLayoutStrategy ;
257243 use crate :: layouts:: zoned:: Zoned ;
258244 use crate :: layouts:: zoned:: schema:: default_bounded_stat_max_bytes;
245+ use crate :: layouts:: zoned:: skip_index:: SkipIndex ;
259246 use crate :: segments:: TestSegments ;
260247 use crate :: sequence:: SequenceId ;
261248 use crate :: sequence:: SequentialArrayStreamExt ;
262249 use crate :: session:: LayoutSession ;
263250
264251 /// Write three zones of primitives through `ctx`, returning the aggregates the zoned
265252 /// layout recorded.
266- fn write_zones ( ctx : LayoutWriterContext ) -> VortexResult < Vec < String > > {
253+ fn write_zones_with_options (
254+ ctx : LayoutWriterContext ,
255+ options : ZonedLayoutOptions ,
256+ ) -> VortexResult < Vec < String > > {
267257 let strategy = ZonedStrategy :: new (
268258 ChunkedLayoutStrategy :: new ( FlatLayoutStrategy :: default ( ) ) ,
269259 FlatLayoutStrategy :: default ( ) ,
270- ZonedLayoutOptions {
271- block_size : NonZeroUsize :: new ( 3 ) . vortex_expect ( "non zero" ) ,
272- ..Default :: default ( )
273- } ,
260+ options,
274261 ) ;
275262 let ( ptr, eof) = SequenceId :: root ( ) . split ( ) ;
276263 let stream = ChunkedArray :: from_iter ( [
@@ -306,6 +293,16 @@ mod tests {
306293 . collect ( ) )
307294 }
308295
296+ fn write_zones ( ctx : LayoutWriterContext ) -> VortexResult < Vec < String > > {
297+ write_zones_with_options (
298+ ctx,
299+ ZonedLayoutOptions {
300+ block_size : NonZeroUsize :: new ( 3 ) . vortex_expect ( "non zero" ) ,
301+ ..Default :: default ( )
302+ } ,
303+ )
304+ }
305+
309306 #[ test]
310307 fn unrestricted_context_writes_the_default_aggregates ( ) -> VortexResult < ( ) > {
311308 let written = write_zones ( LayoutWriterContext :: new ( ArrayContext :: empty ( ) ) ) ?;
@@ -380,4 +377,64 @@ mod tests {
380377 . all( |aggregate_fn| !aggregate_fn. is:: <Sum >( ) )
381378 ) ;
382379 }
380+
381+ // TODO (joacoc): replace with bloom skip index after merge <https://github.com/vortex-data/vortex/pull/9398>
382+ // Just a small skip-index implementation used to test the zoned writer until the bloom filter lands.
383+ struct TestSkipIndex {
384+ aggregate_fn : Option < AggregateFnRef > ,
385+ }
386+
387+ impl TestSkipIndex {
388+ fn new ( aggregate_fn : AggregateFnRef ) -> Self {
389+ Self {
390+ aggregate_fn : Some ( aggregate_fn) ,
391+ }
392+ }
393+
394+ fn unsupported ( ) -> Self {
395+ Self { aggregate_fn : None }
396+ }
397+
398+ fn into_ref ( self ) -> SkipIndexRef {
399+ SkipIndexRef :: new ( Arc :: new ( self ) )
400+ }
401+ }
402+
403+ impl SkipIndex for TestSkipIndex {
404+ fn aggregate_fn ( & self , _input_dtype : & DType ) -> Option < AggregateFnRef > {
405+ self . aggregate_fn . clone ( )
406+ }
407+
408+ fn register ( & self , _session : & VortexSession ) { }
409+ }
410+
411+ #[ test]
412+ fn writer_appends_skip_index_aggregate ( ) -> VortexResult < ( ) > {
413+ let options = ZonedLayoutOptions :: default ( ) . with_skip_index (
414+ TestSkipIndex :: new ( Sum . bind ( NumericalAggregateOpts :: default ( ) ) ) . into_ref ( ) ,
415+ ) ;
416+ let written =
417+ write_zones_with_options ( LayoutWriterContext :: new ( ArrayContext :: empty ( ) ) , options) ?;
418+
419+ assert ! (
420+ written. contains( & Sum . id( ) . to_string( ) ) ,
421+ "expected aggregation id present, wrote {written:?}"
422+ ) ;
423+ Ok ( ( ) )
424+ }
425+
426+ #[ test]
427+ fn writer_rejects_skip_index_unsupported_for_dtype ( ) -> VortexResult < ( ) > {
428+ let options =
429+ ZonedLayoutOptions :: default ( ) . with_skip_index ( TestSkipIndex :: unsupported ( ) . into_ref ( ) ) ;
430+
431+ let error =
432+ write_zones_with_options ( LayoutWriterContext :: new ( ArrayContext :: empty ( ) ) , options)
433+ . expect_err ( "unsupported skip index should fail the write" ) ;
434+ assert ! (
435+ error. to_string( ) . contains( "unsupported for type" ) ,
436+ "unexpected error: {error}"
437+ ) ;
438+ Ok ( ( ) )
439+ }
383440}
0 commit comments