Integrate new bitflags structure (#10123)

* Add `FlagsMode::from_feature_flags`, the mode for newly created flags

Compact in serverless-compatible deployments, dynamic otherwise. Only
creation consults it; opening existing flags detects their mode from
disk.

* Support the compact mode in the read-only flags types

Add `ReadOnlyFlags`, the mode-dispatching union of the two read-only
counterparts, serving the shared `RoaringFlagsRead` surface. Teach
`InMemoryBitvecFlags` to detect the mode it opens; its compact
`reload_appended` decodes the whole (small) file, as the format has no
random access.

* Create flags through mode selection in storages and indexes

Vector storage deleted flags and the bool/null indexes now open through
`open_or_create` with the mode from the feature flags: serverless
deployments create compact flags, dedicated ones keep creating dynamic
flags, and existing flags are opened in their detected mode either way.

* Read flags in either mode in the read-only bool and null indexes

`ReadOnlyFlags` shares the `RoaringFlagsRead` surface and the lifecycle
signatures of the roaring type it replaces, so the swap is a type
rename.

* Add TODO to not lock bitmask structure during flush

* `MutableStoredBitmask::save` returns the number of bytes written

Zero when the skip-clean save wrote nothing. Lets wrappers charge the
actual write to a hardware counter.

* Refuse to open compact flags in a dynamic-mode directory

Creating the compact file next to dynamic files would leave a directory
of both modes behind, which every later open rejects — refuse up front
instead. Both production callers already rule the case out through
`FlagsMode::detect`, so this only removes a foot-gun for future callers.

The open-or-eagerly-create logic moves into
`open_or_create_compact_mask`, shared with the update-only writer next.

* Rewrite `UpdateOnlyStoredFlags` onto the compact bitmask

The update-only flags writer now writes the compact mode — a single
roaring-encoded `compact_flags.dat` through `MutableStoredBitmask` —
instead of rewriting the whole padded dynamic file pair every batch. A
flush with no effective changes now writes nothing at all, where the old
writer rewrote the full mask on any `set`.

This also fixes opening serverless-created segments: the old open
eagerly wrote a `status.dat` into directories the writable side had
created in the compact mode, leaving files of both modes behind and
poisoning the directory for every later open.

A directory already holding dynamic-mode flags is refused loudly rather
than kept current or migrated; rebuild the segment to migrate its flags.
Migration may come later.

Drops the now-dead `InMemoryBitvecFlags::into_bitvec` and
`DynamicFlagsStatus::new`, and demotes `file_size_for` to private.

* Run edge tests with serverless feature flags

The edge fixtures ran with default feature flags, building leader shards
with dynamic-mode flags — a configuration edge never serves in
production, and one the update-only flags writer now refuses. It also
hid that the writer poisoned compact directories: no test exercised
update-only writes over a serverless-created shard.

Feature flags are process-global and first-init-wins, so every fixture
in the binary initializes the same serverless set; the manifest test
folds into it, since serverless implies `write_segment_manifest`.

* Don't use sequencial mode for one shot reads

lib/common/common/src/stored_bitmask/mutable.rs

@@ -150,10 +150,12 @@ impl MutableStoredBitmask {
     /// the write entirely when nothing changed since the mask was opened or
     /// last saved.
     ///
+    /// Returns the number of bytes written, zero when the write was skipped.
+    ///
     /// A failed save leaves the mask dirty, so a later retry writes again.
-    pub fn save(&mut self, fs: &impl UniversalWriteFileOps, path: &Path) -> UioResult<()> {
+    pub fn save(&mut self, fs: &impl UniversalWriteFileOps, path: &Path) -> UioResult<usize> {
         if !self.is_dirty() {
-            return Ok(());
+            return Ok(0);
         }
 
         // Run-optimize in place so the encoder can serialize a borrow.
@@ -163,6 +165,6 @@ impl MutableStoredBitmask {
 
         self.changed = false;
         self.persisted_len = Some(self.logical_len);
-        Ok(())
+        Ok(bytes.len())
     }
 }

lib/common/common/src/universal_io/types.rs

@@ -119,6 +119,7 @@ impl Populate {
 #[derive(Copy, Clone, Debug)]
 pub struct OpenOptions {
     pub writeable: bool,
+    // Not needed for one shot reeds
     pub need_sequential: bool,
     /// Populate RAM cache on open, if applicable for this implementation.
     pub populate: Populate,

lib/edge/src/read_only/tests.rs

@@ -28,7 +28,20 @@ use crate::{CountRequest, EdgeConfig, EdgeShard, RetrieveRequestBuilder, ScrollR
 
 pub(crate) const VECTOR_NAME: &str = "edge-ro-test-vector";
 
+/// Initialize the process-global feature flags the way a serverless
+/// deployment runs, which is what every edge shard serves in production.
+///
+/// Feature flags are first-init-wins for the whole process, so every fixture
+/// in this binary initializes the same set to stay deterministic regardless
+/// of test order. The field is private, set through the same route a config
+/// file takes.
+pub(crate) fn init_serverless_feature_flags() {
+    let flags: FeatureFlags = serde_json::from_str(r#"{ "serverless_compatible": true }"#).unwrap();
+    init_feature_flags(flags);
+}
+
 pub(crate) fn test_config() -> EdgeConfig {
+    init_serverless_feature_flags();
     EdgeConfig {
         on_disk_payload: Some(false),
         vectors: HashMap::from([(
@@ -493,15 +506,10 @@ fn manifest_enumerator_requires_manifest() {
 /// its segments, and a follower opened over the same directory loads through it.
 #[test]
 fn leader_writes_manifest_and_follower_loads_it() {
-    let mut flags = FeatureFlags::default();
-    flags.write_segment_manifest = true;
-    init_feature_flags(flags);
-
-    // Another test in this process may have initialized flags first; the assertions below only hold
-    // with the flag enabled.
-    if !common::flags::feature_flags().write_segment_manifest {
-        return;
-    }
+    // Serverless mode implies `write_segment_manifest`, which the assertions
+    // below rely on.
+    init_serverless_feature_flags();
+    assert!(common::flags::feature_flags().write_segment_manifest);
 
     let dir = tempfile::Builder::new()
         .prefix("edge-ro-manifest-write")

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

@@ -6,17 +6,57 @@ use common::mmap::AdviceSetting;
 use common::stored_bitmask::MutableStoredBitmask;
 use common::types::PointOffsetType;
 use common::universal_io::{
-    OkNotFound, OpenOptions, Populate, UniversalRead, UniversalWrite, UniversalWriteFileOps,
+    OkNotFound, OpenOptions, Populate, UniversalRead, UniversalReadFs, UniversalWrite,
+    UniversalWriteFileOps,
 };
 use parking_lot::Mutex;
 use roaring::RoaringBitmap;
 
+use super::mode::FlagsMode;
 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";
 
+/// Open the compact-mode mask in `directory`, creating the directory and a
+/// persisted empty mask when none exists yet.
+///
+/// The eager persist is deliberate: flags directories double as the marker
+/// that their storage exists, and an empty directory would leave the next
+/// open to pick a mode from feature flags rather than from disk.
+///
+/// The caller must have ruled out dynamic-mode flags in the directory first —
+/// see [`FlagsMode::detect`]. Creating the compact file next to dynamic files
+/// would leave a directory of both modes behind, which every later open
+/// rejects.
+pub(super) fn open_or_create_compact_mask<Fs>(
+    fs: &Fs,
+    directory: &Path,
+    populate: Populate,
+) -> OperationResult<MutableStoredBitmask>
+where
+    Fs: UniversalReadFs + UniversalWriteFileOps,
+{
+    fs.create_dir(directory)?;
+    let path = directory.join(COMPACT_FLAGS_FILE);
+
+    let options = OpenOptions {
+        writeable: false,
+        need_sequential: true,
+        populate,
+        advice: AdviceSetting::Global,
+    };
+    match MutableStoredBitmask::open(fs, &path, options, Default::default()).ok_not_found()? {
+        Some(mask) => Ok(mask),
+        None => {
+            let mut mask = MutableStoredBitmask::new(0);
+            mask.save(fs, &path)?;
+            Ok(mask)
+        }
+    }
+}
+
 /// Flags over a single compact stored-bitmask file, rewritten whole on flush.
 ///
 /// The serverless-compatible counterpart of `DynamicStoredFlags` +
@@ -47,32 +87,22 @@ where
     /// 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.
+    /// missing, so [`Self::files`] always exist on disk. Errors when the
+    /// directory holds dynamic-mode flags instead.
     pub fn open(fs: S::Fs, directory: &Path, populate: Populate) -> OperationResult<Self> {
-        fs.create_dir(directory)?;
-        let path = directory.join(COMPACT_FLAGS_FILE);
+        if FlagsMode::detect(&fs, directory)? == Some(FlagsMode::Dynamic) {
+            return Err(OperationError::service_error(format!(
+                "cannot open compact flags in {}: the directory holds flags of the dynamic mode",
+                directory.display(),
+            )));
+        }
 
-        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
-            }
-        };
+        let mask = open_or_create_compact_mask(&fs, directory, populate)?;
 
         Ok(Self {
             mask: Arc::new(Mutex::new(mask)),
             fs: Arc::new(fs),
-            path,
+            path: directory.join(COMPACT_FLAGS_FILE),
             is_alive_flush_lock: IsAliveLock::new(),
         })
     }
@@ -149,6 +179,8 @@ where
                 ));
             };
 
+            // TODO(serverless): we should not lock here during save
+            // TODO(serverless): currently acceptable because readers and writers are completely separate
             mask_arc.lock().save(&*fs, &path)?;
 
             drop(is_alive_flush_guard);
@@ -187,12 +219,21 @@ mod tests_mod {
     use tempfile::TempDir;
 
     use crate::common::flags::compact_stored_flags::CompactStoredFlags;
+    use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
     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_refuses_dynamic_directory() {
+        let dir = TempDir::new().unwrap();
+        DynamicStoredFlags::<S>::open(&Fs::default(), dir.path(), Populate::No).unwrap();
+        let result = CompactStoredFlags::<S>::open(Fs::default(), dir.path(), Populate::No);
+        assert!(result.is_err(), "expected refusal on a dynamic directory");
+    }
+
     #[test]
     fn open_creates_file_and_lists_it() {
         let dir = TempDir::new().unwrap();

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

@@ -41,14 +41,6 @@ pub struct DynamicFlagsStatus {
 }
 
 impl DynamicFlagsStatus {
-    /// The status of a storage holding `len` flags.
-    pub(super) fn new(len: usize) -> Self {
-        Self {
-            len,
-            current_file_id: 0,
-        }
-    }
-
     /// Number of logical flags (bits) stored.
     pub fn len(&self) -> usize {
         self.len
@@ -92,7 +84,7 @@ impl<S: fmt::Debug> fmt::Debug for DynamicStoredFlags<S> {
 }
 
 /// Based on the number of flags determines the size in bytes for the storage file.
-pub(super) fn file_size_for(num_flags: usize) -> usize {
+fn file_size_for(num_flags: usize) -> usize {
     let number_of_bytes = num_flags.div_ceil(u8::BITS as usize);
 
     max(MINIMAL_MMAP_SIZE, number_of_bytes.next_power_of_two())

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

@@ -3,13 +3,16 @@ use std::path::{Path, PathBuf};
 use common::bitvec::{BitSlice, BitVec};
 use common::mmap::AdviceSetting;
 use common::sorted_slice::SortedSlice;
+use common::stored_bitmask::StoredBitmask;
 use common::stored_bitslice::StoredBitSlice;
 use common::types::PointOffsetType;
 use common::universal_io::{
     CachedReadFs, OpenOptions, Populate, TypedStorage, UniversalRead, UniversalReadFs,
 };
 
+use super::compact_stored_flags::COMPACT_FLAGS_FILE;
 use super::dynamic_stored_flags::{DynamicFlagsStatus, FLAGS_FILE, status_file};
+use super::mode::FlagsMode;
 use crate::common::operation_error::{OperationError, OperationResult};
 
 /// In-memory counterpart of `BitvecFlags`: persisted flags materialized into an
@@ -30,9 +33,10 @@ pub struct InMemoryBitvecFlags {
     bitvec: BitVec,
     /// Set-flag count, kept in sync with `bitvec`.
     count: usize,
-    /// Backing directory of the dynamic flags file, so [`Self::reload_appended`]
-    /// can reopen it. `None` for flags built via [`Self::from_bitvec`].
-    directory: Option<PathBuf>,
+    /// Backing directory and [mode](FlagsMode) of the persisted flags, so
+    /// [`Self::reload_appended`] can reopen them. `None` for flags built via
+    /// [`Self::from_bitvec`].
+    backing: Option<(PathBuf, FlagsMode)>,
 }
 
 /// Read-only mmap options: never writable, lazily paged, nothing populated.
@@ -45,22 +49,45 @@ fn bitslice_open_options(populate: Populate) -> OpenOptions {
     }
 }
 
-impl InMemoryBitvecFlags {
-    /// Schedule background prefetch of the two files [`Self::open`] reads.
-    pub fn preopen(fs: &impl CachedReadFs, directory: &Path) -> OperationResult<()> {
-        // Status file
-        fs.schedule_prefetch(
-            &status_file(directory),
-            Some(bitslice_open_options(Populate::PreferBackground)),
-            None,
-        )?;
+/// Read-only options for the compact bitmask, which is decoded whole.
+fn compact_open_options(populate: Populate) -> OpenOptions {
+    OpenOptions {
+        writeable: false,
+        // Not needed for one shot reeds
+        need_sequential: false,
+        populate,
+        advice: AdviceSetting::Global,
+    }
+}
 
-        // Bitslice
-        fs.schedule_prefetch(
-            &directory.join(FLAGS_FILE),
-            Some(bitslice_open_options(Populate::PreferBackground)),
-            None,
-        )?;
+impl InMemoryBitvecFlags {
+    /// Schedule background prefetch of the files [`Self::open`] reads, in the
+    /// detected [mode](FlagsMode).
+    pub fn preopen(fs: &impl CachedReadFs, directory: &Path) -> OperationResult<()> {
+        match FlagsMode::detect(fs, directory)?.unwrap_or(FlagsMode::Dynamic) {
+            FlagsMode::Dynamic => {
+                // Status file
+                fs.schedule_prefetch(
+                    &status_file(directory),
+                    Some(bitslice_open_options(Populate::PreferBackground)),
+                    None,
+                )?;
+
+                // Bitslice
+                fs.schedule_prefetch(
+                    &directory.join(FLAGS_FILE),
+                    Some(bitslice_open_options(Populate::PreferBackground)),
+                    None,
+                )?;
+            }
+            FlagsMode::Compact => {
+                fs.schedule_prefetch(
+                    &directory.join(COMPACT_FLAGS_FILE),
+                    Some(compact_open_options(Populate::PreferBackground)),
+                    None,
+                )?;
+            }
+        }
         Ok(())
     }
 
@@ -82,12 +109,24 @@ impl InMemoryBitvecFlags {
             .map_or(0, DynamicFlagsStatus::len))
     }
 
-    /// Open persisted flags read-only into an owned `BitVec`; creates and writes
-    /// nothing. The flags file is padded past the logical length (held in the
-    /// status file), so the bitvec is truncated to it and `count` is exact.
+    /// Open persisted flags read-only into an owned `BitVec`, in their
+    /// detected [mode](FlagsMode); creates and writes nothing.
     pub fn open<S: UniversalRead>(
         fs: &impl UniversalReadFs<File = S>,
         directory: &Path,
+    ) -> OperationResult<Self> {
+        match FlagsMode::detect(fs, directory)?.unwrap_or(FlagsMode::Dynamic) {
+            FlagsMode::Dynamic => Self::open_dynamic(fs, directory),
+            FlagsMode::Compact => Self::open_compact(fs, directory),
+        }
+    }
+
+    /// [`Self::open`] for the dynamic mode. The flags file is padded past the
+    /// logical length (held in the status file), so the bitvec is truncated to
+    /// it and `count` is exact.
+    fn open_dynamic<S: UniversalRead>(
+        fs: &impl UniversalReadFs<File = S>,
+        directory: &Path,
     ) -> OperationResult<Self> {
         let len = Self::persisted_len(fs, directory)?;
 
@@ -110,7 +149,33 @@ impl InMemoryBitvecFlags {
         Ok(Self {
             bitvec,
             count,
-            directory: Some(directory.to_path_buf()),
+            backing: Some((directory.to_path_buf(), FlagsMode::Dynamic)),
+        })
+    }
+
+    /// [`Self::open`] for the compact mode: the single bitmask file is decoded
+    /// whole into the bitvec.
+    fn open_compact<S: UniversalRead>(
+        fs: &impl UniversalReadFs<File = S>,
+        directory: &Path,
+    ) -> OperationResult<Self> {
+        let mask = StoredBitmask::<S>::open(
+            fs,
+            directory.join(COMPACT_FLAGS_FILE),
+            compact_open_options(Populate::No),
+            Default::default(),
+        )?;
+        let ones = mask.read_ones()?;
+
+        let mut bitvec = BitVec::repeat(false, mask.bit_len() as usize);
+        for index in &ones {
+            bitvec.set(index as usize, true);
+        }
+
+        Ok(Self {
+            bitvec,
+            count: ones.len() as usize,
+            backing: Some((directory.to_path_buf(), FlagsMode::Compact)),
         })
     }
 
@@ -122,7 +187,7 @@ impl InMemoryBitvecFlags {
         Self {
             bitvec,
             count,
-            directory: None,
+            backing: None,
         }
     }
 
@@ -141,11 +206,6 @@ impl InMemoryBitvecFlags {
         self.bitvec.as_bitslice()
     }
 
-    /// Set flags as an owned `BitVec`, dropping the count.
-    pub fn into_bitvec(self) -> BitVec {
-        self.bitvec
-    }
-
     /// Set `points`, growing as needed and keeping `count` in sync. Folds a
     /// live-reload deletion delta; live offsets aren't passed (they read unset).
     pub fn insert_all(&mut self, points: &[PointOffsetType]) {
@@ -171,14 +231,29 @@ impl InMemoryBitvecFlags {
         fs: &impl UniversalReadFs<File = S>,
         new_points: &SortedSlice<'_, PointOffsetType>,
     ) -> OperationResult<()> {
-        let Some(directory) = self.directory.clone() else {
+        let Some((directory, mode)) = self.backing.clone() else {
             return Ok(());
         };
+        match mode {
+            FlagsMode::Dynamic => self.reload_appended_dynamic(fs, &directory, new_points),
+            FlagsMode::Compact => self.reload_appended_compact(fs, &directory, new_points),
+        }
+    }
+
+    /// [`Self::reload_appended`] for the dynamic mode. `new_points` is sorted,
+    /// so the covering range up to the persisted length is read in one batched
+    /// read.
+    fn reload_appended_dynamic<S: UniversalRead>(
+        &mut self,
+        fs: &impl UniversalReadFs<File = S>,
+        directory: &Path,
+        new_points: &SortedSlice<'_, PointOffsetType>,
+    ) -> OperationResult<()> {
         let (Some(&first), Some(&last)) = (new_points.first(), new_points.last()) else {
             return Ok(());
         };
 
-        let len = Self::persisted_len(fs, &directory)? as u64;
+        let len = Self::persisted_len(fs, directory)? as u64;
         let start = u64::from(first);
         let end = u64::from(last).saturating_add(1).min(len);
         if start >= end {
@@ -204,6 +279,37 @@ impl InMemoryBitvecFlags {
 
         Ok(())
     }
+
+    /// [`Self::reload_appended`] for the compact mode. The bitmask has no
+    /// random access, so the (small) file is decoded whole and the appended
+    /// offsets' bits are folded in from it.
+    fn reload_appended_compact<S: UniversalRead>(
+        &mut self,
+        fs: &impl UniversalReadFs<File = S>,
+        directory: &Path,
+        new_points: &SortedSlice<'_, PointOffsetType>,
+    ) -> OperationResult<()> {
+        if new_points.is_empty() {
+            return Ok(());
+        }
+
+        let mask = StoredBitmask::<S>::open(
+            fs,
+            directory.join(COMPACT_FLAGS_FILE),
+            compact_open_options(Populate::No),
+            Default::default(),
+        )?;
+        let ones = mask.read_ones()?;
+
+        let deleted: Vec<_> = new_points
+            .iter()
+            .copied()
+            .filter(|&point| ones.contains(point))
+            .collect();
+        self.insert_all(&deleted);
+
+        Ok(())
+    }
 }
 
 #[allow(clippy::default_constructed_unit_structs)]
@@ -224,6 +330,7 @@ mod tests_mod {
     use tempfile::Builder;
 
     use super::*;
+    use crate::common::flags::compact_stored_flags::CompactStoredFlags;
     use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
 
     /// Persist `flags` via the writable storage so the read-only path can open it.
@@ -284,4 +391,37 @@ mod tests_mod {
         assert!(flags.get(beyond));
         assert!(!flags.get(beyond + 1));
     }
+
+    #[test]
+    fn open_compact_mode_and_reload_appended() {
+        let dir = Builder::new().prefix("storage_dir").tempdir().unwrap();
+
+        // Persist compact-mode flags via the writable wrapper.
+        let writer =
+            CompactStoredFlags::<S>::open(Fs::default(), dir.path(), Populate::No).unwrap();
+        writer.set(3, true);
+        writer.set(9, false); // grows the flags to 10
+        writer.flusher()().unwrap();
+
+        let mut flags = InMemoryBitvecFlags::open::<S>(&Fs::default(), dir.path()).unwrap();
+        assert_eq!(flags.count(), 1);
+        assert!(flags.get(3));
+        assert!(!flags.get(9));
+        assert_eq!(flags.as_bitslice().len(), 10);
+
+        // Append points 10..=12; one carries a deletion recorded only in the
+        // persisted flags.
+        writer.set(11, true);
+        writer.set(12, false);
+        writer.flusher()().unwrap();
+
+        let new_points = [10, 11, 12];
+        flags
+            .reload_appended::<S>(&Fs::default(), &SortedSlice::new(&new_points).unwrap())
+            .unwrap();
+        assert!(flags.get(11));
+        assert!(!flags.get(10));
+        assert!(!flags.get(12));
+        assert_eq!(flags.count(), 2);
+    }
 }

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

@@ -11,10 +11,13 @@
 //!   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`.
+//! - `read_only_flags`: mode-dispatching union of `read_only_roaring_flags` and
+//!   `read_only_compact_flags`.
 //!
 //! `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.
+//! automatically when opening existing flags. The read-only types (`read_only_flags`,
+//! `in_memory_bitvec_flags`) likewise detect the mode of the flags they open.
 
 pub mod bitvec_flags;
 mod buffered_dynamic_flags;
@@ -23,6 +26,7 @@ pub mod dynamic_stored_flags;
 pub mod in_memory_bitvec_flags;
 mod mode;
 pub mod read_only_compact_flags;
+pub mod read_only_flags;
 pub mod read_only_roaring_flags;
 pub mod roaring_flags;
 mod storage;

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

@@ -1,5 +1,6 @@
 use std::path::Path;
 
+use common::flags::feature_flags;
 use common::universal_io::UniversalReadFileOps;
 
 use super::compact_stored_flags::COMPACT_FLAGS_FILE;
@@ -26,6 +27,19 @@ pub enum FlagsMode {
 }
 
 impl FlagsMode {
+    /// Mode for newly created flags, per the deployment's feature flags:
+    /// compact in serverless-compatible deployments, dynamic otherwise.
+    ///
+    /// Only relevant when creating flags; opening existing flags detects
+    /// their mode from disk regardless of this value.
+    pub fn from_feature_flags() -> Self {
+        if feature_flags().serverless_compatible() {
+            Self::Compact
+        } else {
+            Self::Dynamic
+        }
+    }
+
     /// Detect the mode of the flags in `directory` from the files present,
     /// `None` when no flags exist there yet.
     ///
@@ -97,9 +111,12 @@ mod tests_mod {
 
     #[test]
     fn errors_on_files_of_both_modes() {
+        // Compact first: opening compact flags in a dynamic directory is
+        // refused outright, so this order is the only way to end up with
+        // 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();
+        DynamicStoredFlags::<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_flags.rs

@@ -0,0 +1,156 @@
+use std::path::{Path, PathBuf};
+
+use common::universal_io::{CachedReadFs, Populate, UniversalRead, UniversalReadFs};
+use roaring::RoaringBitmap;
+
+use super::mode::FlagsMode;
+use super::read_only_compact_flags::ReadOnlyCompactFlags;
+use super::read_only_roaring_flags::ReadOnlyRoaringFlags;
+use super::roaring_flags::RoaringFlagsRead;
+use crate::common::operation_error::OperationResult;
+
+/// Read-only flags in either [storage mode](FlagsMode), serving the
+/// [`RoaringFlagsRead`] surface.
+///
+/// Read-side counterpart of the mode dispatch inside the writable wrappers:
+/// [`Self::open`] detects the mode from the files present and every operation
+/// forwards to that variant.
+pub enum ReadOnlyFlags<S: UniversalRead> {
+    /// Flags in the dynamic (mmapped, mutated in place) format.
+    Dynamic(ReadOnlyRoaringFlags<S>),
+
+    /// Flags in the compact stored-bitmask format.
+    Compact(ReadOnlyCompactFlags<S>),
+}
+
+impl<S: UniversalRead> ReadOnlyFlags<S> {
+    /// Schedule background prefetch of the files [`Self::open`] reads, in the
+    /// detected mode.
+    ///
+    /// Returns whether the flags exist.
+    pub fn preopen(
+        fs: &impl CachedReadFs<File = S>,
+        directory: &Path,
+        populate: Populate,
+    ) -> OperationResult<bool> {
+        match FlagsMode::detect(fs, directory)? {
+            None => Ok(false),
+            Some(FlagsMode::Dynamic) => ReadOnlyRoaringFlags::<S>::preopen(fs, directory, populate),
+            Some(FlagsMode::Compact) => ReadOnlyCompactFlags::<S>::preopen(fs, directory, populate),
+        }
+    }
+
+    /// Open persisted flags read-only, in their detected mode.
+    ///
+    /// Returns [`Ok(None)`] when no flags exist in `directory`, matching the
+    /// read path's never-create contract.
+    pub fn open(
+        fs: &impl UniversalReadFs<File = S>,
+        directory: &Path,
+    ) -> OperationResult<Option<Self>> {
+        match FlagsMode::detect(fs, directory)? {
+            None => Ok(None),
+            Some(FlagsMode::Dynamic) => {
+                Ok(ReadOnlyRoaringFlags::open(fs, directory)?.map(Self::Dynamic))
+            }
+            Some(FlagsMode::Compact) => {
+                Ok(ReadOnlyCompactFlags::open(fs, directory)?.map(Self::Compact))
+            }
+        }
+    }
+
+    /// Refresh to the current on-disk state; see the variants' impls for the
+    /// respective reload semantics.
+    pub fn live_reload(&mut self, fs: &impl UniversalReadFs<File = S>) -> OperationResult<()> {
+        match self {
+            Self::Dynamic(flags) => flags.live_reload(fs),
+            Self::Compact(flags) => flags.live_reload(fs),
+        }
+    }
+}
+
+impl<S: UniversalRead> RoaringFlagsRead for ReadOnlyFlags<S> {
+    fn len(&self) -> usize {
+        match self {
+            Self::Dynamic(flags) => flags.len(),
+            Self::Compact(flags) => flags.len(),
+        }
+    }
+
+    fn get_bitmap(&self) -> OperationResult<&RoaringBitmap> {
+        match self {
+            Self::Dynamic(flags) => flags.get_bitmap(),
+            Self::Compact(flags) => flags.get_bitmap(),
+        }
+    }
+
+    fn bitmap_if_materialized(&self) -> Option<&RoaringBitmap> {
+        match self {
+            Self::Dynamic(flags) => flags.bitmap_if_materialized(),
+            Self::Compact(flags) => flags.bitmap_if_materialized(),
+        }
+    }
+
+    fn files(&self) -> Vec<PathBuf> {
+        match self {
+            Self::Dynamic(flags) => flags.files(),
+            Self::Compact(flags) => flags.files(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use common::universal_io::{MmapFile, MmapFs};
+    use tempfile::TempDir;
+
+    use super::*;
+    use crate::common::flags::compact_stored_flags::CompactStoredFlags;
+    use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
+
+    #[test]
+    fn open_dispatches_on_detected_mode() {
+        let tmp = TempDir::new().unwrap();
+
+        // Nothing on disk: no flags.
+        let missing = tmp.path().join("missing");
+        assert!(
+            ReadOnlyFlags::<MmapFile>::open(&MmapFs, &missing)
+                .unwrap()
+                .is_none()
+        );
+
+        // Dynamic-mode flags.
+        let dynamic_dir = tmp.path().join("dynamic");
+        {
+            let mut writer =
+                DynamicStoredFlags::<MmapFile>::open(&MmapFs, &dynamic_dir, Populate::No).unwrap();
+            writer.set_len(&MmapFs, 10).unwrap();
+            writer.set(3, true).unwrap();
+            writer.flusher()().unwrap();
+        }
+        let flags = ReadOnlyFlags::<MmapFile>::open(&MmapFs, &dynamic_dir)
+            .unwrap()
+            .unwrap();
+        assert!(matches!(flags, ReadOnlyFlags::Dynamic(_)));
+        assert_eq!(flags.len(), 10);
+        assert!(flags.get(3).unwrap());
+        assert_eq!(flags.count_trues().unwrap(), 1);
+
+        // Compact-mode flags.
+        let compact_dir = tmp.path().join("compact");
+        {
+            let writer =
+                CompactStoredFlags::<MmapFile>::open(MmapFs, &compact_dir, Populate::No).unwrap();
+            writer.set(7, true);
+            writer.flusher()().unwrap();
+        }
+        let flags = ReadOnlyFlags::<MmapFile>::open(&MmapFs, &compact_dir)
+            .unwrap()
+            .unwrap();
+        assert!(matches!(flags, ReadOnlyFlags::Compact(_)));
+        assert_eq!(flags.len(), 8);
+        assert!(flags.get(7).unwrap());
+        assert_eq!(flags.files().len(), 1);
+    }
+}

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

@@ -1,61 +1,58 @@
 use std::path::{Path, PathBuf};
 
-use common::bitvec::BitVec;
 use common::counter::hardware_counter::HardwareCounterCell;
+use common::stored_bitmask::MutableStoredBitmask;
 use common::types::PointOffsetType;
-use common::universal_io::{
-    OkNotFound as _, UniversalAppend, UniversalReadFileOps as _, UniversalWriteFileOps as _,
-};
+use common::universal_io::{Populate, UniversalAppend};
 
-use super::dynamic_stored_flags::{DynamicFlagsStatus, FLAGS_FILE, file_size_for, status_file};
-use super::in_memory_bitvec_flags::InMemoryBitvecFlags;
-use crate::common::operation_error::OperationResult;
+use super::compact_stored_flags::{COMPACT_FLAGS_FILE, open_or_create_compact_mask};
+use super::mode::FlagsMode;
+use crate::common::operation_error::{OperationError, OperationResult};
 
 /// Short-lived writer for the persisted flags of an update-only segment,
 /// opened for one batch and dropped with it.
 ///
-/// Writes the same two files [`DynamicStoredFlags`][1] does, but rewrites the
-/// mask whole: a bitmask over all points cannot be kept current by appending.
-/// The cost is one bit per point of the segment per batch, which is why only
-/// the bitmask-backed indexes use this.
+/// The update-only counterpart of [`CompactStoredFlags`][1]: the mask is
+/// fully resident in RAM, mutations collect there, and a flush rewrites the
+/// single compact file in one atomic whole-file write — a bitmask over all
+/// points cannot be kept current by appending. A flush with no effective
+/// changes writes nothing.
 ///
-/// [1]: super::dynamic_stored_flags::DynamicStoredFlags
+/// Only the compact storage mode is supported. Opening a directory that
+/// holds dynamic-mode flags fails loudly rather than going stale or leaving
+/// files of both modes behind; rebuild such a segment to migrate its flags.
+///
+/// [1]: super::compact_stored_flags::CompactStoredFlags
 pub struct UpdateOnlyStoredFlags<S: UniversalAppend + 'static> {
     fs: S::Fs,
-    directory: PathBuf,
-    /// Every flag the storage holds, materialized on open. Its length is the
-    /// logical flag count, which the status file publishes.
-    flags: BitVec,
-    dirty: bool,
+    /// Path of the mask file inside the flags directory.
+    path: PathBuf,
+    /// The mask, resident in RAM; tracks its own effective dirtiness.
+    mask: MutableStoredBitmask,
 }
 
 impl<S: UniversalAppend + 'static> UpdateOnlyStoredFlags<S> {
     /// Read the flags at `directory` into memory, ready to be extended. A
-    /// directory that holds none yet opens empty, and is created — status file
-    /// included — right away.
+    /// directory that holds none yet opens empty, and is created — mask file
+    /// included — right away, because storages take the directory as the
+    /// marker that they exist at all.
+    ///
+    /// Errors when the directory holds flags of the dynamic mode.
     pub fn open(fs: S::Fs, directory: &Path) -> OperationResult<Self> {
-        // Materialize the directory on the first open rather than on the first
-        // flag. Storages use it as the marker that they exist at all — the
-        // sparse one takes its absence for "not created yet" and starts over —
-        // so a batch that flags nothing must still leave it behind.
-        if !fs.exists(&status_file(directory))? {
-            fs.create_dir(directory)?;
-            fs.atomic_save(
-                &status_file(directory),
-                bytemuck::bytes_of(&DynamicFlagsStatus::new(0)),
-            )?;
+        if FlagsMode::detect(&fs, directory)? == Some(FlagsMode::Dynamic) {
+            return Err(OperationError::service_error(format!(
+                "flags in {} are in the dynamic mode, which the update-only writer does not \
+                 support; rebuild the segment to migrate its flags",
+                directory.display(),
+            )));
         }
 
-        let flags = InMemoryBitvecFlags::open::<S>(&fs, directory)
-            .ok_not_found()?
-            .map(InMemoryBitvecFlags::into_bitvec)
-            .unwrap_or_default();
+        let mask = open_or_create_compact_mask(&fs, directory, Populate::Blocking)?;
 
         Ok(Self {
             fs,
-            directory: directory.to_owned(),
-            flags,
-            dirty: false,
+            path: directory.join(COMPACT_FLAGS_FILE),
+            mask,
         })
     }
 
@@ -64,40 +61,180 @@ impl<S: UniversalAppend + 'static> UpdateOnlyStoredFlags<S> {
     /// Extends for a `false` as much as for a `true`: the mask's length is what
     /// says which points it has an answer for.
     pub fn set(&mut self, slot: PointOffsetType, value: bool) {
-        let slot = slot as usize;
-        if slot >= self.flags.len() {
-            self.flags.resize(slot + 1, false);
+        if u64::from(slot) >= self.mask.bit_len() {
+            self.mask.set_len(u64::from(slot) + 1);
         }
-        self.flags.set(slot, value);
-        self.dirty = true;
+        self.mask.set(slot, value);
     }
 
-    /// Write both files, the mask before the length that publishes it, so that
-    /// a torn batch never leaves a length pointing past the written flags.
+    /// Persist the mask in one atomic whole-file write, or write nothing when
+    /// it has not effectively changed since it was opened or last flushed.
     pub fn flush(&mut self, hw_counter: &HardwareCounterCell) -> OperationResult<()> {
-        if !self.dirty {
-            return Ok(());
-        }
-
-        self.fs.create_dir(&self.directory)?;
-
-        // Padded to the size the writable side would have allocated, so that a
-        // writer of either kind can go on from here without resizing.
-        let len = self.flags.len();
-        let mut bytes = bytemuck::cast_slice(self.flags.as_raw_slice()).to_vec();
-        bytes.resize(file_size_for(len), 0);
-        self.fs
-            .atomic_save(&self.directory.join(FLAGS_FILE), &bytes)?;
-
-        let status = DynamicFlagsStatus::new(len);
-        self.fs
-            .atomic_save(&status_file(&self.directory), bytemuck::bytes_of(&status))?;
-
+        let bytes_written = self.mask.save(&self.fs, &self.path)?;
         hw_counter
             .payload_index_io_write_counter()
-            .incr_delta(bytes.len() + size_of::<DynamicFlagsStatus>());
-
-        self.dirty = false;
+            .incr_delta(bytes_written);
         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::counter::hardware_counter::HardwareCounterCell;
+    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::{COMPACT_FLAGS_FILE, CompactStoredFlags};
+    use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
+    use crate::common::flags::update_only_stored_flags::UpdateOnlyStoredFlags;
+
+    fn open(dir: &std::path::Path) -> UpdateOnlyStoredFlags<S> {
+        UpdateOnlyStoredFlags::open(Fs::default(), dir).unwrap()
+    }
+
+    fn flush(flags: &mut UpdateOnlyStoredFlags<S>) {
+        flags.flush(&HardwareCounterCell::new()).unwrap();
+    }
+
+    /// Reader for what the writer left behind, through the type the writable
+    /// segment side uses on the same directory.
+    fn read_back(dir: &std::path::Path) -> CompactStoredFlags<S> {
+        CompactStoredFlags::open(Fs::default(), dir, Populate::Blocking).unwrap()
+    }
+
+    #[test]
+    fn open_creates_mask_file_eagerly() {
+        // The flags directory is the marker that its storage exists, so a
+        // batch that flags nothing must still leave it behind, mask included.
+        let dir = TempDir::new().unwrap();
+        let _flags = open(dir.path());
+        assert!(dir.path().join(COMPACT_FLAGS_FILE).exists());
+        assert_eq!(
+            FlagsMode::detect(&Fs::default(), dir.path()).unwrap(),
+            Some(FlagsMode::Compact),
+        );
+    }
+
+    #[test]
+    fn set_flush_reopen_roundtrip() {
+        let dir = TempDir::new().unwrap();
+        {
+            let mut flags = open(dir.path());
+            flags.set(3, true);
+            flags.set(100, false); // grows to 101 without setting anything
+            flush(&mut flags);
+        }
+        let reader = read_back(dir.path());
+        assert_eq!(reader.len(), 101);
+        assert_eq!(reader.count_flags(), 1);
+        assert!(reader.get(3));
+        assert!(!reader.get(100));
+    }
+
+    #[test]
+    fn clean_flush_writes_nothing() {
+        let dir = TempDir::new().unwrap();
+        let mut flags = open(dir.path());
+        flags.set(1, true);
+        flush(&mut flags);
+
+        // 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(dir.path().join(COMPACT_FLAGS_FILE)).unwrap();
+        flush(&mut flags);
+        assert!(!dir.path().join(COMPACT_FLAGS_FILE).exists());
+
+        // Re-setting a flag to its value is not an effective change either.
+        flags.set(1, true);
+        flush(&mut flags);
+        assert!(!dir.path().join(COMPACT_FLAGS_FILE).exists());
+
+        // The next effective change rewrites the whole mask.
+        flags.set(0, true);
+        flush(&mut flags);
+        let reader = read_back(dir.path());
+        assert!(reader.get(0));
+        assert!(reader.get(1));
+    }
+
+    #[test]
+    fn reopen_extends_previous_batch() {
+        let dir = TempDir::new().unwrap();
+        {
+            let mut flags = open(dir.path());
+            flags.set(0, true);
+            flags.set(5, true);
+            flush(&mut flags);
+        }
+        {
+            let mut flags = open(dir.path());
+            flags.set(5, false);
+            flags.set(9, true);
+            flush(&mut flags);
+        }
+        let reader = read_back(dir.path());
+        assert_eq!(reader.len(), 10);
+        assert!(reader.get(0));
+        assert!(!reader.get(5));
+        assert!(reader.get(9));
+    }
+
+    #[test]
+    fn extends_flags_created_by_the_writable_side() {
+        // A serverless-created segment arrives with compact flags written by
+        // the writable side; the update-only writer must pick them up and
+        // leave the directory in the compact mode.
+        let dir = TempDir::new().unwrap();
+        {
+            let writable: CompactStoredFlags<S> =
+                CompactStoredFlags::open(Fs::default(), dir.path(), Populate::Blocking).unwrap();
+            writable.set(2, true);
+            writable.flusher()().unwrap();
+        }
+        {
+            let mut flags = open(dir.path());
+            flags.set(4, true);
+            flush(&mut flags);
+        }
+        assert_eq!(
+            FlagsMode::detect(&Fs::default(), dir.path()).unwrap(),
+            Some(FlagsMode::Compact),
+        );
+        let reader = read_back(dir.path());
+        assert_eq!(reader.len(), 5);
+        assert!(reader.get(2));
+        assert!(reader.get(4));
+    }
+
+    #[test]
+    fn refuses_dynamic_directory() {
+        let dir = TempDir::new().unwrap();
+        DynamicStoredFlags::<S>::open(&Fs::default(), dir.path(), Populate::No).unwrap();
+
+        let result = UpdateOnlyStoredFlags::<S>::open(Fs::default(), dir.path());
+        let message = result
+            .err()
+            .expect("dynamic flags must be refused")
+            .to_string();
+        assert!(
+            message.contains("dynamic"),
+            "refusal should name the dynamic mode: {message}",
+        );
+
+        // The refused directory is left untouched: still cleanly dynamic.
+        assert_eq!(
+            FlagsMode::detect(&Fs::default(), dir.path()).unwrap(),
+            Some(FlagsMode::Dynamic),
+        );
+    }
+}

lib/segment/src/index/field_index/bool_index/mutable_bool_index/lifecycle.rs

@@ -8,7 +8,7 @@ use fs_err as fs;
 
 use super::super::read_ops::BoolIndexRead;
 use super::{FALSES_DIRNAME, MutableBoolIndex, Storage, TRUES_DIRNAME};
-use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
+use crate::common::flags::FlagsMode;
 use crate::common::flags::roaring_flags::{RoaringFlags, RoaringFlagsRead};
 use crate::common::operation_error::{OperationError, OperationResult};
 use crate::index::field_index::{FieldIndexBuilderTrait, PayloadFieldIndex, ValueIndexer};
@@ -46,15 +46,21 @@ impl MutableBoolIndex {
             ))
         })?;
 
-        // Trues bitslice
-        let trues_path = path.join(TRUES_DIRNAME);
-        let trues_slice = DynamicStoredFlags::open(&MmapFs, &trues_path, Populate::No)?;
-        let trues_flags = RoaringFlags::new(MmapFs, trues_slice)?;
+        // Trues flags
+        let trues_flags = RoaringFlags::open_or_create(
+            MmapFs,
+            &path.join(TRUES_DIRNAME),
+            FlagsMode::from_feature_flags(),
+            Populate::No,
+        )?;
 
-        // Falses bitslice
-        let falses_path = path.join(FALSES_DIRNAME);
-        let falses_slice = DynamicStoredFlags::open(&MmapFs, &falses_path, Populate::No)?;
-        let falses_flags = RoaringFlags::new(MmapFs, falses_slice)?;
+        // Falses flags
+        let falses_flags = RoaringFlags::open_or_create(
+            MmapFs,
+            &path.join(FALSES_DIRNAME),
+            FlagsMode::from_feature_flags(),
+            Populate::No,
+        )?;
 
         // Infallible for the writable variant: its bitmaps are materialized by
         // `RoaringFlags::new` above.

lib/segment/src/index/field_index/bool_index/read_only_bool_index/lifecycle.rs

@@ -5,7 +5,7 @@ use common::universal_io::{CachedReadFs, Populate, UniversalReadFs};
 
 use super::super::mutable_bool_index::{FALSES_DIRNAME, TRUES_DIRNAME};
 use super::{ReadOnlyBoolIndex, ReadOnlyStorage};
-use crate::common::flags::read_only_roaring_flags::ReadOnlyRoaringFlags;
+use crate::common::flags::read_only_flags::ReadOnlyFlags;
 use crate::common::operation_error::{OperationError, OperationResult};
 use crate::index::UniversalReadExt;
 
@@ -22,8 +22,8 @@ impl<S: UniversalReadExt> ReadOnlyBoolIndex<S> {
         path: &Path,
         populate: Populate,
     ) -> OperationResult<bool> {
-        let trues = ReadOnlyRoaringFlags::<S>::preopen(fs, &path.join(TRUES_DIRNAME), populate)?;
-        let falses = ReadOnlyRoaringFlags::<S>::preopen(fs, &path.join(FALSES_DIRNAME), populate)?;
+        let trues = ReadOnlyFlags::<S>::preopen(fs, &path.join(TRUES_DIRNAME), populate)?;
+        let falses = ReadOnlyFlags::<S>::preopen(fs, &path.join(FALSES_DIRNAME), populate)?;
         Ok(trues || falses)
     }
 
@@ -46,8 +46,8 @@ impl<S: UniversalReadExt> ReadOnlyBoolIndex<S> {
     pub fn open(fs: &impl UniversalReadFs<File = S>, path: &Path) -> OperationResult<Option<Self>> {
         // Open both directories first so a partial layout can be distinguished
         // from a genuinely absent index, regardless of which half is missing.
-        let trues_flags = ReadOnlyRoaringFlags::<S>::open(fs, &path.join(TRUES_DIRNAME))?;
-        let falses_flags = ReadOnlyRoaringFlags::<S>::open(fs, &path.join(FALSES_DIRNAME))?;
+        let trues_flags = ReadOnlyFlags::<S>::open(fs, &path.join(TRUES_DIRNAME))?;
+        let falses_flags = ReadOnlyFlags::<S>::open(fs, &path.join(FALSES_DIRNAME))?;
 
         match (trues_flags, falses_flags) {
             // Neither directory exists: the index isn't present on disk.

lib/segment/src/index/field_index/bool_index/read_only_bool_index/mod.rs

@@ -3,7 +3,7 @@ use std::sync::OnceLock;
 
 use roaring::RoaringBitmap;
 
-use crate::common::flags::read_only_roaring_flags::ReadOnlyRoaringFlags;
+use crate::common::flags::read_only_flags::ReadOnlyFlags;
 use crate::common::flags::roaring_flags::RoaringFlagsRead;
 use crate::common::operation_error::OperationResult;
 use crate::index::UniversalReadExt;
@@ -21,9 +21,9 @@ mod read_ops;
 /// condition checker / faceting) is shared with the writable variants via
 /// [`BoolIndexRead`][4].
 ///
-/// Each [`ReadOnlyRoaringFlags`] retains its backend `S` so a [`LiveReload`][5]
-/// can reopen the bitslice and apply only the changed positions, instead of
-/// re-materializing the bitmaps from scratch.
+/// Each [`ReadOnlyFlags`] retains its backend `S` so a [`LiveReload`][5]
+/// can reopen the backing files and resync, instead of re-materializing the
+/// bitmaps from scratch.
 ///
 /// [1]: super::mutable_bool_index::MutableBoolIndex
 /// [2]: super::immutable_bool_index::ImmutableBoolIndex
@@ -37,7 +37,7 @@ pub struct ReadOnlyBoolIndex<S: UniversalReadExt> {
     ///
     /// Deriving them eagerly would force both bitmaps to materialize at open,
     /// scanning each flags file end to end — exactly what the lazy
-    /// [`ReadOnlyRoaringFlags`] bitmap exists to avoid. [`LiveReload`][1]
+    /// [`ReadOnlyFlags`] bitmap exists to avoid. [`LiveReload`][1]
     /// refreshes them in place if they are present, and leaves them unset
     /// otherwise.
     ///
@@ -59,9 +59,9 @@ pub(super) struct BoolCounts {
 
 pub(super) struct ReadOnlyStorage<S: UniversalReadExt> {
     /// Points which have at least one `true` value
-    pub(super) trues_flags: ReadOnlyRoaringFlags<S>,
+    pub(super) trues_flags: ReadOnlyFlags<S>,
     /// Points which have at least one `false` value
-    pub(super) falses_flags: ReadOnlyRoaringFlags<S>,
+    pub(super) falses_flags: ReadOnlyFlags<S>,
 }
 
 impl<S: UniversalReadExt> ReadOnlyBoolIndex<S> {

lib/segment/src/index/field_index/bool_index/read_only_bool_index/read_ops.rs

@@ -4,7 +4,7 @@ use common::types::PointOffsetType;
 
 use super::super::read_ops::{self, BoolIndexRead};
 use super::ReadOnlyBoolIndex;
-use crate::common::flags::read_only_roaring_flags::ReadOnlyRoaringFlags;
+use crate::common::flags::read_only_flags::ReadOnlyFlags;
 use crate::common::operation_error::OperationResult;
 use crate::data_types::facets::{FacetHit, FacetValue, FacetValueRef};
 use crate::index::UniversalReadExt;
@@ -28,7 +28,7 @@ impl<S: UniversalReadExt> ReadOnlyBoolIndex<S> {
 }
 
 impl<S: UniversalReadExt> BoolIndexRead for ReadOnlyBoolIndex<S> {
-    type Flags = ReadOnlyRoaringFlags<S>;
+    type Flags = ReadOnlyFlags<S>;
 
     fn trues_flags(&self) -> &Self::Flags {
         &self.storage.trues_flags

lib/segment/src/index/field_index/field_index_base/update_only/tests.rs

@@ -225,8 +225,9 @@ fn rewriting_a_slot_is_rejected() {
     assert!(writer.add_point(0, &[&json!(0)], &hw_counter).is_err());
 }
 
-/// The boolean index is mask-backed: its writer rewrites both masks whole, and
-/// leaves behind the same two files the mutable index writes.
+/// The boolean index is mask-backed: its writer rewrites both masks whole,
+/// leaving compact-mode flags the mutable index picks up through mode
+/// detection.
 #[test]
 fn bool_index_round_trip() {
     let dir = TempDir::with_prefix("update_only_index").unwrap();

lib/segment/src/index/field_index/null_index/mutable_null_index/lifecycle.rs

@@ -10,7 +10,7 @@ use serde_json::Value;
 use super::super::read_ops::NullIndexRead;
 use super::{HAS_VALUES_DIRNAME, IS_NULL_DIRNAME, MutableNullIndex, Storage};
 use crate::common::Flusher;
-use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
+use crate::common::flags::FlagsMode;
 use crate::common::flags::roaring_flags::RoaringFlags;
 use crate::common::operation_error::{OperationError, OperationResult};
 use crate::index::field_index::{FieldIndexBuilderTrait, PayloadFieldIndex};
@@ -76,13 +76,19 @@ impl MutableNullIndex {
             ))
         })?;
 
-        let has_values_path = path.join(HAS_VALUES_DIRNAME);
-        let has_values_mmap = DynamicStoredFlags::open(&MmapFs, &has_values_path, Populate::No)?;
-        let has_values_flags = RoaringFlags::new(MmapFs, has_values_mmap)?;
+        let has_values_flags = RoaringFlags::open_or_create(
+            MmapFs,
+            &path.join(HAS_VALUES_DIRNAME),
+            FlagsMode::from_feature_flags(),
+            Populate::No,
+        )?;
 
-        let is_null_path = path.join(IS_NULL_DIRNAME);
-        let is_null_mmap = DynamicStoredFlags::open(&MmapFs, &is_null_path, Populate::No)?;
-        let is_null_flags = RoaringFlags::new(MmapFs, is_null_mmap)?;
+        let is_null_flags = RoaringFlags::open_or_create(
+            MmapFs,
+            &path.join(IS_NULL_DIRNAME),
+            FlagsMode::from_feature_flags(),
+            Populate::No,
+        )?;
 
         let storage = Storage {
             has_values_flags,

lib/segment/src/index/field_index/null_index/read_only_null_index/lifecycle.rs

@@ -4,7 +4,7 @@ use common::universal_io::{CachedReadFs, Populate, UniversalRead, UniversalReadF
 
 use super::super::mutable_null_index::{HAS_VALUES_DIRNAME, IS_NULL_DIRNAME};
 use super::{ReadOnlyNullIndex, ReadOnlyStorage};
-use crate::common::flags::read_only_roaring_flags::ReadOnlyRoaringFlags;
+use crate::common::flags::read_only_flags::ReadOnlyFlags;
 use crate::common::operation_error::{OperationError, OperationResult};
 
 impl<S: UniversalRead> ReadOnlyNullIndex<S> {
@@ -20,10 +20,8 @@ impl<S: UniversalRead> ReadOnlyNullIndex<S> {
         path: &Path,
         populate: Populate,
     ) -> OperationResult<bool> {
-        let has_values =
-            ReadOnlyRoaringFlags::<S>::preopen(fs, &path.join(HAS_VALUES_DIRNAME), populate)?;
-        let is_null =
-            ReadOnlyRoaringFlags::<S>::preopen(fs, &path.join(IS_NULL_DIRNAME), populate)?;
+        let has_values = ReadOnlyFlags::<S>::preopen(fs, &path.join(HAS_VALUES_DIRNAME), populate)?;
+        let is_null = ReadOnlyFlags::<S>::preopen(fs, &path.join(IS_NULL_DIRNAME), populate)?;
         Ok(has_values || is_null)
     }
 
@@ -50,8 +48,8 @@ impl<S: UniversalRead> ReadOnlyNullIndex<S> {
     ) -> OperationResult<Option<Self>> {
         // Open both directories first so a partial layout can be distinguished
         // from a genuinely absent index, regardless of which half is missing.
-        let has_values_flags = ReadOnlyRoaringFlags::<S>::open(fs, &path.join(HAS_VALUES_DIRNAME))?;
-        let is_null_flags = ReadOnlyRoaringFlags::<S>::open(fs, &path.join(IS_NULL_DIRNAME))?;
+        let has_values_flags = ReadOnlyFlags::<S>::open(fs, &path.join(HAS_VALUES_DIRNAME))?;
+        let is_null_flags = ReadOnlyFlags::<S>::open(fs, &path.join(IS_NULL_DIRNAME))?;
 
         match (has_values_flags, is_null_flags) {
             // Neither directory exists: the index isn't present on disk.

lib/segment/src/index/field_index/null_index/read_only_null_index/mod.rs

@@ -2,7 +2,7 @@ use std::path::PathBuf;
 
 use common::universal_io::UniversalRead;
 
-use crate::common::flags::read_only_roaring_flags::ReadOnlyRoaringFlags;
+use crate::common::flags::read_only_flags::ReadOnlyFlags;
 use crate::index::payload_config::IndexMutability;
 
 mod lifecycle;
@@ -16,9 +16,9 @@ mod read_ops;
 /// bound to [`UniversalRead`][3] only — no buffer, no flusher, no write path. Query logic (filter / cardinality / condition checker) is
 /// shared with the writable variant via the [`NullIndexRead`][4] trait.
 ///
-/// Each [`ReadOnlyRoaringFlags`] retains its backend `S` so a [`LiveReload`][5]
-/// can reopen the bitslice and apply only the changed positions on reload,
-/// instead of re-materializing the bitmaps from scratch.
+/// Each [`ReadOnlyFlags`] retains its backend `S` so a [`LiveReload`][5]
+/// can reopen the backing files and resync on reload, instead of
+/// re-materializing the bitmaps from scratch.
 ///
 /// [1]: super::mutable_null_index::MutableNullIndex
 /// [2]: super::immutable_null_index::ImmutableNullIndex
@@ -33,9 +33,9 @@ pub struct ReadOnlyNullIndex<S: UniversalRead> {
 
 pub(super) struct ReadOnlyStorage<S: UniversalRead> {
     /// Points which have at least one value
-    pub(super) has_values_flags: ReadOnlyRoaringFlags<S>,
+    pub(super) has_values_flags: ReadOnlyFlags<S>,
     /// Points which have null values
-    pub(super) is_null_flags: ReadOnlyRoaringFlags<S>,
+    pub(super) is_null_flags: ReadOnlyFlags<S>,
 }
 
 impl<S: UniversalRead> ReadOnlyNullIndex<S> {

lib/segment/src/index/field_index/null_index/read_only_null_index/read_ops.rs

@@ -5,7 +5,7 @@ use common::universal_io::UniversalRead;
 
 use super::super::read_ops::{self, NullIndexRead};
 use super::ReadOnlyNullIndex;
-use crate::common::flags::read_only_roaring_flags::ReadOnlyRoaringFlags;
+use crate::common::flags::read_only_flags::ReadOnlyFlags;
 use crate::common::operation_error::OperationResult;
 use crate::index::UniversalReadExt;
 use crate::index::condition_checker::ConditionCheckerEnum;
@@ -15,7 +15,7 @@ use crate::index::field_index::{
 use crate::types::{FieldCondition, PayloadKeyType};
 
 impl<S: UniversalRead> NullIndexRead for ReadOnlyNullIndex<S> {
-    type Flags = ReadOnlyRoaringFlags<S>;
+    type Flags = ReadOnlyFlags<S>;
 
     fn has_values_flags(&self) -> &Self::Flags {
         &self.storage.has_values_flags

lib/segment/src/vector_storage/dense/appendable_dense_vector_storage.rs

@@ -12,8 +12,8 @@ use common::universal_io::{MmapFile, MmapFs, Populate, UserData};
 use fs_err as fs;
 
 use crate::common::Flusher;
+use crate::common::flags::FlagsMode;
 use crate::common::flags::bitvec_flags::BitvecFlags;
-use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
 use crate::common::operation_error::{OperationResult, check_process_stopped};
 use crate::data_types::named_vectors::CowVector;
 use crate::data_types::primitive::PrimitiveVectorElement;
@@ -318,9 +318,11 @@ pub fn open_appendable_memmap_vector_storage_impl<T: PrimitiveVectorElement>(
         Populate::from(populate),
     )?;
 
-    let deleted = BitvecFlags::new(
+    let deleted = BitvecFlags::open_or_create(
         MmapFs,
-        DynamicStoredFlags::open(&MmapFs, &deleted_path, Populate::from(populate))?,
+        &deleted_path,
+        FlagsMode::from_feature_flags(),
+        Populate::from(populate),
     )?;
     let deleted_count = deleted.count_trues();
 

lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs

@@ -13,8 +13,8 @@ use fs_err as fs;
 
 use super::buffered_offsets::BufferedOffsets;
 use crate::common::Flusher;
+use crate::common::flags::FlagsMode;
 use crate::common::flags::bitvec_flags::BitvecFlags;
-use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
 use crate::common::operation_error::{OperationError, OperationResult, check_process_stopped};
 use crate::data_types::named_vectors::{CowMultiVector, CowVector};
 use crate::data_types::primitive::PrimitiveVectorElement;
@@ -614,9 +614,11 @@ pub fn open_appendable_memmap_multi_vector_storage_impl<T: PrimitiveVectorElemen
     // the durable `vectors` length, which would corrupt points on reload.
     let offsets = BufferedOffsets::new(offsets);
 
-    let deleted = BitvecFlags::new(
+    let deleted = BitvecFlags::open_or_create(
         MmapFs,
-        DynamicStoredFlags::open(&MmapFs, &deleted_path, Populate::from(populate))?,
+        &deleted_path,
+        FlagsMode::from_feature_flags(),
+        Populate::from(populate),
     )?;
     let deleted_count = deleted.count_trues();
 

lib/segment/src/vector_storage/sparse/mmap_sparse_vector_storage.rs

@@ -16,8 +16,8 @@ use common::universal_io::{MmapFile, MmapFs, Populate, UserData};
 use fs_err as fs;
 use sparse::common::sparse_vector::SparseVector;
 
+use crate::common::flags::FlagsMode;
 use crate::common::flags::bitvec_flags::BitvecFlags;
-use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
 use crate::common::operation_error::{OperationError, OperationResult, check_process_stopped};
 use crate::data_types::named_vectors::CowVector;
 use crate::data_types::vectors::VectorRef;
@@ -71,9 +71,11 @@ impl MmapSparseVectorStorage {
 
         // Deleted flags
         let deleted_path = path.join(DELETED_DIRNAME);
-        let deleted = BitvecFlags::new(
+        let deleted = BitvecFlags::open_or_create(
             MmapFs,
-            DynamicStoredFlags::open(&MmapFs, &deleted_path, populate)?,
+            &deleted_path,
+            FlagsMode::from_feature_flags(),
+            populate,
         )?;
 
         let deleted_count = deleted.count_trues();
@@ -117,9 +119,11 @@ impl MmapSparseVectorStorage {
 
         // Deleted flags
         let deleted_path = path.join(DELETED_DIRNAME);
-        let deleted = BitvecFlags::new(
+        let deleted = BitvecFlags::open_or_create(
             MmapFs,
-            DynamicStoredFlags::open(&MmapFs, &deleted_path, Populate::from(populate))?,
+            &deleted_path,
+            FlagsMode::from_feature_flags(),
+            Populate::from(populate),
         )?;
 
         Ok(Self {

lib/segment/src/vector_storage/turbo/appendable_turbo_vector_storage.rs

@@ -22,8 +22,8 @@ use quantization::{EncodedStorage, EncodedStorageWrite};
 
 use super::shared::{self, DELETED_DIR_PATH, VECTORS_DIR_PATH};
 use crate::common::Flusher;
+use crate::common::flags::FlagsMode;
 use crate::common::flags::bitvec_flags::BitvecFlags;
-use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
 use crate::common::operation_error::{OperationError, OperationResult, check_process_stopped};
 use crate::data_types::named_vectors::CowVector;
 use crate::data_types::vectors::{DenseVector, VectorElementType, VectorRef};
@@ -85,13 +85,11 @@ impl AppendableMmapTurboVectorStorage {
             in_ram,
         )?;
 
-        let deleted = BitvecFlags::new(
+        let deleted = BitvecFlags::open_or_create(
             MmapFs,
-            DynamicStoredFlags::open(
-                &MmapFs,
-                &path.join(DELETED_DIR_PATH),
-                Populate::from(in_ram),
-            )?,
+            &path.join(DELETED_DIR_PATH),
+            FlagsMode::from_feature_flags(),
+            Populate::from(in_ram),
         )?;
         let deleted_count = deleted.count_trues();
         let quantization_buffer = vec![0.0; quantizer.get_padded_dim()];

lib/segment/src/vector_storage/turbo/multi_turbo/appendable_mmap_multi_turbo_vector_storage.rs

@@ -29,8 +29,8 @@ use super::super::shared::{
     DELETED_DIR_PATH, TQDT_BITS, TQDT_MODE, TQDT_ROTATION, VECTORS_DIR_PATH,
 };
 use crate::common::Flusher;
+use crate::common::flags::FlagsMode;
 use crate::common::flags::bitvec_flags::BitvecFlags;
-use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
 use crate::common::operation_error::{OperationError, OperationResult, check_process_stopped};
 use crate::data_types::named_vectors::{CowMultiVector, CowVector};
 use crate::data_types::vectors::{
@@ -135,9 +135,11 @@ pub fn open_appendable_turbo_multi_vector_storage(
         populate,
     )?;
 
-    let deleted = BitvecFlags::new(
+    let deleted = BitvecFlags::open_or_create(
         MmapFs,
-        DynamicStoredFlags::open(&MmapFs, &path.join(DELETED_DIR_PATH), populate)?,
+        &path.join(DELETED_DIR_PATH),
+        FlagsMode::from_feature_flags(),
+        populate,
     )?;
     let deleted_count = deleted.count_trues();
 

lib/segment/src/vector_storage/turbo/turbo_vector_storage.rs

@@ -27,8 +27,8 @@ use quantization::{EncodedStorage, EncodedStorageWrite};
 
 use super::shared::{self, DELETED_DIR_PATH, VECTORS_PATH};
 use crate::common::Flusher;
+use crate::common::flags::FlagsMode;
 use crate::common::flags::bitvec_flags::BitvecFlags;
-use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
 use crate::common::io_uring::{IoUringFallback, use_io_uring};
 use crate::common::operation_error::{OperationError, OperationResult, check_process_stopped};
 use crate::data_types::named_vectors::CowVector;
@@ -164,13 +164,11 @@ impl<S: UniversalRead> TurboVectorStorageImpl<S> {
         populate: bool,
     ) -> OperationResult<Self> {
         fs_err::create_dir_all(path)?;
-        let deleted = BitvecFlags::new(
+        let deleted = BitvecFlags::open_or_create(
             MmapFs,
-            DynamicStoredFlags::open(
-                &MmapFs,
-                &path.join(DELETED_DIR_PATH),
-                Populate::from(populate),
-            )?,
+            &path.join(DELETED_DIR_PATH),
+            FlagsMode::from_feature_flags(),
+            Populate::from(populate),
         )?;
         let deleted_count = deleted.count_trues();
         let quantization_buffer = vec![0.0; quantizer.get_padded_dim()];