qdrant-edge 0.7.0

use std::collections::HashMap;
use std::io::Read;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Instant;

use atomic_refcell::AtomicRefCell;
use crate::common::budget::ResourcePermit;
use crate::common::defaults::log_load_timing;
use crate::common::flags::FeatureFlags;
use crate::common::fs::{safe_delete_with_suffix, sync_parent_dir};
use crate::common::is_alive_lock::IsAliveLock;
use crate::common::mmap::{Advice, AdviceSetting};
use crate::common::progress_tracker::ProgressTracker;
use crate::common::storage_version::StorageVersion;
use crate::common::types::PointOffsetType;
use crate::common::universal_io::MmapFs;
use fs_err as fs;
use fs_err::File;
use log::info;
use parking_lot::Mutex;
use rand::Rng;
use serde::Deserialize;
use uuid::Uuid;

use crate::segment::common::operation_error::{OperationError, OperationResult, check_process_stopped};
use crate::segment::data_types::vectors::DEFAULT_VECTOR_NAME;
use crate::segment::id_tracker::immutable_id_tracker::{self, ImmutableIdTracker};
use crate::segment::id_tracker::mutable_id_tracker::MutableIdTracker;
use crate::segment::id_tracker::{IdTrackerEnum, IdTrackerRead};
use crate::segment::index::VectorIndexEnum;
use crate::segment::index::hnsw_index::gpu::gpu_devices_manager::LockedGpuDevice;
use crate::segment::index::hnsw_index::hnsw::{HNSWIndex, HnswIndexOpenArgs};
use crate::segment::index::plain_vector_index::PlainVectorIndex;
use crate::segment::index::sparse_index::sparse_index_config::SparseIndexType;
use crate::segment::index::sparse_index::sparse_vector_index::{
    SparseVectorIndex, SparseVectorIndexOpenArgs,
};
use crate::segment::index::struct_payload_index::StructPayloadIndex;
use crate::segment::payload_storage::mmap_payload_storage::MmapPayloadStorage;
use crate::segment::payload_storage::payload_storage_enum::PayloadStorageEnum;
use crate::segment::segment::{SEGMENT_STATE_FILE, Segment, SegmentVersion, VectorData};
use crate::segment::types::{
    Distance, HnswGlobalConfig, Indexes, PayloadStorageType, SegmentConfig, SegmentState,
    SegmentType, SeqNumberType, SparseVectorStorageType, VectorDataConfig, VectorName,
    VectorStorageDatatype, VectorStorageType,
};
use crate::segment::vector_storage::dense::dense_vector_storage::{
    open_dense_vector_storage, open_dense_vector_storage_byte, open_dense_vector_storage_half,
};
use crate::segment::vector_storage::multi_dense::appendable_mmap_multi_dense_vector_storage::{
    open_appendable_memmap_multi_vector_storage, open_appendable_memmap_vector_storage,
};
use crate::segment::vector_storage::quantized::quantized_vectors::QuantizedVectors;
use crate::segment::vector_storage::sparse::mmap_sparse_vector_storage::MmapSparseVectorStorage;
use crate::segment::vector_storage::{VectorStorageEnum, VectorStorageRead};

pub const PAYLOAD_INDEX_PATH: &str = "payload_index";
pub const VECTOR_STORAGE_PATH: &str = "vector_storage";
pub const VECTOR_INDEX_PATH: &str = "vector_index";

fn sp<T>(t: T) -> Arc<AtomicRefCell<T>> {
    Arc::new(AtomicRefCell::new(t))
}

pub fn get_vector_name_with_prefix(prefix: &str, vector_name: &VectorName) -> String {
    if !vector_name.is_empty() {
        format!("{prefix}-{vector_name}")
    } else {
        prefix.to_owned()
    }
}

pub fn get_vector_storage_path(segment_path: &Path, vector_name: &VectorName) -> PathBuf {
    segment_path.join(get_vector_name_with_prefix(
        VECTOR_STORAGE_PATH,
        vector_name,
    ))
}

