Skip to content

Commit 7fe27fe

Browse files
committed
Reuse indexed pages for compact Take ranges and sliced execution
Signed-off-by: Daniel King <dan@spiraldb.com>
1 parent eea0ae2 commit 7fe27fe

3 files changed

Lines changed: 327 additions & 10 deletions

File tree

‎encodings/pco/src/array.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ use crate::PcoLogicalRowPrefix;
6666
use crate::PcoMetadata;
6767
use crate::PcoPageInfo;
6868
use crate::page_index::PageIndex;
69+
use crate::take::slice as decode_slice;
6970

7071
// Overall approach here:
7172
// Chunk the array into Pco chunks (currently using the default recommended size
@@ -278,6 +279,12 @@ impl VTable for Pco {
278279
}
279280

280281
fn execute(array: Array<Self>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
282+
if array.metadata.logical_row_prefix.is_some()
283+
&& (array.slice_start() != 0 || array.slice_stop() != array.unsliced_n_rows()) {
284+
return Ok(ExecutionResult::done(
285+
decode_slice(array.as_view(), ctx)?.into_array(),
286+
));
287+
}
281288
let unsliced_validity = array.unsliced_validity();
282289
Ok(ExecutionResult::done(
283290
array

‎encodings/pco/src/decoder.rs‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,14 +9,19 @@ use std::collections::btree_map::Entry;
99
use pco::data_types::Number;
1010
use pco::wrapped::ChunkDecompressor;
1111
use pco::wrapped::FileDecompressor;
12+
use vortex_array::ArrayView;
1213
use vortex_array::ExecutionCtx;
1314
use vortex_array::arrays::PrimitiveArray;
1415
use vortex_array::dtype::NativePType;
1516
use vortex_array::validity::Validity;
1617
use vortex_buffer::BufferMut;
1718
use vortex_error::VortexResult;
19+
use vortex_error::vortex_err;
1820

21+
use crate::Pco;
1922
use crate::array::vortex_err_from_pco;
23+
use crate::page_index::ChunkCursor;
24+
use crate::page_index::IndexedView;
2025

2126
pub(crate) struct PageDecoder<T: Number> {
2227
file: FileDecompressor,
@@ -71,3 +76,51 @@ impl<T: Number + NativePType> PageDecoder<T> {
7176
Ok(())
7277
}
7378
}
79+
80+
/// Operation-owned decoded pages for compact repeated/range selections. The physical
81+
/// descriptor is derived from the chosen mode only when a page is decoded for the first time.
82+
pub(crate) struct DecodedPages<T: Number> {
83+
decoder: PageDecoder<T>,
84+
chunks: ChunkCursor,
85+
pages: BTreeMap<usize, PrimitiveArray>,
86+
}
87+
88+
impl<T: Number + NativePType> DecodedPages<T> {
89+
pub fn new(array: ArrayView<'_, Pco>) -> VortexResult<Self> {
90+
Ok(Self {
91+
decoder: PageDecoder::new(array.metadata.header.as_slice())?,
92+
chunks: ChunkCursor::default(),
93+
pages: BTreeMap::new(),
94+
})
95+
}
96+
pub fn values(
97+
&mut self,
98+
view: IndexedView<'_>,
99+
array: ArrayView<'_, Pco>,
100+
page: usize,
101+
ctx: &mut ExecutionCtx,
102+
) -> VortexResult<&PrimitiveArray> {
103+
let entry = match self.pages.entry(page) {
104+
Entry::Occupied(entry) => entry.into_mut(),
105+
Entry::Vacant(entry) => {
106+
let info = view.physical(page, &mut self.chunks)?;
107+
let buffer = array
108+
.pages
109+
.get(page)
110+
.ok_or_else(|| vortex_err!("Pco compressed page is missing"))?;
111+
let chunk = array
112+
.chunk_metas
113+
.get(info.chunk)
114+
.ok_or_else(|| vortex_err!("Pco chunk metadata is missing"))?;
115+
entry.insert(self.decoder.read(
116+
info.chunk,
117+
chunk.as_slice(),
118+
info.n_values,
119+
buffer.as_slice(),
120+
ctx,
121+
)?)
122+
}
123+
};
124+
Ok(entry)
125+
}
126+
}

0 commit comments

Comments
 (0)