uio mmap posting (#8721)

* [AI] update docstrings

* move support strucutres from mmap_posting.rs

* rename and simplify: MmapPostingValue -> PersistedPostingValue -> ZerocopyPostingValue

* implement `with_view` method for UniversalPostings

* WIP: RawPOstingList + AsPostingListView

* wip: bathcing and pre-collection of posting lists

* wip: batch reading in UniversalPostings

* [AI] implement with_existing_postings

* fmt

* pre-collect in check_compressed_postings_phrase

* batch pre-collect for check_any and check_intersection

* get rid of iter_postings which is collected anyway

* all_postings in UniversalPostings

* missing unique for tokens in phrases

* [AI] replace MmapPostings with UniversalPostings

* mising file

* cfg(test)

* unnecessary Cow

---------

Co-authored-by: Luis Cossío <luis.cossio@outlook.com>

lib/common/common/src/universal_io/error.rs

@@ -12,6 +12,9 @@ pub enum UniversalIoError {
     #[error("Bytemuck cast error: {0:?}")]
     BytemuckCast(bytemuck::PodCastError),
 
+    #[error("Zerocopy size error: {0:?}")]
+    ZerocopySize(String),
+
     #[error(transparent)]
     IoUringNotSupported(io::Error),
 
@@ -59,3 +62,9 @@ impl From<bytemuck::PodCastError> for UniversalIoError {
         Self::BytemuckCast(err)
     }
 }
+
+impl<Src, Dst: ?Sized> From<zerocopy::SizeError<Src, Dst>> for UniversalIoError {
+    fn from(err: zerocopy::SizeError<Src, Dst>) -> Self {
+        Self::ZerocopySize(format!("{err:?}"))
+    }
+}

lib/posting_list/src/lib.rs

@@ -12,7 +12,7 @@ use bitpacking::BitPacker;
 type BitPackerImpl = bitpacking::BitPacker4x;
 
 /// How many elements are packed in a single chunk.
-const CHUNK_LEN: usize = 128;
+pub const CHUNK_LEN: usize = 128;
 const _: () = assert!(128 == BitPackerImpl::BLOCK_LEN);
 
 pub trait SizedValue: Sized + Copy + std::fmt::Debug {}

lib/posting_list/src/view.rs

@@ -12,6 +12,7 @@ use crate::visitor::PostingVisitor;
 use crate::{BitPackerImpl, CHUNK_LEN, PostingChunk, PostingList, SizedTypeFor};
 
 /// A non-owning view of [`PostingList`].
+#[derive(Clone)]
 pub struct PostingListView<'a, V: PostingValue> {
     pub(crate) id_data: &'a [u8],
     chunks: &'a [PostingChunk<SizedTypeFor<V>>],

lib/segment/src/common/operation_error.rs

@@ -162,10 +162,29 @@ impl From<UniversalIoError> for OperationError {
             | UniversalIoError::InvalidFileIndex { .. } => Self::service_error(err.to_string()),
             UniversalIoError::BytemuckCast(_) => Self::service_error(err.to_string()),
             UniversalIoError::Uninitialized { .. } => Self::service_error(err.to_string()),
+            UniversalIoError::ZerocopySize(_) => Self::service_error(err.to_string()),
         }
     }
 }
 
