1use crate::{EsDocument, EsError, EsSearchRequest, EsSync, InMemoryEsSync};
17use serde::{Deserialize, Serialize};
18use std::collections::HashMap;
19use std::sync::RwLock;
20
21#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
27pub enum BulkAction {
28 Index {
30 index: String,
31 id: String,
32 source: serde_json::Value,
33 },
34 Create {
36 index: String,
37 id: String,
38 source: serde_json::Value,
39 },
40 Update {
42 index: String,
43 id: String,
44 doc: serde_json::Value,
45 },
46 Delete { index: String, id: String },
48}
49
50impl BulkAction {
51 pub fn index(&self) -> &str {
53 match self {
54 BulkAction::Index { index, .. }
55 | BulkAction::Create { index, .. }
56 | BulkAction::Update { index, .. }
57 | BulkAction::Delete { index, .. } => index,
58 }
59 }
60
61 pub fn id(&self) -> &str {
63 match self {
64 BulkAction::Index { id, .. }
65 | BulkAction::Create { id, .. }
66 | BulkAction::Update { id, .. }
67 | BulkAction::Delete { id, .. } => id,
68 }
69 }
70}
71
72#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
74pub struct BulkItemResult {
75 pub action: String,
77 pub index: String,
79 pub id: String,
81 pub status: u16,
83 pub error: Option<String>,
85}
86
87impl BulkItemResult {
88 pub fn success(action: &str, index: &str, id: &str, status: u16) -> Self {
89 Self {
90 action: action.to_string(),
91 index: index.to_string(),
92 id: id.to_string(),
93 status,
94 error: None,
95 }
96 }
97
98 pub fn failure(action: &str, index: &str, id: &str, status: u16, error: &str) -> Self {
99 Self {
100 action: action.to_string(),
101 index: index.to_string(),
102 id: id.to_string(),
103 status,
104 error: Some(error.to_string()),
105 }
106 }
107
108 pub fn is_success(&self) -> bool {
109 self.status >= 200 && self.status < 300
110 }
111}
112
113#[derive(Debug, Clone, Serialize, Deserialize)]
115pub struct BulkResult {
116 pub items: Vec<BulkItemResult>,
118 pub took: u64,
120 pub success_count: usize,
122 pub failure_count: usize,
124}
125
126impl BulkResult {
127 pub fn new(items: Vec<BulkItemResult>, took: u64) -> Self {
128 let success_count = items.iter().filter(|r| r.is_success()).count();
129 let failure_count = items.len() - success_count;
130 Self {
131 items,
132 took,
133 success_count,
134 failure_count,
135 }
136 }
137
138 pub fn all_success(&self) -> bool {
140 self.failure_count == 0
141 }
142
143 pub fn failures(&self) -> Vec<&BulkItemResult> {
145 self.items.iter().filter(|r| !r.is_success()).collect()
146 }
147}
148
149pub struct BulkExecutor {
151 backend: InMemoryEsSync,
152}
153
154impl BulkExecutor {
155 pub fn new(backend: InMemoryEsSync) -> Self {
156 Self { backend }
157 }
158
159 pub fn from_new() -> Self {
160 Self::new(InMemoryEsSync::new())
161 }
162
163 pub fn backend(&self) -> &InMemoryEsSync {
165 &self.backend
166 }
167
168 pub fn bulk(&self, actions: Vec<BulkAction>) -> Result<BulkResult, EsError> {
172 let start = std::time::Instant::now();
173 let mut items = Vec::with_capacity(actions.len());
174
175 for action in actions {
176 let result = self.execute_action(action)?;
177 items.push(result);
178 }
179
180 Ok(BulkResult::new(items, start.elapsed().as_millis() as u64))
181 }
182
183 fn execute_action(&self, action: BulkAction) -> Result<BulkItemResult, EsError> {
185 match action {
186 BulkAction::Index { index, id, source } => {
187 let doc = EsDocument::new(&index, source).with_id(&id);
188 self.backend.sync_to_es(vec![doc])?;
189 Ok(BulkItemResult::success("index", &index, &id, 200))
190 }
191 BulkAction::Create { index, id, source } => {
192 let doc = EsDocument::new(&index, source).with_id(&id);
194 self.backend.sync_to_es(vec![doc])?;
195 Ok(BulkItemResult::success("create", &index, &id, 201))
196 }
197 BulkAction::Update { index, id, doc } => {
198 let search_req = EsSearchRequest::new(&index, crate::EsQuery::match_all())
202 .with_pagination(0, 10000);
203 let result = match self.backend.search(search_req) {
204 Ok(r) => r,
205 Err(EsError::IndexNotFound(_)) => {
206 return Ok(BulkItemResult::failure(
207 "update",
208 &index,
209 &id,
210 404,
211 "document not found",
212 ));
213 }
214 Err(e) => return Err(e),
215 };
216 let hit = result.hits.iter().find(|h| h.id == id);
217 if hit.is_none() {
218 return Ok(BulkItemResult::failure(
219 "update",
220 &index,
221 &id,
222 404,
223 "document not found",
224 ));
225 }
226 let hit = hit.unwrap();
227 let mut source = hit.source.clone();
228 if let (Some(obj), Some(updates)) = (source.as_object_mut(), doc.as_object()) {
229 for (k, v) in updates {
230 obj.insert(k.clone(), v.clone());
231 }
232 }
233 let new_doc = EsDocument::new(&index, source).with_id(&id);
234 self.backend.sync_to_es(vec![new_doc])?;
235 Ok(BulkItemResult::success("update", &index, &id, 200))
236 }
237 BulkAction::Delete { index, id } => {
238 match self.backend.delete_from_es(&index, vec![id.clone()]) {
239 Ok(_) => Ok(BulkItemResult::success("delete", &index, &id, 200)),
240 Err(EsError::IndexNotFound(_)) => Ok(BulkItemResult::failure(
241 "delete",
242 &index,
243 &id,
244 404,
245 "index not found",
246 )),
247 Err(e) => Err(e),
248 }
249 }
250 }
251 }
252}
253
254#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
260pub enum SuggesterType {
261 Term,
263 Completion,
265 Phrase,
267}
268
269#[derive(Debug, Clone, Serialize, Deserialize)]
271pub struct SuggestRequest {
272 pub name: String,
274 pub prefix: String,
276 pub suggester_type: SuggesterType,
278 pub field: String,
280 pub size: usize,
282}
283
284impl SuggestRequest {
285 pub fn new(name: impl Into<String>, prefix: impl Into<String>) -> Self {
286 Self {
287 name: name.into(),
288 prefix: prefix.into(),
289 suggester_type: SuggesterType::Term,
290 field: "text".to_string(),
291 size: 5,
292 }
293 }
294
295 pub fn with_field(mut self, field: impl Into<String>) -> Self {
296 self.field = field.into();
297 self
298 }
299
300 pub fn with_size(mut self, size: usize) -> Self {
301 self.size = size;
302 self
303 }
304
305 pub fn with_type(mut self, suggester_type: SuggesterType) -> Self {
306 self.suggester_type = suggester_type;
307 self
308 }
309}
310
311#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
313pub struct SuggestOption {
314 pub text: String,
316 pub freq: u64,
318 pub score: f64,
320}
321
322#[derive(Debug, Clone, Serialize, Deserialize)]
324pub struct SuggestResult {
325 pub name: String,
327 pub options: Vec<SuggestOption>,
329}
330
331pub struct MemorySuggester {
335 backend: InMemoryEsSync,
336}
337
338impl MemorySuggester {
339 pub fn new(backend: InMemoryEsSync) -> Self {
340 Self { backend }
341 }
342
343 pub fn from_new() -> Self {
344 Self::new(InMemoryEsSync::new())
345 }
346
347 pub fn suggest(&self, index: &str, request: &SuggestRequest) -> Result<SuggestResult, EsError> {
351 let search_req =
352 EsSearchRequest::new(index, crate::EsQuery::match_all()).with_pagination(0, 10000);
353 let result = self.backend.search(search_req)?;
354
355 let prefix_lower = request.prefix.to_lowercase();
356 let mut candidates: HashMap<String, u64> = HashMap::new();
357
358 for hit in &result.hits {
359 let text_opt = hit.source.get(&request.field).and_then(|v| v.as_str());
361 if let Some(text) = text_opt {
362 for word in text.split(|c: char| !c.is_alphanumeric()) {
364 if word.is_empty() {
365 continue;
366 }
367 let word_lower = word.to_lowercase();
368 if word_lower.starts_with(&prefix_lower) {
369 *candidates.entry(word_lower).or_insert(0) += 1;
370 }
371 }
372 }
373 }
374
375 let mut options: Vec<SuggestOption> = candidates
376 .into_iter()
377 .map(|(text, freq)| {
378 let prefix_ratio = prefix_lower.len() as f64 / text.len().max(1) as f64;
380 let score = freq as f64 * (1.0 + prefix_ratio);
381 SuggestOption { text, freq, score }
382 })
383 .collect();
384 options.sort_by(|a, b| {
386 b.score
387 .partial_cmp(&a.score)
388 .unwrap_or(std::cmp::Ordering::Equal)
389 });
390 options.truncate(request.size);
391
392 Ok(SuggestResult {
393 name: request.name.clone(),
394 options,
395 })
396 }
397
398 pub fn index_doc(&self, index: &str, id: &str, doc: serde_json::Value) -> Result<(), EsError> {
400 let doc = EsDocument::new(index, doc).with_id(id);
401 self.backend.sync_to_es(vec![doc])?;
402 Ok(())
403 }
404}
405
406#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
412pub enum AggregationType {
413 Terms,
415 Range,
417 Sum,
419 Avg,
421 Max,
423 Min,
425 Histogram,
427 ValueCount,
429}
430
431#[derive(Debug, Clone, Serialize, Deserialize)]
433pub struct Aggregation {
434 pub name: String,
436 pub agg_type: AggregationType,
438 pub field: String,
440 pub size: usize,
442 pub ranges: Vec<AggRange>,
444 pub interval: Option<f64>,
446}
447
448impl Aggregation {
449 pub fn terms(name: impl Into<String>, field: impl Into<String>) -> Self {
450 Self {
451 name: name.into(),
452 agg_type: AggregationType::Terms,
453 field: field.into(),
454 size: 10,
455 ranges: Vec::new(),
456 interval: None,
457 }
458 }
459
460 pub fn range(name: impl Into<String>, field: impl Into<String>) -> Self {
461 Self {
462 name: name.into(),
463 agg_type: AggregationType::Range,
464 field: field.into(),
465 size: 0,
466 ranges: Vec::new(),
467 interval: None,
468 }
469 }
470
471 pub fn sum(name: impl Into<String>, field: impl Into<String>) -> Self {
472 Self {
473 name: name.into(),
474 agg_type: AggregationType::Sum,
475 field: field.into(),
476 size: 0,
477 ranges: Vec::new(),
478 interval: None,
479 }
480 }
481
482 pub fn avg(name: impl Into<String>, field: impl Into<String>) -> Self {
483 Self {
484 name: name.into(),
485 agg_type: AggregationType::Avg,
486 field: field.into(),
487 size: 0,
488 ranges: Vec::new(),
489 interval: None,
490 }
491 }
492
493 pub fn max(name: impl Into<String>, field: impl Into<String>) -> Self {
494 Self {
495 name: name.into(),
496 agg_type: AggregationType::Max,
497 field: field.into(),
498 size: 0,
499 ranges: Vec::new(),
500 interval: None,
501 }
502 }
503
504 pub fn min(name: impl Into<String>, field: impl Into<String>) -> Self {
505 Self {
506 name: name.into(),
507 agg_type: AggregationType::Min,
508 field: field.into(),
509 size: 0,
510 ranges: Vec::new(),
511 interval: None,
512 }
513 }
514
515 pub fn histogram(name: impl Into<String>, field: impl Into<String>, interval: f64) -> Self {
516 Self {
517 name: name.into(),
518 agg_type: AggregationType::Histogram,
519 field: field.into(),
520 size: 0,
521 ranges: Vec::new(),
522 interval: Some(interval),
523 }
524 }
525
526 pub fn count(name: impl Into<String>, field: impl Into<String>) -> Self {
527 Self {
528 name: name.into(),
529 agg_type: AggregationType::ValueCount,
530 field: field.into(),
531 size: 0,
532 ranges: Vec::new(),
533 interval: None,
534 }
535 }
536
537 pub fn with_size(mut self, size: usize) -> Self {
538 self.size = size;
539 self
540 }
541
542 pub fn with_range(mut self, from: Option<f64>, to: Option<f64>) -> Self {
543 self.ranges.push(AggRange { from, to });
544 self
545 }
546}
547
548#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
550pub struct AggRange {
551 pub from: Option<f64>,
552 pub to: Option<f64>,
553}
554
555impl AggRange {
556 pub fn new(from: Option<f64>, to: Option<f64>) -> Self {
557 Self { from, to }
558 }
559
560 pub fn contains(&self, value: f64) -> bool {
562 if let Some(from) = self.from {
563 if value < from {
564 return false;
565 }
566 }
567 if let Some(to) = self.to {
568 if value >= to {
569 return false;
570 }
571 }
572 true
573 }
574
575 pub fn label(&self) -> String {
577 match (self.from, self.to) {
578 (Some(from), Some(to)) => format!("{}-{}", from, to),
579 (Some(from), None) => format!("{}-*", from),
580 (None, Some(to)) => format!("*-{}", to),
581 (None, None) => "*-*".to_string(),
582 }
583 }
584}
585
586#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
588pub struct AggBucket {
589 pub key: String,
591 pub doc_count: u64,
593 pub value: Option<f64>,
595}
596
597#[derive(Debug, Clone, Serialize, Deserialize)]
599pub struct AggregationResult {
600 pub name: String,
602 pub agg_type: AggregationType,
604 pub buckets: Vec<AggBucket>,
606 pub value: Option<f64>,
608}
609
610pub struct MemoryAggregator {
612 backend: InMemoryEsSync,
613}
614
615impl MemoryAggregator {
616 pub fn new(backend: InMemoryEsSync) -> Self {
617 Self { backend }
618 }
619
620 pub fn from_new() -> Self {
621 Self::new(InMemoryEsSync::new())
622 }
623
624 pub fn aggregate(
628 &self,
629 index: &str,
630 query: crate::EsQuery,
631 aggregations: &[Aggregation],
632 ) -> Result<Vec<AggregationResult>, EsError> {
633 let search_req = EsSearchRequest::new(index, query).with_pagination(0, 10000);
634 let result = self.backend.search(search_req)?;
635 let docs: Vec<&serde_json::Value> = result.hits.iter().map(|h| &h.source).collect();
636
637 let mut results = Vec::with_capacity(aggregations.len());
638 for agg in aggregations {
639 let result = self.compute_aggregation(agg, &docs);
640 results.push(result);
641 }
642 Ok(results)
643 }
644
645 fn compute_aggregation(
647 &self,
648 agg: &Aggregation,
649 docs: &[&serde_json::Value],
650 ) -> AggregationResult {
651 match agg.agg_type {
652 AggregationType::Terms => self.compute_terms(agg, docs),
653 AggregationType::Range => self.compute_range(agg, docs),
654 AggregationType::Sum => {
655 let sum = self.sum_field(agg, docs);
656 AggregationResult {
657 name: agg.name.clone(),
658 agg_type: agg.agg_type.clone(),
659 buckets: Vec::new(),
660 value: Some(sum),
661 }
662 }
663 AggregationType::Avg => {
664 let avg = self.avg_field(agg, docs);
665 AggregationResult {
666 name: agg.name.clone(),
667 agg_type: agg.agg_type.clone(),
668 buckets: Vec::new(),
669 value: avg,
670 }
671 }
672 AggregationType::Max => {
673 let max = self.max_field(agg, docs);
674 AggregationResult {
675 name: agg.name.clone(),
676 agg_type: agg.agg_type.clone(),
677 buckets: Vec::new(),
678 value: max,
679 }
680 }
681 AggregationType::Min => {
682 let min = self.min_field(agg, docs);
683 AggregationResult {
684 name: agg.name.clone(),
685 agg_type: agg.agg_type.clone(),
686 buckets: Vec::new(),
687 value: min,
688 }
689 }
690 AggregationType::Histogram => self.compute_histogram(agg, docs),
691 AggregationType::ValueCount => {
692 let count = self.count_field(agg, docs);
693 AggregationResult {
694 name: agg.name.clone(),
695 agg_type: agg.agg_type.clone(),
696 buckets: Vec::new(),
697 value: Some(count as f64),
698 }
699 }
700 }
701 }
702
703 fn compute_terms(&self, agg: &Aggregation, docs: &[&serde_json::Value]) -> AggregationResult {
705 let mut counts: HashMap<String, u64> = HashMap::new();
706 for doc in docs {
707 if let Some(value) = doc.get(&agg.field) {
708 let key = match value {
709 serde_json::Value::String(s) => s.clone(),
710 serde_json::Value::Number(n) => n.to_string(),
711 serde_json::Value::Bool(b) => b.to_string(),
712 _ => continue,
713 };
714 *counts.entry(key).or_insert(0) += 1;
715 }
716 }
717 let mut buckets: Vec<AggBucket> = counts
718 .into_iter()
719 .map(|(key, doc_count)| AggBucket {
720 key,
721 doc_count,
722 value: None,
723 })
724 .collect();
725 buckets.sort_by_key(|b| std::cmp::Reverse(b.doc_count));
727 buckets.truncate(agg.size);
728 AggregationResult {
729 name: agg.name.clone(),
730 agg_type: agg.agg_type.clone(),
731 buckets,
732 value: None,
733 }
734 }
735
736 fn compute_range(&self, agg: &Aggregation, docs: &[&serde_json::Value]) -> AggregationResult {
738 let mut buckets: Vec<AggBucket> = agg
739 .ranges
740 .iter()
741 .map(|r| AggBucket {
742 key: r.label(),
743 doc_count: 0,
744 value: None,
745 })
746 .collect();
747
748 for doc in docs {
749 if let Some(value) = doc.get(&agg.field).and_then(|v| v.as_f64()) {
750 for (i, range) in agg.ranges.iter().enumerate() {
751 if range.contains(value) {
752 buckets[i].doc_count += 1;
753 }
754 }
755 }
756 }
757 AggregationResult {
758 name: agg.name.clone(),
759 agg_type: agg.agg_type.clone(),
760 buckets,
761 value: None,
762 }
763 }
764
765 fn compute_histogram(
767 &self,
768 agg: &Aggregation,
769 docs: &[&serde_json::Value],
770 ) -> AggregationResult {
771 let interval = agg.interval.unwrap_or(1.0);
772 if interval <= 0.0 {
773 return AggregationResult {
774 name: agg.name.clone(),
775 agg_type: agg.agg_type.clone(),
776 buckets: Vec::new(),
777 value: None,
778 };
779 }
780 let mut buckets_map: HashMap<i64, u64> = HashMap::new();
781 for doc in docs {
782 if let Some(value) = doc.get(&agg.field).and_then(|v| v.as_f64()) {
783 let bucket_key = (value / interval).floor() as i64;
784 *buckets_map.entry(bucket_key).or_insert(0) += 1;
785 }
786 }
787 let mut buckets: Vec<AggBucket> = buckets_map
788 .into_iter()
789 .map(|(key, count)| AggBucket {
790 key: (key as f64 * interval).to_string(),
791 doc_count: count,
792 value: None,
793 })
794 .collect();
795 buckets.sort_by(|a, b| a.key.cmp(&b.key));
797 AggregationResult {
798 name: agg.name.clone(),
799 agg_type: agg.agg_type.clone(),
800 buckets,
801 value: None,
802 }
803 }
804
805 fn sum_field(&self, agg: &Aggregation, docs: &[&serde_json::Value]) -> f64 {
807 docs.iter()
808 .filter_map(|doc| doc.get(&agg.field).and_then(|v| v.as_f64()))
809 .sum()
810 }
811
812 fn avg_field(&self, agg: &Aggregation, docs: &[&serde_json::Value]) -> Option<f64> {
814 let values: Vec<f64> = docs
815 .iter()
816 .filter_map(|doc| doc.get(&agg.field).and_then(|v| v.as_f64()))
817 .collect();
818 if values.is_empty() {
819 None
820 } else {
821 Some(values.iter().sum::<f64>() / values.len() as f64)
822 }
823 }
824
825 fn max_field(&self, agg: &Aggregation, docs: &[&serde_json::Value]) -> Option<f64> {
827 docs.iter()
828 .filter_map(|doc| doc.get(&agg.field).and_then(|v| v.as_f64()))
829 .fold(None, |acc, v| Some(acc.map_or(v, |a: f64| a.max(v))))
830 }
831
832 fn min_field(&self, agg: &Aggregation, docs: &[&serde_json::Value]) -> Option<f64> {
834 docs.iter()
835 .filter_map(|doc| doc.get(&agg.field).and_then(|v| v.as_f64()))
836 .fold(None, |acc, v| Some(acc.map_or(v, |a: f64| a.min(v))))
837 }
838
839 fn count_field(&self, agg: &Aggregation, docs: &[&serde_json::Value]) -> u64 {
841 docs.iter()
842 .filter(|doc| doc.get(&agg.field).is_some())
843 .count() as u64
844 }
845}
846
847#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
853pub struct AliasDefinition {
854 pub name: String,
856 pub index: String,
858 pub filter: Option<crate::EsQuery>,
860 pub routing: Option<String>,
862 pub is_write_index: bool,
864}
865
866impl AliasDefinition {
867 pub fn new(name: impl Into<String>, index: impl Into<String>) -> Self {
868 Self {
869 name: name.into(),
870 index: index.into(),
871 filter: None,
872 routing: None,
873 is_write_index: false,
874 }
875 }
876
877 pub fn with_filter(mut self, filter: crate::EsQuery) -> Self {
878 self.filter = Some(filter);
879 self
880 }
881
882 pub fn with_routing(mut self, routing: impl Into<String>) -> Self {
883 self.routing = Some(routing.into());
884 self
885 }
886
887 pub fn with_write_index(mut self, is_write: bool) -> Self {
888 self.is_write_index = is_write;
889 self
890 }
891}
892
893#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
895pub enum AliasAction {
896 Add(AliasDefinition),
898 Remove { name: String, index: String },
900}
901
902pub struct AliasManager {
906 aliases: RwLock<Vec<AliasDefinition>>,
908}
909
910impl AliasManager {
911 pub fn new() -> Self {
912 Self {
913 aliases: RwLock::new(Vec::new()),
914 }
915 }
916
917 pub fn update_aliases(&self, actions: Vec<AliasAction>) -> Result<(), EsError> {
919 let mut aliases = self
920 .aliases
921 .write()
922 .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
923 for action in actions {
924 match action {
925 AliasAction::Add(def) => {
926 let exists = aliases
928 .iter()
929 .any(|a| a.name == def.name && a.index == def.index);
930 if !exists {
931 aliases.push(def);
932 }
933 }
934 AliasAction::Remove { name, index } => {
935 aliases.retain(|a| !(a.name == name && a.index == index));
936 }
937 }
938 }
939 Ok(())
940 }
941
942 pub fn add_alias(&self, def: AliasDefinition) -> Result<(), EsError> {
944 self.update_aliases(vec![AliasAction::Add(def)])
945 }
946
947 pub fn remove_alias(&self, name: &str, index: &str) -> Result<(), EsError> {
949 self.update_aliases(vec![AliasAction::Remove {
950 name: name.to_string(),
951 index: index.to_string(),
952 }])
953 }
954
955 pub fn resolve_index(&self, alias: &str) -> Result<Vec<String>, EsError> {
957 let aliases = self
958 .aliases
959 .read()
960 .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
961 let indices: Vec<String> = aliases
962 .iter()
963 .filter(|a| a.name == alias)
964 .map(|a| a.index.clone())
965 .collect();
966 Ok(indices)
967 }
968
969 pub fn get_aliases(&self, index: &str) -> Result<Vec<AliasDefinition>, EsError> {
971 let aliases = self
972 .aliases
973 .read()
974 .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
975 Ok(aliases
976 .iter()
977 .filter(|a| a.index == index)
978 .cloned()
979 .collect())
980 }
981
982 pub fn list_aliases(&self) -> Result<Vec<AliasDefinition>, EsError> {
984 let aliases = self
985 .aliases
986 .read()
987 .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
988 Ok(aliases.clone())
989 }
990
991 pub fn swap_alias(&self, alias: &str, old_index: &str, new_index: &str) -> Result<(), EsError> {
995 let actions = vec![
996 AliasAction::Add(AliasDefinition::new(alias, new_index)),
997 AliasAction::Remove {
998 name: alias.to_string(),
999 index: old_index.to_string(),
1000 },
1001 ];
1002 self.update_aliases(actions)
1003 }
1004
1005 pub fn get_write_index(&self, alias: &str) -> Result<Option<String>, EsError> {
1007 let aliases = self
1008 .aliases
1009 .read()
1010 .map_err(|e| EsError::SyncError(format!("lock error: {}", e)))?;
1011 Ok(aliases
1012 .iter()
1013 .find(|a| a.name == alias && a.is_write_index)
1014 .map(|a| a.index.clone()))
1015 }
1016}
1017
1018impl Default for AliasManager {
1019 fn default() -> Self {
1020 Self::new()
1021 }
1022}
1023
1024#[cfg(test)]
1025mod tests {
1026 use super::*;
1027 use crate::EsSync;
1028 use serde_json::json;
1029
1030 #[test]
1033 fn test_bulk_action_index() {
1034 let action = BulkAction::Index {
1035 index: "test".to_string(),
1036 id: "1".to_string(),
1037 source: json!({"name": "test"}),
1038 };
1039 assert_eq!(action.index(), "test");
1040 assert_eq!(action.id(), "1");
1041 }
1042
1043 #[test]
1044 fn test_bulk_action_create() {
1045 let action = BulkAction::Create {
1046 index: "test".to_string(),
1047 id: "1".to_string(),
1048 source: json!({}),
1049 };
1050 assert_eq!(action.index(), "test");
1051 }
1052
1053 #[test]
1054 fn test_bulk_action_update() {
1055 let action = BulkAction::Update {
1056 index: "test".to_string(),
1057 id: "1".to_string(),
1058 doc: json!({"field": "value"}),
1059 };
1060 assert_eq!(action.id(), "1");
1061 }
1062
1063 #[test]
1064 fn test_bulk_action_delete() {
1065 let action = BulkAction::Delete {
1066 index: "test".to_string(),
1067 id: "1".to_string(),
1068 };
1069 assert_eq!(action.index(), "test");
1070 }
1071
1072 #[test]
1075 fn test_bulk_item_result_success() {
1076 let result = BulkItemResult::success("index", "test", "1", 200);
1077 assert!(result.is_success());
1078 assert!(result.error.is_none());
1079 }
1080
1081 #[test]
1082 fn test_bulk_item_result_failure() {
1083 let result = BulkItemResult::failure("update", "test", "1", 404, "not found");
1084 assert!(!result.is_success());
1085 assert_eq!(result.error, Some("not found".to_string()));
1086 }
1087
1088 #[test]
1089 fn test_bulk_result_new() {
1090 let items = vec![
1091 BulkItemResult::success("index", "test", "1", 200),
1092 BulkItemResult::success("index", "test", "2", 201),
1093 BulkItemResult::failure("delete", "test", "3", 404, "not found"),
1094 ];
1095 let result = BulkResult::new(items, 10);
1096 assert_eq!(result.success_count, 2);
1097 assert_eq!(result.failure_count, 1);
1098 assert!(!result.all_success());
1099 assert_eq!(result.failures().len(), 1);
1100 }
1101
1102 #[test]
1103 fn test_bulk_result_all_success() {
1104 let items = vec![
1105 BulkItemResult::success("index", "test", "1", 200),
1106 BulkItemResult::success("index", "test", "2", 201),
1107 ];
1108 let result = BulkResult::new(items, 5);
1109 assert!(result.all_success());
1110 assert_eq!(result.failures().len(), 0);
1111 }
1112
1113 #[test]
1116 fn test_bulk_executor_index_actions() {
1117 let executor = BulkExecutor::from_new();
1118 let actions = vec![
1119 BulkAction::Index {
1120 index: "docs".to_string(),
1121 id: "1".to_string(),
1122 source: json!({"title": "hello"}),
1123 },
1124 BulkAction::Index {
1125 index: "docs".to_string(),
1126 id: "2".to_string(),
1127 source: json!({"title": "world"}),
1128 },
1129 ];
1130 let result = executor.bulk(actions).unwrap();
1131 assert_eq!(result.success_count, 2);
1132 assert_eq!(result.failure_count, 0);
1133 assert_eq!(executor.backend().count("docs").unwrap(), 2);
1134 }
1135
1136 #[test]
1137 fn test_bulk_executor_create_action() {
1138 let executor = BulkExecutor::from_new();
1139 let action = BulkAction::Create {
1140 index: "docs".to_string(),
1141 id: "1".to_string(),
1142 source: json!({"name": "test"}),
1143 };
1144 let result = executor.bulk(vec![action]).unwrap();
1145 assert_eq!(result.success_count, 1);
1146 assert_eq!(result.items[0].status, 201);
1147 }
1148
1149 #[test]
1150 fn test_bulk_executor_delete_action() {
1151 let executor = BulkExecutor::from_new();
1152 executor
1153 .bulk(vec![BulkAction::Index {
1154 index: "docs".to_string(),
1155 id: "1".to_string(),
1156 source: json!({"v": 1}),
1157 }])
1158 .unwrap();
1159 let result = executor
1160 .bulk(vec![BulkAction::Delete {
1161 index: "docs".to_string(),
1162 id: "1".to_string(),
1163 }])
1164 .unwrap();
1165 assert_eq!(result.success_count, 1);
1166 assert_eq!(executor.backend().count("docs").unwrap(), 0);
1167 }
1168
1169 #[test]
1170 fn test_bulk_executor_update_action() {
1171 let executor = BulkExecutor::from_new();
1172 executor
1174 .bulk(vec![BulkAction::Index {
1175 index: "docs".to_string(),
1176 id: "1".to_string(),
1177 source: json!({"name": "old", "count": 10}),
1178 }])
1179 .unwrap();
1180 let result = executor
1182 .bulk(vec![BulkAction::Update {
1183 index: "docs".to_string(),
1184 id: "1".to_string(),
1185 doc: json!({"count": 20}),
1186 }])
1187 .unwrap();
1188 assert_eq!(result.success_count, 1);
1189 let search_req = EsSearchRequest::new("docs", crate::EsQuery::match_all());
1191 let search_result = executor.backend().search(search_req).unwrap();
1192 assert_eq!(search_result.hits[0].source["name"], "old");
1193 assert_eq!(search_result.hits[0].source["count"], 20);
1194 }
1195
1196 #[test]
1197 fn test_bulk_executor_update_missing_doc() {
1198 let executor = BulkExecutor::from_new();
1199 let result = executor
1200 .bulk(vec![BulkAction::Update {
1201 index: "docs".to_string(),
1202 id: "999".to_string(),
1203 doc: json!({"v": 1}),
1204 }])
1205 .unwrap();
1206 assert_eq!(result.failure_count, 1);
1207 assert_eq!(result.items[0].status, 404);
1208 }
1209
1210 #[test]
1211 fn test_bulk_executor_mixed_actions() {
1212 let executor = BulkExecutor::from_new();
1213 let actions = vec![
1214 BulkAction::Index {
1215 index: "docs".to_string(),
1216 id: "1".to_string(),
1217 source: json!({"v": 1}),
1218 },
1219 BulkAction::Index {
1220 index: "docs".to_string(),
1221 id: "2".to_string(),
1222 source: json!({"v": 2}),
1223 },
1224 BulkAction::Delete {
1225 index: "docs".to_string(),
1226 id: "1".to_string(),
1227 },
1228 BulkAction::Update {
1229 index: "docs".to_string(),
1230 id: "999".to_string(),
1231 doc: json!({}),
1232 },
1233 ];
1234 let result = executor.bulk(actions).unwrap();
1235 assert_eq!(result.success_count, 3);
1236 assert_eq!(result.failure_count, 1);
1237 assert_eq!(executor.backend().count("docs").unwrap(), 1);
1238 }
1239
1240 #[test]
1243 fn test_suggest_request_new() {
1244 let req = SuggestRequest::new("my_suggest", "hel");
1245 assert_eq!(req.name, "my_suggest");
1246 assert_eq!(req.prefix, "hel");
1247 assert_eq!(req.suggester_type, SuggesterType::Term);
1248 assert_eq!(req.field, "text");
1249 assert_eq!(req.size, 5);
1250 }
1251
1252 #[test]
1253 fn test_suggest_request_builder() {
1254 let req = SuggestRequest::new("s", "ru")
1255 .with_field("title")
1256 .with_size(3)
1257 .with_type(SuggesterType::Completion);
1258 assert_eq!(req.field, "title");
1259 assert_eq!(req.size, 3);
1260 assert_eq!(req.suggester_type, SuggesterType::Completion);
1261 }
1262
1263 #[test]
1264 fn test_memory_suggester_basic() {
1265 let suggester = MemorySuggester::from_new();
1266 suggester
1267 .index_doc("docs", "1", json!({"title": "hello world help"}))
1268 .unwrap();
1269 suggester
1270 .index_doc("docs", "2", json!({"title": "hello rust"}))
1271 .unwrap();
1272 suggester
1273 .index_doc("docs", "3", json!({"title": "help care"}))
1274 .unwrap();
1275
1276 let req = SuggestRequest::new("s", "hel").with_field("title");
1277 let result = suggester.suggest("docs", &req).unwrap();
1278 assert_eq!(result.name, "s");
1279 assert!(
1282 result.options.iter().any(|o| o.text == "hello"),
1283 "expected hello in options: {:?}",
1284 result.options
1285 );
1286 assert!(
1287 result.options.iter().any(|o| o.text == "help"),
1288 "expected help in options: {:?}",
1289 result.options
1290 );
1291 }
1292
1293 #[test]
1294 fn test_memory_suggester_size_limit() {
1295 let suggester = MemorySuggester::from_new();
1296 suggester
1297 .index_doc("docs", "1", json!({"title": "apple apply apollo"}))
1298 .unwrap();
1299 let req = SuggestRequest::new("s", "ap")
1300 .with_field("title")
1301 .with_size(2);
1302 let result = suggester.suggest("docs", &req).unwrap();
1303 assert!(result.options.len() <= 2);
1304 }
1305
1306 #[test]
1307 fn test_memory_suggester_no_match() {
1308 let suggester = MemorySuggester::from_new();
1309 suggester
1310 .index_doc("docs", "1", json!({"title": "hello world"}))
1311 .unwrap();
1312 let req = SuggestRequest::new("s", "xyz").with_field("title");
1313 let result = suggester.suggest("docs", &req).unwrap();
1314 assert!(result.options.is_empty());
1315 }
1316
1317 #[test]
1318 fn test_memory_suggester_freq_count() {
1319 let suggester = MemorySuggester::from_new();
1320 suggester
1321 .index_doc("docs", "1", json!({"title": "rust rust rust"}))
1322 .unwrap();
1323 suggester
1324 .index_doc("docs", "2", json!({"title": "ruby"}))
1325 .unwrap();
1326 let req = SuggestRequest::new("s", "ru").with_field("title");
1327 let result = suggester.suggest("docs", &req).unwrap();
1328 let rust = result.options.iter().find(|o| o.text == "rust").unwrap();
1329 assert_eq!(rust.freq, 3);
1330 let ruby = result.options.iter().find(|o| o.text == "ruby").unwrap();
1331 assert_eq!(ruby.freq, 1);
1332 }
1333
1334 #[test]
1335 fn test_memory_suggester_case_insensitive() {
1336 let suggester = MemorySuggester::from_new();
1337 suggester
1338 .index_doc("docs", "1", json!({"title": "Hello HELLO hello"}))
1339 .unwrap();
1340 let req = SuggestRequest::new("s", "HEL").with_field("title");
1341 let result = suggester.suggest("docs", &req).unwrap();
1342 assert!(result.options.iter().any(|o| o.text == "hello"));
1344 }
1345
1346 #[test]
1349 fn test_aggregation_terms() {
1350 let agg = Aggregation::terms("by_category", "category").with_size(5);
1351 assert_eq!(agg.name, "by_category");
1352 assert_eq!(agg.agg_type, AggregationType::Terms);
1353 assert_eq!(agg.field, "category");
1354 assert_eq!(agg.size, 5);
1355 }
1356
1357 #[test]
1358 fn test_aggregation_range() {
1359 let agg = Aggregation::range("price_ranges", "price")
1360 .with_range(Some(0.0), Some(100.0))
1361 .with_range(Some(100.0), Some(500.0))
1362 .with_range(Some(500.0), None);
1363 assert_eq!(agg.agg_type, AggregationType::Range);
1364 assert_eq!(agg.ranges.len(), 3);
1365 }
1366
1367 #[test]
1368 fn test_aggregation_sum() {
1369 let agg = Aggregation::sum("total_price", "price");
1370 assert_eq!(agg.agg_type, AggregationType::Sum);
1371 }
1372
1373 #[test]
1374 fn test_aggregation_avg() {
1375 let agg = Aggregation::avg("avg_price", "price");
1376 assert_eq!(agg.agg_type, AggregationType::Avg);
1377 }
1378
1379 #[test]
1380 fn test_aggregation_histogram() {
1381 let agg = Aggregation::histogram("price_hist", "price", 10.0);
1382 assert_eq!(agg.agg_type, AggregationType::Histogram);
1383 assert_eq!(agg.interval, Some(10.0));
1384 }
1385
1386 #[test]
1387 fn test_aggregation_count() {
1388 let agg = Aggregation::count("doc_count", "title");
1389 assert_eq!(agg.agg_type, AggregationType::ValueCount);
1390 }
1391
1392 #[test]
1395 fn test_agg_range_contains() {
1396 let range = AggRange::new(Some(10.0), Some(20.0));
1397 assert!(!range.contains(5.0));
1398 assert!(range.contains(10.0));
1399 assert!(range.contains(15.0));
1400 assert!(!range.contains(20.0));
1401 assert!(!range.contains(25.0));
1402 }
1403
1404 #[test]
1405 fn test_agg_range_open_ended() {
1406 let range = AggRange::new(None, Some(10.0));
1407 assert!(range.contains(-100.0));
1408 assert!(range.contains(0.0));
1409 assert!(!range.contains(10.0));
1410
1411 let range = AggRange::new(Some(10.0), None);
1412 assert!(!range.contains(5.0));
1413 assert!(range.contains(10.0));
1414 assert!(range.contains(100.0));
1415 }
1416
1417 #[test]
1418 fn test_agg_range_label() {
1419 assert_eq!(AggRange::new(Some(10.0), Some(20.0)).label(), "10-20");
1420 assert_eq!(AggRange::new(Some(10.0), None).label(), "10-*");
1421 assert_eq!(AggRange::new(None, Some(20.0)).label(), "*-20");
1422 assert_eq!(AggRange::new(None, None).label(), "*-*");
1423 }
1424
1425 #[test]
1428 fn test_memory_aggregator_terms() {
1429 let _agg = MemoryAggregator::from_new();
1430 let backend = InMemoryEsSync::new();
1431 let docs = vec![
1432 EsDocument::new("docs", json!({"category": "tech", "price": 100})).with_id("1"),
1433 EsDocument::new("docs", json!({"category": "tech", "price": 200})).with_id("2"),
1434 EsDocument::new("docs", json!({"category": "food", "price": 50})).with_id("3"),
1435 ];
1436 backend.sync_to_es(docs).unwrap();
1437 let aggregator = MemoryAggregator::new(backend);
1438 let aggs = vec![Aggregation::terms("by_cat", "category")];
1439 let results = aggregator
1440 .aggregate("docs", crate::EsQuery::match_all(), &aggs)
1441 .unwrap();
1442 assert_eq!(results.len(), 1);
1443 assert_eq!(results[0].buckets.len(), 2);
1444 let tech = results[0].buckets.iter().find(|b| b.key == "tech").unwrap();
1445 assert_eq!(tech.doc_count, 2);
1446 }
1447
1448 #[test]
1449 fn test_memory_aggregator_sum() {
1450 let backend = InMemoryEsSync::new();
1451 backend
1452 .sync_to_es(vec![
1453 EsDocument::new("docs", json!({"price": 100})).with_id("1"),
1454 EsDocument::new("docs", json!({"price": 200})).with_id("2"),
1455 EsDocument::new("docs", json!({"price": 300})).with_id("3"),
1456 ])
1457 .unwrap();
1458 let aggregator = MemoryAggregator::new(backend);
1459 let aggs = vec![Aggregation::sum("total", "price")];
1460 let results = aggregator
1461 .aggregate("docs", crate::EsQuery::match_all(), &aggs)
1462 .unwrap();
1463 assert_eq!(results[0].value, Some(600.0));
1464 }
1465
1466 #[test]
1467 fn test_memory_aggregator_avg() {
1468 let backend = InMemoryEsSync::new();
1469 backend
1470 .sync_to_es(vec![
1471 EsDocument::new("docs", json!({"price": 100})).with_id("1"),
1472 EsDocument::new("docs", json!({"price": 200})).with_id("2"),
1473 ])
1474 .unwrap();
1475 let aggregator = MemoryAggregator::new(backend);
1476 let results = aggregator
1477 .aggregate(
1478 "docs",
1479 crate::EsQuery::match_all(),
1480 &[Aggregation::avg("avg", "price")],
1481 )
1482 .unwrap();
1483 assert_eq!(results[0].value, Some(150.0));
1484 }
1485
1486 #[test]
1487 fn test_memory_aggregator_max_min() {
1488 let backend = InMemoryEsSync::new();
1489 backend
1490 .sync_to_es(vec![
1491 EsDocument::new("docs", json!({"price": 100})).with_id("1"),
1492 EsDocument::new("docs", json!({"price": 500})).with_id("2"),
1493 EsDocument::new("docs", json!({"price": 50})).with_id("3"),
1494 ])
1495 .unwrap();
1496 let aggregator = MemoryAggregator::new(backend);
1497 let results = aggregator
1498 .aggregate(
1499 "docs",
1500 crate::EsQuery::match_all(),
1501 &[
1502 Aggregation::max("max_p", "price"),
1503 Aggregation::min("min_p", "price"),
1504 ],
1505 )
1506 .unwrap();
1507 assert_eq!(results[0].value, Some(500.0));
1508 assert_eq!(results[1].value, Some(50.0));
1509 }
1510
1511 #[test]
1512 fn test_memory_aggregator_range() {
1513 let backend = InMemoryEsSync::new();
1514 backend
1515 .sync_to_es(vec![
1516 EsDocument::new("docs", json!({"price": 50})).with_id("1"),
1517 EsDocument::new("docs", json!({"price": 150})).with_id("2"),
1518 EsDocument::new("docs", json!({"price": 600})).with_id("3"),
1519 ])
1520 .unwrap();
1521 let aggregator = MemoryAggregator::new(backend);
1522 let agg = Aggregation::range("price_ranges", "price")
1523 .with_range(Some(0.0), Some(100.0))
1524 .with_range(Some(100.0), Some(500.0))
1525 .with_range(Some(500.0), None);
1526 let results = aggregator
1527 .aggregate("docs", crate::EsQuery::match_all(), &[agg])
1528 .unwrap();
1529 assert_eq!(results[0].buckets.len(), 3);
1530 assert_eq!(results[0].buckets[0].doc_count, 1); assert_eq!(results[0].buckets[1].doc_count, 1); assert_eq!(results[0].buckets[2].doc_count, 1); }
1534
1535 #[test]
1536 fn test_memory_aggregator_histogram() {
1537 let backend = InMemoryEsSync::new();
1538 backend
1539 .sync_to_es(vec![
1540 EsDocument::new("docs", json!({"price": 5})).with_id("1"),
1541 EsDocument::new("docs", json!({"price": 15})).with_id("2"),
1542 EsDocument::new("docs", json!({"price": 25})).with_id("3"),
1543 EsDocument::new("docs", json!({"price": 35})).with_id("4"),
1544 ])
1545 .unwrap();
1546 let aggregator = MemoryAggregator::new(backend);
1547 let agg = Aggregation::histogram("hist", "price", 10.0);
1548 let results = aggregator
1549 .aggregate("docs", crate::EsQuery::match_all(), &[agg])
1550 .unwrap();
1551 assert_eq!(results[0].buckets.len(), 4);
1553 }
1554
1555 #[test]
1556 fn test_memory_aggregator_value_count() {
1557 let backend = InMemoryEsSync::new();
1558 backend
1559 .sync_to_es(vec![
1560 EsDocument::new("docs", json!({"name": "a", "price": 100})).with_id("1"),
1561 EsDocument::new("docs", json!({"name": "b", "price": 200})).with_id("2"),
1562 EsDocument::new("docs", json!({"name": "c"})).with_id("3"),
1563 ])
1564 .unwrap();
1565 let aggregator = MemoryAggregator::new(backend);
1566 let results = aggregator
1567 .aggregate(
1568 "docs",
1569 crate::EsQuery::match_all(),
1570 &[Aggregation::count("cnt", "price")],
1571 )
1572 .unwrap();
1573 assert_eq!(results[0].value, Some(2.0)); }
1575
1576 #[test]
1577 fn test_memory_aggregator_avg_empty() {
1578 let backend = InMemoryEsSync::new();
1579 backend
1580 .sync_to_es(vec![
1581 EsDocument::new("docs", json!({"name": "a"})).with_id("1")
1582 ])
1583 .unwrap();
1584 let aggregator = MemoryAggregator::new(backend);
1585 let results = aggregator
1586 .aggregate(
1587 "docs",
1588 crate::EsQuery::match_all(),
1589 &[Aggregation::avg("avg", "price")],
1590 )
1591 .unwrap();
1592 assert_eq!(results[0].value, None); }
1594
1595 #[test]
1598 fn test_alias_definition_new() {
1599 let def = AliasDefinition::new("alias1", "index1");
1600 assert_eq!(def.name, "alias1");
1601 assert_eq!(def.index, "index1");
1602 assert!(def.filter.is_none());
1603 assert!(def.routing.is_none());
1604 assert!(!def.is_write_index);
1605 }
1606
1607 #[test]
1608 fn test_alias_definition_builder() {
1609 let def = AliasDefinition::new("alias1", "index1")
1610 .with_routing("routing_key")
1611 .with_write_index(true);
1612 assert_eq!(def.routing, Some("routing_key".to_string()));
1613 assert!(def.is_write_index);
1614 }
1615
1616 #[test]
1617 fn test_alias_definition_with_filter() {
1618 let def = AliasDefinition::new("alias1", "index1")
1619 .with_filter(crate::EsQuery::term("status", json!("active")));
1620 assert!(def.filter.is_some());
1621 }
1622
1623 #[test]
1626 fn test_alias_manager_add_and_resolve() {
1627 let manager = AliasManager::new();
1628 manager
1629 .add_alias(AliasDefinition::new("alias1", "index1"))
1630 .unwrap();
1631 let indices = manager.resolve_index("alias1").unwrap();
1632 assert_eq!(indices, vec!["index1"]);
1633 }
1634
1635 #[test]
1636 fn test_alias_manager_remove() {
1637 let manager = AliasManager::new();
1638 manager
1639 .add_alias(AliasDefinition::new("alias1", "index1"))
1640 .unwrap();
1641 manager.remove_alias("alias1", "index1").unwrap();
1642 let indices = manager.resolve_index("alias1").unwrap();
1643 assert!(indices.is_empty());
1644 }
1645
1646 #[test]
1647 fn test_alias_manager_multiple_indices() {
1648 let manager = AliasManager::new();
1649 manager
1650 .add_alias(AliasDefinition::new("alias1", "index1"))
1651 .unwrap();
1652 manager
1653 .add_alias(AliasDefinition::new("alias1", "index2"))
1654 .unwrap();
1655 let indices = manager.resolve_index("alias1").unwrap();
1656 assert_eq!(indices.len(), 2);
1657 assert!(indices.contains(&"index1".to_string()));
1658 assert!(indices.contains(&"index2".to_string()));
1659 }
1660
1661 #[test]
1662 fn test_alias_manager_get_aliases() {
1663 let manager = AliasManager::new();
1664 manager
1665 .add_alias(AliasDefinition::new("alias1", "index1"))
1666 .unwrap();
1667 manager
1668 .add_alias(AliasDefinition::new("alias2", "index1"))
1669 .unwrap();
1670 let aliases = manager.get_aliases("index1").unwrap();
1671 assert_eq!(aliases.len(), 2);
1672 }
1673
1674 #[test]
1675 fn test_alias_manager_list_all() {
1676 let manager = AliasManager::new();
1677 manager
1678 .add_alias(AliasDefinition::new("alias1", "index1"))
1679 .unwrap();
1680 manager
1681 .add_alias(AliasDefinition::new("alias2", "index2"))
1682 .unwrap();
1683 let all = manager.list_aliases().unwrap();
1684 assert_eq!(all.len(), 2);
1685 }
1686
1687 #[test]
1688 fn test_alias_manager_swap() {
1689 let manager = AliasManager::new();
1690 manager
1691 .add_alias(AliasDefinition::new("alias1", "old_index"))
1692 .unwrap();
1693 manager
1694 .swap_alias("alias1", "old_index", "new_index")
1695 .unwrap();
1696 let indices = manager.resolve_index("alias1").unwrap();
1697 assert_eq!(indices, vec!["new_index"]);
1698 }
1699
1700 #[test]
1701 fn test_alias_manager_write_index() {
1702 let manager = AliasManager::new();
1703 manager
1704 .add_alias(AliasDefinition::new("alias1", "index1").with_write_index(true))
1705 .unwrap();
1706 manager
1707 .add_alias(AliasDefinition::new("alias1", "index2"))
1708 .unwrap();
1709 let write_index = manager.get_write_index("alias1").unwrap();
1710 assert_eq!(write_index, Some("index1".to_string()));
1711 }
1712
1713 #[test]
1714 fn test_alias_manager_no_write_index() {
1715 let manager = AliasManager::new();
1716 manager
1717 .add_alias(AliasDefinition::new("alias1", "index1"))
1718 .unwrap();
1719 let write_index = manager.get_write_index("alias1").unwrap();
1720 assert_eq!(write_index, None);
1721 }
1722
1723 #[test]
1724 fn test_alias_manager_duplicate_add() {
1725 let manager = AliasManager::new();
1726 manager
1727 .add_alias(AliasDefinition::new("alias1", "index1"))
1728 .unwrap();
1729 manager
1731 .add_alias(AliasDefinition::new("alias1", "index1"))
1732 .unwrap();
1733 let all = manager.list_aliases().unwrap();
1734 assert_eq!(all.len(), 1);
1735 }
1736
1737 #[test]
1738 fn test_alias_manager_batch_update() {
1739 let manager = AliasManager::new();
1740 let actions = vec![
1741 AliasAction::Add(AliasDefinition::new("alias1", "index1")),
1742 AliasAction::Add(AliasDefinition::new("alias2", "index2")),
1743 AliasAction::Remove {
1744 name: "alias1".to_string(),
1745 index: "index1".to_string(),
1746 },
1747 ];
1748 manager.update_aliases(actions).unwrap();
1749 let all = manager.list_aliases().unwrap();
1750 assert_eq!(all.len(), 1);
1751 assert_eq!(all[0].name, "alias2");
1752 }
1753}