Add segment level type for serverless bitflags (#10121)

* Add `CompactStoredFlags`, segment wrapper over the mutable bitmask

RAM-resident flags with a Flusher (skips the write when clean, cancels after drop) and files lister, backed by one compact stored-bitmask file rewritten whole on flush. Not integrated yet.

* Add `FlagsMode`, detecting the storage mode of a flags directory

`Dynamic` is the existing mmap stack for dedicated deployments, `Compact`
the compact stored-bitmask file for serverless ones; detection probes
which files are present. Also add the clippy allow the compact flags
tests were missing.

* Support the compact storage mode in `BitvecFlags` and `RoaringFlags`

The wrappers keep their in-memory read state in both modes; the new
`FlagsStorage` dispatches the write side between `BufferedDynamicFlags`
and `CompactStoredFlags`. `open_or_create` opens existing flags in their
detected mode and only applies `mode_if_create` to fresh ones — existing
call sites keep constructing the dynamic stack through `new`.

* Add `ReadOnlyCompactFlags`, read-only counterpart of compact flags

Bound to `UniversalRead`: opens on the bitmask header alone,
materializes the bitmap lazily on first query, and never creates a
missing file. Implements `RoaringFlagsRead` for the shared query
surface; `live_reload` reopens a fresh handle, as flushes replace the
file whole but cached handles keep serving the bytes they were opened
on. Not integrated yet.

* Skip compact live-reload tests on Windows, which forbids the rename

Both tests replace the compact flags file behind a reader whose disk
cache keeps the "remote" file mapped. On Unix the rename-over succeeds
and the mapping serves the old inode — the staleness under test — but
Windows forbids renaming over a mapped file, failing the writer's flush
with access denied. A limitation of the local-mmap remote stand-in, not
of the reload logic, which stays covered on the other targets.

* Don't check legacy flag file

lib/segment/src/common/flags/bitvec_flags.rs

@@ -1,11 +1,14 @@
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 
 use common::bitvec::{BitSlice, BitVec};
 use common::types::PointOffsetType;
-use common::universal_io::{UniversalRead, UniversalWrite};
+use common::universal_io::{Populate, UniversalRead, UniversalWrite};
 
 use super::buffered_dynamic_flags::BufferedDynamicFlags;
+use super::compact_stored_flags::CompactStoredFlags;
 use super::dynamic_stored_flags::DynamicStoredFlags;
+use super::mode::FlagsMode;
+use super::storage::FlagsStorage;
 use crate::common::Flusher;
 use crate::common::operation_error::OperationResult;
 
@@ -18,8 +21,8 @@ use crate::common::operation_error::OperationResult;
 /// [1]: super::roaring_flags::RoaringFlags
 #[derive(Debug)]
 pub struct BitvecFlags<S: UniversalRead> {
-    /// Buffered persisted flags.
-    storage: BufferedDynamicFlags<S>,
+    /// Persisted flags, in either storage mode.
+    storage: FlagsStorage<S>,
 
     /// In-memory bitvec of true and false flags.
     bitvec: BitVec,
@@ -33,6 +36,28 @@ where
     S: UniversalWrite + Send + 'static,
     S::Fs: Send + Sync + 'static,
 {
+    /// Open the flags in `directory`, or create them when none exist there yet.
+    ///
+    /// The mode of existing flags is detected automatically from the files
+    /// present; `mode_if_create` only applies when creating fresh flags.
+    pub fn open_or_create(
+        fs: S::Fs,
+        directory: &Path,
+        mode_if_create: FlagsMode,
+        populate: Populate,
+    ) -> OperationResult<Self> {
+        match FlagsMode::detect(&fs, directory)?.unwrap_or(mode_if_create) {
+            FlagsMode::Dynamic => {
+                let dynamic_flags = DynamicStoredFlags::open(&fs, directory, populate)?;
+                Self::new(fs, dynamic_flags)
+            }
+            FlagsMode::Compact => {
+                let compact_flags = CompactStoredFlags::open(fs, directory, populate)?;
+                Ok(Self::from_compact(compact_flags))
+            }
+        }
+    }
+
     pub fn new(fs: S::Fs, dynamic_flags: DynamicStoredFlags<S>) -> OperationResult<Self> {
         // load flags into memory
         let bitvec = BitVec::from_bitslice(&*dynamic_flags.get_bitslice()?);
@@ -43,11 +68,27 @@ where
 
         Ok(Self {
             len: dynamic_flags.len(),
-            storage: BufferedDynamicFlags::new(fs, dynamic_flags),
+            storage: FlagsStorage::Dynamic(BufferedDynamicFlags::new(fs, dynamic_flags)),
             bitvec,
         })
     }
 
+    fn from_compact(compact_flags: CompactStoredFlags<S>) -> Self {
+        let len = compact_flags.len();
+
+        // load flags into memory
+        let mut bitvec = BitVec::repeat(false, len);
+        for index in compact_flags.to_bitmap() {
+            bitvec.set(index as usize, true);
+        }
+
+        Self {
+            storage: FlagsStorage::Compact(compact_flags),
+            bitvec,
+            len,
+        }
+    }
+
     pub fn len(&self) -> usize {
         self.len
     }
@@ -89,8 +130,8 @@ where
     /// Set the value of a flag at the given index, grows the bitvec if needed.
     /// Returns the previous value of the flag.
     pub fn set(&mut self, index: PointOffsetType, value: bool) -> bool {
-        // queue write in buffer
-        self.storage.buffer_set(index, value);
+        // record write in persisted storage
+        self.storage.set(index, value);
 
         // update length if needed
         let index_usize = index as usize;
@@ -136,7 +177,9 @@ mod tests_mod {
     #[cfg_predicate]
     use common::universal_io::{Fs, S};
 
+    use crate::common::flags::FlagsMode;
     use crate::common::flags::bitvec_flags::BitvecFlags;
+    use crate::common::flags::compact_stored_flags::COMPACT_FLAGS_FILE;
     use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
 
     #[test]
@@ -207,4 +250,87 @@ mod tests_mod {
             assert_eq!(all_indices, expected_all);
         }
     }
+
+    #[test]
+    fn test_compact_mode_roundtrip() {
+        let dir = tempfile::Builder::new()
+            .prefix("bitvec_flags_compact")
+            .tempdir()
+            .unwrap();
+
+        {
+            let mut flags = BitvecFlags::<S>::open_or_create(
+                Fs::default(),
+                dir.path(),
+                FlagsMode::Compact,
+                Populate::No,
+            )
+            .unwrap();
+            assert!(!flags.set(0, true));
+            assert!(!flags.set(5, true));
+            assert!(!flags.set(9, false)); // grows the flags to 10
+            assert!(flags.set(5, true)); // previous value
+            assert_eq!(flags.len(), 10);
+
+            let files = flags.files();
+            assert_eq!(files.len(), 1);
+            assert!(files[0].ends_with(COMPACT_FLAGS_FILE));
+
+            flags.flusher()().unwrap();
+        }
+
+        // Requesting dynamic mode on existing flags keeps the compact mode.
+        {
+            let flags = BitvecFlags::<S>::open_or_create(
+                Fs::default(),
+                dir.path(),
+                FlagsMode::Dynamic,
+                Populate::Blocking,
+            )
+            .unwrap();
+            assert_eq!(flags.files().len(), 1);
+            assert_eq!(flags.len(), 10);
+            assert_eq!(flags.count_trues(), 2);
+            assert_eq!(flags.iter_trues().collect::<Vec<_>>(), vec![0, 5]);
+            assert!(flags.get(0));
+            assert!(!flags.get(9));
+            assert_eq!(flags.get_bitslice().count_ones(), 2);
+        }
+    }
+
+    #[test]
+    fn test_dynamic_mode_kept_when_compact_requested() {
+        let dir = tempfile::Builder::new()
+            .prefix("bitvec_flags_dynamic")
+            .tempdir()
+            .unwrap();
+
+        {
+            let mut flags = BitvecFlags::<S>::open_or_create(
+                Fs::default(),
+                dir.path(),
+                FlagsMode::Dynamic,
+                Populate::No,
+            )
+            .unwrap();
+            flags.set(1, true);
+            flags.flusher()().unwrap();
+
+            // dynamic stack: flags file plus status file
+            assert!(flags.files().len() > 1);
+        }
+
+        {
+            let flags = BitvecFlags::<S>::open_or_create(
+                Fs::default(),
+                dir.path(),
+                FlagsMode::Compact,
+                Populate::Blocking,
+            )
+            .unwrap();
+            assert!(flags.files().len() > 1);
+            assert_eq!(flags.len(), 2);
+            assert!(flags.get(1));
+        }
+    }
 }

lib/segment/src/common/flags/compact_stored_flags.rs

@@ -0,0 +1,291 @@
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use common::is_alive_lock::IsAliveLock;
+use common::mmap::AdviceSetting;
+use common::stored_bitmask::MutableStoredBitmask;
+use common::types::PointOffsetType;
+use common::universal_io::{
+    OkNotFound, OpenOptions, Populate, UniversalRead, UniversalWrite, UniversalWriteFileOps,
+};
+use parking_lot::Mutex;
+use roaring::RoaringBitmap;
+
+use crate::common::Flusher;
+use crate::common::operation_error::{OperationError, OperationResult};
+
+/// Name of the single file holding the mask, inside the storage directory.
+pub(super) const COMPACT_FLAGS_FILE: &str = "compact_flags.dat";
+
+/// Flags over a single compact stored-bitmask file, rewritten whole on flush.
+///
+/// The serverless-compatible counterpart of `DynamicStoredFlags` +
+/// `BufferedDynamicFlags`: the mask is fully resident in RAM, mutations
+/// collect there, and a flush rewrites the file in one atomic whole-file
+/// write — no in-place mmap mutation, so it also works on object stores.
+/// A flush with no effective changes skips the write entirely.
+#[derive(Debug)]
+pub struct CompactStoredFlags<S: UniversalRead> {
+    /// The mask, resident in RAM.
+    mask: Arc<Mutex<MutableStoredBitmask>>,
+
+    /// Filesystem handle used to rewrite the file on flush.
+    fs: Arc<S::Fs>,
+
+    /// Path of the mask file inside the storage directory.
+    path: PathBuf,
+
+    /// Lock to prevent concurrent flush and drop
+    is_alive_flush_lock: IsAliveLock,
+}
+
+impl<S> CompactStoredFlags<S>
+where
+    S: UniversalWrite + Send + 'static,
+    S::Fs: Send + Sync + 'static,
+{
+    /// Open the flags in `directory`, materializing the whole mask into RAM.
+    ///
+    /// Creates the directory and persists an empty mask when the file is
+    /// missing, so [`Self::files`] always exist on disk.
+    pub fn open(fs: S::Fs, directory: &Path, populate: Populate) -> OperationResult<Self> {
+        fs.create_dir(directory)?;
+        let path = directory.join(COMPACT_FLAGS_FILE);
+
+        let options = OpenOptions {
+            writeable: false,
+            need_sequential: true,
+            populate,
+            advice: AdviceSetting::Global,
+        };
+        let mask = match MutableStoredBitmask::open(&fs, &path, options, Default::default())
+            .ok_not_found()?
+        {
+            Some(mask) => mask,
+            None => {
+                let mut mask = MutableStoredBitmask::new(0);
+                mask.save(&fs, &path)?;
+                mask
+            }
+        };
+
+        Ok(Self {
+            mask: Arc::new(Mutex::new(mask)),
+            fs: Arc::new(fs),
+            path,
+            is_alive_flush_lock: IsAliveLock::new(),
+        })
+    }
+
+    /// Number of logical flags in the mask.
+    pub fn len(&self) -> usize {
+        self.mask.lock().bit_len() as usize
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Value of the flag at `index`; `false` at and beyond [`Self::len`].
+    pub fn get(&self, index: PointOffsetType) -> bool {
+        self.mask.lock().get(index)
+    }
+
+    /// Number of set flags.
+    pub fn count_flags(&self) -> usize {
+        self.mask.lock().count_ones() as usize
+    }
+
+    /// Snapshot of the set flags as a roaring bitmap.
+    pub fn to_bitmap(&self) -> RoaringBitmap {
+        self.mask.lock().ones().clone()
+    }
+
+    /// Set the flag at `index`, returning its previous value. Grows the mask
+    /// when `index` is at or beyond its length.
+    pub fn set(&self, index: PointOffsetType, value: bool) -> bool {
+        let mut mask = self.mask.lock();
+        if u64::from(index) >= mask.bit_len() {
+            mask.set_len(u64::from(index) + 1);
+        }
+        mask.set(index, value)
+    }
+
+    /// Grow the mask to `new_len` flags; the new flags are unset.
+    ///
+    /// # Panics
+    ///
+    /// If `new_len` would shrink the mask.
+    pub fn set_len(&self, new_len: usize) {
+        self.mask.lock().set_len(new_len as u64);
+    }
+
+    /// Flusher that persists the mask as it is at the moment of flushing.
+    ///
+    /// The write is skipped when nothing effectively changed since the mask
+    /// was opened or last flushed. Flushing after the instance was dropped
+    /// is cancelled.
+    pub fn flusher(&self) -> Flusher {
+        // The mask persists itself and knows when it is clean: nothing is
+        // snapshotted here. Clean right now means this flush cycle has
+        // nothing to write; later mutations are for the next cycle.
+        if !self.mask.lock().is_dirty() {
+            return Box::new(|| Ok(()));
+        }
+
+        // Weak reference to detect when the storage has been deleted
+        let mask_weak = Arc::downgrade(&self.mask);
+        let fs = Arc::clone(&self.fs);
+        let path = self.path.clone();
+        let is_alive_flush_lock = self.is_alive_flush_lock.handle();
+
+        Box::new(move || {
+            let (Some(is_alive_flush_guard), Some(mask_arc)) =
+                (is_alive_flush_lock.lock_if_alive(), mask_weak.upgrade())
+            else {
+                log::trace!("CompactStoredFlags was dropped, cancelling flush");
+                return Err(OperationError::cancelled(
+                    "Aborted flushing on a dropped CompactStoredFlags instance",
+                ));
+            };
+
+            mask_arc.lock().save(&*fs, &path)?;
+
+            drop(is_alive_flush_guard);
+            Ok(())
+        })
+    }
+
+    /// The single file backing the flags; guaranteed to exist on disk.
+    pub fn files(&self) -> Vec<PathBuf> {
+        vec![self.path.clone()]
+    }
+
+    /// No-op: the mask is fully resident from open.
+    pub fn populate(&self) -> OperationResult<()> {
+        Ok(())
+    }
+
+    /// No-op: the resident mask is the authoritative state, not a cache.
+    pub fn clear_cache(&self) -> OperationResult<()> {
+        Ok(())
+    }
+}
+
+#[allow(clippy::default_constructed_unit_structs)]
+#[duplicate::duplicate_item(
+    tests_mod       S               Fs              cfg_predicate;
+    [tests_mmap]    [MmapFile]      [MmapFs]        [cfg(all())];
+    [tests_uring]   [IoUringFile]   [IoUringFs]     [cfg(target_os = "linux")];
+)]
+#[cfg_predicate]
+#[cfg(test)]
+mod tests_mod {
+    use common::universal_io::Populate;
+    #[cfg_predicate]
+    use common::universal_io::{Fs, S};
+    use tempfile::TempDir;
+
+    use crate::common::flags::compact_stored_flags::CompactStoredFlags;
+    use crate::common::operation_error::OperationError;
+
+    fn open(dir: &std::path::Path) -> CompactStoredFlags<S> {
+        CompactStoredFlags::open(Fs::default(), dir, Populate::Blocking).unwrap()
+    }
+
+    #[test]
+    fn open_creates_file_and_lists_it() {
+        let dir = TempDir::new().unwrap();
+        let flags = open(dir.path());
+        assert_eq!(flags.len(), 0);
+        assert_eq!(flags.count_flags(), 0);
+
+        let files = flags.files();
+        assert_eq!(files.len(), 1);
+        assert!(files[0].exists()); // eagerly persisted empty mask
+        assert!(files[0].starts_with(dir.path()));
+    }
+
+    #[test]
+    fn set_flush_reopen_roundtrip() {
+        let dir = TempDir::new().unwrap();
+        {
+            let flags = open(dir.path());
+            assert!(!flags.set(3, true));
+            assert!(!flags.set(100, true)); // grows to 101
+            assert!(flags.set(3, true)); // previous value
+            assert!(!flags.set(50, false));
+            assert_eq!(flags.len(), 101);
+            assert_eq!(flags.count_flags(), 2);
+            flags.flusher()().unwrap();
+        }
+        {
+            let flags = open(dir.path());
+            assert_eq!(flags.len(), 101);
+            assert_eq!(flags.count_flags(), 2);
+            assert!(flags.get(3));
+            assert!(flags.get(100));
+            assert!(!flags.get(50));
+        }
+    }
+
+    #[test]
+    fn clean_flusher_skips_write() {
+        let dir = TempDir::new().unwrap();
+        let flags = open(dir.path());
+        flags.set(1, true);
+        flags.flusher()().unwrap();
+
+        // A clean flush must not touch storage: with the file deleted behind
+        // its back, only an actual write could bring it back.
+        fs_err::remove_file(&flags.files()[0]).unwrap();
+        flags.flusher()().unwrap();
+        assert!(!flags.files()[0].exists());
+
+        // The next effective change rewrites the whole mask.
+        flags.set(2, true);
+        flags.flusher()().unwrap();
+        let reopened = open(dir.path());
+        assert!(reopened.get(1));
+        assert!(reopened.get(2));
+    }
+
+    #[test]
+    fn flusher_flushes_call_time_state() {
+        let dir = TempDir::new().unwrap();
+        let flags = open(dir.path());
+        flags.set(1, true);
+        let flusher = flags.flusher();
+        flags.set(2, true); // after flusher creation, before the flush
+        flusher().unwrap();
+
+        let reopened = open(dir.path());
+        assert!(reopened.get(1));
+        assert!(reopened.get(2));
+    }
+
+    #[test]
+    fn flush_after_drop_is_cancelled() {
+        let dir = TempDir::new().unwrap();
+        let flags = open(dir.path());
+        flags.set(1, true);
+        let flusher = flags.flusher();
+        drop(flags);
+        assert!(matches!(flusher(), Err(OperationError::Cancelled { .. })));
+    }
+
+    #[test]
+    fn set_len_grows_and_persists() {
+        let dir = TempDir::new().unwrap();
+        {
+            let flags = open(dir.path());
+            flags.set_len(10);
+            assert_eq!(flags.len(), 10);
+            assert_eq!(flags.count_flags(), 0);
+            flags.flusher()().unwrap();
+        }
+        let flags = open(dir.path());
+        assert_eq!(flags.len(), 10);
+        assert_eq!(flags.count_flags(), 0);
+    }
+}

lib/segment/src/common/flags/dynamic_stored_flags.rs

@@ -21,7 +21,7 @@ const MINIMAL_MMAP_SIZE: usize = 128; // 128 bytes -> 1024 flags
 const MINIMAL_MMAP_SIZE: usize = 1024 * 1024; // 1Mb
 
 pub(super) const FLAGS_FILE: &str = "flags_a.dat";
-const FLAGS_FILE_LEGACY: &str = "flags_b.dat";
+pub(super) const FLAGS_FILE_LEGACY: &str = "flags_b.dat";
 
 pub(super) const STATUS_FILE_NAME: &str = "status.dat";
 

lib/segment/src/common/flags/mod.rs

@@ -7,11 +7,25 @@
 //! - `roaring_flags`: `buffered_dynamic_flags` with in-memory roaring bitmap for reads.
 //! - `in_memory_bitvec_flags`: in-memory counterpart of `bitvec_flags`, bound to `UniversalRead`.
 //! - `read_only_roaring_flags`: read-only counterpart of `roaring_flags`, bound to `UniversalRead`.
+//! - `compact_stored_flags`: RAM-resident flags over a single compact stored-bitmask file,
+//!   rewritten whole on flush; serverless-compatible counterpart of the dynamic + buffered stack.
+//! - `read_only_compact_flags`: read-only counterpart of `compact_stored_flags`, bound to
+//!   `UniversalRead`.
+//!
+//! `bitvec_flags` and `roaring_flags` persist either through the dynamic stack or through
+//! `compact_stored_flags`, selected by [`FlagsMode`] when flags are created and detected
+//! automatically when opening existing flags.
 
 pub mod bitvec_flags;
 mod buffered_dynamic_flags;
+pub mod compact_stored_flags;
 pub mod dynamic_stored_flags;
 pub mod in_memory_bitvec_flags;
+mod mode;
+pub mod read_only_compact_flags;
 pub mod read_only_roaring_flags;
 pub mod roaring_flags;
+mod storage;
 pub mod update_only_stored_flags;
+
+pub use mode::FlagsMode;

lib/segment/src/common/flags/mode.rs

@@ -0,0 +1,105 @@
+use std::path::Path;
+
+use common::universal_io::UniversalReadFileOps;
+
+use super::compact_stored_flags::COMPACT_FLAGS_FILE;
+use super::dynamic_stored_flags::{FLAGS_FILE, status_file};
+use crate::common::operation_error::{OperationError, OperationResult};
+
+/// Storage mode of a flags directory, selecting the persistence flavor.
+///
+/// Specified when flags are created, and automatically detected from the
+/// files present when opening existing flags — see [`Self::detect`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FlagsMode {
+    /// Flags in mmapped files, mutated in place with buffered writes.
+    ///
+    /// The mode for dedicated deployments.
+    Dynamic,
+
+    /// Flags in a single compact stored-bitmask file, fully RAM-resident and
+    /// rewritten whole on flush.
+    ///
+    /// The mode for serverless deployments: it needs no in-place writes, so
+    /// it also works on object stores.
+    Compact,
+}
+
+impl FlagsMode {
+    /// Detect the mode of the flags in `directory` from the files present,
+    /// `None` when no flags exist there yet.
+    ///
+    /// Errors when files of both modes are present.
+    pub fn detect(
+        fs: &impl UniversalReadFileOps,
+        directory: &Path,
+    ) -> OperationResult<Option<Self>> {
+        let dynamic =
+            fs.exists(&status_file(directory))? || fs.exists(&directory.join(FLAGS_FILE))?;
+        let compact = fs.exists(&directory.join(COMPACT_FLAGS_FILE))?;
+        match (dynamic, compact) {
+            (false, false) => Ok(None),
+            (true, false) => Ok(Some(Self::Dynamic)),
+            (false, true) => Ok(Some(Self::Compact)),
+            (true, true) => Err(OperationError::service_error(format!(
+                "flags in {} have files of both the dynamic and the compact mode",
+                directory.display(),
+            ))),
+        }
+    }
+}
+
+#[allow(clippy::default_constructed_unit_structs)]
+#[duplicate::duplicate_item(
+    tests_mod       S               Fs              cfg_predicate;
+    [tests_mmap]    [MmapFile]      [MmapFs]        [cfg(all())];
+    [tests_uring]   [IoUringFile]   [IoUringFs]     [cfg(target_os = "linux")];
+)]
+#[cfg_predicate]
+#[cfg(test)]
+mod tests_mod {
+    use common::universal_io::Populate;
+    #[cfg_predicate]
+    use common::universal_io::{Fs, S};
+    use tempfile::TempDir;
+
+    use crate::common::flags::FlagsMode;
+    use crate::common::flags::compact_stored_flags::CompactStoredFlags;
+    use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
+
+    #[test]
+    fn detects_nothing_in_missing_or_empty_directory() {
+        let dir = TempDir::new().unwrap();
+        let missing = dir.path().join("missing");
+        assert_eq!(FlagsMode::detect(&Fs::default(), &missing).unwrap(), None);
+        assert_eq!(FlagsMode::detect(&Fs::default(), dir.path()).unwrap(), None);
+    }
+
+    #[test]
+    fn detects_dynamic_flags() {
+        let dir = TempDir::new().unwrap();
+        DynamicStoredFlags::<S>::open(&Fs::default(), dir.path(), Populate::No).unwrap();
+        assert_eq!(
+            FlagsMode::detect(&Fs::default(), dir.path()).unwrap(),
+            Some(FlagsMode::Dynamic),
+        );
+    }
+
+    #[test]
+    fn detects_compact_flags() {
+        let dir = TempDir::new().unwrap();
+        CompactStoredFlags::<S>::open(Fs::default(), dir.path(), Populate::No).unwrap();
+        assert_eq!(
+            FlagsMode::detect(&Fs::default(), dir.path()).unwrap(),
+            Some(FlagsMode::Compact),
+        );
+    }
+
+    #[test]
+    fn errors_on_files_of_both_modes() {
+        let dir = TempDir::new().unwrap();
+        DynamicStoredFlags::<S>::open(&Fs::default(), dir.path(), Populate::No).unwrap();
+        CompactStoredFlags::<S>::open(Fs::default(), dir.path(), Populate::No).unwrap();
+        assert!(FlagsMode::detect(&Fs::default(), dir.path()).is_err());
+    }
+}