pub fn get_vector_index_path(segment_path: &Path, vector_name: &VectorName) -> PathBuf {
    segment_path.join(get_vector_name_with_prefix(VECTOR_INDEX_PATH, vector_name))
}

fn open_mmap_vector_storage(
    vector_storage_path: &Path,
    vector_config: &VectorDataConfig,
    madvise: AdviceSetting,
    populate: bool,
) -> OperationResult<VectorStorageEnum> {
    let storage_element_type = vector_config.datatype.unwrap_or_default();
    if let Some(multi_vec_config) = &vector_config.multivector_config {
        // there are no mmap multi vector storages, appendable only
        open_appendable_memmap_multi_vector_storage(
            storage_element_type,
            vector_storage_path,
            vector_config.size,
            vector_config.distance,
            *multi_vec_config,
            madvise,
            populate,
        )
    } else {
        match storage_element_type {
            VectorStorageDatatype::Float32 => open_dense_vector_storage(
                vector_storage_path,
                vector_config.size,
                vector_config.distance,
                populate,
            ),
            VectorStorageDatatype::Uint8 => open_dense_vector_storage_byte(
                vector_storage_path,
                vector_config.size,
                vector_config.distance,
                populate,
            ),
            VectorStorageDatatype::Float16 => open_dense_vector_storage_half(
                vector_storage_path,
                vector_config.size,
                vector_config.distance,
                populate,
            ),
        }
    }
}

fn open_chunked_mmap_vector_storage(
    vector_storage_path: &Path,
    vector_config: &VectorDataConfig,
    madvise: AdviceSetting,
    populate: bool,
) -> OperationResult<VectorStorageEnum> {
    let storage_element_type = vector_config.datatype.unwrap_or_default();
    if let Some(multi_vec_config) = &vector_config.multivector_config {
        open_appendable_memmap_multi_vector_storage(
            storage_element_type,
            vector_storage_path,
            vector_config.size,
            vector_config.distance,
            *multi_vec_config,
            madvise,
            populate,
        )
    } else {
        open_appendable_memmap_vector_storage(
            storage_element_type,
            vector_storage_path,
            vector_config.size,
            vector_config.distance,
            madvise,
            populate,
        )
    }
}

pub(crate) fn open_vector_storage(
    vector_config: &VectorDataConfig,
    vector_storage_path: &Path,
) -> OperationResult<VectorStorageEnum> {
    match vector_config.storage_type {
        VectorStorageType::Memory => Err(OperationError::service_error(
            "Failed to load 'Memory' storage type, RocksDB is not supported in this Qdrant version",
        )),

        // Mmap on disk, not appendable
        VectorStorageType::Mmap => open_mmap_vector_storage(
            vector_storage_path,
            vector_config,
            AdviceSetting::Global,
            false,
        ),
        VectorStorageType::InRamMmap => open_mmap_vector_storage(
            vector_storage_path,
            vector_config,
            AdviceSetting::from(Advice::Normal),
            true,
        ),

        // Chunked mmap on disk, appendable
        VectorStorageType::ChunkedMmap => open_chunked_mmap_vector_storage(
            vector_storage_path,
            vector_config,
            AdviceSetting::Global,
            false,
        ),
        VectorStorageType::InRamChunkedMmap => open_chunked_mmap_vector_storage(
            vector_storage_path,
            vector_config,
            AdviceSetting::from(Advice::Normal),
            true,
        ),

        // Empty placeholder storage, no files on disk
        VectorStorageType::Empty => {
            use crate::segment::vector_storage::dense::empty_dense_vector_storage::new_empty_dense_vector_storage;
            Ok(new_empty_dense_vector_storage(
                vector_config.size,
                vector_config.distance,
                vector_config.datatype.unwrap_or_default(),
                vector_config.storage_type.is_on_disk(),
                vector_config.multivector_config,
                0, // num_points set after id_tracker is loaded
            ))
        }
    }
}

