Use PCO logical-page indexes for retained scalar probes

encodings/pco/src/access.rs

@@ -160,6 +160,20 @@ impl AccessState {
             mask: page.mask.mask.slice(start_in_page..end_in_page),
         })
     }
+    pub fn page_id(&self) -> Option<usize> {
+        self.last.as_ref().map(|(page_index, _)| *page_index)
+    }
+    pub fn cached_position(&self, row: usize) -> Option<Option<(usize, usize)>> {
+        let (page_index, page) = self.last.as_ref()?;
+        if !page.rows.contains(&row) {
+            return None;
+        }
+        Some(
+            page.mask
+                .value_index::<true>(row - page.rows.start)
+                .map(|value| (*page_index, value)),
+        )
+    }
 }
 pub(crate) struct PageSpan {
     pub page_index: usize,

encodings/pco/src/array.rs

@@ -66,6 +66,7 @@ use crate::PcoLogicalRowPrefix;
 use crate::PcoMetadata;
 use crate::PcoPageInfo;
 use crate::page_index::PageIndex;
+use crate::probe::PcoProbeState;
 use crate::range_take::slice as decode_slice;
 
 // Overall approach here:
@@ -839,7 +840,7 @@ impl ValidityVTable<Pco> for Pco {
 }
 
 impl OperationsVTable<Pco> for Pco {
-    type ProbeState = crate::probe::PcoProbeState;
+    type ProbeState = PcoProbeState;
 
     fn probe_scalar(
         state: &mut ProbeState<'_, Pco>,

encodings/pco/src/decoder.rs

@@ -116,6 +116,7 @@ pub(crate) struct DecodedPages<T: Number> {
     decoder: PageDecoder<T>,
     lookup: PageLookup,
     pages: BTreeMap<usize, PrimitiveArray>,
+    last: Option<(usize, PrimitiveArray)>,
 }
 
 impl<T: Number + NativePType> DecodedPages<T> {
@@ -124,18 +125,26 @@ impl<T: Number + NativePType> DecodedPages<T> {
             decoder: PageDecoder::new(array.metadata.header.as_slice())?,
             lookup: PageLookup::default(),
             pages: BTreeMap::new(),
+            last: None,
         })
     }
+    pub fn last_values(&self, page: usize) -> Option<&PrimitiveArray> {
+        self.last
+            .as_ref()
+            .filter(|(last, _)| *last == page)
+            .map(|(_, values)| values)
+    }
     pub fn values(
         &mut self,
         view: &LogicalPageIndex,
         array: ArrayView<'_, Pco>,
         page: usize,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<&PrimitiveArray> {
-        let entry = match self.pages.entry(page) {
-            Entry::Occupied(entry) => entry.into_mut(),
-            Entry::Vacant(entry) => {
+        if !self.last.as_ref().is_some_and(|(last, _)| *last == page) {
+            let decoded = if let Some(decoded) = self.pages.remove(&page) {
+                decoded
+            } else {
                 let (chunk_id, n_values) = self.lookup.resolve(view, array, page)?;
                 let buffer = array
                     .pages
@@ -145,15 +154,23 @@ impl<T: Number + NativePType> DecodedPages<T> {
                     .chunk_metas
                     .get(chunk_id)
                     .ok_or_else(|| vortex_err!("Pco chunk metadata is missing"))?;
-                entry.insert(self.decoder.read(
+                self.decoder.read(
                     chunk_id,
                     chunk.as_slice(),
                     n_values,
                     buffer.as_slice(),
                     ctx,
-                )?)
+                )?
+            };
+            if let Some((page, previous)) = self.last.replace((page, decoded)) {
+                self.pages.insert(page, previous);
             }
-        };
+        }
+        let entry = &self
+            .last
+            .as_ref()
+            .vortex_expect("decoded page prepared above")
+            .1;
         Ok(entry)
     }
 }

encodings/pco/src/indexed_probe.rs

@@ -0,0 +1,152 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Operation-owned probes bind indexed or legacy addressing on first use.
+
+use std::collections::BTreeMap;
+use std::mem::replace;
+use std::sync::Arc;
+
+use pco::data_types::Number;
+use pco::data_types::NumberType;
+use pco::match_number_enum;
+use vortex_array::ArrayView;
+use vortex_array::ExecutionCtx;
+use vortex_array::dtype::NativePType;
+use vortex_array::dtype::half;
+use vortex_array::scalar::PValue;
+use vortex_array::scalar::Scalar;
+use vortex_error::VortexExpect;
+use vortex_error::VortexResult;
+
+use crate::Pco;
+use crate::access::AccessState;
+use crate::array::number_type_from_ptype;
+use crate::decoder::DecodedPages;
+use crate::page_index::LogicalPageIndex;
+use crate::page_index::PageIndex;
+
+/// Construction is empty; first use binds this state to one array owner and a typed decoder.
+/// Touched decoded pages, validity masks, and rank tables are retained across probes.
+#[derive(Default)]
+pub(crate) struct IndexedProbeState {
+    bound: Option<Box<dyn BoundProbe>>,
+}
+
+trait BoundProbe: Send + Sync {
+    fn scalar(
+        &mut self,
+        array: ArrayView<'_, Pco>,
+        row: usize,
+        ranked: bool,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Scalar>;
+}
+
+struct TypedProbe<T: Number> {
+    owner: Arc<LogicalPageIndex>,
+    access: AccessState,
+    prepared: BTreeMap<usize, AccessState>,
+    decoded: Option<DecodedPages<T>>,
+}
+
+impl<T> BoundProbe for TypedProbe<T>
+where
+    T: Number + NativePType + Into<PValue>,
+{
+    fn scalar(
+        &mut self,
+        array: ArrayView<'_, Pco>,
+        row: usize,
+        ranked: bool,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Scalar> {
+        if ranked && let Some(position) = self.access.cached_position(row) {
+            match position {
+                None => return Ok(Scalar::null(array.dtype().clone())),
+                Some((page, value)) => {
+                    if let Some(values) = self
+                        .decoded
+                        .as_ref()
+                        .and_then(|decoded| decoded.last_values(page))
+                    {
+                        return Ok(Scalar::primitive(
+                            values.as_slice::<T>()[value],
+                            array.dtype().nullability(),
+                        ));
+                    }
+                }
+            }
+        }
+        let view = self.owner.as_ref();
+        let page = view.find_row(row);
+        if self.access.page_id() != Some(page) {
+            let next = self.prepared.remove(&page).unwrap_or_default();
+            let previous = replace(&mut self.access, next);
+            if let Some(previous_page) = previous.page_id() {
+                self.prepared.insert(previous_page, previous);
+            }
+        }
+        let position = if ranked {
+            self.access.resolve::<true>(view, array, row, ctx)?
+        } else {
+            self.access.resolve::<false>(view, array, row, ctx)?
+        };
+        let Some((page, value)) = position else {
+            return Ok(Scalar::null(array.dtype().clone()));
+        };
+        if self.decoded.is_none() {
+            self.decoded = Some(DecodedPages::new(array)?);
+        }
+        let decoded = self
+            .decoded
+            .as_mut()
+            .vortex_expect("decoder prepared above");
+        let values = decoded.values(view, array, page, ctx)?;
+        Ok(Scalar::primitive(
+            values.as_slice::<T>()[value],
+            array.dtype().nullability(),
+        ))
+    }
+}
+
+fn bind<T: Number + NativePType + Into<PValue>>(owner: Arc<LogicalPageIndex>) -> Box<dyn BoundProbe> {
+    Box::new(TypedProbe::<T> {
+        owner,
+        access: AccessState::default(),
+        prepared: BTreeMap::new(),
+        decoded: None,
+    })
+}
+
+pub(crate) fn scalar_at(
+    array: ArrayView<'_, Pco>,
+    index: usize,
+    state: Option<&mut IndexedProbeState>,
+    ctx: &mut ExecutionCtx,
+) -> VortexResult<Scalar> {
+    // The public probe dispatcher already checks visible bounds. A zero-page array
+    // stores no defined values, so no validity/index/decoder preparation is needed.
+    if array.pages.is_empty() {
+        return Ok(Scalar::null(array.dtype().clone()));
+    }
+    let mut once = IndexedProbeState::default();
+    let ranked = state.is_some();
+    let state = state.unwrap_or(&mut once);
+    if state.bound.is_none() {
+        let PageIndex::IndexedFast(owner) = array.page_index()? else {
+            unreachable!("indexed probes are dispatched only for a present prefix")
+        };
+        state.bound = Some(
+            match_number_enum!(number_type_from_ptype(array.dtype().as_ptype()), NumberType<T> => {
+                bind::<T>(Arc::clone(owner))
+            }),
+        );
+    }
+    state.bound.as_mut().vortex_expect("bound above").scalar(
+        array,
+        array.slice_start() + index,
+        ranked,
+        ctx,
+    )
+}

encodings/pco/src/lib.rs

@@ -31,6 +31,7 @@ mod access;
 mod array;
 mod compute;
 mod decoder;
+mod indexed_probe;
 mod page_index;
 mod probe;
 mod range_take;

encodings/pco/src/probe.rs

@@ -27,16 +27,20 @@ use crate::Pco;
 use crate::PcoArrayExt;
 use crate::array::number_type_from_ptype;
 use crate::array::vortex_err_from_pco;
+use crate::indexed_probe::IndexedProbeState;
+use crate::indexed_probe::scalar_at as indexed_scalar_at;
 
 const RANK_STRIDE: usize = 512;
 
 /// State retained by repeated PCO probes. Construction performs no allocation.
 ///
 /// The mask and rank index use unsliced logical rows; page boundaries use compacted value
 /// positions. Every page decoded so far is retained, so the state grows with the pages touched
-/// and never beyond the decompressed array.
+/// along with the indexed path's prepared validity masks and rank tables. The probe dispatcher
+/// binds this state to one array; callers must not reuse it with another array.
 #[derive(Default)]
 pub struct PcoProbeState {
+    indexed: IndexedProbeState,
     validity: Option<Mask>,
     rank: Vec<usize>,
     pages: Vec<Page>,
@@ -56,6 +60,9 @@ pub(crate) fn scalar_at(
     state: Option<&mut PcoProbeState>,
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<Scalar> {
+    if array.metadata.logical_row_prefix.is_some() {
+        return indexed_scalar_at(array, index, state.map(|state| &mut state.indexed), ctx);
+    }
     let Some(state) = state else {
         let validity = array.unsliced_validity();
         return array