+impl<Src, Dst: ?Sized> From<zerocopy::SizeError<Src, Dst>> for OperationError
+where
+    zerocopy::SizeError<Src, Dst>: std::fmt::Display,
+{
+    fn from(err: zerocopy::SizeError<Src, Dst>) -> Self {
+        Self::service_error(format!("Zerocopy size error: {err}"))
+    }
+}
+
+impl<A, S, V> From<zerocopy::ConvertError<A, S, V>> for OperationError
+where
+    zerocopy::ConvertError<A, S, V>: std::fmt::Display,
+{
+    fn from(err: zerocopy::ConvertError<A, S, V>) -> Self {
+        Self::service_error(format!("Zerocopy convert error: {err}"))
+    }
+}
+
 impl From<serde_cbor::Error> for OperationError {
     fn from(err: serde_cbor::Error) -> Self {
         Self::service_error(format!("Failed to parse data: {err}"))

lib/segment/src/index/field_index/full_text_index/immutable_text_index.rs

@@ -22,8 +22,8 @@ pub(super) enum Storage {
 
 impl ImmutableFullTextIndex {
     /// Open and load immutable full text index from mmap storage
-    pub fn open_mmap(index: MmapFullTextIndex) -> Self {
-        let inverted_index = ImmutableInvertedIndex::from(&index.inverted_index);
+    pub fn open_mmap(index: MmapFullTextIndex) -> OperationResult<Self> {
+        let inverted_index = ImmutableInvertedIndex::try_from(&index.inverted_index)?;
 
         // Index is now loaded into memory, clear cache of backing mmap storage
         if let Err(err) = index.clear_cache() {
@@ -36,7 +36,7 @@ impl ImmutableFullTextIndex {
             cached_ram_usage_bytes: 0,
         };
         result.cached_ram_usage_bytes = result.inverted_index.ram_usage_bytes();
-        result
+        Ok(result)
     }
 
     pub fn remove_point(&mut self, id: PointOffsetType) {

lib/segment/src/index/field_index/full_text_index/inverted_index/immutable_inverted_index.rs

@@ -21,6 +21,22 @@ use crate::index::field_index::full_text_index::inverted_index::postings_iterato
     check_compressed_postings_phrase, intersect_compressed_postings_phrase_iterator,
 };
 
+/// Collect posting-list views for every token in `token_ids`.
+/// Returns `None` as soon as any token id is out of range.
+fn get_all_or_none<'a, V: PostingValue>(
+    postings: &'a [PostingList<V>],
+    token_ids: &[TokenId],
+) -> Option<Vec<(TokenId, PostingListView<'a, V>)>> {
+    token_ids
+        .iter()
+        .map(|&token_id| {
+            postings
+                .get(token_id as usize)
+                .map(|list| (token_id, list.view()))
+        })
+        .collect()
+}
+
 #[cfg_attr(test, derive(Clone))]
 #[derive(Debug)]
 pub struct ImmutableInvertedIndex {
@@ -188,11 +204,19 @@ impl ImmutableInvertedIndex {
 
         match &self.postings {
             ImmutablePostings::WithPositions(postings) => {
-                Either::Right(intersect_compressed_postings_phrase_iterator(
-                    phrase,
-                    |token_id| postings.get(*token_id as usize).map(PostingList::view),
-                    is_active,
-                ))
+                // Deduplicate phrase tokens: repeated tokens (e.g. "zn zn") must
+                // not fetch the same posting list twice, otherwise positions get
+                // added twice in `phrase_in_all_postings`.
+                let unique_tokens = phrase.to_token_set();
+                if let Some(selected_postings) = get_all_or_none(postings, unique_tokens.tokens()) {
+                    Either::Right(intersect_compressed_postings_phrase_iterator(
+                        phrase,
+                        selected_postings,
+                        is_active,
+                    ))
+                } else {
+                    Either::Left(std::iter::empty())
+                }
             }
             // cannot do phrase matching if there's no positional information
             ImmutablePostings::Ids(_postings) => Either::Left(std::iter::empty()),
@@ -212,9 +236,13 @@ impl ImmutableInvertedIndex {
 
         match &self.postings {
             ImmutablePostings::WithPositions(postings) => {
-                check_compressed_postings_phrase(phrase, point_id, |token_id| {
-                    postings.get(*token_id as usize).map(PostingList::view)
-                })
+                let unique_tokens = phrase.to_token_set();
+                let Some(selected_postings) = get_all_or_none(postings, unique_tokens.tokens())
+                else {
+                    return false;
+                };
+
+                check_compressed_postings_phrase(phrase, point_id, selected_postings)
             }
             // cannot do phrase matching if there's no positional information
             ImmutablePostings::Ids(_postings) => false,
@@ -270,24 +298,35 @@ impl InvertedIndex for ImmutableInvertedIndex {
         }
     }
 
-    fn get_posting_len(&self, token_id: TokenId, _: &HardwareCounterCell) -> Option<usize> {
-        self.postings.posting_len(token_id)
+    fn get_posting_len(
+        &self,
+        token_id: TokenId,
+        _: &HardwareCounterCell,
+    ) -> OperationResult<Option<usize>> {
+        Ok(self.postings.posting_len(token_id))
     }
 
-    fn vocab_with_postings_len_iter(&self) -> impl Iterator<Item = (&str, usize)> + '_ {
+    fn vocab_with_postings_len_iter(
+        &self,
+    ) -> impl Iterator<Item = OperationResult<(&str, usize)>> + '_ {
         self.vocab.iter().filter_map(|(token, &token_id)| {
             self.postings
                 .posting_len(token_id)
-                .map(|len| (token.as_str(), len))
+                .map(|len| Ok((token.as_str(), len)))
         })
     }
 
-    fn check_match(&self, parsed_query: &ParsedQuery, point_id: PointOffsetType) -> bool {
-        match parsed_query {
+    fn check_match(
+        &self,
+        parsed_query: &ParsedQuery,
+        point_id: PointOffsetType,
+    ) -> OperationResult<bool> {
+        let matched = match parsed_query {
             ParsedQuery::AllTokens(tokens) => self.check_has_subset(tokens, point_id),
             ParsedQuery::Phrase(phrase) => self.check_has_phrase(phrase, point_id),
             ParsedQuery::AnyTokens(tokens) => self.check_has_any(tokens, point_id),
-        }
+        };
+        Ok(matched)
     }
 
     fn values_is_empty(&self, point_id: PointOffsetType) -> bool {
@@ -446,21 +485,15 @@ fn create_compressed_postings_with_positions(
             .collect()
 }
 
-impl From<&MmapInvertedIndex> for ImmutableInvertedIndex {
-    fn from(index: &MmapInvertedIndex) -> Self {
+impl TryFrom<&MmapInvertedIndex> for ImmutableInvertedIndex {
+    type Error = OperationError;
+
+    fn try_from(index: &MmapInvertedIndex) -> OperationResult<Self> {
         let postings = match &index.storage.postings {
-            MmapPostingsEnum::Ids(postings) => ImmutablePostings::Ids(
-                postings
-                    .iter_postings()
-                    .map(PostingListView::to_owned)
-                    .collect(),
-            ),
-            MmapPostingsEnum::WithPositions(postings) => ImmutablePostings::WithPositions(
-                postings
-                    .iter_postings()
-                    .map(PostingListView::to_owned)
-                    .collect(),
-            ),
+            MmapPostingsEnum::Ids(postings) => ImmutablePostings::Ids(postings.all_postings()?),
+            MmapPostingsEnum::WithPositions(postings) => {
+                ImmutablePostings::WithPositions(postings.all_postings()?)
+            }
         };
 
         let vocab: HashMap<String, TokenId> = index
@@ -475,12 +508,12 @@ impl From<&MmapInvertedIndex> for ImmutableInvertedIndex {
             "postings and vocab must be the same size",
         );
 
-        ImmutableInvertedIndex {
+        Ok(ImmutableInvertedIndex {
             postings,
             vocab,
             point_to_tokens_count: index.storage.point_to_tokens_count.to_vec(),
             points_count: index.points_count(),
-        }
+        })
     }
 }
 

lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/create_postings.rs

@@ -0,0 +1,169 @@
+use std::io;
+use std::io::Write;
+use std::path::{Path, PathBuf};
+
+use common::zeros::WriteZerosExt;
+use fs_err::File;
+use posting_list::{PostingList, PostingListComponents};
+use zerocopy::IntoBytes;
+
+use crate::index::field_index::full_text_index::inverted_index::mmap_inverted_index::types::{
+    ALIGNMENT, PostingListHeader, PostingsHeader, ZerocopyPostingValue,
+};
+
+/// Write a vector of compressed posting lists to `path` in the format expected
+/// by [`UniversalPostings`](super::uio_postings::UniversalPostings).
+///
+/// On-disk layout:
+///
+/// ```text
+/// offset 0
+/// ┌────────────────────────────────────────────────┐
+/// │ PostingsHeader                                 │  size_of::<PostingsHeader>()
+/// │   posting_count: usize    (one per term)       │  = 8 + 32 = 40 bytes
+/// │   _reserved: [u8; 32]                          │
+/// ├────────────────────────────────────────────────┤
+/// │ PostingListHeader[0]   ── token_id 0           │  each
+/// │ PostingListHeader[1]   ── token_id 1           │  size_of::<PostingListHeader>()
+/// │ ...                                            │  = 24 bytes
+/// │ PostingListHeader[N-1] ── token_id N-1         │
+/// ├────────────────────────────────────────────────┤
+/// │ CompressedPostingList[0]                       │  variable
+/// │ CompressedPostingList[1]                       │  size; located via
+/// │ ...                                            │  header.offset
+/// │ CompressedPostingList[N-1]                     │
+/// └────────────────────────────────────────────────┘
+/// ```
+///
+/// One `CompressedPostingList`:
+///
+/// ```text
+/// header.offset ─►
+/// ┌──────────────────────────────────────────────┐
+/// │ last_doc_id : PointOffsetType (u32)          │  4 B
+/// ├──────────────────────────────────────────────┤
+/// │ chunks: [PostingChunk<Sized<V>>;             │  chunks_count *
+/// │          chunks_count]                       │  size_of::<PostingChunk>
+/// │   each chunk = (initial_id, sized_value,     │
+/// │                 offset into id_data)         │
+/// ├──────────────────────────────────────────────┤
+/// │ id_data: [u8; ids_data_bytes_count]          │  bit-packed deltas for
+/// │                                              │  CHUNK_LEN ids per chunk
+/// ├──────────────────────────────────────────────┤
+/// │ var_size_data: [u8; var_size_data_bytes]     │  optional (Positions);
+/// │                                              │  empty for V = ()
+/// ├──────────────────────────────────────────────┤
+/// │ alignment: [0u8; alignment_bytes_count]      │  pad so next region is
+/// │                                              │  4-byte aligned
+/// ├──────────────────────────────────────────────┤
+/// │ remainders:[RemainderPosting<Sized<V>>;      │  tail < CHUNK_LEN ids
+/// │             remainder_count]                 │  stored uncompressed
+/// └──────────────────────────────────────────────┘
+/// ```
+///
+/// The alignment slot exists because `id_data` + `var_size_data` are byte
+/// streams of arbitrary length, but `RemainderPosting` is `#[repr(C)]` and
+/// read via zerocopy, so its start needs 4-byte alignment.
+///
+/// Two flavors of `V`:
+/// - `V = ()` — ids-only postings; `var_size_data` is empty and
+///   `SizedTypeFor<V>` is zero-sized.
+/// - `V = Positions` — phrase index; per-doc token positions live in
+///   `var_size_data`, with each `PostingChunk` / `RemainderPosting` carrying
+///   a `Sized` handle (offset/len) into that blob.
+pub fn create_postings_file<V: ZerocopyPostingValue>(
+    path: PathBuf,
+    compressed_postings: &[PostingList<V>],
+) -> io::Result<()> {
+    // Create a new empty file, where we will write the compressed posting lists and the header
+    let (file, temp_path) = tempfile::Builder::new()
+        .prefix(path.file_name().ok_or(io::ErrorKind::InvalidInput)?)
+        .tempfile_in(path.parent().ok_or(io::ErrorKind::InvalidInput)?)?
+        .into_parts();
+    let file = File::from_parts::<&Path>(file, temp_path.as_ref());
+    let mut bufw = io::BufWriter::new(&file);
+
+    let postings_header = PostingsHeader {
+        posting_count: compressed_postings.len(),
+        _reserved: [0; 32],
+    };
+
+    bufw.write_all(postings_header.as_bytes())?;
+
+    let postings_lists_headers_size = compressed_postings.len() * size_of::<PostingListHeader>();
+    let mut posting_offset = size_of::<PostingsHeader>() + postings_lists_headers_size;
+
+    for compressed_posting in compressed_postings {
+        let view = compressed_posting.view();
+        let PostingListComponents {
+            id_data,
+            chunks,
+            var_size_data,
+            remainders,
+            last_id: _, // not used for the header
+        } = view.components();
+
+        let id_data_len = id_data.len();
+        let var_size_data_len = var_size_data.len();
+        let data_len = id_data_len + var_size_data_len;
+        let alignment_len = data_len.next_multiple_of(ALIGNMENT) - data_len;
+
+        let posting_list_header = PostingListHeader {
+            offset: posting_offset as u64,
+            chunks_count: chunks.len() as u32,
+            ids_data_bytes_count: id_data_len as u32,
+            var_size_data_bytes_count: var_size_data_len as u32,
+            alignment_bytes_count: alignment_len as u8,
+            remainder_count: remainders.len() as u8,
+            _reserved: [0; 2],
+        };
+
+        bufw.write_all(posting_list_header.as_bytes())?;
+
+        posting_offset += posting_list_header.posting_size::<V>();
+    }
+
+    for compressed_posting in compressed_postings {
+        let view = compressed_posting.view();
+        let PostingListComponents {
+            id_data,
+            chunks,
+            var_size_data,
+            remainders,
+            last_id,
+        } = view.components();
+
+        bufw.write_all(
+            last_id
+                .expect("posting must have at least one element")
+                .as_bytes(),
+        )?;
+
+        for chunk in chunks {
+            bufw.write_all(chunk.as_bytes())?;
+        }
+
+        bufw.write_all(id_data)?;
+
+        if !var_size_data.is_empty() {
+            bufw.write_all(var_size_data)?;
+        }
+
+        // alignment padding so the next `RemainderPosting` is 4-byte aligned
+        let data_len = id_data.len() + var_size_data.len();
+        bufw.write_zeros(data_len.next_multiple_of(ALIGNMENT) - data_len)?;
+
+        for element in remainders {
+            bufw.write_all(element.as_bytes())?;
+        }
+    }
+
+    bufw.flush()?;
+    drop(bufw);
+
+    file.sync_all()?;
+    drop(file);
+    temp_path.persist(path)?;
+
+    Ok(())
+}

lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/mmap_postings.rs

@@ -1,304 +0,0 @@
-use std::fmt::Debug;
-use std::io;
-use std::io::Write;
-use std::marker::PhantomData;
-use std::path::{Path, PathBuf};
-
-use common::mmap::{Advice, AdviceSetting, Madviseable, open_read_mmap};
-use common::types::PointOffsetType;
-use common::zeros::WriteZerosExt;
-use fs_err::File;
-use memmap2::Mmap;
-use posting_list::{
-    PostingChunk, PostingList, PostingListComponents, PostingListView, PostingValue,
-    RemainderPosting, SizedTypeFor, ValueHandler,
-};
-use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout, Unaligned};
-
-use crate::index::field_index::full_text_index::inverted_index::TokenId;
-use crate::index::field_index::full_text_index::inverted_index::positions::Positions;
-
-const ALIGNMENT: usize = 4;
-
-/// Trait marker to enrich [`posting_list::PostingValue`] for handling mmap files with the posting list.
-pub(in crate::index::field_index::full_text_index) trait MmapPostingValue:
-    PostingValue<
-    Handler: ValueHandler<Sized: FromBytes + Immutable + IntoBytes + KnownLayout + Unaligned>
-                 + Clone
-                 + Debug,
->
-{
-}
-
-impl MmapPostingValue for () {}
-
-impl MmapPostingValue for Positions {}
-
-#[derive(Debug, Default, Clone, FromBytes, Immutable, IntoBytes, KnownLayout)]
-#[repr(C)]
-struct PostingsHeader {
-    /// Number of posting lists. One posting list per term
-    pub posting_count: usize,
-    _reserved: [u8; 32],
-}
-
-/// This data structure should contain all the necessary information to
-/// construct `PostingListView<V>` from the mmap file.
-#[derive(Debug, Default, Clone, FromBytes, Immutable, IntoBytes, KnownLayout)]
-#[repr(C)]
-pub(in crate::index::field_index::full_text_index) struct PostingListHeader {
-    /// Offset in bytes from the start of the mmap file
-    /// where the posting list data starts
-    offset: u64,
-    /// Amount of chunks in compressed posting list
-    chunks_count: u32,
-    /// Length in bytes for the compressed postings data
-    ids_data_bytes_count: u32,
-    /// Length in bytes for the alignment bytes
-    alignment_bytes_count: u8,
-    /// Length in bytes for the remainder postings
-    remainder_count: u8,
-
-    _reserved: [u8; 2],
-
-    /// Length in bytes for the var-sized data. Add-on for phrase matching, otherwise 0
-    var_size_data_bytes_count: u32,
-}
-
-impl PostingListHeader {
-    fn posting_size<V: PostingValue>(&self) -> usize {
-        self.ids_data_bytes_count as usize
-            + self.var_size_data_bytes_count as usize
-            + self.alignment_bytes_count as usize
-            + self.remainder_count as usize * size_of::<RemainderPosting<SizedTypeFor<V>>>()
-            + self.chunks_count as usize * size_of::<PostingChunk<SizedTypeFor<V>>>()
-            + size_of::<PointOffsetType>() // last_doc_id
-    }
-}
-
-/// MmapPostings Structure on disk:
-///
-///
-/// `| PostingsHeader |
-/// [ PostingListHeader, PostingListHeader, ... ] |
-/// [ CompressedMmapPostingList, CompressedMmapPostingList, ... ] |`
-pub struct MmapPostings<V: MmapPostingValue> {
-    _path: PathBuf,
-    mmap: Mmap,
-    header: PostingsHeader,
-    _value_type: PhantomData<V>,
-}
-
-impl<V: MmapPostingValue> MmapPostings<V> {
-    fn get_header(&self, token_id: TokenId) -> Option<&PostingListHeader> {
-        if self.header.posting_count <= token_id as usize {
-            return None;
-        }
-
-        let header_offset =
-            size_of::<PostingsHeader>() + token_id as usize * size_of::<PostingListHeader>();
-
-        PostingListHeader::ref_from_prefix(self.mmap.get(header_offset..)?)
-            .ok()
-            .map(|(header, _)| header)
-    }
-
-    /// Create PostingListView<V> from the given header
-    ///
-    /// Assume the following layout:
-    ///
-    /// ```ignore
-    /// last_doc_id: &'a PointOffsetType,
-    /// chunks_index: &'a [PostingChunk<()>],
-    /// data: &'a [u8],
-    /// var_size_data: &'a [u8], // might be empty in case of only ids
-    /// _alignment: &'a [u8], // 0-3 extra bytes to align the data
-    /// remainder_postings: &'a [PointOffsetType],
-    /// ```
-    fn get_view<'a>(&'a self, header: &'a PostingListHeader) -> Option<PostingListView<'a, V>> {
-        let bytes = self.mmap.get(header.offset as usize..)?;
-
-        let (last_doc_id, bytes) = PointOffsetType::read_from_prefix(bytes).ok()?;
-
-        let (chunks, bytes) = <[PostingChunk<SizedTypeFor<V>>]>::ref_from_prefix_with_elems(
-            bytes,
-            header.chunks_count as usize,
-        )
-        .ok()?;
-
-        let (id_data, bytes) = bytes.split_at(header.ids_data_bytes_count as usize);
-
-        let (var_size_data, bytes) = bytes.split_at(header.var_size_data_bytes_count as usize);
-
-        // skip padding
-        let bytes = bytes.get(header.alignment_bytes_count as usize..)?;
-
-        let (remainder_postings, _) =
-            <[RemainderPosting<SizedTypeFor<V>>]>::ref_from_prefix_with_elems(
-                bytes,
-                header.remainder_count as usize,
-            )
-            .ok()?;
-
-        Some(PostingListView::from_components(
-            id_data,
-            chunks,
-            var_size_data,
-            remainder_postings,
-            Some(last_doc_id),
-        ))
-    }
-
-    pub fn get<'a>(&'a self, token_id: TokenId) -> Option<PostingListView<'a, V>> {
-        let header = self.get_header(token_id)?;
-        self.get_view(header)
-    }
-
-    /// Given a vector of compressed posting lists, this function writes them to the `path` file.
-    /// The format of the file is compatible with the `MmapPostings` structure.
-    pub fn create(path: PathBuf, compressed_postings: &[PostingList<V>]) -> io::Result<()> {
-        // Create a new empty file, where we will write the compressed posting lists and the header
-        let (file, temp_path) = tempfile::Builder::new()
-            .prefix(path.file_name().ok_or(io::ErrorKind::InvalidInput)?)
-            .tempfile_in(path.parent().ok_or(io::ErrorKind::InvalidInput)?)?
-            .into_parts();
-        let file = File::from_parts::<&Path>(file, temp_path.as_ref());
-        let mut bufw = io::BufWriter::new(&file);
-
-        let postings_header = PostingsHeader {
-            posting_count: compressed_postings.len(),
-            _reserved: [0; 32],
-        };
-
-        // Write the header to the buffer
-        bufw.write_all(postings_header.as_bytes())?;
-
-        let postings_lists_headers_size =
-            compressed_postings.len() * size_of::<PostingListHeader>();
-        let mut posting_offset = size_of::<PostingsHeader>() + postings_lists_headers_size;
-
-        for compressed_posting in compressed_postings {
-            let view = compressed_posting.view();
-            let PostingListComponents {
-                id_data,
-                chunks,
-                var_size_data,
-                remainders,
-                last_id: _, // not used for the header
-            } = view.components();
-
-            let id_data_len = id_data.len();
-            let var_size_data_len = var_size_data.len();
-            let data_len = id_data_len + var_size_data_len;
-            let alignment_len = data_len.next_multiple_of(ALIGNMENT) - data_len;
-
-            let posting_list_header = PostingListHeader {
-                offset: posting_offset as u64,
-                chunks_count: chunks.len() as u32,
-                ids_data_bytes_count: id_data_len as u32,
-                var_size_data_bytes_count: var_size_data_len as u32,
-                alignment_bytes_count: alignment_len as u8,
-                remainder_count: remainders.len() as u8,
-                _reserved: [0; 2],
-            };
-
-            // Write the posting list header to the buffer
-            bufw.write_all(posting_list_header.as_bytes())?;
-
-            posting_offset += posting_list_header.posting_size::<V>();
-        }
-
-        for compressed_posting in compressed_postings {
-            let view = compressed_posting.view();
-            let PostingListComponents {
-                id_data,
-                chunks,
-                var_size_data, // not used with just ids postings
-                remainders,
-                last_id,
-            } = view.components();
-
-            bufw.write_all(
-                last_id
-                    .expect("posting must have at least one element")
-                    .as_bytes(),
-            )?;
-
-            for chunk in chunks {
-                bufw.write_all(chunk.as_bytes())?;
-            }
-
-            // write all unaligned data together
-            bufw.write_all(id_data)?;
-
-            // write var_size_data if it exists
-            if !var_size_data.is_empty() {
-                bufw.write_all(var_size_data)?;
-            }
-
-            // write alignment padding
-            // Example:
-            // For data size = 5, alignment = 3 as (5 + 3 = 8)
-            // alignment = 8 - 5 = 3
-            let data_len = id_data.len() + var_size_data.len();
-            bufw.write_zeros(data_len.next_multiple_of(ALIGNMENT) - data_len)?;
-
-            for element in remainders {
-                bufw.write_all(element.as_bytes())?;
-            }
-        }
-
-        // Explicitly flush write buffer so we can catch IO errors
-        bufw.flush()?;
-        drop(bufw);
-
-        file.sync_all()?;
-        drop(file);
-        temp_path.persist(path)?;
-
-        Ok(())
-    }
-
-    pub fn open(path: impl Into<PathBuf>, populate: bool) -> io::Result<Self> {
-        let path = path.into();
-        let mmap = open_read_mmap(&path, AdviceSetting::Advice(Advice::Normal), populate)?;
-
-        let (header, _) = PostingsHeader::read_from_prefix(&mmap).map_err(|_| {
-            io::Error::new(
-                io::ErrorKind::InvalidData,
-                format!("Invalid header deserialization in {}", path.display()),
-            )
-        })?;
-
-        Ok(Self {
-            _path: path,
-            mmap,
-            header,
-            _value_type: PhantomData,
-        })
-    }
-
-    /// Populate all pages in the mmap.
-    /// Block until all pages are populated.
-    pub fn populate(&self) {
-        self.mmap.populate();
-    }
-
-    /// Hint to the OS that pages backing this mmap can be reclaimed.
-    pub fn clear_cache(&self) {
-        let Self {
-            _path,
-            mmap,
-            header: _,
-            _value_type,
-        } = self;
-        mmap.clear_cache();
-    }
-
-    /// Iterate over posting lists, returning a view for each
-    pub fn iter_postings<'a>(&'a self) -> impl Iterator<Item = PostingListView<'a, V>> {
-        (0..self.header.posting_count as u32)
-            // we are iterating over existing posting lists, all of them should return `Some`
-            .filter_map(|posting_idx| self.get(posting_idx))
-    }
-}

lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/mmap_postings_enum.rs

@@ -1,36 +1,36 @@
 #[cfg(test)]
 use common::types::PointOffsetType;
+use common::universal_io::MmapFile;
 
-use crate::index::field_index::full_text_index::inverted_index::TokenId;
-use crate::index::field_index::full_text_index::inverted_index::mmap_inverted_index::mmap_postings::MmapPostings;
 use super::super::positions::Positions;
+use crate::common::operation_error::OperationResult;
+use crate::index::field_index::full_text_index::inverted_index::TokenId;
+use crate::index::field_index::full_text_index::inverted_index::mmap_inverted_index::uio_postings::UniversalPostings;
 
 pub enum MmapPostingsEnum {
-    Ids(MmapPostings<()>),
-    WithPositions(MmapPostings<Positions>),
+    Ids(UniversalPostings<(), MmapFile>),
+    WithPositions(UniversalPostings<Positions, MmapFile>),
 }
 
 impl MmapPostingsEnum {
-    pub fn populate(&self) {
+    pub fn populate(&self) -> OperationResult<()> {
         match self {
             MmapPostingsEnum::Ids(postings) => postings.populate(),
             MmapPostingsEnum::WithPositions(postings) => postings.populate(),
         }
     }
 
-    pub fn clear_cache(&self) {
+    pub fn clear_cache(&self) -> OperationResult<()> {
         match self {
             MmapPostingsEnum::Ids(postings) => postings.clear_cache(),
             MmapPostingsEnum::WithPositions(postings) => postings.clear_cache(),
         }
     }
 
-    pub fn posting_len(&self, token_id: TokenId) -> Option<usize> {
+    pub fn posting_len(&self, token_id: TokenId) -> OperationResult<Option<usize>> {
         match self {
-            MmapPostingsEnum::Ids(postings) => postings.get(token_id).map(|view| view.len()),
-            MmapPostingsEnum::WithPositions(postings) => {
-                postings.get(token_id).map(|view| view.len())
-            }
+            MmapPostingsEnum::Ids(postings) => postings.posting_len(token_id),
+            MmapPostingsEnum::WithPositions(postings) => postings.posting_len(token_id),
         }
     }
 
@@ -39,15 +39,20 @@ impl MmapPostingsEnum {
         &'a self,
         token_id: TokenId,
     ) -> Option<Box<dyn Iterator<Item = PointOffsetType> + 'a>> {
-        match self {
-            MmapPostingsEnum::Ids(postings) => postings.get(token_id).map(|view| {
-                Box::new(view.into_iter().map(|elem| elem.id))
-                    as Box<dyn Iterator<Item = PointOffsetType>>
-            }),
-            MmapPostingsEnum::WithPositions(postings) => postings.get(token_id).map(|view| {
-                Box::new(view.into_iter().map(|elem| elem.id))
-                    as Box<dyn Iterator<Item = PointOffsetType>>
-            }),
-        }
+        // Collect ids upfront so the borrowed `RawPostingList` bytes don't have
+        // to outlive this call. Acceptable because UniversalPostings is on disk.
+        let ids: Vec<PointOffsetType> = match self {
+            MmapPostingsEnum::Ids(postings) => {
+                let raw = postings.get(token_id).unwrap()?;
+                let view = raw.as_view::<()>().unwrap();
+                view.into_iter().map(|elem| elem.id).collect()
+            }
+            MmapPostingsEnum::WithPositions(postings) => {
+                let raw = postings.get(token_id).unwrap()?;
+                let view = raw.as_view::<Positions>().unwrap();
+                view.into_iter().map(|elem| elem.id).collect()
+            }
+        };
+        Some(Box::new(ids.into_iter()))
     }
 }

lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/mod.rs

@@ -3,13 +3,14 @@ use std::path::PathBuf;
 
 use common::bitvec::BitVec;
 use common::counter::hardware_counter::HardwareCounterCell;
-use common::mmap::{self, AdviceSetting, MmapSlice, create_and_ensure_length};
+use common::mmap::{self, Advice, AdviceSetting, MmapSlice, create_and_ensure_length};
 use common::mmap_hashmap::{MmapHashMap, READ_ENTRY_OVERHEAD};
 use common::types::PointOffsetType;
 use common::universal_io::{MmapFile, OpenOptions};
-use itertools::Either;
-use mmap_postings::{MmapPostingValue, MmapPostings};
+use types::ZerocopyPostingValue;
+use uio_postings::UniversalPostings;
 
+use self::create_postings::create_postings_file;
 use super::immutable_inverted_index::ImmutableInvertedIndex;
 use super::immutable_postings_enum::ImmutablePostings;
 use super::mmap_inverted_index::mmap_postings_enum::MmapPostingsEnum;
@@ -27,8 +28,11 @@ use crate::index::field_index::full_text_index::inverted_index::postings_iterato
     check_compressed_postings_phrase, intersect_compressed_postings_phrase_iterator,
 };
 
-pub(super) mod mmap_postings;
+mod create_postings;
 pub mod mmap_postings_enum;
+mod raw_posting_list;
+pub(in crate::index::field_index::full_text_index) mod types;
+mod uio_postings;
 
 const POSTINGS_FILE: &str = "postings.dat";
 const VOCAB_FILE: &str = "vocab.dat";
@@ -68,9 +72,9 @@ impl MmapInvertedIndex {
         let deleted_points_path = path.join(DELETED_POINTS_FILE);
 
         match postings {
-            ImmutablePostings::Ids(postings) => MmapPostings::create(postings_path, postings)?,
+            ImmutablePostings::Ids(postings) => create_postings_file(postings_path, postings)?,
             ImmutablePostings::WithPositions(postings) => {
-                MmapPostings::create(postings_path, postings)?
+                create_postings_file(postings_path, postings)?
             }
         }
 
@@ -126,12 +130,25 @@ impl MmapInvertedIndex {
             return Ok(None);
         }
 
+        let postings_open_options = OpenOptions {
+            writeable: false,
+            need_sequential: false,
+            disk_parallel: None,
+            populate: Some(populate),
+            advice: Some(AdviceSetting::Advice(Advice::Normal)),
+            prevent_caching: None,
+        };
         let postings = match has_positions {
-            false => MmapPostingsEnum::Ids(MmapPostings::<()>::open(&postings_path, populate)?),
-            true => MmapPostingsEnum::WithPositions(MmapPostings::<Positions>::open(
+            false => MmapPostingsEnum::Ids(UniversalPostings::<(), MmapFile>::open(
                 &postings_path,
-                populate,
+                postings_open_options,
             )?),
+            true => {
+                MmapPostingsEnum::WithPositions(UniversalPostings::<Positions, MmapFile>::open(
+                    &postings_path,
+                    postings_open_options,
+                )?)
+            }
         };
         let vocab = MmapHashMap::<str, TokenId>::open(&vocab_path, false)?;
 
@@ -185,39 +202,33 @@ impl MmapInvertedIndex {
         !is_deleted
     }
 
-    /// Iterate over point ids whose documents contain all given tokens
-    pub fn filter_has_all<'a>(
-        &'a self,
-        tokens: TokenSet,
-    ) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
+    /// Iterate over point ids whose documents contain all given tokens.
+    ///
+    /// Pre-collected upfront because [`UniversalPostings`] exposes posting
+    /// views via a `FnOnce` callback. Acceptable since this index lives
+    /// on disk.
+    pub fn filter_has_all(&self, tokens: TokenSet) -> OperationResult<Vec<PointOffsetType>> {
         // in case of mmap immutable index, deleted points are still in the postings
         let filter = move |idx| self.is_active(idx);
 
-        fn intersection<'a, V: MmapPostingValue>(
-            postings: &'a MmapPostings<V>,
+        fn intersection<V: ZerocopyPostingValue>(
+            postings: &UniversalPostings<V, MmapFile>,
             tokens: TokenSet,
-            filter: impl Fn(u32) -> bool + 'a,
-        ) -> Box<dyn Iterator<Item = PointOffsetType> + 'a> {
-            let postings_opt: Option<Vec<_>> = tokens
-                .tokens()
-                .iter()
-                .map(|&token_id| postings.get(token_id))
-                .collect();
-
-            let Some(posting_readers) = postings_opt else {
-                // There are unseen tokens -> no matches
-                return Box::new(std::iter::empty());
-            };
-
-            if posting_readers.is_empty() {
-                // Empty request -> no matches
-                return Box::new(std::iter::empty());
-            }
-
-            Box::new(intersect_compressed_postings_iterator(
-                posting_readers,
-                filter,
-            ))
+            filter: impl Fn(PointOffsetType) -> bool,
+        ) -> OperationResult<Vec<PointOffsetType>> {
+            let result =
+                postings.with_all_or_none_postings(tokens.tokens(), |posting_readers| {
+                    if posting_readers.is_empty() {
+                        return Ok(Vec::new());
+                    }
+                    let posting_readers = posting_readers
+                        .into_iter()
+                        .map(|(_token_id, posting_list_view)| posting_list_view)
+                        .collect();
+                    Ok(intersect_compressed_postings_iterator(posting_readers, filter).collect())
+                })?;
+            // Some token has no posting list -> no matches
+            Ok(result.unwrap_or_default())
         }
 
         match &self.storage.postings {
@@ -227,65 +238,60 @@ impl MmapInvertedIndex {
     }
 
     /// Iterate over point ids whose documents contain at least one of the given tokens
-    fn filter_has_any<'a>(
-        &'a self,
-        tokens: TokenSet,
-    ) -> impl Iterator<Item = PointOffsetType> + 'a {
+    fn filter_has_any(&self, tokens: TokenSet) -> OperationResult<Vec<PointOffsetType>> {
         // in case of immutable index, deleted documents are still in the postings
         let is_active = move |idx| self.is_active(idx);
 
-        fn merge<'a, V: MmapPostingValue>(
-            postings: &'a MmapPostings<V>,
+        fn merge<V: ZerocopyPostingValue>(
+            postings: &UniversalPostings<V, MmapFile>,
             tokens: TokenSet,
-            is_active: impl Fn(PointOffsetType) -> bool + 'a,
-        ) -> impl Iterator<Item = PointOffsetType> + 'a {
-            let postings: Vec<_> = tokens
-                .tokens()
-                .iter()
-                .filter_map(|&token_id| postings.get(token_id))
-                .collect();
-
-            // Query must not be empty
-            if postings.is_empty() {
-                return Either::Left(std::iter::empty());
-            };
-
-            Either::Right(merge_compressed_postings_iterator(postings, is_active))
+            is_active: impl Fn(PointOffsetType) -> bool,
+        ) -> OperationResult<Vec<PointOffsetType>> {
+            postings.with_existing_postings(tokens.tokens(), |posting_readers| {
+                if posting_readers.is_empty() {
+                    return Ok(Vec::new());
+                }
+                let posting_readers = posting_readers
+                    .into_iter()
+                    .map(|(_token_id, posting_list_view)| posting_list_view)
+                    .collect();
+                Ok(merge_compressed_postings_iterator(posting_readers, is_active).collect())
+            })
         }
 
         match &self.storage.postings {
-            MmapPostingsEnum::Ids(postings) => Either::Left(merge(postings, tokens, is_active)),
-            MmapPostingsEnum::WithPositions(postings) => {
-                Either::Right(merge(postings, tokens, is_active))
-            }
+            MmapPostingsEnum::Ids(postings) => merge(postings, tokens, is_active),
+            MmapPostingsEnum::WithPositions(postings) => merge(postings, tokens, is_active),
         }
     }
 
-    fn check_has_subset(&self, tokens: &TokenSet, point_id: PointOffsetType) -> bool {
+    fn check_has_subset(
+        &self,
+        tokens: &TokenSet,
+        point_id: PointOffsetType,
+    ) -> OperationResult<bool> {
         // check non-empty query
         if tokens.is_empty() {
-            return false;
+            return Ok(false);
         }
 
         // check presence of the document
         if self.values_is_empty(point_id) {
-            return false;
+            return Ok(false);
         }
 
-        fn check_intersection<V: MmapPostingValue>(
-            postings: &MmapPostings<V>,
+        fn check_intersection<V: ZerocopyPostingValue>(
+            postings: &UniversalPostings<V, MmapFile>,
             tokens: &TokenSet,
             point_id: PointOffsetType,
-        ) -> bool {
-            // Check that all tokens are in document
-            tokens.tokens().iter().all(|query_token| {
-                postings
-                    .get(*query_token)
-                    // unwrap safety: all tokens exist in the vocabulary, otherwise there'd be no query tokens
-                    .unwrap()
-                    .visitor()
-                    .contains(point_id)
-            })
+        ) -> OperationResult<bool> {
+            let result = postings.with_all_or_none_postings(tokens.tokens(), |all_postings| {
+                Ok(all_postings
+                    .into_iter()
+                    .all(|(_token_id, posting)| posting.visitor().contains(point_id)))
+            })?;
+            // Some token has no posting list -> no match
+            Ok(result.unwrap_or(false))
         }
 
         match &self.storage.postings {
@@ -296,25 +302,25 @@ impl MmapInvertedIndex {
         }
     }
 
-    fn check_has_any(&self, tokens: &TokenSet, point_id: PointOffsetType) -> bool {
+    fn check_has_any(&self, tokens: &TokenSet, point_id: PointOffsetType) -> OperationResult<bool> {
         if tokens.is_empty() {
-            return false;
+            return Ok(false);
         }
 
         // check presence of the document
         if self.values_is_empty(point_id) {
-            return false;
+            return Ok(false);
         }
 
-        fn check_any<V: MmapPostingValue>(
-            postings: &MmapPostings<V>,
+        fn check_any<V: ZerocopyPostingValue>(
+            postings: &UniversalPostings<V, MmapFile>,
             tokens: &TokenSet,
             point_id: PointOffsetType,
-        ) -> bool {
-            // Check that at least one token is in document
-            tokens.tokens().iter().any(|token_id| {
-                let posting_list = postings.get(*token_id).unwrap();
-                posting_list.visitor().contains(point_id)
+        ) -> OperationResult<bool> {
+            postings.with_existing_postings(tokens.tokens(), |all_postings| {
+                Ok(all_postings
+                    .into_iter()
+                    .any(|(_token_id, posting)| posting.visitor().contains(point_id)))
             })
         }
 
@@ -325,40 +331,63 @@ impl MmapInvertedIndex {
     }
 
     /// Iterate over point ids whose documents contain all given tokens in the same order they are provided
-    pub fn filter_has_phrase<'a>(
-        &'a self,
-        phrase: Document,
-    ) -> impl Iterator<Item = PointOffsetType> + 'a {
+    pub fn filter_has_phrase(&self, phrase: Document) -> OperationResult<Vec<PointOffsetType>> {
         // in case of mmap immutable index, deleted points are still in the postings
         let is_active = move |idx| self.is_active(idx);
 
         match &self.storage.postings {
             MmapPostingsEnum::WithPositions(postings) => {
-                Either::Right(intersect_compressed_postings_phrase_iterator(
-                    phrase,
-                    |token_id| postings.get(*token_id),
-                    is_active,
-                ))
+                // Deduplicate phrase tokens: repeated tokens (e.g. "zn zn") must
+                // not fetch the same posting list twice, otherwise positions get
+                // added twice in `phrase_in_all_postings`.
+                let unique_tokens = phrase.to_token_set();
+                let result = postings.with_all_or_none_postings(
+                    unique_tokens.tokens(),
+                    |selected_postings| {
+                        Ok(intersect_compressed_postings_phrase_iterator(
+                            phrase,
+                            selected_postings,
+                            is_active,
+                        )
+                        .collect())
+                    },
+                )?;
+                // Some token has no posting list -> no matches
+                Ok(result.unwrap_or_default())
             }
             // cannot do phrase matching if there's no positional information
-            MmapPostingsEnum::Ids(_postings) => Either::Left(std::iter::empty()),
+            MmapPostingsEnum::Ids(_postings) => Ok(Vec::new()),
         }
     }
 
-    pub fn check_has_phrase(&self, phrase: &Document, point_id: PointOffsetType) -> bool {
+    pub fn check_has_phrase(
+        &self,
+        phrase: &Document,
+        point_id: PointOffsetType,
+    ) -> OperationResult<bool> {
         // in case of mmap immutable index, deleted points are still in the postings
         if !self.is_active(point_id) {
-            return false;
+            return Ok(false);
         }
 
         match &self.storage.postings {
             MmapPostingsEnum::WithPositions(postings) => {
-                check_compressed_postings_phrase(phrase, point_id, |token_id| {
-                    postings.get(*token_id)
-                })
+                let unique_tokens = phrase.to_token_set();
+                let result = postings.with_all_or_none_postings(
+                    unique_tokens.tokens(),
+                    |selected_postings| {
+                        Ok(check_compressed_postings_phrase(
+                            phrase,
+                            point_id,
+                            selected_postings,
+                        ))
+                    },
+                )?;
+                // Some token has no posting list -> no match
+                Ok(result.unwrap_or(false))
             }
             // cannot do phrase matching if there's no positional information
-            MmapPostingsEnum::Ids(_postings) => false,
+            MmapPostingsEnum::Ids(_postings) => Ok(false),
         }
     }
 
@@ -386,7 +415,7 @@ impl MmapInvertedIndex {
     /// Populate all pages in the mmap.
     /// Block until all pages are populated.
     pub fn populate(&self) -> OperationResult<()> {
-        self.storage.postings.populate();
+        self.storage.postings.populate()?;
         self.storage.vocab.populate()?;
         self.storage.point_to_tokens_count.populate()?;
         Ok(())
@@ -406,7 +435,7 @@ impl MmapInvertedIndex {
             point_to_tokens_count,
             deleted_points,
         } = storage;
-        postings.clear_cache();
+        postings.clear_cache()?;
         vocab.clear_cache()?;
         point_to_tokens_count.clear_cache()?;
         deleted_points.clear_cache()?;
@@ -467,31 +496,41 @@ impl InvertedIndex for MmapInvertedIndex {
         query: ParsedQuery,
         _hw_counter: &HardwareCounterCell,
     ) -> OperationResult<Box<dyn Iterator<Item = PointOffsetType> + 'a>> {
-        match query {
-            ParsedQuery::AllTokens(tokens) => Ok(self.filter_has_all(tokens)),
-            ParsedQuery::Phrase(phrase) => Ok(Box::new(self.filter_has_phrase(phrase))),
-            ParsedQuery::AnyTokens(tokens) => Ok(Box::new(self.filter_has_any(tokens))),
-        }
+        let ids = match query {
+            ParsedQuery::AllTokens(tokens) => self.filter_has_all(tokens)?,
+            ParsedQuery::Phrase(phrase) => self.filter_has_phrase(phrase)?,
+            ParsedQuery::AnyTokens(tokens) => self.filter_has_any(tokens)?,
+        };
+        Ok(Box::new(ids.into_iter()))
     }
 
     fn get_posting_len(
         &self,
         token_id: TokenId,
         _hw_counter: &HardwareCounterCell,
-    ) -> Option<usize> {
+    ) -> OperationResult<Option<usize>> {
         self.storage.postings.posting_len(token_id)
     }
 
-    fn vocab_with_postings_len_iter(&self) -> impl Iterator<Item = (&str, usize)> + '_ {
+    fn vocab_with_postings_len_iter(
+        &self,
+    ) -> impl Iterator<Item = OperationResult<(&str, usize)>> + '_ {
         self.iter_vocab().filter_map(move |(token, &token_id)| {
-            self.storage
-                .postings
-                .posting_len(token_id)
-                .map(|posting_len| (token, posting_len))
+            // Surface read errors as iterator items; drop tokens with no
+            // posting list silently (same as the in-memory variants).
+            match self.storage.postings.posting_len(token_id) {
+                Ok(Some(posting_len)) => Some(Ok((token, posting_len))),
+                Ok(None) => None,
+                Err(err) => Some(Err(err)),
+            }
         })
     }
 
-    fn check_match(&self, parsed_query: &ParsedQuery, point_id: PointOffsetType) -> bool {
+    fn check_match(
+        &self,
+        parsed_query: &ParsedQuery,
+        point_id: PointOffsetType,
+    ) -> OperationResult<bool> {
         match parsed_query {
             ParsedQuery::AllTokens(tokens) => self.check_has_subset(tokens, point_id),
             ParsedQuery::Phrase(phrase) => self.check_has_phrase(phrase, point_id),

lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/raw_posting_list.rs

@@ -0,0 +1,56 @@
+use std::borrow::Cow;
+
+use common::types::PointOffsetType;
+use posting_list::{PostingChunk, PostingListView, RemainderPosting, SizedTypeFor};
+use zerocopy::FromBytes;
+
+use crate::common::operation_error::OperationResult;
+use crate::index::field_index::full_text_index::inverted_index::mmap_inverted_index::types::{
+    PostingListHeader, ZerocopyPostingValue,
+};
+
+/// Raw byte representation of posting list, which can be converted into [`PostingListView`]
+pub struct RawPostingList<'a> {
+    bytes: Cow<'a, [u8]>,
+    header: PostingListHeader,
+}
+
+impl<'a> RawPostingList<'a> {
+    pub fn new(bytes: Cow<'a, [u8]>, header: PostingListHeader) -> RawPostingList<'a> {
+        RawPostingList { bytes, header }
+    }
+}
+
+impl<'a> RawPostingList<'a> {
+    pub fn as_view<V: ZerocopyPostingValue>(&'a self) -> OperationResult<PostingListView<'a, V>> {
+        let (last_doc_id, bytes) = PointOffsetType::read_from_prefix(self.bytes.as_ref())?;
+
+        let (chunks, bytes) = <[PostingChunk<SizedTypeFor<V>>]>::ref_from_prefix_with_elems(
+            bytes,
+            self.header.chunks_count as usize,
+        )?;
+
+        let (id_data, bytes) = bytes.split_at(self.header.ids_data_bytes_count as usize);
+
+        let (var_size_data, bytes) = bytes.split_at(self.header.var_size_data_bytes_count as usize);
+
+        // skip padding
+        let bytes = &bytes[self.header.alignment_bytes_count as usize..];
+
+        let (remainder_postings, _) =
+            <[RemainderPosting<SizedTypeFor<V>>]>::ref_from_prefix_with_elems(
+                bytes,
+                self.header.remainder_count as usize,
+            )?;
+
+        let posting_list_view = PostingListView::from_components(
+            id_data,
+            chunks,
+            var_size_data,
+            remainder_postings,
+            Some(last_doc_id),
+        );
+
+        Ok(posting_list_view)
+    }
+}

lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/types.rs

@@ -0,0 +1,68 @@
+use common::types::PointOffsetType;
+use posting_list::{
+    CHUNK_LEN, PostingChunk, PostingValue, RemainderPosting, SizedTypeFor, ValueHandler,
+};
+use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout, Unaligned};
+
+use crate::index::field_index::full_text_index::inverted_index::positions::Positions;
+
+pub const ALIGNMENT: usize = 4;
+
+/// A [`PostingValue`] whose sized payload can be zerocopy-read from an mmap.
+pub(in crate::index::field_index::full_text_index) trait ZerocopyPostingValue:
+    PostingValue<
+    Handler: ValueHandler<Sized: FromBytes + IntoBytes + Immutable + KnownLayout + Unaligned>,
+>
+{
+}
+
+impl ZerocopyPostingValue for () {}
+
+impl ZerocopyPostingValue for Positions {}
+
+#[derive(Debug, Default, Clone, FromBytes, Immutable, IntoBytes, KnownLayout)]
+#[repr(C)]
+pub(in crate::index::field_index::full_text_index) struct PostingsHeader {
+    /// Number of posting lists. One posting list per term
+    pub posting_count: usize,
+    pub _reserved: [u8; 32],
+}
+
+/// This data structure should contain all the necessary information to
+/// construct `PostingListView<V>` from the mmap file.
+#[derive(Debug, Default, Clone, FromBytes, Immutable, IntoBytes, KnownLayout)]
+#[repr(C)]
+pub(in crate::index::field_index::full_text_index) struct PostingListHeader {
+    /// Offset in bytes from the start of the mmap file
+    /// where the posting list data starts
+    pub offset: u64,
+    /// Amount of chunks in compressed posting list
+    pub chunks_count: u32,
+    /// Length in bytes for the compressed postings data
+    pub ids_data_bytes_count: u32,
+    /// Length in bytes for the alignment bytes
+    pub alignment_bytes_count: u8,
+    /// Number of `RemainderPosting` elements (tail that doesn't fill a chunk)
+    pub remainder_count: u8,
+
+    pub _reserved: [u8; 2],
+
+    /// Length in bytes for the var-sized data. Add-on for phrase matching, otherwise 0
+    pub var_size_data_bytes_count: u32,
+}
+
+impl PostingListHeader {
+    pub fn posting_size<V: PostingValue>(&self) -> usize {
+        self.ids_data_bytes_count as usize
+            + self.var_size_data_bytes_count as usize
+            + self.alignment_bytes_count as usize
+            + self.remainder_count as usize * size_of::<RemainderPosting<SizedTypeFor<V>>>()
+            + self.chunks_count as usize * size_of::<PostingChunk<SizedTypeFor<V>>>()
+            + size_of::<PointOffsetType>() // last_doc_id
+    }
+
+    /// Number of elements in the posting list. Matches `PostingListView::len`.
+    pub fn posting_len(&self) -> usize {
+        self.chunks_count as usize * CHUNK_LEN + self.remainder_count as usize
+    }
+}

lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/uio_postings.rs

@@ -0,0 +1,268 @@
+use std::cell::Cell;
+use std::marker::PhantomData;
+use std::path::PathBuf;
+
+use common::generic_consts::{Random, Sequential};
+use common::universal_io::{OpenOptions, ReadRange, UniversalIoError, UniversalRead};
+use posting_list::{PostingList, PostingListView};
+use zerocopy::FromBytes;
+
+use crate::common::operation_error::OperationResult;
+use crate::index::field_index::full_text_index::inverted_index::TokenId;
+use crate::index::field_index::full_text_index::inverted_index::mmap_inverted_index::raw_posting_list::RawPostingList;
+use crate::index::field_index::full_text_index::inverted_index::mmap_inverted_index::types::{
+    PostingListHeader, PostingsHeader, ZerocopyPostingValue,
+};
+
+/// Posting lists stored on disk and accessed via the [`UniversalRead`]
+/// abstraction. The on-disk layout is produced by
+/// [`create_postings_file`](super::create_postings::create_postings_file)
+/// and matches the layout documented there.
+///
+/// `get_header(token_id)` indexes the per-token header array directly:
+/// `size_of::<PostingsHeader>() + token_id * size_of::<PostingListHeader>()`.
+/// Each [`PostingListHeader`] then points (via absolute `offset`) into the
+/// posting-data region.
+pub struct UniversalPostings<V: ZerocopyPostingValue, S: UniversalRead<u8>> {
+    _path: PathBuf,
+    storage: S,
+    header: PostingsHeader,
+    _value_type: PhantomData<V>,
+}
+
+type HeaderResult = Result<(TokenId, PostingListHeader), UniversalIoError>;
+
+/// Output of [`UniversalPostings::headers_iter`]: a lazy iterator over the
+/// headers that are in range, paired with the pre-filtered list of token ids
+/// that were outside the valid range.
+struct HeadersBatch<'a> {
+    iter: Box<dyn Iterator<Item = HeaderResult> + 'a>,
+    missing: Vec<TokenId>,
+}
+
+impl<V: ZerocopyPostingValue, S: UniversalRead<u8>> UniversalPostings<V, S> {
+    /// Open the postings file at `path` via the `S` storage backend.
+    pub fn open(path: impl Into<PathBuf>, options: OpenOptions) -> OperationResult<Self> {
+        let path = path.into();
+        let storage = S::open(&path, options)?;
+
+        let header_bytes = storage.read::<Sequential>(ReadRange {
+            byte_offset: 0,
+            length: size_of::<PostingsHeader>() as u64,
+        })?;
+
+        let (header, _) = PostingsHeader::read_from_prefix(header_bytes.as_ref())?;
+
+        Ok(Self {
+            _path: path,
+            storage,
+            header,
+            _value_type: PhantomData,
+        })
+    }
+
+    /// Hint the storage backend to populate any RAM cache backing this file.
+    /// For mmap-backed storage this is `madvise(MADV_POPULATE_READ)` and blocks
+    /// until pages are populated.
+    pub fn populate(&self) -> OperationResult<()> {
+        self.storage.populate()?;
+        Ok(())
+    }
+
+    /// Hint the storage backend to drop any RAM cache backing this file.
+    /// For mmap-backed storage this is `madvise(MADV_PAGEOUT)`.
+    pub fn clear_cache(&self) -> OperationResult<()> {
+        self.storage.clear_ram_cache()?;
+        Ok(())
+    }
+
+    /// Stream posting list headers for a batch of token ids.
+    ///
+    /// The returned [`HeadersBatch`] pairs a lazy iterator over the in-range
+    /// `(token_id, header)` pairs with the list of token ids that were
+    /// out-of-range. Iterator order is not guaranteed to match the input.
+    fn headers_iter(&self, token_ids: &[TokenId]) -> Result<HeadersBatch<'_>, UniversalIoError> {
+        let header_length = size_of::<PostingListHeader>() as u64;
+        let posting_count = self.header.posting_count;
+
+        let mut valid_ranges: Vec<(TokenId, ReadRange)> = Vec::with_capacity(token_ids.len());
+        let mut filtered_out: Vec<TokenId> = Vec::new();
+
+        for &token_id in token_ids {
+            if posting_count <= token_id as usize {
+                filtered_out.push(token_id);
+                continue;
+            }
+            let header_offset =
+                size_of::<PostingsHeader>() as u64 + u64::from(token_id) * header_length;
+
+            valid_ranges.push((
+                token_id,
+                ReadRange {
+                    byte_offset: header_offset,
+                    length: header_length,
+                },
+            ));
+        }
+
+        let valid_iter = self
+            .storage
+            .read_iter::<Random, _>(valid_ranges)?
+            .map(|res| {
+                let (token_id, bytes) = res?;
+                let (header, _) = PostingListHeader::read_from_prefix(bytes.as_ref())?;
+                Ok((token_id, header))
+            });
+
+        Ok(HeadersBatch {
+            iter: Box::new(valid_iter),
+            missing: filtered_out,
+        })
+    }
+
+    fn get_header(&self, token_id: TokenId) -> OperationResult<Option<PostingListHeader>> {
+        let HeadersBatch { mut iter, .. } = self.headers_iter(&[token_id])?;
+        let Some(entry) = iter.next() else {
+            return Ok(None);
+        };
+        let (_, header) = entry?;
+        Ok(Some(header))
+    }
+
+    /// Create PostingListView<V> from the given header
+    ///
+    /// Assume the following layout:
+    ///
+    /// ```ignore
+    /// last_doc_id: &'a PointOffsetType,
+    /// chunks_index: &'a [PostingChunk<()>],
+    /// data: &'a [u8],
+    /// var_size_data: &'a [u8], // might be empty in case of only ids
+    /// _alignment: &'a [u8], // 0-3 extra bytes to align the data
+    /// remainder_postings: &'a [PointOffsetType],
+    /// ```
+    #[cfg(test)]
+    fn raw_posting<'a>(
+        &'a self,
+        header: PostingListHeader,
+    ) -> Result<RawPostingList<'a>, UniversalIoError> {
+        let read_range = ReadRange {
+            byte_offset: header.offset,
+            length: header.posting_size::<V>() as u64,
+        };
+        let bytes = self.storage.read::<Sequential>(read_range)?;
+        let result = RawPostingList::new(bytes, header);
+        Ok(result)
+    }
+
+    #[cfg(test)]
+    pub fn get(&self, token_id: TokenId) -> OperationResult<Option<RawPostingList<'_>>> {
+        let header = self.get_header(token_id)?;
+        if let Some(header) = header {
+            let posting = self.raw_posting(header).map(Some)?;
+            return Ok(posting);
+        }
+        Ok(None)
+    }
+
+    /// Number of elements in the posting list for `token_id`. Reads only the
+    /// per-token header, not the posting bytes.
+    pub fn posting_len(&self, token_id: TokenId) -> OperationResult<Option<usize>> {
+        Ok(self.get_header(token_id)?.map(|h| h.posting_len()))
+    }
+
+    /// Read the posting lists for every header yielded by `header_iter` and
+    /// hand the resulting views to `callback`. Header iteration is pipelined
+    /// into the posting reads - submissions are scheduled lazily as headers
+    /// arrive.
+    fn with_posting_views<T>(
+        &self,
+        header_iter: Box<dyn Iterator<Item = HeaderResult> + '_>,
+        expected_capacity: usize,
+        callback: impl FnOnce(Vec<(TokenId, PostingListView<'_, V>)>) -> OperationResult<T>,
+    ) -> OperationResult<T> {
+        let header_err: Cell<Option<UniversalIoError>> = Cell::new(None);
+
+        let range_iter = header_iter.filter_map(|header_res| match header_res {
+            Ok((token_id, header)) => {
+                let range = ReadRange {
+                    byte_offset: header.offset,
+                    length: header.posting_size::<V>() as u64,
+                };
+                Some(((token_id, header), range))
+            }
+            Err(err) => {
+                header_err.set(Some(err));
+                None
+            }
+        });
+
+        let mut raw_postings: Vec<(TokenId, RawPostingList<'_>)> =
+            Vec::with_capacity(expected_capacity);
+
+        for entry in self.storage.read_iter::<Sequential, _>(range_iter)? {
+            let ((token_id, header), bytes) = entry?;
+            raw_postings.push((token_id, RawPostingList::new(bytes, header)));
+        }
+
+        if let Some(err) = header_err.take() {
+            return Err(err.into());
+        }
+
+        let views = raw_postings
+            .iter()
+            .map(|(token_id, raw)| raw.as_view::<V>().map(|view| (*token_id, view)))
+            .collect::<OperationResult<Vec<_>>>()?;
+
+        callback(views)
+    }
+
+    /// Fetch posting lists for the given token ids and hand them to `callback`.
+    /// If any of posting lists is not found - return `None`.
+    pub fn with_all_or_none_postings<T>(
+        &self,
+        token_ids: &[TokenId],
+        callback: impl FnOnce(Vec<(TokenId, PostingListView<'_, V>)>) -> OperationResult<T>,
+    ) -> OperationResult<Option<T>> {
+        let HeadersBatch {
+            iter: header_iter,
+            missing,
+        } = self.headers_iter(token_ids)?;
+
+        if !missing.is_empty() {
+            return Ok(None);
+        }
+
+        self.with_posting_views(header_iter, token_ids.len(), callback)
+            .map(Some)
+    }
+
+    /// Fetch all existing posting lists for the given token ids and hand them
+    /// to `callback`. Token ids without a posting list are silently ignored.
+    pub fn with_existing_postings<T>(
+        &self,
+        token_ids: &[TokenId],
+        callback: impl FnOnce(Vec<(TokenId, PostingListView<'_, V>)>) -> OperationResult<T>,
+    ) -> OperationResult<T> {
+        let HeadersBatch {
+            iter: header_iter, ..
+        } = self.headers_iter(token_ids)?;
+
+        self.with_posting_views(header_iter, token_ids.len(), callback)
+    }
+
+    pub fn all_postings(&self) -> OperationResult<Vec<PostingList<V>>> {
+        let mut result = Vec::new();
+        let all_tokens = (0..self.header.posting_count as TokenId).collect::<Vec<_>>();
+
+        self.with_existing_postings(&all_tokens, |views| {
+            for (_, view) in views {
+                let posting_list = view.to_owned();
+                result.push(posting_list);
+            }
+            Ok(())
+        })?;
+
+        Ok(result)
+    }
+}

lib/segment/src/index/field_index/full_text_index/inverted_index/mod.rs

@@ -230,15 +230,18 @@ pub trait InvertedIndex {
         hw_counter: &'a HardwareCounterCell,
     ) -> OperationResult<Box<dyn Iterator<Item = PointOffsetType> + 'a>>;
 
-    fn get_posting_len(&self, token_id: TokenId, hw_counter: &HardwareCounterCell)
-    -> Option<usize>;
+    fn get_posting_len(
+        &self,
+        token_id: TokenId,
+        hw_counter: &HardwareCounterCell,
+    ) -> OperationResult<Option<usize>>;
 
     fn estimate_cardinality(
         &self,
         query: &ParsedQuery,
         condition: &FieldCondition,
         hw_counter: &HardwareCounterCell,
-    ) -> CardinalityEstimation {
+    ) -> OperationResult<CardinalityEstimation> {
         match query {
             ParsedQuery::AllTokens(tokens) => {
                 self.estimate_has_subset_cardinality(tokens, condition, hw_counter)
@@ -257,31 +260,31 @@ pub trait InvertedIndex {
         tokens: &TokenSet,
         condition: &FieldCondition,
         hw_counter: &HardwareCounterCell,
-    ) -> CardinalityEstimation {
+    ) -> OperationResult<CardinalityEstimation> {
         let points_count = self.points_count();
 
         let posting_lengths: Option<Vec<usize>> = tokens
             .tokens()
             .iter()
             .map(|&vocab_idx| self.get_posting_len(vocab_idx, hw_counter))
-            .collect();
+            .collect::<OperationResult<Option<Vec<usize>>>>()?;
         if posting_lengths.is_none() || points_count == 0 {
             // There are unseen tokens -> no matches
-            return CardinalityEstimation::exact(0)
-                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone())));
+            return Ok(CardinalityEstimation::exact(0)
+                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone()))));
         }
         let postings = posting_lengths.unwrap();
         if postings.is_empty() {
             // Empty request -> no matches
-            return CardinalityEstimation::exact(0)
-                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone())));
+            return Ok(CardinalityEstimation::exact(0)
+                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone()))));
         }
         // Smallest posting is the largest possible cardinality
         let smallest_posting = postings.iter().min().copied().unwrap();
 
         if postings.len() == 1 {
-            return CardinalityEstimation::exact(smallest_posting)
-                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone())));
+            return Ok(CardinalityEstimation::exact(smallest_posting)
+                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone()))));
         }
 
         let expected_frac: f64 = postings
@@ -289,12 +292,12 @@ pub trait InvertedIndex {
             .map(|posting| *posting as f64 / points_count as f64)
             .product();
         let exp = (expected_frac * points_count as f64) as usize;
-        CardinalityEstimation {
+        Ok(CardinalityEstimation {
             primary_clauses: vec![PrimaryCondition::Condition(Box::new(condition.clone()))],
             min: 0, // ToDo: make better estimation
             exp,
             max: smallest_posting,
-        }
+        })
     }
 
     fn estimate_has_any_cardinality(
@@ -302,39 +305,39 @@ pub trait InvertedIndex {
         tokens: &TokenSet,
         condition: &FieldCondition,
         hw_counter: &HardwareCounterCell,
-    ) -> CardinalityEstimation {
+    ) -> OperationResult<CardinalityEstimation> {
         let points_count = self.points_count();
 
-        let posting_lengths: Vec<_> = tokens
+        let posting_lengths: Vec<usize> = tokens
             .tokens()
             .iter()
-            .filter_map(|&vocab_idx| self.get_posting_len(vocab_idx, hw_counter))
-            .collect();
+            .filter_map(|&vocab_idx| self.get_posting_len(vocab_idx, hw_counter).transpose())
+            .collect::<OperationResult<Vec<usize>>>()?;
 
         if posting_lengths.is_empty() {
             // Empty request -> no matches
-            return CardinalityEstimation::exact(0)
-                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone())));
+            return Ok(CardinalityEstimation::exact(0)
+                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone()))));
         }
 
         // At least one posting is the largest possible cardinality
         let largest_posting = posting_lengths.iter().max().copied().unwrap();
 
         if posting_lengths.len() == 1 {
-            return CardinalityEstimation::exact(largest_posting)
-                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone())));
+            return Ok(CardinalityEstimation::exact(largest_posting)
+                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone()))));
         }
 
         let sum: usize = posting_lengths.iter().sum();
 
         let exp = expected_should_estimation(posting_lengths.into_iter(), points_count);
 