pub(crate) fn create_payload_storage(
    segment_path: &Path,
    config: &SegmentConfig,
) -> OperationResult<PayloadStorageEnum> {
    let payload_storage = match config.payload_storage_type {
        PayloadStorageType::Mmap => PayloadStorageEnum::from(MmapPayloadStorage::open_or_create(
            segment_path.to_path_buf(),
            false,
        )?),
        PayloadStorageType::InRamMmap => PayloadStorageEnum::from(
            MmapPayloadStorage::open_or_create(segment_path.to_path_buf(), true)?,
        ),
    };
    Ok(payload_storage)
}

pub(crate) fn create_mutable_id_tracker(
    segment_path: &Path,
    deferred_internal_id: Option<PointOffsetType>,
) -> OperationResult<MutableIdTracker> {
    MutableIdTracker::open(segment_path, deferred_internal_id)
}

pub(crate) fn get_payload_index_path(segment_path: &Path) -> PathBuf {
    segment_path.join(PAYLOAD_INDEX_PATH)
}

pub(crate) struct VectorIndexOpenArgs<'a> {
    pub path: &'a Path,
    pub id_tracker: Arc<AtomicRefCell<IdTrackerEnum>>,
    pub vector_storage: Arc<AtomicRefCell<VectorStorageEnum>>,
    pub payload_index: Arc<AtomicRefCell<StructPayloadIndex>>,
    pub quantized_vectors: Arc<AtomicRefCell<Option<QuantizedVectors>>>,
}

pub struct VectorIndexBuildArgs<'a, R: Rng + ?Sized> {
    pub permit: Arc<ResourcePermit>,
    /// Vector indices from other segments, used to speed up index building.
    /// May or may not contain the same vectors.
    pub old_indices: &'a [Arc<AtomicRefCell<VectorIndexEnum>>],
    pub gpu_device: Option<&'a LockedGpuDevice<'a>>,
    pub rng: &'a mut R,
    pub stopped: &'a AtomicBool,
    pub hnsw_global_config: &'a HnswGlobalConfig,
    pub feature_flags: FeatureFlags,
    pub progress: ProgressTracker,
}

pub(crate) fn open_vector_index(
    vector_config: &VectorDataConfig,
    open_args: VectorIndexOpenArgs,
) -> OperationResult<VectorIndexEnum> {
    let VectorIndexOpenArgs {
        path,
        id_tracker,
        vector_storage,
        payload_index,
        quantized_vectors,
    } = open_args;
    Ok(match &vector_config.index {
        Indexes::Plain {} => VectorIndexEnum::Plain(PlainVectorIndex::new(
            id_tracker,
            vector_storage,
            quantized_vectors,
            payload_index,
        )),
        Indexes::Hnsw(hnsw_config) => VectorIndexEnum::Hnsw(HNSWIndex::open(HnswIndexOpenArgs {
            path,
            id_tracker,
            vector_storage,
            quantized_vectors,
            payload_index,
            hnsw_config: *hnsw_config,
        })?),
    })
}

pub(crate) fn build_vector_index<R: Rng + ?Sized>(
    vector_config: &VectorDataConfig,
    open_args: VectorIndexOpenArgs,
    build_args: VectorIndexBuildArgs<R>,
) -> OperationResult<VectorIndexEnum> {
    let VectorIndexOpenArgs {
        path,
        id_tracker,
        vector_storage,
        payload_index,
        quantized_vectors,
    } = open_args;
    Ok(match &vector_config.index {
        Indexes::Plain {} => VectorIndexEnum::Plain(PlainVectorIndex::new(
            id_tracker,
            vector_storage,
            quantized_vectors,
            payload_index,
        )),
        Indexes::Hnsw(hnsw_config) => VectorIndexEnum::Hnsw(HNSWIndex::build(
            HnswIndexOpenArgs {
                path,
                id_tracker,
                vector_storage,
                quantized_vectors,
                payload_index,
                hnsw_config: *hnsw_config,
            },
            build_args,
        )?),
    })
}

