lance_io/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3use std::{
4    ops::{Range, RangeFrom, RangeFull, RangeTo},
5    sync::Arc,
6};
7
8use arrow::datatypes::UInt32Type;
9use arrow_array::{PrimitiveArray, UInt32Array};
10use snafu::location;
11
12use lance_core::{Error, Result};
13
14pub mod encodings;
15pub mod ffi;
16pub mod local;
17pub mod object_reader;
18pub mod object_store;
19pub mod object_writer;
20pub mod scheduler;
21pub mod stream;
22#[cfg(test)]
23pub mod testing;
24pub mod traits;
25pub mod utils;
26
27pub use scheduler::{bytes_read_counter, iops_counter};
28
29/// Defines a selection of rows to read from a file/batch
30#[derive(Debug, Clone, PartialEq, Default)]
31pub enum ReadBatchParams {
32    /// Select a contiguous range of rows
33    Range(Range<usize>),
34    /// Select multiple contiguous ranges of rows
35    Ranges(Arc<[Range<u64>]>),
36    /// Select all rows (this is the default)
37    #[default]
38    RangeFull,
39    /// Select all rows up to a given index
40    RangeTo(RangeTo<usize>),
41    /// Select all rows starting at a given index
42    RangeFrom(RangeFrom<usize>),
43    /// Select scattered non-contiguous rows
44    Indices(UInt32Array),
45}
46
47impl std::fmt::Display for ReadBatchParams {
48    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
49        match self {
50            Self::Range(r) => write!(f, "Range({}..{})", r.start, r.end),
51            Self::Ranges(ranges) => {
52                let mut ranges_str = ranges.iter().fold(String::new(), |mut acc, r| {
53                    acc.push_str(&format!("{}..{}", r.start, r.end));
54                    acc.push(',');
55                    acc
56                });
57                // Remove the trailing comma
58                if !ranges_str.is_empty() {
59                    ranges_str.pop();
60                }
61                write!(f, "Ranges({})", ranges_str)
62            }
63            Self::RangeFull => write!(f, "RangeFull"),
64            Self::RangeTo(r) => write!(f, "RangeTo({})", r.end),
65            Self::RangeFrom(r) => write!(f, "RangeFrom({})", r.start),
66            Self::Indices(indices) => {
67                let mut indices_str = indices.values().iter().fold(String::new(), |mut acc, v| {
68                    acc.push_str(&v.to_string());
69                    acc.push(',');
70                    acc
71                });
72                if !indices_str.is_empty() {
73                    indices_str.pop();
74                }
75                write!(f, "Indices({})", indices_str)
76            }
77        }
78    }
79}
80
81impl From<&[u32]> for ReadBatchParams {
82    fn from(value: &[u32]) -> Self {
83        Self::Indices(UInt32Array::from_iter_values(value.iter().copied()))
84    }
85}
86
87impl From<UInt32Array> for ReadBatchParams {
88    fn from(value: UInt32Array) -> Self {
89        Self::Indices(value)
90    }
91}
92
93impl From<RangeFull> for ReadBatchParams {
94    fn from(_: RangeFull) -> Self {
95        Self::RangeFull
96    }
97}
98
99impl From<Range<usize>> for ReadBatchParams {
100    fn from(r: Range<usize>) -> Self {
101        Self::Range(r)
102    }
103}
104
105impl From<RangeTo<usize>> for ReadBatchParams {
106    fn from(r: RangeTo<usize>) -> Self {
107        Self::RangeTo(r)
108    }
109}
110
111impl From<RangeFrom<usize>> for ReadBatchParams {
112    fn from(r: RangeFrom<usize>) -> Self {
113        Self::RangeFrom(r)
114    }
115}
116
117impl From<&Self> for ReadBatchParams {
118    fn from(params: &Self) -> Self {
119        params.clone()
120    }
121}
122
123impl ReadBatchParams {
124    /// Validate that the selection is valid given the length of the batch
125    pub fn valid_given_len(&self, len: usize) -> bool {
126        match self {
127            Self::Indices(indices) => indices.iter().all(|i| i.unwrap_or(0) < len as u32),
128            Self::Range(r) => r.start < len && r.end <= len,
129            Self::Ranges(ranges) => ranges.iter().all(|r| r.end <= len as u64),
130            Self::RangeFull => true,
131            Self::RangeTo(r) => r.end <= len,
132            Self::RangeFrom(r) => r.start < len,
133        }
134    }
135
136    /// Slice the selection
137    ///
138    /// For example, given ReadBatchParams::RangeFull and slice(10, 20), the output will be
139    /// ReadBatchParams::Range(10..20)
140    ///
141    /// Given ReadBatchParams::Range(10..20) and slice(5, 3), the output will be
142    /// ReadBatchParams::Range(15..18)
143    ///
144    /// Given ReadBatchParams::RangeTo(20) and slice(10, 5), the output will be
145    /// ReadBatchParams::Range(10..15)
146    ///
147    /// Given ReadBatchParams::RangeFrom(20) and slice(10, 5), the output will be
148    /// ReadBatchParams::Range(30..35)
149    ///
150    /// Given ReadBatchParams::Indices([1, 3, 5, 7, 9]) and slice(1, 3), the output will be
151    /// ReadBatchParams::Indices([3, 5, 7])
152    ///
153    /// You cannot slice beyond the bounds of the selection and an attempt to do so will
154    /// return an error.
155    pub fn slice(&self, start: usize, length: usize) -> Result<Self> {
156        let out_of_bounds = |size: usize| {
157            Err(Error::InvalidInput {
158                source: format!(
159                    "Cannot slice from {} with length {} given a selection of size {}",
160                    start, length, size
161                )
162                .into(),
163                location: location!(),
164            })
165        };
166
167        match self {
168            Self::Indices(indices) => {
169                if start + length > indices.len() {
170                    return out_of_bounds(indices.len());
171                }
172                Ok(Self::Indices(indices.slice(start, length)))
173            }
174            Self::Range(r) => {
175                if (r.start + start + length) > r.end {
176                    return out_of_bounds(r.end - r.start);
177                }
178                Ok(Self::Range((r.start + start)..(r.start + start + length)))
179            }
180            Self::Ranges(ranges) => {
181                let mut new_ranges = Vec::with_capacity(ranges.len());
182                let mut to_skip = start as u64;
183                let mut to_take = length as u64;
184                let mut total_num_rows = 0;
185                for r in ranges.as_ref() {
186                    let num_rows = r.end - r.start;
187                    total_num_rows += num_rows;
188                    if to_skip > num_rows {
189                        to_skip -= num_rows;
190                        continue;
191                    }
192                    let new_start = r.start + to_skip;
193                    let to_take_this_range = (num_rows - to_skip).min(to_take);
194                    new_ranges.push(new_start..(new_start + to_take_this_range));
195                    to_skip = 0;
196                    to_take -= to_take_this_range;
197                    if to_take == 0 {
198                        break;
199                    }
200                }
201                if to_take > 0 {
202                    out_of_bounds(total_num_rows as usize)
203                } else {
204                    Ok(Self::Ranges(new_ranges.into()))
205                }
206            }
207            Self::RangeFull => Ok(Self::Range(start..(start + length))),
208            Self::RangeTo(range) => {
209                if start + length > range.end {
210                    return out_of_bounds(range.end);
211                }
212                Ok(Self::Range(start..(start + length)))
213            }
214            Self::RangeFrom(r) => {
215                // No way to validate out_of_bounds, assume caller will do so
216                Ok(Self::Range((r.start + start)..(r.start + start + length)))
217            }
218        }
219    }
220
221    /// Convert a read range into a vector of row offsets
222    ///
223    /// RangeFull and RangeFrom are unbounded and cannot be converted into row offsets
224    /// and any attempt to do so will return an error.  Call slice first
225    pub fn to_offsets(&self) -> Result<PrimitiveArray<UInt32Type>> {
226        match self {
227            Self::Indices(indices) => Ok(indices.clone()),
228            Self::Range(r) => Ok(UInt32Array::from(Vec::from_iter(
229                r.start as u32..r.end as u32,
230            ))),
231            Self::Ranges(ranges) => {
232                let num_rows = ranges
233                    .iter()
234                    .map(|r| (r.end - r.start) as usize)
235                    .sum::<usize>();
236                let mut offsets = Vec::with_capacity(num_rows);
237                for r in ranges.as_ref() {
238                    offsets.extend(r.start as u32..r.end as u32);
239                }
240                Ok(UInt32Array::from(offsets))
241            }
242            Self::RangeFull => Err(Error::invalid_input(
243                "cannot materialize RangeFull",
244                location!(),
245            )),
246            Self::RangeTo(r) => Ok(UInt32Array::from(Vec::from_iter(0..r.end as u32))),
247            Self::RangeFrom(_) => Err(Error::invalid_input(
248                "cannot materialize RangeFrom",
249                location!(),
250            )),
251        }
252    }
253
254    pub fn iter_offset_ranges<'a>(
255        &'a self,
256    ) -> Result<Box<dyn Iterator<Item = Range<u32>> + Send + 'a>> {
257        match self {
258            Self::Indices(indices) => Ok(Box::new(indices.values().iter().map(|i| *i..(*i + 1)))),
259            Self::Range(r) => Ok(Box::new(std::iter::once(r.start as u32..r.end as u32))),
260            Self::Ranges(ranges) => Ok(Box::new(
261                ranges.iter().map(|r| r.start as u32..r.end as u32),
262            )),
263            Self::RangeFull => Err(Error::invalid_input(
264                "cannot materialize RangeFull",
265                location!(),
266            )),
267            Self::RangeTo(r) => Ok(Box::new(std::iter::once(0..r.end as u32))),
268            Self::RangeFrom(_) => Err(Error::invalid_input(
269                "cannot materialize RangeFrom",
270                location!(),
271            )),
272        }
273    }
274
275    /// Convert a read range into a vector of row ranges
276    pub fn to_ranges(&self) -> Result<Vec<Range<u64>>> {
277        match self {
278            Self::Indices(indices) => Ok(indices
279                .values()
280                .iter()
281                .map(|i| *i as u64..(*i + 1) as u64)
282                .collect()),
283            Self::Range(r) => Ok(vec![r.start as u64..r.end as u64]),
284            Self::Ranges(ranges) => Ok(ranges.to_vec()),
285            Self::RangeFull => Err(Error::invalid_input(
286                "cannot materialize RangeFull",
287                location!(),
288            )),
289            Self::RangeTo(r) => Ok(vec![0..r.end as u64]),
290            Self::RangeFrom(_) => Err(Error::invalid_input(
291                "cannot materialize RangeFrom",
292                location!(),
293            )),
294        }
295    }
296
297    /// Same thing as to_offsets but the caller knows the total number of rows in the file
298    ///
299    /// This makes it possible to materialize RangeFull / RangeFrom
300    pub fn to_offsets_total(&self, total: u32) -> PrimitiveArray<UInt32Type> {
301        match self {
302            Self::Indices(indices) => indices.clone(),
303            Self::Range(r) => UInt32Array::from_iter_values(r.start as u32..r.end as u32),
304            Self::Ranges(ranges) => {
305                let num_rows = ranges
306                    .iter()
307                    .map(|r| (r.end - r.start) as usize)
308                    .sum::<usize>();
309                let mut offsets = Vec::with_capacity(num_rows);
310                for r in ranges.as_ref() {
311                    offsets.extend(r.start as u32..r.end as u32);
312                }
313                UInt32Array::from(offsets)
314            }
315            Self::RangeFull => UInt32Array::from_iter_values(0_u32..total),
316            Self::RangeTo(r) => UInt32Array::from_iter_values(0..r.end as u32),
317            Self::RangeFrom(r) => UInt32Array::from_iter_values(r.start as u32..total),
318        }
319    }
320}
321
322#[cfg(test)]
323mod test {
324    use std::ops::{RangeFrom, RangeTo};
325
326    use arrow_array::UInt32Array;
327
328    use crate::ReadBatchParams;
329
330    #[test]
331    fn test_params_slice() {
332        let params = ReadBatchParams::Ranges(vec![0..15, 20..40].into());
333        let sliced = params.slice(10, 10).unwrap();
334        assert_eq!(sliced, ReadBatchParams::Ranges(vec![10..15, 20..25].into()));
335    }
336
337    #[test]
338    fn test_params_to_offsets() {
339        let check = |params: ReadBatchParams, base_offset, length, expected: Vec<u32>| {
340            let offsets = params
341                .slice(base_offset, length)
342                .unwrap()
343                .to_offsets()
344                .unwrap();
345            let expected = UInt32Array::from(expected);
346            assert_eq!(offsets, expected);
347        };
348
349        check(ReadBatchParams::RangeFull, 0, 100, (0..100).collect());
350        check(ReadBatchParams::RangeFull, 50, 100, (50..150).collect());
351        check(
352            ReadBatchParams::RangeFrom(RangeFrom { start: 500 }),
353            0,
354            100,
355            (500..600).collect(),
356        );
357        check(
358            ReadBatchParams::RangeFrom(RangeFrom { start: 500 }),
359            100,
360            100,
361            (600..700).collect(),
362        );
363        check(
364            ReadBatchParams::RangeTo(RangeTo { end: 800 }),
365            0,
366            100,
367            (0..100).collect(),
368        );
369        check(
370            ReadBatchParams::RangeTo(RangeTo { end: 800 }),
371            200,
372            100,
373            (200..300).collect(),
374        );
375        check(
376            ReadBatchParams::Indices(UInt32Array::from(vec![1, 3, 5, 7, 9])),
377            0,
378            2,
379            vec![1, 3],
380        );
381        check(
382            ReadBatchParams::Indices(UInt32Array::from(vec![1, 3, 5, 7, 9])),
383            2,
384            2,
385            vec![5, 7],
386        );
387
388        let check_error = |params: ReadBatchParams, base_offset, length| {
389            assert!(params.slice(base_offset, length).is_err());
390        };
391
392        check_error(ReadBatchParams::Indices(UInt32Array::from(vec![1])), 0, 2);
393        check_error(ReadBatchParams::Indices(UInt32Array::from(vec![1])), 1, 1);
394        check_error(ReadBatchParams::Range(0..10), 5, 6);
395        check_error(ReadBatchParams::RangeTo(RangeTo { end: 10 }), 5, 6);
396
397        assert!(ReadBatchParams::RangeFull.to_offsets().is_err());
398        assert!(ReadBatchParams::RangeFrom(RangeFrom { start: 10 })
399            .to_offsets()
400            .is_err());
401    }
402}
lance_io/lib.rs

lance_io/
lib.rs