-        CardinalityEstimation {
+        Ok(CardinalityEstimation {
             primary_clauses: vec![PrimaryCondition::Condition(Box::new(condition.clone()))],
             min: largest_posting,
             exp,
             max: min(sum, points_count),
-        }
+        })
     }
 
     fn estimate_has_phrase_cardinality(
@@ -342,44 +345,46 @@ pub trait InvertedIndex {
         phrase: &Document,
         condition: &FieldCondition,
         hw_counter: &HardwareCounterCell,
-    ) -> CardinalityEstimation {
+    ) -> OperationResult<CardinalityEstimation> {
         if phrase.is_empty() {
-            return CardinalityEstimation::exact(0)
-                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone())));
+            return Ok(CardinalityEstimation::exact(0)
+                .with_primary_clause(PrimaryCondition::Condition(Box::new(condition.clone()))));
         }
 
         // Start with same cardinality estimation as has_subset
         let tokenset = phrase.to_token_set();
         let subset_estimation =
-            self.estimate_has_subset_cardinality(&tokenset, condition, hw_counter);
+            self.estimate_has_subset_cardinality(&tokenset, condition, hw_counter)?;
 
         // But we can restrict it by considering the phrase length
         let phrase_sq = phrase.len() * phrase.len();
 
-        CardinalityEstimation {
+        Ok(CardinalityEstimation {
             primary_clauses: vec![PrimaryCondition::Condition(Box::new(condition.clone()))],
             min: subset_estimation.min / phrase_sq,
             exp: subset_estimation.exp / phrase_sq,
             max: subset_estimation.max / phrase_sq,
-        }
+        })
     }
 