#[cfg(feature = "testing")]
pub fn create_sparse_vector_index_test(
    args: SparseVectorIndexOpenArgs<impl FnMut()>,
) -> OperationResult<VectorIndexEnum> {
    create_sparse_vector_index(args)
}

pub(crate) fn create_sparse_vector_index(
    args: SparseVectorIndexOpenArgs<impl FnMut()>,
) -> OperationResult<VectorIndexEnum> {
    let effective_index_type = match args.config.index_type {
        SparseIndexType::ImmutableRam => {
            // Low-memory mode downgrades `ImmutableRam` (which copies the inverted
            // index from mmap files into heap RAM at load) to `Mmap` (which keeps
            // it on disk). The two variants share the same on-disk file format, so
            // flipping at load time is safe without rebuild. The persisted
            // `SparseIndexConfig.index_type` is not modified — `try_load` re-reads
            // it from disk and the loaded config is kept for future persistence.
            if crate::common::low_memory::low_memory_mode().prefer_disk() {
                SparseIndexType::Mmap
            } else {
                SparseIndexType::ImmutableRam
            }
        }
        SparseIndexType::MutableRam => SparseIndexType::MutableRam,
        SparseIndexType::Mmap => SparseIndexType::Mmap,
    };
    let vector_index = match (
        effective_index_type,
        args.config.datatype.unwrap_or_default(),
    ) {
        (SparseIndexType::MutableRam, _) => {
            VectorIndexEnum::SparseRam(SparseVectorIndex::open(args)?)
        }

        (SparseIndexType::ImmutableRam, VectorStorageDatatype::Float32) => {
            VectorIndexEnum::SparseCompressedImmutableRamF32(SparseVectorIndex::open(args)?)
        }
        (SparseIndexType::Mmap, VectorStorageDatatype::Float32) => {
            VectorIndexEnum::SparseCompressedMmapF32(SparseVectorIndex::open(args)?)
        }
        (SparseIndexType::ImmutableRam, VectorStorageDatatype::Float16) => {
            VectorIndexEnum::SparseCompressedImmutableRamF16(SparseVectorIndex::open(args)?)
        }
        (SparseIndexType::Mmap, VectorStorageDatatype::Float16) => {
            VectorIndexEnum::SparseCompressedMmapF16(SparseVectorIndex::open(args)?)
        }
        (SparseIndexType::ImmutableRam, VectorStorageDatatype::Uint8) => {
            VectorIndexEnum::SparseCompressedImmutableRamU8(SparseVectorIndex::open(args)?)
        }
        (SparseIndexType::Mmap, VectorStorageDatatype::Uint8) => {
            VectorIndexEnum::SparseCompressedMmapU8(SparseVectorIndex::open(args)?)
        }
    };

    Ok(vector_index)
}

pub(crate) fn create_sparse_vector_storage(
    path: &Path,
    storage_type: &SparseVectorStorageType,
) -> OperationResult<VectorStorageEnum> {
    match storage_type {
        SparseVectorStorageType::Mmap => {
            let mmap_storage = MmapSparseVectorStorage::open_or_create(path)?;
            Ok(VectorStorageEnum::SparseMmap(mmap_storage))
        }
        SparseVectorStorageType::Empty => {
            use crate::segment::vector_storage::sparse::empty_sparse_vector_storage::new_empty_sparse_vector_storage;
            Ok(new_empty_sparse_vector_storage(0))
        }
    }
}

