1use std::sync::Arc;
2use std::sync::atomic::AtomicBool;
3
4use crate::common::budget::ResourceBudget;
5use crate::common::progress_tracker::new_progress_tracker;
6use crate::segment::common::operation_error::{OperationError, OperationResult};
7use crate::segment::types::HnswGlobalConfig;
8use crate::shard::optimizers::config::{
9 DEFAULT_DELETED_THRESHOLD, DEFAULT_VACUUM_MIN_VECTOR_NUMBER, TEMP_SEGMENTS_PATH,
10};
11use crate::shard::optimizers::config_mismatch_optimizer::ConfigMismatchOptimizer;
12use crate::shard::optimizers::indexing_optimizer::IndexingOptimizer;
13use crate::shard::optimizers::merge_optimizer::MergeOptimizer;
14use crate::shard::optimizers::segment_optimizer::{
15 Optimizer, max_num_indexing_threads, plan_optimizations,
16};
17use crate::shard::optimizers::vacuum_optimizer::VacuumOptimizer;
18use uuid::Uuid;
19
20use crate::edge::{EdgeShard, SEGMENTS_PATH};
21
22impl EdgeShard {
23 pub fn optimize(&self) -> OperationResult<bool> {
27 let optimizers = self.build_blocking_optimizers();
28 let stopped = AtomicBool::new(false);
29 let mut optimized_any = false;
30
31 loop {
32 let planned = {
33 let segments = self.segments.read();
34 plan_optimizations(&segments, &optimizers)
35 };
36
37 if planned.is_empty() {
38 return Ok(optimized_any);
39 }
40
41 let mut optimized_in_iteration = false;
42
43 for (optimizer, segment_ids) in planned {
44 let num_indexing_threads = optimizer.num_indexing_threads();
45 let desired_io = num_indexing_threads;
46 let budget = ResourceBudget::new(num_indexing_threads, desired_io);
48 let permit = budget.try_acquire(0, desired_io).ok_or_else(|| {
49 OperationError::service_error(format!(
50 "failed to acquire resource permit for {} optimizer",
51 optimizer.name(),
52 ))
53 })?;
54
55 let (_, progress) = new_progress_tracker();
56 let points_optimized = optimizer.as_ref().optimize(
57 self.segments.clone(),
58 segment_ids,
59 Uuid::new_v4(),
60 permit,
61 budget,
62 &stopped,
63 progress,
64 Box::new(|| ()),
65 )?;
66
67 if points_optimized > 0 {
68 optimized_in_iteration = true;
69 optimized_any = true;
70 }
71 }
72
73 if !optimized_in_iteration {
75 return Ok(optimized_any);
76 }
77 }
78 }
79
80 fn build_blocking_optimizers(&self) -> Vec<Arc<Optimizer>> {
81 let segments_path = self.path.join(SEGMENTS_PATH);
82 let temp_segments_path = self.path.join(TEMP_SEGMENTS_PATH);
83
84 let cfg = self.config();
85 let segment_optimizer_config = cfg.segment_optimizer_config();
86 let global_hnsw_config = cfg.hnsw_config;
87 let hnsw_global_config = HnswGlobalConfig::default();
88 let num_indexing_threads = max_num_indexing_threads(&segment_optimizer_config);
89 let threshold_config = cfg.optimizer_thresholds(num_indexing_threads);
90 let default_segments_number = cfg.optimizers.get_number_segments();
91
92 vec![
93 Arc::new(MergeOptimizer::new(
94 default_segments_number,
95 threshold_config,
96 segments_path.clone(),
97 temp_segments_path.clone(),
98 segment_optimizer_config.clone(),
99 hnsw_global_config.clone(),
100 )),
101 Arc::new(IndexingOptimizer::new(
102 default_segments_number,
103 threshold_config,
104 segments_path.clone(),
105 temp_segments_path.clone(),
106 segment_optimizer_config.clone(),
107 hnsw_global_config.clone(),
108 )),
109 Arc::new(VacuumOptimizer::new(
110 cfg.optimizers
111 .deleted_threshold
112 .unwrap_or(DEFAULT_DELETED_THRESHOLD),
113 cfg.optimizers
114 .vacuum_min_vector_number
115 .unwrap_or(DEFAULT_VACUUM_MIN_VECTOR_NUMBER),
116 threshold_config,
117 segments_path.clone(),
118 temp_segments_path.clone(),
119 segment_optimizer_config.clone(),
120 hnsw_global_config.clone(),
121 )),
122 Arc::new(ConfigMismatchOptimizer::new(
123 threshold_config,
124 segments_path,
125 temp_segments_path,
126 segment_optimizer_config,
127 global_hnsw_config,
128 hnsw_global_config,
129 )),
130 ]
131 }
132}
133
134#[cfg(test)]
139mod tests {
140 #![expect(clippy::wildcard_enum_match_arm, reason = "test code")]
141
142 use std::collections::HashMap;
143 use std::path::Path;
144
145 use fs_err as fs;
146 use crate::segment::data_types::vectors::{VectorInternal, VectorStructInternal};
147 use crate::segment::types::{Distance, ExtendedPointId, WithPayloadInterface, WithVector};
148 use crate::shard::count::CountRequestInternal;
149 use crate::shard::operations::CollectionUpdateOperations::PointOperation;
150 use crate::shard::operations::point_ops::PointInsertOperationsInternal::PointsList;
151 use crate::shard::operations::point_ops::PointOperations::{DeletePoints, UpsertPoints};
152 use crate::shard::operations::point_ops::{PointStructPersisted, VectorStructPersisted};
153 use crate::shard::optimizers::config::default_segment_number;
154 use uuid::Uuid;
155
156 use crate::edge::config::vectors::EdgeVectorParams;
157 use crate::edge::{EdgeConfig, EdgeShard};
158
159 const VECTOR_NAME: &str = "edge-test-vector";
160
161 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
162 #[test]
163 fn does_not_force_merge_all_segments_into_one() {
164 let dir = tempfile::Builder::new()
165 .prefix("edge-opt-do-not-force-one")
166 .tempdir()
167 .unwrap();
168
169 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
170 shard
171 .update(PointOperation(UpsertPoints(PointsList(vec![point(1)]))))
172 .unwrap();
173 drop(shard);
174
175 duplicate_single_segment(dir.path());
176
177 let reopened = EdgeShard::load(dir.path(), None).unwrap();
178 assert_eq!(reopened.info().segments_count, 2);
179
180 let optimized = reopened.optimize().unwrap();
181 assert!(!optimized, "optimizer should not force-merge all segments");
182 assert_eq!(reopened.info().segments_count, 2);
183
184 assert_points_retrievable_with_vectors(&reopened, &[1]);
185 }
186
187 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
188 #[test]
189 fn vacuum_optimizer_runs_in_blocking_mode_until_idle() {
190 let dir = tempfile::Builder::new()
191 .prefix("edge-opt-vacuum")
192 .tempdir()
193 .unwrap();
194
195 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
196
197 let points = (1..=1000).map(point).collect::<Vec<_>>();
198 shard
199 .update(PointOperation(UpsertPoints(PointsList(points))))
200 .unwrap();
201
202 let deleted_ids = (1..=250).map(ExtendedPointId::NumId).collect::<Vec<_>>();
204 shard
205 .update(PointOperation(DeletePoints { ids: deleted_ids }))
206 .unwrap();
207
208 let optimized = shard.optimize().unwrap();
209 assert!(optimized, "vacuum candidate should be optimized");
210
211 let optimized_again = shard.optimize().unwrap();
212 assert!(
213 !optimized_again,
214 "second run should be idle after blocking optimization"
215 );
216
217 assert_points_retrievable_with_vectors(&shard, &[251, 500, 999, 1000]);
219 }
220
221 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
224 #[test]
225 fn no_op_on_single_segment_without_deletions() {
226 let dir = tempfile::Builder::new()
227 .prefix("edge-opt-noop-single")
228 .tempdir()
229 .unwrap();
230
231 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
232
233 let points = (1..=100).map(point).collect::<Vec<_>>();
234 shard
235 .update(PointOperation(UpsertPoints(PointsList(points))))
236 .unwrap();
237
238 let optimized = shard.optimize().unwrap();
239 assert!(!optimized, "single clean segment should not be optimized");
240 assert_eq!(shard.info().points_count, 100);
241 assert_eq!(shard.info().segments_count, 1);
242
243 assert_points_retrievable_with_vectors(&shard, &[1, 50, 100]);
244 }
245
246 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
248 #[test]
249 fn no_op_on_empty_shard() {
250 let dir = tempfile::Builder::new()
251 .prefix("edge-opt-noop-empty")
252 .tempdir()
253 .unwrap();
254
255 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
256
257 let optimized = shard.optimize().unwrap();
258 assert!(!optimized, "empty shard should not trigger optimization");
259 assert_eq!(shard.info().points_count, 0);
260 }
261
262 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
265 #[test]
266 fn merge_reduces_excess_segments() {
267 let target_count = default_segment_number() + 6;
268
269 let dir = tempfile::Builder::new()
270 .prefix("edge-opt-merge-excess")
271 .tempdir()
272 .unwrap();
273
274 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
275 shard
276 .update(PointOperation(UpsertPoints(PointsList(vec![point(1)]))))
277 .unwrap();
278 drop(shard);
279
280 multiply_segments(dir.path(), target_count);
281
282 let reopened = EdgeShard::load(dir.path(), None).unwrap();
283 reopened.optimize().unwrap();
284 let info = reopened.info();
285 assert!(
286 info.segments_count <= default_segment_number() + 1,
287 "segments should be reduced after merge: got {} segments, \
288 expected at most {} (default_segment_number={}, +1 for appendable)",
289 info.segments_count,
290 default_segment_number() + 1,
291 default_segment_number(),
292 );
293
294 let count = reopened
299 .count(CountRequestInternal {
300 filter: None,
301 exact: true,
302 })
303 .unwrap();
304 assert!(count >= 1, "shard should still have data after merge");
305
306 assert_points_retrievable_with_vectors(&reopened, &[1]);
307 }
308
309 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
311 #[test]
312 fn optimization_is_idempotent_after_merge() {
313 let target_count = default_segment_number() + 6;
314
315 let dir = tempfile::Builder::new()
316 .prefix("edge-opt-merge-idempotent")
317 .tempdir()
318 .unwrap();
319
320 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
321 shard
322 .update(PointOperation(UpsertPoints(PointsList(vec![point(1)]))))
323 .unwrap();
324 drop(shard);
325
326 multiply_segments(dir.path(), target_count);
327
328 let reopened = EdgeShard::load(dir.path(), None).unwrap();
329 reopened.optimize().unwrap();
331 let segments_after_first = reopened.info().segments_count;
332
333 let optimized = reopened.optimize().unwrap();
335 assert!(
336 !optimized,
337 "second optimization run should be idle after merge"
338 );
339 assert_eq!(reopened.info().segments_count, segments_after_first);
340
341 assert_points_retrievable_with_vectors(&reopened, &[1]);
342 }
343
344 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
347 #[test]
348 fn vacuum_below_threshold_is_noop() {
349 let dir = tempfile::Builder::new()
350 .prefix("edge-opt-vacuum-below")
351 .tempdir()
352 .unwrap();
353
354 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
355
356 let points = (1..=1000).map(point).collect::<Vec<_>>();
357 shard
358 .update(PointOperation(UpsertPoints(PointsList(points))))
359 .unwrap();
360
361 let deleted_ids = (1..=50).map(ExtendedPointId::NumId).collect::<Vec<_>>();
363 shard
364 .update(PointOperation(DeletePoints { ids: deleted_ids }))
365 .unwrap();
366
367 let optimized = shard.optimize().unwrap();
368 assert!(
369 !optimized,
370 "5% deletion should not trigger vacuum (threshold is 20%)"
371 );
372
373 assert_points_retrievable_with_vectors(&shard, &[51, 500, 1000]);
375 }
376
377 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
380 #[test]
381 fn vacuum_below_min_vector_count_is_noop() {
382 let dir = tempfile::Builder::new()
383 .prefix("edge-opt-vacuum-min-vecs")
384 .tempdir()
385 .unwrap();
386
387 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
388
389 let points = (1..=100).map(point).collect::<Vec<_>>();
391 shard
392 .update(PointOperation(UpsertPoints(PointsList(points))))
393 .unwrap();
394
395 let deleted_ids = (1..=50).map(ExtendedPointId::NumId).collect::<Vec<_>>();
397 shard
398 .update(PointOperation(DeletePoints { ids: deleted_ids }))
399 .unwrap();
400
401 let optimized = shard.optimize().unwrap();
402 assert!(
403 !optimized,
404 "high deletion ratio with only 100 total points should not trigger vacuum \
405 (min_vectors_number=1000)"
406 );
407
408 assert_points_retrievable_with_vectors(&shard, &[51, 75, 100]);
409 }
410
411 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
414 #[test]
415 fn vacuum_preserves_remaining_points() {
416 let dir = tempfile::Builder::new()
417 .prefix("edge-opt-vacuum-data")
418 .tempdir()
419 .unwrap();
420
421 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
422
423 let points = (1..=1000).map(point).collect::<Vec<_>>();
424 shard
425 .update(PointOperation(UpsertPoints(PointsList(points))))
426 .unwrap();
427
428 let deleted_ids = (1..=250).map(ExtendedPointId::NumId).collect::<Vec<_>>();
430 shard
431 .update(PointOperation(DeletePoints {
432 ids: deleted_ids.clone(),
433 }))
434 .unwrap();
435
436 let optimized = shard.optimize().unwrap();
437 assert!(optimized, "25% deletion should trigger vacuum");
438
439 let count = shard
441 .count(CountRequestInternal {
442 filter: None,
443 exact: true,
444 })
445 .unwrap();
446 assert_eq!(count, 750, "should have 750 remaining points after vacuum");
447
448 let deleted_results = shard
450 .retrieve(
451 &deleted_ids,
452 Some(WithPayloadInterface::Bool(false)),
453 Some(WithVector::Bool(false)),
454 )
455 .unwrap();
456 assert!(
457 deleted_results.is_empty(),
458 "deleted points should not be retrievable"
459 );
460
461 assert_points_retrievable_with_vectors(&shard, &[251, 500, 750, 1000]);
463 }
464
465 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
471 #[test]
472 fn vacuum_after_all_points_deleted() {
473 let dir = tempfile::Builder::new()
474 .prefix("edge-opt-vacuum-all-deleted")
475 .tempdir()
476 .unwrap();
477
478 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
479
480 let points = (1..=1000).map(point).collect::<Vec<_>>();
481 shard
482 .update(PointOperation(UpsertPoints(PointsList(points))))
483 .unwrap();
484
485 let deleted_ids = (1..=1000).map(ExtendedPointId::NumId).collect();
487 shard
488 .update(PointOperation(DeletePoints { ids: deleted_ids }))
489 .unwrap();
490
491 let _optimized = shard.optimize().unwrap();
494
495 let count = shard
496 .count(CountRequestInternal {
497 filter: None,
498 exact: true,
499 })
500 .unwrap();
501 assert_eq!(count, 0, "all points should be gone after vacuum");
502
503 shard
505 .update(PointOperation(UpsertPoints(PointsList(vec![point(9999)]))))
506 .unwrap();
507 let count = shard
508 .count(CountRequestInternal {
509 filter: None,
510 exact: true,
511 })
512 .unwrap();
513 assert_eq!(count, 1, "shard should accept new points after full vacuum");
514
515 assert_points_retrievable_with_vectors(&shard, &[9999]);
516 }
517
518 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
522 #[test]
523 fn vacuum_at_exact_threshold_boundary_is_noop() {
524 let dir = tempfile::Builder::new()
525 .prefix("edge-opt-vacuum-boundary")
526 .tempdir()
527 .unwrap();
528
529 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
530
531 let points = (1..=1000).map(point).collect::<Vec<_>>();
532 shard
533 .update(PointOperation(UpsertPoints(PointsList(points))))
534 .unwrap();
535
536 let deleted_ids = (1..=200).map(ExtendedPointId::NumId).collect();
538 shard
539 .update(PointOperation(DeletePoints { ids: deleted_ids }))
540 .unwrap();
541
542 let optimized = shard.optimize().unwrap();
543 assert!(
544 !optimized,
545 "exactly 20% deletion (not strictly greater) should not trigger vacuum"
546 );
547
548 assert_points_retrievable_with_vectors(&shard, &[201, 500, 1000]);
549 }
550
551 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
553 #[test]
554 fn vacuum_just_above_threshold_triggers() {
555 let dir = tempfile::Builder::new()
556 .prefix("edge-opt-vacuum-above")
557 .tempdir()
558 .unwrap();
559
560 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
561
562 let points = (1..=1000).map(point).collect::<Vec<_>>();
563 shard
564 .update(PointOperation(UpsertPoints(PointsList(points))))
565 .unwrap();
566
567 let deleted_ids = (1..=201).map(ExtendedPointId::NumId).collect();
569 shard
570 .update(PointOperation(DeletePoints { ids: deleted_ids }))
571 .unwrap();
572
573 let optimized = shard.optimize().unwrap();
574 assert!(
575 optimized,
576 "20.1% deletion should trigger vacuum (threshold is >20%)"
577 );
578
579 assert_points_retrievable_with_vectors(&shard, &[202, 500, 1000]);
581 }
582
583 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
586 #[test]
587 fn merge_and_vacuum_cooperate() {
588 let target_count = default_segment_number() + 6;
589
590 let dir = tempfile::Builder::new()
591 .prefix("edge-opt-merge-vacuum")
592 .tempdir()
593 .unwrap();
594
595 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
596
597 let points = (1..=1000).map(point).collect::<Vec<_>>();
599 shard
600 .update(PointOperation(UpsertPoints(PointsList(points))))
601 .unwrap();
602 let deleted_ids = (1..=250).map(ExtendedPointId::NumId).collect();
603 shard
604 .update(PointOperation(DeletePoints { ids: deleted_ids }))
605 .unwrap();
606 drop(shard);
607
608 multiply_segments(dir.path(), target_count);
610
611 let reopened = EdgeShard::load(dir.path(), None).unwrap();
613 reopened.optimize().unwrap();
614
615 let info = reopened.info();
616 assert!(
617 info.segments_count <= default_segment_number() + 1,
618 "excess segments should be merged: got {}",
619 info.segments_count,
620 );
621
622 let count = reopened
627 .count(CountRequestInternal {
628 filter: None,
629 exact: true,
630 })
631 .unwrap();
632 assert!(
633 count >= 750,
634 "merged shard should preserve surviving points"
635 );
636
637 assert_points_retrievable_with_vectors(&reopened, &[251, 500, 1000]);
639
640 let optimized = reopened.optimize().unwrap();
642 assert!(!optimized, "second run should be idle after merge+vacuum");
643 }
644
645 #[cfg_attr(target_os = "windows", ignore = "slow on Windows, not OS-specific")]
647 #[test]
648 fn data_survives_optimize_and_reload() {
649 let dir = tempfile::Builder::new()
650 .prefix("edge-opt-reload")
651 .tempdir()
652 .unwrap();
653
654 let shard = EdgeShard::new(dir.path(), test_config()).unwrap();
655
656 let points = (1..=1000).map(point).collect::<Vec<_>>();
657 shard
658 .update(PointOperation(UpsertPoints(PointsList(points))))
659 .unwrap();
660
661 let deleted_ids = (1..=250).map(ExtendedPointId::NumId).collect();
663 shard
664 .update(PointOperation(DeletePoints { ids: deleted_ids }))
665 .unwrap();
666
667 let optimized = shard.optimize().unwrap();
668 assert!(optimized);
669 drop(shard);
670
671 let reopened = EdgeShard::load(dir.path(), None).unwrap();
673
674 let count = reopened
675 .count(CountRequestInternal {
676 filter: None,
677 exact: true,
678 })
679 .unwrap();
680 assert_eq!(count, 750, "point count should be preserved across reload");
681
682 assert_points_retrievable_with_vectors(&reopened, &[251, 500, 750, 1000]);
684 }
685
686 fn assert_points_retrievable_with_vectors(shard: &EdgeShard, ids: &[u64]) {
689 let point_ids = ids
690 .iter()
691 .map(|id| ExtendedPointId::NumId(*id))
692 .collect::<Vec<_>>();
693 let results = shard
694 .retrieve(
695 &point_ids,
696 Some(WithPayloadInterface::Bool(false)),
697 Some(WithVector::Bool(true)),
698 )
699 .unwrap();
700 assert_eq!(
701 results.len(),
702 ids.len(),
703 "expected {} retrievable points, got {}",
704 ids.len(),
705 results.len(),
706 );
707 for (result, &expected_id) in results.iter().zip(ids) {
708 assert_eq!(result.id, ExtendedPointId::NumId(expected_id));
709 let vectors = match result.vector.as_ref().expect("vector should be present") {
710 VectorStructInternal::Named(named) => named,
711 other => panic!("expected Named vectors, got {other:?}"),
712 };
713 let vec = match vectors.get(VECTOR_NAME).expect("vector name should exist") {
714 VectorInternal::Dense(v) => v,
715 other => panic!("expected Dense vector, got {other:?}"),
716 };
717 assert_eq!(
718 vec,
719 &vec![expected_id as f32],
720 "vector value mismatch for point {expected_id}"
721 );
722 }
723 }
724
725 fn test_config() -> EdgeConfig {
726 EdgeConfig {
727 on_disk_payload: false,
728 vectors: HashMap::from([(
729 VECTOR_NAME.to_string(),
730 EdgeVectorParams {
731 size: 1,
732 distance: Distance::Dot,
733 quantization_config: None,
734 multivector_config: None,
735 datatype: None,
736 on_disk: None,
737 hnsw_config: None,
738 },
739 )]),
740 sparse_vectors: HashMap::new(),
741 hnsw_config: Default::default(),
742 quantization_config: None,
743 optimizers: Default::default(),
744 wal_options: None,
745 }
746 }
747
748 fn point(id: u64) -> PointStructPersisted {
749 PointStructPersisted {
750 id: ExtendedPointId::NumId(id),
751 vector: VectorStructPersisted::from(VectorStructInternal::Named(HashMap::from([(
752 VECTOR_NAME.to_string(),
753 VectorInternal::from(vec![id as f32]),
754 )]))),
755 payload: None,
756 }
757 }
758
759 fn multiply_segments(shard_dir: &Path, target_count: usize) {
761 let segments_path = shard_dir.join("segments");
762 let segment_dirs = fs::read_dir(&segments_path)
763 .unwrap()
764 .filter_map(Result::ok)
765 .map(|entry| entry.path())
766 .filter(|path| path.is_dir())
767 .collect::<Vec<_>>();
768 assert!(!segment_dirs.is_empty(), "need at least one source segment");
769
770 let source = &segment_dirs[0];
771 let current_count = segment_dirs.len();
772 for _ in current_count..target_count {
773 let target = segments_path.join(Uuid::new_v4().to_string());
774 copy_dir_recursive(source, &target);
775 }
776 }
777
778 fn duplicate_single_segment(shard_dir: &Path) {
779 let segments_path = shard_dir.join("segments");
780 let segment_dirs = fs::read_dir(&segments_path)
781 .unwrap()
782 .filter_map(Result::ok)
783 .map(|entry| entry.path())
784 .filter(|path| path.is_dir())
785 .collect::<Vec<_>>();
786 assert_eq!(segment_dirs.len(), 1, "expected exactly one source segment");
787
788 let source = &segment_dirs[0];
789 let target = segments_path.join(Uuid::new_v4().to_string());
790 copy_dir_recursive(source, &target);
791 }
792
793 fn copy_dir_recursive(from: &Path, to: &Path) {
794 fs::create_dir_all(to).unwrap();
795 for entry in fs::read_dir(from).unwrap().filter_map(Result::ok) {
796 let from_path = entry.path();
797 let to_path = to.join(entry.file_name());
798 if entry.file_type().unwrap().is_dir() {
799 copy_dir_recursive(&from_path, &to_path);
800 } else {
801 fs::copy(&from_path, &to_path).unwrap();
802 }
803 }
804 }
805}