-    fn vocab_with_postings_len_iter(&self) -> impl Iterator<Item = (&str, usize)> + '_;
+    fn vocab_with_postings_len_iter(
+        &self,
+    ) -> impl Iterator<Item = OperationResult<(&str, usize)>> + '_;
 
     fn payload_blocks(
         &self,
         threshold: usize,
         key: PayloadKeyType,
     ) -> impl Iterator<Item = OperationResult<PayloadBlockCondition>> + '_ {
-        let map_filter_condition = move |(token, postings_len): (&str, usize)| {
-            if postings_len >= threshold {
+        let map_filter_condition = move |item: OperationResult<(&str, usize)>| match item {
+            Ok((token, postings_len)) if postings_len >= threshold => {
                 Some(Ok(PayloadBlockCondition {
                     condition: FieldCondition::new_match(key.clone(), Match::new_text(token)),
                     cardinality: postings_len,
                 }))
-            } else {
-                None
             }
+            Ok(_) => None,
+            Err(err) => Some(Err(err)),
         };
 
         // It might be very hard to predict possible combinations of conditions,
@@ -388,7 +393,11 @@ pub trait InvertedIndex {
             .filter_map(map_filter_condition)
     }
 
-    fn check_match(&self, parsed_query: &ParsedQuery, point_id: PointOffsetType) -> bool;
+    fn check_match(
+        &self,
+        parsed_query: &ParsedQuery,
+        point_id: PointOffsetType,
+    ) -> OperationResult<bool>;
 
     fn values_is_empty(&self, point_id: PointOffsetType) -> bool;
 
@@ -544,7 +553,7 @@ mod tests {
             .unwrap()
             .unwrap();
 
-        let imm_mmap = ImmutableInvertedIndex::from(&mmap);
+        let imm_mmap = ImmutableInvertedIndex::try_from(&mmap).unwrap();
 
         // Check same vocabulary
         for (token, token_id) in &immutable.vocab {
@@ -611,7 +620,7 @@ mod tests {
                 .unwrap()
                 .unwrap();
 
-        let mut imm_mmap_index = ImmutableInvertedIndex::from(&mmap_index);
+        let mut imm_mmap_index = ImmutableInvertedIndex::try_from(&mmap_index).unwrap();
 
         let queries: Vec<_> = (0..100).map(|_| generate_query()).collect();
 

lib/segment/src/index/field_index/full_text_index/inverted_index/mutable_inverted_index.rs

@@ -222,23 +222,33 @@ impl InvertedIndex for MutableInvertedIndex {
         }
     }
 
-    fn get_posting_len(&self, token_id: TokenId, _: &HardwareCounterCell) -> Option<usize> {
-        self.postings.get(token_id as usize).map(|x| x.len())
+    fn get_posting_len(
+        &self,
+        token_id: TokenId,
+        _: &HardwareCounterCell,
+    ) -> OperationResult<Option<usize>> {
+        Ok(self.postings.get(token_id as usize).map(|x| x.len()))
     }
 
-    fn vocab_with_postings_len_iter(&self) -> impl Iterator<Item = (&str, usize)> + '_ {
+    fn vocab_with_postings_len_iter(
+        &self,
+    ) -> impl Iterator<Item = OperationResult<(&str, usize)>> + '_ {
         self.vocab.iter().filter_map(|(token, &posting_idx)| {
             self.postings
                 .get(posting_idx as usize)
-                .map(|postings| (token.as_str(), postings.len()))
+                .map(|postings| Ok((token.as_str(), postings.len())))
         })
     }
 
-    fn check_match(&self, parsed_query: &ParsedQuery, point_id: PointOffsetType) -> bool {
-        match parsed_query {
+    fn check_match(
+        &self,
+        parsed_query: &ParsedQuery,
+        point_id: PointOffsetType,
+    ) -> OperationResult<bool> {
+        let matched = match parsed_query {
             ParsedQuery::AllTokens(query) => {
                 let Some(doc) = self.get_tokens(point_id) else {
-                    return false;
+                    return Ok(false);
                 };
 
                 // Check that all tokens are in document
@@ -246,7 +256,7 @@ impl InvertedIndex for MutableInvertedIndex {
             }
             ParsedQuery::Phrase(document) => {
                 let Some(doc) = self.get_document(point_id) else {
-                    return false;
+                    return Ok(false);
                 };
 
                 // Check that all tokens are in document, in order
@@ -254,13 +264,14 @@ impl InvertedIndex for MutableInvertedIndex {
             }
             ParsedQuery::AnyTokens(query) => {
                 let Some(doc) = self.get_tokens(point_id) else {
-                    return false;
+                    return Ok(false);
                 };
 
                 // Check that at least one token is in document
                 doc.has_any(query)
             }
-        }
+        };
+        Ok(matched)
     }
 
     fn values_is_empty(&self, point_id: PointOffsetType) -> bool {

lib/segment/src/index/field_index/full_text_index/inverted_index/postings_iterator.rs

@@ -87,7 +87,7 @@ pub fn merge_compressed_postings_iterator<'a, V: PostingValue + 'a>(
 /// Returns an iterator over the points that match the given phrase query.
 pub fn intersect_compressed_postings_phrase_iterator<'a>(
     phrase: Document,
-    token_to_posting: impl Fn(&TokenId) -> Option<PostingListView<'a, Positions>>,
+    mut postings: Vec<(TokenId, PostingListView<'a, Positions>)>,
     is_active: impl Fn(PointOffsetType) -> bool + 'a,
 ) -> impl Iterator<Item = PointOffsetType> + 'a {
     if phrase.is_empty() {
@@ -95,18 +95,6 @@ pub fn intersect_compressed_postings_phrase_iterator<'a>(
         return Either::Left(std::iter::empty());
     }
 
-    let postings_opt: Option<Vec<_>> = phrase
-        .to_token_set()
-        .tokens()
-        .iter()
-        .map(|token_id| token_to_posting(token_id).map(|posting| (*token_id, posting)))
-        .collect();
-
-    let Some(mut postings) = postings_opt else {
-        // There are unseen tokens -> no matches
-        return Either::Left(std::iter::empty());
-    };
-
     let smallest_posting_idx = postings
         .iter()
         .enumerate()
@@ -174,21 +162,15 @@ fn phrase_in_all_postings<'a>(
     PartialDocument::new(tokens_positions).has_phrase(phrase)
 }
 
-pub fn check_compressed_postings_phrase<'a>(
+pub fn check_compressed_postings_phrase(
     phrase: &Document,
     point_id: PointOffsetType,
-    token_to_posting: impl Fn(&TokenId) -> Option<PostingListView<'a, Positions>>,
+    token_to_posting: Vec<(TokenId, PostingListView<'_, Positions>)>,
 ) -> bool {
-    let Some(mut posting_iterators): Option<Vec<_>> = phrase
-        .to_token_set()
-        .tokens()
-        .iter()
-        .map(|token_id| token_to_posting(token_id).map(|posting| (*token_id, posting.into_iter())))
-        .collect()
-    else {
-        // not all tokens are present in the index
-        return false;
-    };
+    let mut posting_iterators = token_to_posting
+        .into_iter()
+        .map(|(token_id, posting)| (token_id, posting.into_iter()))
+        .collect::<Vec<_>>();
 
     phrase_in_all_postings(point_id, phrase, Vec::new(), &mut posting_iterators)
 }

lib/segment/src/index/field_index/full_text_index/mmap_text_index.rs

@@ -209,7 +209,7 @@ impl FieldIndexBuilderTrait for FullTextMmapIndexBuilder {
         let text_index = if is_on_disk {
             FullTextIndex::Mmap(Box::new(mmap_index))
         } else {
-            FullTextIndex::Immutable(ImmutableFullTextIndex::open_mmap(mmap_index))
+            FullTextIndex::Immutable(ImmutableFullTextIndex::open_mmap(mmap_index)?)
         };
 
         Ok(text_index)

lib/segment/src/index/field_index/full_text_index/tests/mod.rs

@@ -188,7 +188,7 @@ fn test_prefix_search() {
     let query = index.parse_text_query("ROBO", &hw_counter).unwrap();
 
     for idx in res.iter().copied() {
-        assert!(index.check_match(&query, idx));
+        assert!(index.check_match(&query, idx).unwrap());
     }
 
     assert_eq!(res.len(), 3);
@@ -253,9 +253,9 @@ fn test_phrase_matching() {
         let text_query = index
             .parse_text_query("quick brown fox", &hw_counter)
             .unwrap();
-        assert!(index.check_match(&text_query, 0));
-        assert!(index.check_match(&text_query, 1));
-        assert!(index.check_match(&text_query, 2));
+        assert!(index.check_match(&text_query, 0).unwrap());
+        assert!(index.check_match(&text_query, 1).unwrap());
+        assert!(index.check_match(&text_query, 2).unwrap());
 
         let text_results: Vec<_> = index
             .filter_query(text_query, &hw_counter)
@@ -272,8 +272,8 @@ fn test_phrase_matching() {
         let phrase_query = index
             .parse_phrase_query("quick brown fox", &hw_counter)
             .unwrap();
-        assert!(index.check_match(&phrase_query, 0));
-        assert!(index.check_match(&phrase_query, 2));
+        assert!(index.check_match(&phrase_query, 0).unwrap());
+        assert!(index.check_match(&phrase_query, 2).unwrap());
 
         let phrase_results: Vec<_> = index
             .filter_query(phrase_query, &hw_counter)
@@ -307,7 +307,7 @@ fn test_phrase_matching() {
         let phrase_query = index
             .parse_phrase_query("brown brown fox", &hw_counter)
             .unwrap();
-        assert!(index.check_match(&phrase_query, 4));
+        assert!(index.check_match(&phrase_query, 4).unwrap());
 
         // Should only match document 4
         let filter_results: Vec<_> = index
@@ -378,7 +378,7 @@ fn test_ascii_folding_in_full_text_index_word() {
 
     // ASCII-only queries should match only when folding is enabled
     let query_enabled = index_enabled.parse_text_query("acao", &hw_counter).unwrap();
-    assert!(index_enabled.check_match(&query_enabled, 0));
+    assert!(index_enabled.check_match(&query_enabled, 0).unwrap());
 
     let results_enabled: Vec<_> = index_enabled
         .filter_query(query_enabled, &hw_counter)
@@ -398,7 +398,7 @@ fn test_ascii_folding_in_full_text_index_word() {
 
     // Non-folded query must work in both
     let query_acento = index_enabled.parse_text_query("ação", &hw_counter).unwrap();
-    assert!(index_enabled.check_match(&query_acento, 0));
+    assert!(index_enabled.check_match(&query_acento, 0).unwrap());
     let results_acento: Vec<_> = index_enabled
         .filter_query(query_acento, &hw_counter)
         .unwrap()

lib/segment/src/index/field_index/full_text_index/tests/test_congruence.rs

@@ -344,8 +344,8 @@ fn test_congruence(
 
             for point_id in 0..POINT_COUNT as PointOffsetType {
                 assert_eq!(
-                    index_a.check_match(&parsed_query_a, point_id),
-                    index_b.check_match(&parsed_query_b, point_id),
+                    index_a.check_match(&parsed_query_a, point_id).unwrap(),
+                    index_b.check_match(&parsed_query_b, point_id).unwrap(),
                 );
             }
 
@@ -387,8 +387,8 @@ fn test_congruence(
 
                 for point_id in 0..POINT_COUNT as PointOffsetType {
                     assert_eq!(
-                        index_a.check_match(&parsed_query_a, point_id),
-                        index_b.check_match(&parsed_query_b, point_id),
+                        index_a.check_match(&parsed_query_a, point_id).unwrap(),
+                        index_b.check_match(&parsed_query_b, point_id).unwrap(),
                     );
                 }
 
@@ -453,7 +453,7 @@ fn check_phrase<const KEYWORD_COUNT: usize>(
 
             let parsed_query = parse_query(phrase, phrase_matching, index);
 
-            assert!(index.check_match(&parsed_query, *exp_id));
+            assert!(index.check_match(&parsed_query, *exp_id).unwrap());
 
             let result = index
                 .filter_query(parsed_query, &hw_counter)

lib/segment/src/index/field_index/full_text_index/text_index.rs

@@ -54,7 +54,7 @@ impl FullTextIndex {
             // Load into RAM, use mmap as backing storage
             Some(Self::Immutable(ImmutableFullTextIndex::open_mmap(
                 mmap_index,
-            )))
+            )?))
         };
         Ok(index)
     }
@@ -156,7 +156,7 @@ impl FullTextIndex {
         query: &ParsedQuery,
         condition: &FieldCondition,
         hw_counter: &HardwareCounterCell,
-    ) -> CardinalityEstimation {
+    ) -> OperationResult<CardinalityEstimation> {
         match self {
             Self::Mutable(index) => index
                 .inverted_index
@@ -170,7 +170,11 @@ impl FullTextIndex {
         }
     }
 
-    pub fn check_match(&self, query: &ParsedQuery, point_id: PointOffsetType) -> bool {
+    pub fn check_match(
+        &self,
+        query: &ParsedQuery,
+        point_id: PointOffsetType,
+    ) -> OperationResult<bool> {
         match self {
             Self::Mutable(index) => index.inverted_index.check_match(query, point_id),
             Self::Immutable(index) => index.inverted_index.check_match(query, point_id),
@@ -522,7 +526,7 @@ impl PayloadFieldIndex for FullTextIndex {
             &parsed_query,
             condition,
             hw_counter,
-        )))
+        )?))
     }
 
     fn payload_blocks(

lib/segment/src/index/field_index/geo_index/mmap_geo_index.rs

@@ -294,7 +294,7 @@ impl<S: StoredGeoMapIndexStorage> StoredGeoMapIndex<S> {
                     .point_to_values
                     .check_values_any(idx, |v| check_fn(v), &hw_counter)
             })
-            .map(|r| r.unwrap_or(false))
+            .map(|r| r.unwrap_or(false)) // FIXME: don't silently ignore error
             .unwrap_or(false)
     }
 

lib/segment/src/index/query_optimization/condition_converter/match_converter.rs

@@ -281,7 +281,10 @@ fn get_match_text_checker(
             };
 
             Some(Box::new(move |point_id: PointOffsetType| {
-                full_text_index.check_match(&parsed_query, point_id)
+                // FIXME: don't silently ignore errors. Log error? Update ConditionCheckerFn?
+                full_text_index
+                    .check_match(&parsed_query, point_id)
+                    .unwrap_or(false)
             }))
         }
         FieldIndex::BoolIndex(_)

src/tonic/api/storage_read_api/helpers.rs

@@ -212,5 +212,6 @@ pub fn io_error_to_status(e: UniversalIoError) -> Status {
             Status::internal(format!("Uninitialized: {description}"))
         }
         UniversalIoError::BytemuckCast(e) => Status::internal(format!("Bytemuck cast error: {e}")),
+        UniversalIoError::ZerocopySize(e) => Status::internal(e),
     }
 }