#[allow(clippy::too_many_arguments)]
fn create_segment(
    initial_version: Option<SeqNumberType>,
    version: Option<SeqNumberType>,
    segment_path: &Path,
    uuid: Uuid,
    deferred_internal_id: Option<PointOffsetType>,
    config: &SegmentConfig,
    stopped: &AtomicBool,
    create: bool,
) -> OperationResult<Segment> {
    let started = Instant::now();
    let payload_storage = sp(create_payload_storage(segment_path, config)?);
    log_load_timing(segment_path, "payload_storage", started);

    let appendable_flag = config.is_appendable();

    // Limit deferred segment feature to appendable segments.
    let deferred_internal_id = deferred_internal_id.filter(|_| appendable_flag);

    let use_mutable_id_tracker =
        appendable_flag || !immutable_id_tracker::mappings_path(segment_path).is_file();
    let started = Instant::now();
    let id_tracker =
        create_segment_id_tracker(use_mutable_id_tracker, segment_path, deferred_internal_id)?;
    log_load_timing(segment_path, "id_tracker", started);

    let mut vector_storages = HashMap::new();

    for (vector_name, vector_config) in &config.vector_data {
        let vector_storage_path = get_vector_storage_path(segment_path, vector_name);

        let started = Instant::now();
        let vector_storage = sp(open_vector_storage(vector_config, &vector_storage_path)?);
        log_load_timing(
            segment_path,
            &format!("vector_storage dense '{vector_name}'"),
            started,
        );

        vector_storages.insert(vector_name.to_owned(), vector_storage);
    }

    for (vector_name, sparse_config) in config.sparse_vector_data.iter() {
        let vector_storage_path = get_vector_storage_path(segment_path, vector_name);

        let started = Instant::now();
        let vector_storage = sp(create_sparse_vector_storage(
            &vector_storage_path,
            &sparse_config.storage_type,
        )?);
        log_load_timing(
            segment_path,
            &format!("vector_storage sparse '{vector_name}'"),
            started,
        );

        vector_storages.insert(vector_name.to_owned(), vector_storage);
    }

    let payload_index_path = get_payload_index_path(segment_path);
    let started = Instant::now();
    let payload_index: Arc<AtomicRefCell<StructPayloadIndex>> = sp(StructPayloadIndex::open(
        payload_storage.clone(),
        id_tracker.clone(),
        vector_storages.clone(),
        &payload_index_path,
        appendable_flag,
        create,
    )?);
    log_load_timing(segment_path, "payload_index", started);

    let mut vector_data = HashMap::new();
    for (vector_name, vector_config) in &config.vector_data {
        let vector_storage_path = get_vector_storage_path(segment_path, vector_name);
        let vector_storage = vector_storages.remove(vector_name).unwrap();

        let vector_index_path = get_vector_index_path(segment_path, vector_name);
        // Ensure vector storage is sized to match the id_tracker's point count.
        // This can be out of sync when a named vector was added to an existing segment.
        let point_count = id_tracker.borrow().total_point_count();
        let vector_count = vector_storage.borrow().total_vector_count();
        if vector_count != point_count {
            log::debug!(
                "Mismatch of point and vector counts ({point_count} != {vector_count}, storage: {}), pre-filling deleted entries",
                vector_storage_path.display(),
            );
            vector_storage
                .borrow_mut()
                .prefill_deleted_entries(point_count)?;
        }

        let started = Instant::now();
        let quantized_vectors = sp(
            if let Some(quantization_config) = config.quantization_config(vector_name) {
                let quantized_data_path = vector_storage_path;
                QuantizedVectors::load(
                    quantization_config,
                    &vector_storage.borrow(),
                    &quantized_data_path,
                    stopped,
                )?
            } else {
                None
            },
        );
        log_load_timing(
            segment_path,
            &format!("quantized_vectors '{vector_name}'"),
            started,
        );

        let started = Instant::now();
        let vector_index: Arc<AtomicRefCell<VectorIndexEnum>> = sp(open_vector_index(
            vector_config,
            VectorIndexOpenArgs {
                path: &vector_index_path,
                id_tracker: id_tracker.clone(),
                vector_storage: vector_storage.clone(),
                payload_index: payload_index.clone(),
                quantized_vectors: quantized_vectors.clone(),
            },
        )?);
        log_load_timing(
            segment_path,
            &format!("vector_index dense '{vector_name}'"),
            started,
        );

        check_process_stopped(stopped)?;

        vector_data.insert(
            vector_name.to_owned(),
            VectorData {
                vector_index,
                vector_storage,
                quantized_vectors,
            },
        );
    }

    for (vector_name, sparse_vector_config) in &config.sparse_vector_data {
        let vector_storage_path = get_vector_storage_path(segment_path, vector_name);
        let vector_index_path = get_vector_index_path(segment_path, vector_name);
        let vector_storage = vector_storages.remove(vector_name).unwrap();

        // Ensure vector storage is sized to match the id_tracker's point count.
        // This can be out of sync when a named vector was added to an existing segment.
        let point_count = id_tracker.borrow().total_point_count();
        let vector_count = vector_storage.borrow().total_vector_count();
        if vector_count != point_count {
            log::debug!(
                "Mismatch of point and vector counts ({point_count} != {vector_count}, storage: {}), pre-filling deleted entries",
                vector_storage_path.display(),
            );
            vector_storage
                .borrow_mut()
                .prefill_deleted_entries(point_count)?;
        }

        let started = Instant::now();
        let vector_index = sp(create_sparse_vector_index(SparseVectorIndexOpenArgs {
            config: sparse_vector_config.index,
            id_tracker: id_tracker.clone(),
            vector_storage: vector_storage.clone(),
            payload_index: payload_index.clone(),
            path: &vector_index_path,
            stopped,
            tick_progress: || (),
        })?);
        log_load_timing(
            segment_path,
            &format!("vector_index sparse '{vector_name}'"),
            started,
        );

        check_process_stopped(stopped)?;

        vector_data.insert(
            vector_name.to_owned(),
            VectorData {
                vector_storage,
                vector_index,
                quantized_vectors: sp(None),
            },
        );
    }

    let segment_type = if config.is_any_vector_indexed() {
        SegmentType::Indexed
    } else {
        SegmentType::Plain
    };

    Ok(Segment {
        uuid,
        initial_version,
        version,
        persisted_version: Arc::new(Mutex::new(version)),
        is_alive_flush_lock: IsAliveLock::new(),
        segment_path: segment_path.to_owned(),
        version_tracker: Default::default(),
        id_tracker,
        vector_data,
        segment_type,
        appendable_flag,
        payload_index,
        payload_storage,
        segment_config: config.clone(),
        error_status: None,
    })
}

