qdrant-edge 0.8.0

use std::borrow::Cow;
use std::fmt::Debug;
use std::ops::Range;
use std::path::Path;

use super::{Item, ReadPipeline, UniversalReadFs, UserData};
use crate::common::ext::aligned_vec::ACow;
use crate::common::generic_consts::{AccessPattern, Sequential};
use crate::common::universal_io::cached_fs::FileInfo;
use crate::common::universal_io::{ReadBytesItem, ReadRange, UioResult, UniversalIoError, UniversalKind};

/// Per-file handle for universal read access.
///
/// Concrete file handles (`MmapFile`, `IoUringFile`, `CachedSlice`, ...)
/// implement this trait. Instances are produced by a corresponding
/// [`UniversalReadFs`](super::UniversalReadFs) backend via
/// [`UniversalReadFs::open`](super::UniversalReadFs::open).
///
/// This trait deliberately does *not* extend
/// [`UniversalReadFileOps`](super::UniversalReadFileOps): a file handle is
/// not a filesystem, and not every filesystem-level backend produces
/// `UniversalRead` handles (e.g. a metadata-only listing service). The
/// link to the producing filesystem is exposed via the [`Self::Fs`]
/// associated type so generic-over-`<S: UniversalRead>` code can refer
/// to the matching filesystem handle as `S::Fs` without an extra generic
/// parameter.
///
/// # Alignment
///
/// Some methods accept an `align` parameter.
/// - When returning [`ACow::Owned`], an implementation should honor that
///   alignment, meaning that the result can be casted to [`Vec<T>`].
/// - When returning [`ACow::Borrowed`], an implementation will ignore the
///   alignment. Practically mmaps are aligned by 4 KiB, which is more than
///   enough for the majority of types.
#[expect(clippy::len_without_is_empty)]
pub trait UniversalRead: Sized + Debug + Send + Sync {
    /// The canonical filesystem handle type that opens `Self`-typed file
    /// handles via [`UniversalReadFs::open`](UniversalReadFs::open).
    ///
    /// Pinned in this direction only (`Self::Fs::File = Self`): every file
    /// type names exactly one canonical backend, but other filesystems may
    /// produce the same file type (e.g.
    /// [`CachedReadFs`](crate::universal_io::CachedReadFs) opens the
    /// wrapped backend's files). Code that should accept any of them takes
    /// `&impl UniversalReadFs<File = S>` instead of `&S::Fs`. Wrappers such
    /// as `ReadOnly<S>` declare a phantom `ReadOnlyFs<S::Fs>` to satisfy
    /// this constraint at the type level.
    type Fs: UniversalReadFs<File = Self>;

    /// Read-pipeline implementation for this backend.
    type ReadPipeline<'file, U>: ReadPipeline<'file, U, File = Self>
    where
        Self: 'file,
        U: UserData;

    /// Enables live-reloading of files. Append-only files can make the
    /// underlying file larger, so reopening can account for this growth.
    ///
    /// This may be a no-op in some implementations.
    fn reopen(&mut self) -> UioResult<()>;

    /// Stage the work that the next [`reopen`](Self::reopen) must do, reading
    /// the file's current length via `get_file_info` — typically backed by a
    /// [`CachedReadFs`] listing snapshot. The implementation resolves its own
    /// path, so there is nothing to mispair.
    ///
    /// Lets a caller submit the fetches of many reopens up front and only pay
    /// the (already in-flight) tail of the wait when applying them. Contract:
    /// must not wait on the data fetch (resolving a pending open-time prefill
    /// is the one bounded exception), and staging must be invisible to
    /// readers of an already-live mirror — no length change, no cache
    /// invalidation.
    ///
    /// Defaults to a no-op: local backends' `reopen` is a stat plus a remap,
    /// so there is nothing worth pre-staging. Only [`DiskCache`] overrides it.
    ///
    /// [`CachedReadFs`]: crate::universal_io::CachedReadFs
    /// [`DiskCache`]: crate::universal_io::DiskCache
    fn schedule_reopen<F: FnOnce(&Path) -> Option<FileInfo>>(
        &mut self,
        get_file_info: F,
    ) -> UioResult<()> {
        let _ = get_file_info;
        Ok(())
    }

