qdrant_edge/shard/files/
mod.rs

1use std::path::{Path, PathBuf};
2
3use fs_err as fs;
4
5pub const WAL_PATH: &str = "wal";
6pub const SEGMENTS_PATH: &str = "segments";
7pub const NEWEST_CLOCKS_PATH: &str = "newest_clocks.json";
8pub const OLDEST_CLOCKS_PATH: &str = "oldest_clocks.json";
9pub const APPLIED_SEQ_FILE: &str = "applied_seq.json";
10
11/// Shard represents all files, associated with a shard data (excluding configs)
12/// Useful to not forget some files, while making operations on shard data
13#[derive(Debug, Clone)]
14pub struct ShardDataFiles {
15    pub wal_path: PathBuf,
16    pub segments_path: PathBuf,
17    pub newest_clocks_path: PathBuf,
18    pub oldest_clocks_path: PathBuf,
19    pub applied_seq_path: PathBuf,
20}
21
22#[inline]
23pub fn segments_path(shard_path: &Path) -> PathBuf {
24    shard_path.join(SEGMENTS_PATH)
25}
26
27#[inline]
28pub fn wal_path(shard_path: &Path) -> PathBuf {
29    shard_path.join(WAL_PATH)
30}
31
32#[inline]
33pub fn newest_clocks_path(shard_path: &Path) -> PathBuf {
34    shard_path.join(NEWEST_CLOCKS_PATH)
35}
36
37#[inline]
38pub fn oldest_clocks_path(shard_path: &Path) -> PathBuf {
39    shard_path.join(OLDEST_CLOCKS_PATH)
40}
41
42#[inline]
43pub fn applied_seq_path(shard_path: &Path) -> PathBuf {
44    shard_path.join(APPLIED_SEQ_FILE)
45}
46
47#[inline]
48pub fn get_shard_data_files(shard_path: &Path) -> ShardDataFiles {
49    ShardDataFiles {
50        wal_path: wal_path(shard_path),
51        segments_path: segments_path(shard_path),
52        newest_clocks_path: newest_clocks_path(shard_path),
53        oldest_clocks_path: oldest_clocks_path(shard_path),
54        applied_seq_path: applied_seq_path(shard_path),
55    }
56}
57
58/// Checks if path have local shard data present
59pub fn check_data(shard_path: &Path) -> bool {
60    let wal_path = wal_path(shard_path);
61    let segments_path = segments_path(shard_path);
62    wal_path.exists() && segments_path.exists()
63}
64
65pub fn clear_data(shard_path: &Path) -> std::io::Result<()> {
66    let shard_data_files = get_shard_data_files(shard_path);
67    let ShardDataFiles {
68        wal_path,
69        segments_path,
70        newest_clocks_path,
71        oldest_clocks_path,
72        applied_seq_path,
73    } = shard_data_files;
74
75    if wal_path.exists() {
76        fs::remove_dir_all(wal_path)?;
77    }
78
79    if segments_path.exists() {
80        fs::remove_dir_all(segments_path)?;
81    }
82
83    if newest_clocks_path.exists() {
84        fs::remove_file(newest_clocks_path)?;
85    }
86
87    if oldest_clocks_path.exists() {
88        fs::remove_file(oldest_clocks_path)?;
89    }
90
91    if applied_seq_path.exists() {
92        fs::remove_file(applied_seq_path)?;
93    }
94
95    Ok(())
96}
97
98pub fn move_data(from_shard_path: &Path, to_shard_path: &Path) -> std::io::Result<()> {
99    let from_shard_data_files = get_shard_data_files(from_shard_path);
100    let to_shard_data_files = get_shard_data_files(to_shard_path);
101
102    let ShardDataFiles {
103        wal_path: from_wal_path,
104        segments_path: from_segments_path,
105        newest_clocks_path: from_newest_clocks_path,
106        oldest_clocks_path: from_oldest_clocks_path,
107        applied_seq_path: from_applied_seq_path,
108    } = from_shard_data_files;
109
110    let ShardDataFiles {
111        wal_path: to_wal_path,
112        segments_path: to_segments_path,
113        newest_clocks_path: to_newest_clocks_path,
114        oldest_clocks_path: to_oldest_clocks_path,
115        applied_seq_path: to_applied_seq_path,
116    } = to_shard_data_files;
117
118    if from_wal_path.exists() {
119        crate::common::fs::move_dir(&from_wal_path, &to_wal_path)?;
120    }
121
122    if from_segments_path.exists() {
123        crate::common::fs::move_dir(&from_segments_path, &to_segments_path)?;
124    }
125
126    if from_newest_clocks_path.exists() {
127        crate::common::fs::move_file(&from_newest_clocks_path, &to_newest_clocks_path)?;
128    }
129
130    if from_oldest_clocks_path.exists() {
131        crate::common::fs::move_file(&from_oldest_clocks_path, &to_oldest_clocks_path)?;
132    }
133
134    if from_applied_seq_path.exists() {
135        crate::common::fs::move_file(&from_applied_seq_path, &to_applied_seq_path)?;
136    }
137
138    Ok(())
139}
140
141pub const PAYLOAD_INDEX_CONFIG_FILE: &str = "payload_index.json";
qdrant_edge/shard/files/mod.rs

qdrant_edge/shard/files/
mod.rs