fn create_segment_id_tracker(
    mutable_id_tracker: bool,
    segment_path: &Path,
    deferred_internal_id: Option<PointOffsetType>,
) -> OperationResult<Arc<AtomicRefCell<IdTrackerEnum>>> {
    if !mutable_id_tracker {
        return Ok(sp(IdTrackerEnum::ImmutableIdTracker(
            ImmutableIdTracker::open(&MmapFs, segment_path)?,
        )));
    }

    Ok(sp(IdTrackerEnum::MutableIdTracker(
        create_mutable_id_tracker(segment_path, deferred_internal_id)?,
    )))
}

/// Normalize segment directory.
///
/// Might delete or rename the directory.
/// Returns `None` if the segment directory was deleted.
pub fn normalize_segment_dir(path: &Path) -> OperationResult<Option<(PathBuf, Uuid)>> {
    // 1. Delete dirs like `5345474d-454e-54f0-9f98-ba206e616d65.deleted`.
    // These are leftovers from rename-then-delete approach.
    if path
        .extension()
        .and_then(|ext| ext.to_str())
        .map(|ext| ext == "deleted")
        .unwrap_or(false)
    {
        log::warn!("Deleting leftover segment: {}", path.display());
        safe_delete_with_suffix(path).map_err(|err| {
            OperationError::service_error(format!("failed to delete leftover segment: {err}"))
        })?;
        return Ok(None);
    }

    // 2. Delete dirs without proper `version.info` file inside.
    // These segments are not properly saved.
    // Likely, the server crashed during saving.
    if SegmentVersion::load(path)?.is_none() {
        log::warn!("Deleting segment without version file: {}", path.display());
        safe_delete_with_suffix(path).map_err(|err| {
            OperationError::service_error(format!("failed to delete leftover segment: {err}"))
        })?;
        return Ok(None);
    }

    // 3. Force directory name to be a valid UUID.
    // Rename if necessary.
    let file_name = path
        .file_name()
        .and_then(|fname| fname.to_str())
        .ok_or_else(|| {
            OperationError::service_error(format!(
                "Failed to get segment folder name: {}",
                path.display()
            ))
        })?;
    match Uuid::try_parse(file_name) {
        Ok(uuid) => Ok(Some((path.to_path_buf(), uuid))),
        Err(_) => {
            let segment_uuid = Uuid::new_v4();
            let new_path = path.with_file_name(segment_uuid.to_string());
            log::warn!(
                "Segment name is not a valid UUID: {}. Renaming to {segment_uuid}",
                path.display(),
            );
            fs::rename(path, &new_path)?;
            sync_parent_dir(&new_path)?;
            Ok(Some((new_path, segment_uuid)))
        }
    }
}