lib/segment/src/common/flags/read_only_compact_flags.rs

@@ -0,0 +1,316 @@
+use std::path::{Path, PathBuf};
+use std::sync::OnceLock;
+
+use common::mmap::AdviceSetting;
+use common::stored_bitmask::StoredBitmask;
+use common::universal_io::{
+    CachedReadFs, OkNotFound, OpenOptions, Populate, UniversalRead, UniversalReadFs,
+};
+use roaring::RoaringBitmap;
+
+use super::compact_stored_flags::COMPACT_FLAGS_FILE;
+use super::roaring_flags::RoaringFlagsRead;
+use crate::common::operation_error::OperationResult;
+
+/// Read-only counterpart of [`CompactStoredFlags`][1], and thereby of the
+/// compact [mode](super::FlagsMode) of the writable wrappers.
+///
+/// Materializes the persisted flags into an in-memory roaring bitmap on first
+/// use — *not* on open, which reads only the fixed-size file header. The
+/// backing [`StoredBitmask`] handle is kept for that lazy read and is replaced
+/// wholesale by [`Self::live_reload`]. There is no write path: opening never
+/// creates a missing file, unlike the writable [`CompactStoredFlags::open`][2].
+///
+/// [1]: super::compact_stored_flags::CompactStoredFlags
+/// [2]: super::compact_stored_flags::CompactStoredFlags::open
+pub struct ReadOnlyCompactFlags<S: UniversalRead> {
+    /// In-memory bitmap of true flags, materialized from the backing file on
+    /// first access and resynced by [`Self::live_reload`].
+    ///
+    /// Lazy so that opening the flags reads only the file header: a segment
+    /// open would otherwise decode every flags file whole, which defeats
+    /// prefetching only the bytes a query actually needs.
+    ///
+    /// [`OnceLock`] rather than a plain cell because the flags are queried
+    /// through `&self` from many threads. On a race both threads may build a
+    /// bitmap; the first to finish wins and the loser's copy is dropped.
+    bitmap: OnceLock<RoaringBitmap>,
+    /// Backing bitmask, used by the lazy [`Self::bitmap`] read. Replaced with
+    /// a freshly opened handle on every [`Self::live_reload`].
+    storage: StoredBitmask<S>,
+    /// Total length of the flags, including trailing falses. Read from the
+    /// bitmask header.
+    len: usize,
+    directory: PathBuf,
+}
+
+fn open_options(populate: Populate) -> OpenOptions {
+    OpenOptions {
+        writeable: false,
+        need_sequential: true,
+        populate,
+        advice: AdviceSetting::Global,
+    }
+}
+
+impl<S: UniversalRead> ReadOnlyCompactFlags<S> {
+    /// Schedule background prefetch of the single file this storage reads.
+    ///
+    /// Returns whether the flags exist.
+    pub fn preopen(
+        fs: &impl CachedReadFs<File = S>,
+        directory: &Path,
+        populate: Populate,
+    ) -> OperationResult<bool> {
+        Ok(fs
+            .schedule_prefetch(
+                &directory.join(COMPACT_FLAGS_FILE),
+                Some(open_options(populate)),
+                None,
+            )
+            .ok_not_found()?
+            .is_some())
+    }
+
+    /// Open persisted flags read-only, retaining the bitmask handle for the
+    /// lazy [`RoaringFlagsRead::get_bitmap`].
+    ///
+    /// Returns [`Ok(None)`] when the flags file doesn't exist, matching the
+    /// read path's never-create contract.
+    pub fn open(
+        fs: &impl UniversalReadFs<File = S>,
+        directory: &Path,
+    ) -> OperationResult<Option<Self>> {
+        // A missing file means the flags aren't present on disk.
+        let Some(storage) = StoredBitmask::<S>::open(
+            fs,
+            directory.join(COMPACT_FLAGS_FILE),
+            open_options(Populate::No),
+            Default::default(),
+        )
+        .ok_not_found()?
+        else {
+            return Ok(None);
+        };
+
+        Ok(Some(Self {
+            bitmap: OnceLock::new(),
+            len: storage.bit_len() as usize,
+            storage,
+            directory: directory.to_path_buf(),
+        }))
+    }
+
+    /// The in-memory bitmap of set positions, decoding the backing file to
+    /// build it on the first call and returning the cached one afterwards.
+    ///
+    /// This is the whole-file read that [`Self::open`] avoids. It is deferred
+    /// to the first query rather than paid per segment open — many segments
+    /// hold flag indexes that no query ever touches.
+    fn bitmap(&self) -> OperationResult<&RoaringBitmap> {
+        // `OnceLock::get_or_try_init` is still unstable, so build outside the
+        // lock and let `get_or_init` arbitrate. A racing thread's bitmap is
+        // simply dropped: both are built from the same bytes.
+        if let Some(bitmap) = self.bitmap.get() {
+            return Ok(bitmap);
+        }
+
+        let bitmap = self.storage.read_ones()?;
+        Ok(self.bitmap.get_or_init(|| bitmap))
+    }
+
+    /// Refresh to the current on-disk state.
+    ///
+    /// The compact file is never mutated in place — every flush replaces it
+    /// whole — but the held handle keeps serving the bytes it was opened on,
+    /// on caching backends forever. So a *fresh* handle is opened (a fresh
+    /// open always mirrors the current remote bytes), the materialized
+    /// bitmap, if any, is resynced from it, and it replaces the old handle.
+    ///
+    /// While the bitmap is still unmaterialized there is nothing to resync:
+    /// the eventual first read decodes the fresh storage installed here.
+    pub fn live_reload(&mut self, fs: &impl UniversalReadFs<File = S>) -> OperationResult<()> {
+        // Once the flags exist their file always does, so absence here is a
+        // genuine not-found (segment removed mid-reload), not a lazy file.
+        let storage = StoredBitmask::<S>::open(
+            fs,
+            self.directory.join(COMPACT_FLAGS_FILE),
+            open_options(Populate::No),
+            Default::default(),
+        )?;
+
+        if let Some(bitmap) = self.bitmap.get_mut() {
+            *bitmap = storage.read_ones()?;
+        }
+
+        // The logical length grows as points are appended; refresh it so
+        // length-driven readers stay correct.
+        self.len = storage.bit_len() as usize;
+        self.storage = storage;
+
+        Ok(())
+    }
+}
+
+impl<S: UniversalRead> RoaringFlagsRead for ReadOnlyCompactFlags<S> {
+    fn len(&self) -> usize {
+        self.len
+    }
+
+    fn get_bitmap(&self) -> OperationResult<&RoaringBitmap> {
+        self.bitmap()
+    }
+
+    fn bitmap_if_materialized(&self) -> Option<&RoaringBitmap> {
+        self.bitmap.get()
+    }
+
+    fn files(&self) -> Vec<PathBuf> {
+        vec![self.directory.join(COMPACT_FLAGS_FILE)]
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[cfg(not(windows))]
+    use std::sync::Arc;
+
+    #[cfg(not(windows))]
+    use common::universal_io::{
+        DiskCache, DiskCacheConfig, DiskCacheFs, DiskCacheFsContext, UniversalReadFileOps,
+    };
+    use common::universal_io::{MmapFile, MmapFs};
+    use tempfile::Builder;
+
+    use super::*;
+    use crate::common::flags::compact_stored_flags::CompactStoredFlags;
+
+    #[test]
+    fn open_reads_persisted_flags_lazily_and_masks_missing() {
+        let tmp = Builder::new().prefix("compact_read").tempdir().unwrap();
+        let dir = tmp.path().join("flags");
+
+        // Never-create contract: nothing on disk, nothing opened.
+        assert!(
+            ReadOnlyCompactFlags::<MmapFile>::open(&MmapFs, &dir)
+                .unwrap()
+                .is_none()
+        );
+
+        let writer = CompactStoredFlags::<MmapFile>::open(MmapFs, &dir, Populate::No).unwrap();
+        writer.set(3, true);
+        writer.set(9, false); // grows the flags to 10
+        writer.flusher()().unwrap();
+
+        let flags = ReadOnlyCompactFlags::<MmapFile>::open(&MmapFs, &dir)
+            .unwrap()
+            .unwrap();
+        assert_eq!(flags.len(), 10);
+        assert!(flags.bitmap_if_materialized().is_none()); // open decodes nothing
+
+        assert!(flags.get(3).unwrap());
+        assert!(!flags.get(9).unwrap());
+        assert_eq!(flags.count_trues().unwrap(), 1);
+        assert!(flags.bitmap_if_materialized().is_some());
+        assert_eq!(flags.files(), vec![dir.join(COMPACT_FLAGS_FILE)]);
+    }
+
+    /// Every flush replaces the compact file whole, which the handle held by
+    /// the reader on a caching backend never picks up. `live_reload` opens a
+    /// fresh handle instead — this drives it over `DiskCacheFs`, where the
+    /// stale-cache failure actually reproduces.
+    ///
+    /// Not run on Windows: the flush renames over a file the reader's disk
+    /// cache keeps mapped, which Windows forbids. That limitation is specific
+    /// to this setup — a local mmap file standing in as the "remote" — and
+    /// the reload logic stays covered on the other targets.
+    #[cfg(not(windows))]
+    #[test]
+    fn live_reload_over_disk_cache_sees_replaced_file() {
+        let tmp = Builder::new().prefix("compact_reload").tempdir().unwrap();
+        let remote_root = tmp.path().join("remote");
+        let local_root = tmp.path().join("local");
+        let dir = remote_root.join("flags");
+        fs_err::create_dir_all(&dir).unwrap();
+        fs_err::create_dir_all(&local_root).unwrap();
+
+        // The writer works on the "remote" directly; the reader mirrors it
+        // into `local_root` through the disk cache.
+        let writer = CompactStoredFlags::<MmapFile>::open(MmapFs, &dir, Populate::No).unwrap();
+        writer.set(5, true);
+        writer.set_len(100);
+        writer.flusher()().unwrap();
+
+        let cache_fs = DiskCacheFs::<MmapFile>::from_context(DiskCacheFsContext {
+            config: Arc::new(DiskCacheConfig::new(remote_root, local_root).unwrap()),
+            remote: Default::default(),
+        })
+        .unwrap();
+        let mut flags = ReadOnlyCompactFlags::<DiskCache<MmapFile>>::open(&cache_fs, &dir)
+            .unwrap()
+            .unwrap();
+
+        // Materialize the bitmap: reads (and locally caches) the whole file —
+        // the pre-write state this test must escape.
+        assert!(flags.get_bitmap().unwrap().contains(5));
+        assert_eq!(flags.len(), 100);
+
+        // Replace the file behind the reader's back: new bits, a cleared bit,
+        // and growth.
+        writer.set(6, true);
+        writer.set(50, true);
+        writer.set(5, false);
+        writer.set_len(120);
+        writer.flusher()().unwrap();
+
+        flags.live_reload(&cache_fs).unwrap();
+
+        let bitmap = flags.get_bitmap().unwrap();
+        assert!(bitmap.contains(6));
+        assert!(bitmap.contains(50));
+        assert!(
+            !bitmap.contains(5),
+            "cleared flag must not survive a reload"
+        );
+        assert_eq!(flags.len(), 120);
+    }
+
+    /// A bitmap that was never materialized needs no resync: the reload just
+    /// swaps in the fresh handle, and the eventual first read decodes it.
+    ///
+    /// Not run on Windows: same rename-over-mapped-file limitation as
+    /// `live_reload_over_disk_cache_sees_replaced_file`.
+    #[cfg(not(windows))]
+    #[test]
+    fn live_reload_before_materialization_reads_fresh_state() {
+        let tmp = Builder::new().prefix("compact_reload").tempdir().unwrap();
+        let remote_root = tmp.path().join("remote");
+        let local_root = tmp.path().join("local");
+        let dir = remote_root.join("flags");
+        fs_err::create_dir_all(&dir).unwrap();
+        fs_err::create_dir_all(&local_root).unwrap();
+
+        let writer = CompactStoredFlags::<MmapFile>::open(MmapFs, &dir, Populate::No).unwrap();
+        writer.set(5, true);
+        writer.flusher()().unwrap();
+
+        let cache_fs = DiskCacheFs::<MmapFile>::from_context(DiskCacheFsContext {
+            config: Arc::new(DiskCacheConfig::new(remote_root, local_root).unwrap()),
+            remote: Default::default(),
+        })
+        .unwrap();
+        let mut flags = ReadOnlyCompactFlags::<DiskCache<MmapFile>>::open(&cache_fs, &dir)
+            .unwrap()
+            .unwrap();
+
+        // No `get_bitmap` here: the bitmap stays unmaterialized.
+        writer.set(6, true);
+        writer.flusher()().unwrap();
+
+        flags.live_reload(&cache_fs).unwrap();
+
+        let bitmap = flags.get_bitmap().unwrap();
+        assert!(bitmap.contains(5));
+        assert!(bitmap.contains(6));
+    }
+}

lib/segment/src/common/flags/roaring_flags.rs

@@ -1,11 +1,14 @@
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 
 use common::types::PointOffsetType;
-use common::universal_io::{UniversalRead, UniversalWrite};
+use common::universal_io::{Populate, UniversalRead, UniversalWrite};
 use roaring::RoaringBitmap;
 
 use super::buffered_dynamic_flags::BufferedDynamicFlags;
+use super::compact_stored_flags::CompactStoredFlags;
 use super::dynamic_stored_flags::DynamicStoredFlags;
+use super::mode::FlagsMode;
+use super::storage::FlagsStorage;
 use crate::common::Flusher;
 use crate::common::operation_error::OperationResult;
 
@@ -97,8 +100,8 @@ pub trait RoaringFlagsRead {
 ///
 /// [1]: super::bitvec_flags::BitvecFlags
 pub struct RoaringFlags<S: UniversalRead> {
-    /// Buffered persisted flags.
-    storage: BufferedDynamicFlags<S>,
+    /// Persisted flags, in either storage mode.
+    storage: FlagsStorage<S>,
 
     /// In-memory bitmap of true flags.
     // Potential optimization: add a secondary bitmap for false values for faster iter_falses implementation.
@@ -146,6 +149,28 @@ where
     S: UniversalWrite + Send + 'static,
     S::Fs: Send + Sync + 'static,
 {
+    /// Open the flags in `directory`, or create them when none exist there yet.
+    ///
+    /// The mode of existing flags is detected automatically from the files
+    /// present; `mode_if_create` only applies when creating fresh flags.
+    pub fn open_or_create(
+        fs: S::Fs,
+        directory: &Path,
+        mode_if_create: FlagsMode,
+        populate: Populate,
+    ) -> OperationResult<Self> {
+        match FlagsMode::detect(&fs, directory)?.unwrap_or(mode_if_create) {
+            FlagsMode::Dynamic => {
+                let dynamic_flags = DynamicStoredFlags::open(&fs, directory, populate)?;
+                Self::new(fs, dynamic_flags)
+            }
+            FlagsMode::Compact => {
+                let compact_flags = CompactStoredFlags::open(fs, directory, populate)?;
+                Ok(Self::from_compact(compact_flags))
+            }
+        }
+    }
+
     pub fn new(fs: S::Fs, dynamic_flags: DynamicStoredFlags<S>) -> OperationResult<Self> {
         // load flags into memory
         let bitmap = RoaringBitmap::from_sorted_iter(dynamic_flags.iter_trues()?)
@@ -157,16 +182,28 @@ where
 
         Ok(Self {
             len: dynamic_flags.len(),
-            storage: BufferedDynamicFlags::new(fs, dynamic_flags),
+            storage: FlagsStorage::Dynamic(BufferedDynamicFlags::new(fs, dynamic_flags)),
             bitmap,
         })
     }
 
+    fn from_compact(compact_flags: CompactStoredFlags<S>) -> Self {
+        // load flags into memory
+        let len = compact_flags.len();
+        let bitmap = compact_flags.to_bitmap();
+
+        Self {
+            storage: FlagsStorage::Compact(compact_flags),
+            bitmap,
+            len,
+        }
+    }
+
     /// Set the value of a flag at the given index.
     /// Returns the previous value of the flag.
     pub fn set(&mut self, index: PointOffsetType, value: bool) -> bool {
-        // queue write in buffer
-        self.storage.buffer_set(index, value);
+        // record write in persisted storage
+        self.storage.set(index, value);
 
         // update length if needed
         let index_usize = index as usize;
@@ -218,6 +255,8 @@ mod tests_mod {
     #[cfg_predicate]
     use common::universal_io::{Fs, S};
 
+    use crate::common::flags::FlagsMode;
+    use crate::common::flags::compact_stored_flags::COMPACT_FLAGS_FILE;
     use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
     use crate::common::flags::roaring_flags::{RoaringFlags, RoaringFlagsRead};
 
@@ -280,4 +319,85 @@ mod tests_mod {
             assert_eq!(all_indices, expected_all);
         }
     }
+
+    #[test]
+    fn test_compact_mode_roundtrip() {
+        let dir = tempfile::Builder::new()
+            .prefix("roaring_flags_compact")
+            .tempdir()
+            .unwrap();
+
+        {
+            let mut flags = RoaringFlags::<S>::open_or_create(
+                Fs::default(),
+                dir.path(),
+                FlagsMode::Compact,
+                Populate::No,
+            )
+            .unwrap();
+            assert!(!flags.set(3, true));
+            assert!(!flags.set(7, true));
+            assert!(flags.set(7, false)); // previous value
+            assert_eq!(flags.len(), 8);
+
+            let files = flags.files();
+            assert_eq!(files.len(), 1);
+            assert!(files[0].ends_with(COMPACT_FLAGS_FILE));
+
+            flags.flusher()().unwrap();
+        }
+
+        // Requesting dynamic mode on existing flags keeps the compact mode.
+        {
+            let flags = RoaringFlags::<S>::open_or_create(
+                Fs::default(),
+                dir.path(),
+                FlagsMode::Dynamic,
+                Populate::Blocking,
+            )
+            .unwrap();
+            assert_eq!(flags.files().len(), 1);
+            assert_eq!(flags.len(), 8);
+            assert_eq!(flags.count_trues().unwrap(), 1);
+            assert_eq!(flags.iter_trues().unwrap().collect::<Vec<_>>(), vec![3]);
+            assert!(flags.get(3).unwrap());
+            assert!(!flags.get(7).unwrap());
+        }
+    }
+
+    #[test]
+    fn test_dynamic_mode_kept_when_compact_requested() {
+        let dir = tempfile::Builder::new()
+            .prefix("roaring_flags_dynamic")
+            .tempdir()
+            .unwrap();
+
+        {
+            let mut flags = RoaringFlags::<S>::open_or_create(
+                Fs::default(),
+                dir.path(),
+                FlagsMode::Dynamic,
+                Populate::No,
+            )
+            .unwrap();
+            flags.set(1, true);
+            flags.flusher()().unwrap();
+
+            // dynamic stack: flags file plus status file
+            assert!(flags.files().len() > 1);
+        }
+
+        {
+            let flags = RoaringFlags::<S>::open_or_create(
+                Fs::default(),
+                dir.path(),
+                FlagsMode::Compact,
+                Populate::Blocking,
+            )
+            .unwrap();
+            assert!(flags.files().len() > 1);
+            assert_eq!(flags.len(), 2);
+            assert!(flags.get(1).unwrap());
+        }
+    }
 }

lib/segment/src/common/flags/storage.rs

@@ -0,0 +1,63 @@
+use std::path::PathBuf;
+
+use common::types::PointOffsetType;
+use common::universal_io::{UniversalRead, UniversalWrite};
+
+use super::buffered_dynamic_flags::BufferedDynamicFlags;
+use super::compact_stored_flags::CompactStoredFlags;
+use crate::common::Flusher;
+use crate::common::operation_error::OperationResult;
+
+/// Write side of a flags stack, dispatching between the storage modes.
+///
+/// The in-memory read state lives in the wrappers ([`BitvecFlags`] /
+/// [`RoaringFlags`]); this enum only persists their mutations, in the flavor
+/// of the [`FlagsMode`](super::FlagsMode) the flags were created in.
+///
+/// [`BitvecFlags`]: super::bitvec_flags::BitvecFlags
+/// [`RoaringFlags`]: super::roaring_flags::RoaringFlags
+#[derive(Debug)]
+pub(crate) enum FlagsStorage<S: UniversalRead> {
+    /// Mmapped files mutated in place, with changes buffered until flush.
+    Dynamic(BufferedDynamicFlags<S>),
+
+    /// RAM-resident compact bitmask, whole-file rewrite on flush.
+    Compact(CompactStoredFlags<S>),
+}
+
+impl<S> FlagsStorage<S>
+where
+    S: UniversalWrite + Send + 'static,
+    S::Fs: Send + Sync + 'static,
+{
+    /// Record setting the flag at `index`, to be persisted on the next flush.
+    pub fn set(&self, index: PointOffsetType, value: bool) {
+        match self {
+            Self::Dynamic(storage) => storage.buffer_set(index, value),
+            Self::Compact(storage) => {
+                storage.set(index, value);
+            }
+        }
+    }
+
+    pub fn files(&self) -> Vec<PathBuf> {
+        match self {
+            Self::Dynamic(storage) => storage.files(),
+            Self::Compact(storage) => storage.files(),
+        }
+    }
+
+    pub fn flusher(&self) -> Flusher {
+        match self {
+            Self::Dynamic(storage) => storage.flusher(),
+            Self::Compact(storage) => storage.flusher(),
+        }
+    }
+
+    pub fn clear_cache(&self) -> OperationResult<()> {
+        match self {
+            Self::Dynamic(storage) => storage.clear_cache(),
+            Self::Compact(storage) => storage.clear_cache(),
+        }
+    }
+}