    /// Prefer [`read_batch`] if you need high performance.
    #[inline]
    fn read<P: AccessPattern, T: Item>(
        &self,
        range: ReadRange,
        access_pattern: P,
    ) -> UioResult<Cow<'_, [T]>> {
        let bytes = self.read_bytes(
            range.into_byte_range::<T>(),
            access_pattern,
            align_of::<T>(),
        )?;
        Ok(bytes.try_cast_bytemuck().unwrap())
    }

    fn read_bytes<P: AccessPattern>(
        &self,
        range: Range<u64>,
        access_pattern: P,
        align: usize,
    ) -> UioResult<ACow<'_>>;

    /// Read the entire file in one logical access.
    ///
    /// Implementations may override this to avoid the two accesses that would
    /// result from `len()` followed by `read(0..len())`. Default implementation
    /// does exactly that.
    fn read_whole<T: Item>(&self) -> UioResult<Cow<'_, [T]>> {
        let range = ReadRange {
            byte_offset: 0,
            length: self.len::<T>()?,
        };

        self.read(range, Sequential)
    }

    fn read_batch<P, T, U, E>(
        &self,
        ranges: impl IntoIterator<Item = (U, ReadRange)>,
        _access_pattern: P,
        mut callback: impl FnMut(U, &[T]) -> Result<(), E>,
    ) -> Result<(), E>
    where
        P: AccessPattern,
        T: Item,
        U: UserData,
        E: From<UniversalIoError>,
    {
        let mut pipeline = Self::ReadPipeline::<'_, U>::new()?;
        let mut ranges = ranges.into_iter();

        loop {
            while pipeline.can_schedule()
                && let Some((user_data, range)) = ranges.next()
            {
                let range = range.into_byte_range::<T>();
                pipeline.schedule::<P>(user_data, self, range, align_of::<T>())?;
            }

            let Some((user_data, data)) = pipeline.wait_bytemuck()? else {
                break;
            };
            callback(user_data, &data)?;
        }

        Ok(())
    }

    /// Like [`read_batch`](Self::read_batch), but returns a fallible iterator
    /// instead of accepting a callback.
    fn read_iter<P: AccessPattern, T: Item, U: UserData>(
        &self,
        ranges: impl IntoIterator<Item = (U, ReadRange)>,
        _access_pattern: P,
    ) -> UioResult<impl Iterator<Item = UioResult<(U, Cow<'_, [T]>)>>> {
        let mut pipeline = Self::ReadPipeline::<'_, U>::new()?;
        let mut ranges = ranges.into_iter();

        Ok(std::iter::from_fn(move || {
            while pipeline.can_schedule()
                && let Some((user_data, range)) = ranges.next()
            {
                let range = range.into_byte_range::<T>();
                if let Err(err) = pipeline.schedule::<P>(user_data, self, range, align_of::<T>()) {
                    return Some(Err(err));
                }
            }

            pipeline.wait_bytemuck().transpose()
        }))
    }

    fn read_bytes_iter<P: AccessPattern, U: UserData>(
        &self,
        ranges: impl IntoIterator<Item = ReadBytesItem<U>>,
        _access_pattern: P,
    ) -> UioResult<impl Iterator<Item = UioResult<(U, ACow<'_>)>>> {
        let mut pipeline = Self::ReadPipeline::<'_, U>::new()?;
        let mut ranges = ranges.into_iter();

        Ok(std::iter::from_fn(move || {
            while pipeline.can_schedule()
                && let Some(item) = ranges.next()
            {
                let ReadBytesItem {
                    user_data,
                    range,
                    align,
                } = item;
                if let Err(err) = pipeline.schedule::<P>(user_data, self, range, align) {
                    return Some(Err(err));
                }
            }

            pipeline.wait().transpose()
        }))
    }

    fn len<T>(&self) -> UioResult<u64>;

    /// Fill RAM cache with related data, if applicable for this implementation.
    ///
    /// For example in MMAP-based files we do `madvise` with `MADV_POPULATE_READ`.
    fn populate(&self) -> UioResult<()>;

    /// Whether the backend chooses to populate when using `Populate::Auto`
    fn populate_auto() -> bool;

    /// Ask to evict related data from RAM cache, if applicable for this implementation.
    ///
    /// For example in MMAP-based files we do `madvise` with `MADV_PAGEOUT`.
    fn clear_ram_cache(&self) -> UioResult<()>;

    fn kind() -> UniversalKind;

    // When adding provided methods, don't forget to update impls in crate::universal_io::wrappers::*.
}