/// Load segment from given `path`.
///
/// Preferably, the `uuid` should match the last component of `path`.
/// In production use [`normalize_segment_dir`] to obtain correct path and UUID.
/// In tests it is acceptable to pass an arbitrary UUID, e.g., [`Uuid::nil()`].
pub fn load_segment(
    path: &Path,
    uuid: Uuid,
    deferred_internal_id: Option<PointOffsetType>,
    stopped: &AtomicBool,
) -> OperationResult<Segment> {
    let total_started = Instant::now();

    let stored_version = SegmentVersion::load(path)?.ok_or_else(|| {
        OperationError::service_error(format!(
            "Segment version file not found in segment: {}",
            path.display()
        ))
    })?;

    let app_version = SegmentVersion::current();

    if stored_version != app_version {
        info!("Migrating segment {stored_version} -> {app_version}");

        if stored_version > app_version {
            return Err(OperationError::service_error(format!(
                "Data version {stored_version} is newer than application version {app_version}. \
                Please upgrade the application. Compatibility is not guaranteed."
            )));
        }

        if stored_version.major == 0 && stored_version.minor < 3 {
            return Err(OperationError::service_error(format!(
                "Segment version({stored_version}) is not compatible with current version({app_version})"
            )));
        }

        if stored_version.major == 0 && stored_version.minor == 3 {
            let segment_state = load_segment_state_v3(path)?;
            Segment::save_state(&segment_state, path)?;
        } else if stored_version.major == 0 && stored_version.minor <= 5 {
            let segment_state = load_segment_state_v5(path)?;
            Segment::save_state(&segment_state, path)?;
        }

        SegmentVersion::save(path)?
    }

    let started = Instant::now();
    let segment_state = Segment::load_state(path)?;
    log_load_timing(path, "load_state", started);

    let segment = create_segment(
        segment_state.initial_version,
        segment_state.version,
        path,
        uuid,
        deferred_internal_id,
        &segment_state.config,
        stopped,
        false,
    )?;

    log_load_timing(path, "total", total_started);

    Ok(segment)
}

/// Build segment instance using given configuration.
/// Builder will generate folder for the segment and store all segment information inside it.
///
/// # Arguments
///
/// * `segments_path` - Path to the segments directory. Segment folder will be created in this directory
/// * `config` - Segment configuration
/// * `ready` - Whether the segment is ready after building; will save segment version
///
/// To load a segment, saving the segment version is required. If `ready` is false, the version
/// will not be stored. Then the segment is skipped on restart when trying to load it again. In
/// that case, the segment version must be stored manually to make it ready.
pub fn build_segment(
    segments_path: &Path,
    config: &SegmentConfig,
    deferred_internal_id: Option<PointOffsetType>,
    ready: bool,
) -> OperationResult<Segment> {
    let uuid = Uuid::new_v4();
    let segment_path = segments_path.join(uuid.to_string());
    let stopped = AtomicBool::new(false);

    fs::create_dir_all(&segment_path)?;
    let segment = create_segment(
        None,
        None,
        &segment_path,
        uuid,
        deferred_internal_id,
        config,
        &stopped,
        true,
    )?;
    segment.save_current_state()?;

    // Version is the last file to save, as it will be used to check if segment was built correctly.
    // If it is not saved, segment will be skipped.
    if ready {
        SegmentVersion::save(&segment_path)?;
    }

    Ok(segment)
}

/// Load v0.3.* segment data and migrate to current version
#[allow(deprecated)]
fn load_segment_state_v3(segment_path: &Path) -> OperationResult<SegmentState> {
    use crate::segment::compat::{SegmentConfigV5, StorageTypeV5, VectorDataConfigV5};

    #[derive(Deserialize)]
    #[serde(rename_all = "snake_case")]
    #[deprecated]
    pub struct SegmentStateV3 {
        pub version: SeqNumberType,
        pub config: SegmentConfigV3,
    }

    #[derive(Deserialize)]
    #[serde(rename_all = "snake_case")]
    #[deprecated]
    pub struct SegmentConfigV3 {
        /// Size of a vectors used
        pub vector_size: usize,
        /// Type of distance function used for measuring distance between vectors
        pub distance: Distance,
        /// Type of index used for search
        pub index: Indexes,
        /// Type of vector storage
        pub storage_type: StorageTypeV5,
        /// Defines payload storage type
        #[serde(default)]
        pub payload_storage_type: Option<PayloadStorageType>,
    }

    let path = segment_path.join(SEGMENT_STATE_FILE);

    let mut contents = String::new();

    let mut file = File::open(&path)?;
    file.read_to_string(&mut contents)?;

    serde_json::from_str::<SegmentStateV3>(&contents)
        .map(|state| {
            // Construct V5 version, then convert into current
            let vector_data = VectorDataConfigV5 {
                size: state.config.vector_size,
                distance: state.config.distance,
                hnsw_config: None,
                quantization_config: None,
                on_disk: None,
            };

            let segment_config = SegmentConfigV5 {
                vector_data: HashMap::from([(DEFAULT_VECTOR_NAME.to_owned(), vector_data)]),
                index: state.config.index,
                storage_type: state.config.storage_type,
                payload_storage_type: state.config.payload_storage_type,
                quantization_config: None,
            };

            SegmentState {
                initial_version: None,
                version: Some(state.version),
                config: segment_config.into(),
            }
        })
        .map_err(|err| {
            OperationError::service_error(format!(
                "Failed to read segment {}. Error: {}",
                path.to_str().unwrap(),
                err
            ))
        })
}

/// Load v0.5.0 segment data and migrate to current version
#[allow(deprecated)]
fn load_segment_state_v5(segment_path: &Path) -> OperationResult<SegmentState> {
    use crate::segment::compat::SegmentStateV5;

    let path = segment_path.join(SEGMENT_STATE_FILE);

    let mut contents = String::new();

    let mut file = File::open(&path)?;
    file.read_to_string(&mut contents)?;

    serde_json::from_str::<SegmentStateV5>(&contents)
        .map(SegmentStateV5::into)
        .map_err(|err| {
            OperationError::service_error(format!(
                "Failed to read segment {}. Error: {}",
                path.to_str().unwrap(),
                err
            ))
        })
}