@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
rs 3,345 lines 98.6 KB
Raw
1 //! Integration tests for multi-file pagination with PaginationState.
2
3 use journal_common::Seconds;
4 use journal_core::file::{JournalFile, JournalFileOptions, JournalWriter};
5 use journal_core::repository::File;
6 use journal_engine::logs::query::LogQuery;
7 use journal_index::{
8 Anchor, Direction, FieldName, FieldValuePair, FileIndexer, Filter, Microseconds,
9 };
10 use std::collections::HashSet;
11 use std::fs;
12 use std::path::PathBuf;
13 use tempfile::TempDir;
14 use uuid::Uuid;
15
16 /// Test journal entry specification
17 struct TestEntry {
18 timestamp: Microseconds,
19 fields: Vec<(String, String)>,
20 }
21
22 impl TestEntry {
23 fn new(timestamp: Microseconds) -> Self {
24 Self {
25 timestamp,
26 fields: Vec::new(),
27 }
28 }
29
30 fn with_field(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
31 self.fields.push((name.into(), value.into()));
32 self
33 }
34 }
35
36 /// Create a test journal file path with a specific name
37 fn create_test_journal_path(temp_dir: &TempDir, filename: &str) -> PathBuf {
38 let machine_id = Uuid::from_u128(0x12345678_1234_1234_1234_123456789abc);
39 let machine_dir = temp_dir.path().join(machine_id.to_string());
40 fs::create_dir_all(&machine_dir).expect("create machine dir");
41 machine_dir.join(filename)
42 }
43
44 /// Helper to create a test journal file with specified entries
45 fn create_test_journal(
46 temp_dir: &TempDir,
47 filename: &str,
48 entries: Vec<TestEntry>,
49 ) -> Result<File, Box<dyn std::error::Error>> {
50 let journal_path = create_test_journal_path(temp_dir, filename);
51
52 let file =
53 File::from_path(&journal_path).ok_or("Failed to create repository File from path")?;
54
55 let machine_id = Uuid::from_u128(0x12345678_1234_1234_1234_123456789abc);
56 let boot_id = Uuid::from_u128(0x11111111_1111_1111_1111_111111111111);
57 let seqnum_id = Uuid::from_u128(0x22222222_2222_2222_2222_222222222222);
58
59 let options = JournalFileOptions::new(machine_id, boot_id, seqnum_id);
60
61 let mut journal_file = JournalFile::create(&file, options)?;
62 let mut writer = JournalWriter::new(&mut journal_file, 1, boot_id)?;
63
64 for entry in entries {
65 let mut entry_data = Vec::new();
66
67 // Add _SOURCE_REALTIME_TIMESTAMP first
68 entry_data.push(format!("_SOURCE_REALTIME_TIMESTAMP={}", entry.timestamp.0).into_bytes());
69
70 // Add all other fields
71 for (field, value) in entry.fields {
72 entry_data.push(format!("{}={}", field, value).into_bytes());
73 }
74
75 let entry_refs: Vec<&[u8]> = entry_data.iter().map(|v| v.as_slice()).collect();
76
77 writer.add_entry(
78 &mut journal_file,
79 &entry_refs,
80 entry.timestamp.0,
81 entry.timestamp.0,
82 )?;
83 }
84
85 Ok(file)
86 }
87
88 #[test]
89 fn test_multi_file_pagination_forward_non_overlapping() {
90 // Create temporary directory
91 let temp_dir = TempDir::new().unwrap();
92
93 // File 1: entries at t=100..200 microseconds (100 entries)
94 let entries_file1: Vec<TestEntry> = (100..200)
95 .map(|i| {
96 TestEntry::new(Microseconds(i))
97 .with_field("MESSAGE", format!("File1 Entry {}", i))
98 .with_field("FILE", "1")
99 })
100 .collect();
101
102 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
103
104 // File 2: entries at t=200..300 microseconds (100 entries)
105 let entries_file2: Vec<TestEntry> = (200..300)
106 .map(|i| {
107 TestEntry::new(Microseconds(i))
108 .with_field("MESSAGE", format!("File2 Entry {}", i))
109 .with_field("FILE", "2")
110 })
111 .collect();
112
113 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
114
115 // Index both files
116 let mut indexer = FileIndexer::default();
117 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
118 let file_field = FieldName::new("FILE").unwrap();
119
120 let index1 = indexer
121 .index(
122 &file1,
123 Some(&source_timestamp_field),
124 &[file_field.clone()],
125 Seconds(3600),
126 )
127 .unwrap();
128
129 let index2 = indexer
130 .index(
131 &file2,
132 Some(&source_timestamp_field),
133 &[file_field],
134 Seconds(3600),
135 )
136 .unwrap();
137
138 let file_indexes = vec![index1, index2];
139
140 // First page: limit=150, should get all 100 from file1 + 50 from file2
141 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
142 .with_limit(150)
143 .execute_page(None)
144 .unwrap();
145
146 assert_eq!(
147 first_page.len(),
148 150,
149 "First page should contain exactly 150 entries"
150 );
151
152 // Verify timestamps are in ascending order
153 for i in 1..first_page.len() {
154 assert!(
155 first_page[i - 1].timestamp <= first_page[i].timestamp,
156 "Entries should be in ascending timestamp order"
157 );
158 }
159
160 // First entry should be at timestamp 100
161 assert_eq!(
162 first_page.first().unwrap().timestamp,
163 100,
164 "First entry should be at timestamp 100"
165 );
166
167 // Last entry should be at timestamp 249 (100-199 from file1, then 200-249 from file2)
168 assert_eq!(
169 first_page.last().unwrap().timestamp,
170 249,
171 "Last entry of first page should be at timestamp 249"
172 );
173
174 // State should track positions for files we read from
175 assert!(
176 !state1.file_positions.is_empty(),
177 "State should track positions"
178 );
179
180 // Second page: use state to get remaining 50 entries
181 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
182 .with_limit(150)
183 .execute_page(Some(&state1))
184 .unwrap();
185
186 assert_eq!(
187 second_page.len(),
188 50,
189 "Second page should contain remaining 50 entries"
190 );
191
192 // Verify timestamps continue from where first page left off
193 assert_eq!(
194 second_page.first().unwrap().timestamp,
195 250,
196 "Second page should start at timestamp 250"
197 );
198
199 assert_eq!(
200 second_page.last().unwrap().timestamp,
201 299,
202 "Second page should end at timestamp 299"
203 );
204
205 // Verify no duplicates across both pages
206 let mut all_timestamps = HashSet::new();
207 for entry in &first_page {
208 assert!(
209 all_timestamps.insert(entry.timestamp),
210 "Found duplicate timestamp: {}",
211 entry.timestamp
212 );
213 }
214 for entry in &second_page {
215 assert!(
216 all_timestamps.insert(entry.timestamp),
217 "Found duplicate timestamp: {}",
218 entry.timestamp
219 );
220 }
221
222 // Verify we got all 200 unique entries
223 assert_eq!(
224 all_timestamps.len(),
225 200,
226 "Should have retrieved all 200 unique entries"
227 );
228
229 // Third page should be empty
230 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
231 .with_limit(150)
232 .execute_page(Some(&state2))
233 .unwrap();
234
235 assert_eq!(
236 third_page.len(),
237 0,
238 "Third page should be empty (no more entries)"
239 );
240 }
241
242 #[test]
243 fn test_multi_file_pagination_same_timestamps() {
244 // Create temporary directory
245 let temp_dir = TempDir::new().unwrap();
246
247 // File 1: 150 entries all at timestamp 1000
248 let entries_file1: Vec<TestEntry> = (0..150)
249 .map(|i| {
250 TestEntry::new(Microseconds(1000))
251 .with_field("MESSAGE", format!("File1 Entry {}", i))
252 .with_field("ENTRY_ID", format!("file1_{}", i))
253 .with_field("FILE", "1")
254 })
255 .collect();
256
257 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
258
259 // File 2: 150 entries all at timestamp 1000
260 let entries_file2: Vec<TestEntry> = (0..150)
261 .map(|i| {
262 TestEntry::new(Microseconds(1000))
263 .with_field("MESSAGE", format!("File2 Entry {}", i))
264 .with_field("ENTRY_ID", format!("file2_{}", i))
265 .with_field("FILE", "2")
266 })
267 .collect();
268
269 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
270
271 // Index both files
272 let mut indexer = FileIndexer::default();
273 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
274 let file_field = FieldName::new("FILE").unwrap();
275 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
276
277 let index1 = indexer
278 .index(
279 &file1,
280 Some(&source_timestamp_field),
281 &[file_field.clone(), entry_id_field.clone()],
282 Seconds(3600),
283 )
284 .unwrap();
285
286 let index2 = indexer
287 .index(
288 &file2,
289 Some(&source_timestamp_field),
290 &[file_field, entry_id_field],
291 Seconds(3600),
292 )
293 .unwrap();
294
295 let file_indexes = vec![index1, index2];
296
297 // First page: limit=200, should get all 150 from file1 + 50 from file2
298 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
299 .with_limit(200)
300 .execute_page(None)
301 .unwrap();
302
303 assert_eq!(
304 first_page.len(),
305 200,
306 "First page should contain exactly 200 entries"
307 );
308
309 // All timestamps should be 1000
310 for entry in &first_page {
311 assert_eq!(
312 entry.timestamp, 1000,
313 "All entries should have timestamp 1000"
314 );
315 }
316
317 // State should track positions for both files
318 assert_eq!(
319 state1.file_positions.len(),
320 2,
321 "State should track positions for both files"
322 );
323
324 // Second page: use state to get remaining 100 entries
325 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
326 .with_limit(200)
327 .execute_page(Some(&state1))
328 .unwrap();
329
330 assert_eq!(
331 second_page.len(),
332 100,
333 "Second page should contain remaining 100 entries"
334 );
335
336 // All timestamps should still be 1000
337 for entry in &second_page {
338 assert_eq!(
339 entry.timestamp, 1000,
340 "All entries should have timestamp 1000"
341 );
342 }
343
344 // Collect all ENTRY_ID values to verify uniqueness
345 let mut all_entry_ids = HashSet::new();
346
347 for entry in &first_page {
348 for field in &entry.fields {
349 if field.field() == "ENTRY_ID" {
350 assert!(
351 all_entry_ids.insert(field.value().to_string()),
352 "Found duplicate ENTRY_ID: {}",
353 field.value()
354 );
355 }
356 }
357 }
358
359 for entry in &second_page {
360 for field in &entry.fields {
361 if field.field() == "ENTRY_ID" {
362 assert!(
363 all_entry_ids.insert(field.value().to_string()),
364 "Found duplicate ENTRY_ID: {}",
365 field.value()
366 );
367 }
368 }
369 }
370
371 // Verify we got all 300 unique entries
372 assert_eq!(
373 all_entry_ids.len(),
374 300,
375 "Should have retrieved all 300 unique entries"
376 );
377
378 // Verify we have entries from both files
379 let file1_entries: usize = all_entry_ids
380 .iter()
381 .filter(|id| id.starts_with("file1_"))
382 .count();
383 let file2_entries: usize = all_entry_ids
384 .iter()
385 .filter(|id| id.starts_with("file2_"))
386 .count();
387
388 assert_eq!(file1_entries, 150, "Should have 150 entries from file1");
389 assert_eq!(file2_entries, 150, "Should have 150 entries from file2");
390
391 // Third page should be empty
392 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
393 .with_limit(200)
394 .execute_page(Some(&state2))
395 .unwrap();
396
397 assert_eq!(
398 third_page.len(),
399 0,
400 "Third page should be empty (no more entries)"
401 );
402 }
403
404 #[test]
405 fn test_multi_file_pagination_overlapping_timestamps() {
406 // Create temporary directory
407 let temp_dir = TempDir::new().unwrap();
408
409 // File 1: entries at t=100..200 (100 entries)
410 let entries_file1: Vec<TestEntry> = (100..200)
411 .map(|i| {
412 TestEntry::new(Microseconds(i))
413 .with_field("MESSAGE", format!("File1 Entry at {}", i))
414 .with_field("ENTRY_ID", format!("file1_{}", i))
415 .with_field("FILE", "1")
416 })
417 .collect();
418
419 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
420
421 // File 2: entries at t=150..250 (100 entries) - overlaps with file1 from 150-199
422 let entries_file2: Vec<TestEntry> = (150..250)
423 .map(|i| {
424 TestEntry::new(Microseconds(i))
425 .with_field("MESSAGE", format!("File2 Entry at {}", i))
426 .with_field("ENTRY_ID", format!("file2_{}", i))
427 .with_field("FILE", "2")
428 })
429 .collect();
430
431 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
432
433 // Index both files
434 let mut indexer = FileIndexer::default();
435 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
436 let file_field = FieldName::new("FILE").unwrap();
437 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
438
439 let index1 = indexer
440 .index(
441 &file1,
442 Some(&source_timestamp_field),
443 &[file_field.clone(), entry_id_field.clone()],
444 Seconds(3600),
445 )
446 .unwrap();
447
448 let index2 = indexer
449 .index(
450 &file2,
451 Some(&source_timestamp_field),
452 &[file_field, entry_id_field],
453 Seconds(3600),
454 )
455 .unwrap();
456
457 let file_indexes = vec![index1, index2];
458
459 // First page: limit=120
460 // Expected: 50 from file1 (100-149) + 50 interleaved from both (150-199) + 20 from file2 (200-219)
461 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
462 .with_limit(120)
463 .execute_page(None)
464 .unwrap();
465
466 assert_eq!(
467 first_page.len(),
468 120,
469 "First page should contain exactly 120 entries"
470 );
471
472 // Verify timestamps are in ascending order
473 for i in 1..first_page.len() {
474 assert!(
475 first_page[i - 1].timestamp <= first_page[i].timestamp,
476 "Entries should be in ascending timestamp order"
477 );
478 }
479
480 // First entry should be at timestamp 100
481 assert_eq!(
482 first_page.first().unwrap().timestamp,
483 100,
484 "First entry should be at timestamp 100"
485 );
486
487 // State should track positions for both files
488 assert!(
489 !state1.file_positions.is_empty(),
490 "State should track positions"
491 );
492
493 // Second page: get remaining entries
494 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
495 .with_limit(120)
496 .execute_page(Some(&state1))
497 .unwrap();
498
499 assert_eq!(
500 second_page.len(),
501 80,
502 "Second page should contain remaining 80 entries"
503 );
504
505 // Verify timestamps continue in order
506 for i in 1..second_page.len() {
507 assert!(
508 second_page[i - 1].timestamp <= second_page[i].timestamp,
509 "Entries should be in ascending timestamp order"
510 );
511 }
512
513 // Verify no timestamp gap between pages
514 if !first_page.is_empty() && !second_page.is_empty() {
515 assert!(
516 first_page.last().unwrap().timestamp <= second_page.first().unwrap().timestamp,
517 "Second page should continue from first page timestamp"
518 );
519 }
520
521 // Last entry should be at timestamp 249
522 assert_eq!(
523 second_page.last().unwrap().timestamp,
524 249,
525 "Last entry should be at timestamp 249"
526 );
527
528 // Collect all ENTRY_ID values to verify uniqueness and completeness
529 let mut all_entry_ids = HashSet::new();
530
531 for entry in &first_page {
532 for field in &entry.fields {
533 if field.field() == "ENTRY_ID" {
534 assert!(
535 all_entry_ids.insert(field.value().to_string()),
536 "Found duplicate ENTRY_ID: {}",
537 field.value()
538 );
539 }
540 }
541 }
542
543 for entry in &second_page {
544 for field in &entry.fields {
545 if field.field() == "ENTRY_ID" {
546 assert!(
547 all_entry_ids.insert(field.value().to_string()),
548 "Found duplicate ENTRY_ID: {}",
549 field.value()
550 );
551 }
552 }
553 }
554
555 // Verify we got all 200 unique entries (100 from each file)
556 assert_eq!(
557 all_entry_ids.len(),
558 200,
559 "Should have retrieved all 200 unique entries"
560 );
561
562 // Verify we have entries from both files
563 let file1_entries: usize = all_entry_ids
564 .iter()
565 .filter(|id| id.starts_with("file1_"))
566 .count();
567 let file2_entries: usize = all_entry_ids
568 .iter()
569 .filter(|id| id.starts_with("file2_"))
570 .count();
571
572 assert_eq!(file1_entries, 100, "Should have 100 entries from file1");
573 assert_eq!(file2_entries, 100, "Should have 100 entries from file2");
574
575 // Verify all timestamps from 100-249 are represented
576 let mut all_timestamps = HashSet::new();
577 for entry in first_page.iter().chain(second_page.iter()) {
578 all_timestamps.insert(entry.timestamp);
579 }
580
581 // We should have entries at all timestamps from 100-249 (150 unique timestamps)
582 for ts in 100..250 {
583 assert!(all_timestamps.contains(&ts), "Missing timestamp: {}", ts);
584 }
585
586 // Third page should be empty
587 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
588 .with_limit(120)
589 .execute_page(Some(&state2))
590 .unwrap();
591
592 assert_eq!(
593 third_page.len(),
594 0,
595 "Third page should be empty (no more entries)"
596 );
597 }
598
599 #[test]
600 fn test_multi_file_pagination_three_files() {
601 // Create temporary directory
602 let temp_dir = TempDir::new().unwrap();
603
604 // File 1: entries at t=100..200 (100 entries)
605 let entries_file1: Vec<TestEntry> = (100..200)
606 .map(|i| {
607 TestEntry::new(Microseconds(i))
608 .with_field("MESSAGE", format!("File1 Entry {}", i))
609 .with_field("ENTRY_ID", format!("file1_{}", i))
610 .with_field("FILE", "1")
611 })
612 .collect();
613
614 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
615
616 // File 2: entries at t=200..300 (100 entries)
617 let entries_file2: Vec<TestEntry> = (200..300)
618 .map(|i| {
619 TestEntry::new(Microseconds(i))
620 .with_field("MESSAGE", format!("File2 Entry {}", i))
621 .with_field("ENTRY_ID", format!("file2_{}", i))
622 .with_field("FILE", "2")
623 })
624 .collect();
625
626 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
627
628 // File 3: entries at t=300..400 (100 entries)
629 let entries_file3: Vec<TestEntry> = (300..400)
630 .map(|i| {
631 TestEntry::new(Microseconds(i))
632 .with_field("MESSAGE", format!("File3 Entry {}", i))
633 .with_field("ENTRY_ID", format!("file3_{}", i))
634 .with_field("FILE", "3")
635 })
636 .collect();
637
638 let file3 = create_test_journal(&temp_dir, "file3.journal", entries_file3).unwrap();
639
640 // Index all three files
641 let mut indexer = FileIndexer::default();
642 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
643 let file_field = FieldName::new("FILE").unwrap();
644 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
645
646 let index1 = indexer
647 .index(
648 &file1,
649 Some(&source_timestamp_field),
650 &[file_field.clone(), entry_id_field.clone()],
651 Seconds(3600),
652 )
653 .unwrap();
654
655 let index2 = indexer
656 .index(
657 &file2,
658 Some(&source_timestamp_field),
659 &[file_field.clone(), entry_id_field.clone()],
660 Seconds(3600),
661 )
662 .unwrap();
663
664 let index3 = indexer
665 .index(
666 &file3,
667 Some(&source_timestamp_field),
668 &[file_field, entry_id_field],
669 Seconds(3600),
670 )
671 .unwrap();
672
673 let file_indexes = vec![index1, index2, index3];
674
675 // First page: limit=125, should get all 100 from file1 + 25 from file2
676 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
677 .with_limit(125)
678 .execute_page(None)
679 .unwrap();
680
681 assert_eq!(
682 first_page.len(),
683 125,
684 "First page should contain exactly 125 entries"
685 );
686
687 assert_eq!(first_page.first().unwrap().timestamp, 100);
688 assert_eq!(first_page.last().unwrap().timestamp, 224);
689
690 // State should track positions for file1 and file2
691 assert_eq!(
692 state1.file_positions.len(),
693 2,
694 "State should track positions for 2 files"
695 );
696
697 // Second page: limit=125, should get 75 from file2 + 50 from file3
698 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
699 .with_limit(125)
700 .execute_page(Some(&state1))
701 .unwrap();
702
703 assert_eq!(
704 second_page.len(),
705 125,
706 "Second page should contain exactly 125 entries"
707 );
708
709 assert_eq!(second_page.first().unwrap().timestamp, 225);
710 assert_eq!(second_page.last().unwrap().timestamp, 349);
711
712 // State should now track all 3 files
713 assert_eq!(
714 state2.file_positions.len(),
715 3,
716 "State should track positions for all 3 files"
717 );
718
719 // Third page: remaining 50 from file3
720 let (third_page, state3) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
721 .with_limit(125)
722 .execute_page(Some(&state2))
723 .unwrap();
724
725 assert_eq!(
726 third_page.len(),
727 50,
728 "Third page should contain remaining 50 entries"
729 );
730
731 assert_eq!(third_page.first().unwrap().timestamp, 350);
732 assert_eq!(third_page.last().unwrap().timestamp, 399);
733
734 // Collect all ENTRY_ID values to verify uniqueness
735 let mut all_entry_ids = HashSet::new();
736
737 for entry in first_page
738 .iter()
739 .chain(second_page.iter())
740 .chain(third_page.iter())
741 {
742 for field in &entry.fields {
743 if field.field() == "ENTRY_ID" {
744 assert!(
745 all_entry_ids.insert(field.value().to_string()),
746 "Found duplicate ENTRY_ID: {}",
747 field.value()
748 );
749 }
750 }
751 }
752
753 // Verify we got all 300 unique entries
754 assert_eq!(
755 all_entry_ids.len(),
756 300,
757 "Should have retrieved all 300 unique entries"
758 );
759
760 // Verify distribution: 100 from each file
761 let file1_count = all_entry_ids
762 .iter()
763 .filter(|id| id.starts_with("file1_"))
764 .count();
765 let file2_count = all_entry_ids
766 .iter()
767 .filter(|id| id.starts_with("file2_"))
768 .count();
769 let file3_count = all_entry_ids
770 .iter()
771 .filter(|id| id.starts_with("file3_"))
772 .count();
773
774 assert_eq!(file1_count, 100, "Should have 100 entries from file1");
775 assert_eq!(file2_count, 100, "Should have 100 entries from file2");
776 assert_eq!(file3_count, 100, "Should have 100 entries from file3");
777
778 // Fourth page should be empty
779 let (fourth_page, _state4) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
780 .with_limit(125)
781 .execute_page(Some(&state3))
782 .unwrap();
783
784 assert_eq!(
785 fourth_page.len(),
786 0,
787 "Fourth page should be empty (no more entries)"
788 );
789 }
790
791 #[test]
792 fn test_multi_file_pagination_small_limit() {
793 // Create temporary directory
794 let temp_dir = TempDir::new().unwrap();
795
796 // File 1: 100 entries at t=100..200
797 let entries_file1: Vec<TestEntry> = (100..200)
798 .map(|i| {
799 TestEntry::new(Microseconds(i))
800 .with_field("MESSAGE", format!("File1 Entry {}", i))
801 .with_field("ENTRY_ID", format!("file1_{}", i))
802 })
803 .collect();
804
805 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
806
807 // File 2: 100 entries at t=200..300
808 let entries_file2: Vec<TestEntry> = (200..300)
809 .map(|i| {
810 TestEntry::new(Microseconds(i))
811 .with_field("MESSAGE", format!("File2 Entry {}", i))
812 .with_field("ENTRY_ID", format!("file2_{}", i))
813 })
814 .collect();
815
816 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
817
818 // Index both files
819 let mut indexer = FileIndexer::default();
820 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
821 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
822
823 let index1 = indexer
824 .index(
825 &file1,
826 Some(&source_timestamp_field),
827 &[entry_id_field.clone()],
828 Seconds(3600),
829 )
830 .unwrap();
831
832 let index2 = indexer
833 .index(
834 &file2,
835 Some(&source_timestamp_field),
836 &[entry_id_field],
837 Seconds(3600),
838 )
839 .unwrap();
840
841 let file_indexes = vec![index1, index2];
842
843 // Use very small limit=30, need multiple pages for file1 alone
844 let mut all_entry_ids = HashSet::new();
845 let mut state = None;
846 let mut page_count = 0;
847
848 // Paginate through all entries with small page size
849 loop {
850 let (page, new_state) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
851 .with_limit(30)
852 .execute_page(state.as_ref())
853 .unwrap();
854
855 if page.is_empty() {
856 break;
857 }
858
859 page_count += 1;
860
861 // Verify order within page
862 for i in 1..page.len() {
863 assert!(
864 page[i - 1].timestamp <= page[i].timestamp,
865 "Page {} entries should be in ascending order",
866 page_count
867 );
868 }
869
870 // Collect ENTRY_IDs
871 for entry in &page {
872 for field in &entry.fields {
873 if field.field() == "ENTRY_ID" {
874 assert!(
875 all_entry_ids.insert(field.value().to_string()),
876 "Found duplicate ENTRY_ID: {}",
877 field.value()
878 );
879 }
880 }
881 }
882
883 state = Some(new_state);
884 }
885
886 // Should need 7 pages: 30+30+30+30+30+30+20 = 200 entries
887 assert_eq!(page_count, 7, "Should need exactly 7 pages");
888
889 // Verify we got all 200 unique entries
890 assert_eq!(
891 all_entry_ids.len(),
892 200,
893 "Should have retrieved all 200 unique entries"
894 );
895
896 // Verify distribution
897 let file1_count = all_entry_ids
898 .iter()
899 .filter(|id| id.starts_with("file1_"))
900 .count();
901 let file2_count = all_entry_ids
902 .iter()
903 .filter(|id| id.starts_with("file2_"))
904 .count();
905
906 assert_eq!(file1_count, 100, "Should have 100 entries from file1");
907 assert_eq!(file2_count, 100, "Should have 100 entries from file2");
908 }
909
910 #[test]
911 fn test_multi_file_pagination_limit_one() {
912 // Create temporary directory
913 let temp_dir = TempDir::new().unwrap();
914
915 // File 1: 10 entries at t=100..110
916 let entries_file1: Vec<TestEntry> = (100..110)
917 .map(|i| {
918 TestEntry::new(Microseconds(i))
919 .with_field("MESSAGE", format!("File1 Entry {}", i))
920 .with_field("ENTRY_ID", format!("file1_{}", i))
921 })
922 .collect();
923
924 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
925
926 // File 2: 10 entries at t=110..120
927 let entries_file2: Vec<TestEntry> = (110..120)
928 .map(|i| {
929 TestEntry::new(Microseconds(i))
930 .with_field("MESSAGE", format!("File2 Entry {}", i))
931 .with_field("ENTRY_ID", format!("file2_{}", i))
932 })
933 .collect();
934
935 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
936
937 // Index both files
938 let mut indexer = FileIndexer::default();
939 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
940 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
941
942 let index1 = indexer
943 .index(
944 &file1,
945 Some(&source_timestamp_field),
946 &[entry_id_field.clone()],
947 Seconds(3600),
948 )
949 .unwrap();
950
951 let index2 = indexer
952 .index(
953 &file2,
954 Some(&source_timestamp_field),
955 &[entry_id_field],
956 Seconds(3600),
957 )
958 .unwrap();
959
960 let file_indexes = vec![index1, index2];
961
962 // Paginate with limit=1 (extreme case)
963 let mut all_entry_ids = HashSet::new();
964 let mut all_timestamps = Vec::new();
965 let mut state = None;
966 let mut page_count = 0;
967
968 loop {
969 let (page, new_state) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
970 .with_limit(1)
971 .execute_page(state.as_ref())
972 .unwrap();
973
974 if page.is_empty() {
975 break;
976 }
977
978 page_count += 1;
979
980 // Each page should have exactly 1 entry
981 assert_eq!(page.len(), 1, "Each page should have exactly 1 entry");
982
983 // Collect ENTRY_ID and timestamp
984 for entry in &page {
985 all_timestamps.push(entry.timestamp);
986 for field in &entry.fields {
987 if field.field() == "ENTRY_ID" {
988 assert!(
989 all_entry_ids.insert(field.value().to_string()),
990 "Found duplicate ENTRY_ID: {}",
991 field.value()
992 );
993 }
994 }
995 }
996
997 state = Some(new_state);
998 }
999
1000 // Should need 20 pages for 20 entries
1001 assert_eq!(page_count, 20, "Should need exactly 20 pages");
1002
1003 // Verify we got all 20 unique entries
1004 assert_eq!(
1005 all_entry_ids.len(),
1006 20,
1007 "Should have retrieved all 20 unique entries"
1008 );
1009
1010 // Verify timestamps are in ascending order
1011 for i in 1..all_timestamps.len() {
1012 assert!(
1013 all_timestamps[i - 1] <= all_timestamps[i],
1014 "Timestamps should be in ascending order"
1015 );
1016 }
1017
1018 // Verify we got all timestamps from 100-119
1019 let unique_timestamps: HashSet<_> = all_timestamps.into_iter().collect();
1020 assert_eq!(
1021 unique_timestamps.len(),
1022 20,
1023 "Should have 20 unique timestamps"
1024 );
1025 for ts in 100..120 {
1026 assert!(unique_timestamps.contains(&ts), "Missing timestamp: {}", ts);
1027 }
1028 }
1029
1030 #[test]
1031 fn test_multi_file_pagination_with_empty_file() {
1032 // Create temporary directory
1033 let temp_dir = TempDir::new().unwrap();
1034
1035 // File 1: 50 entries at t=100..150
1036 let entries_file1: Vec<TestEntry> = (100..150)
1037 .map(|i| {
1038 TestEntry::new(Microseconds(i))
1039 .with_field("MESSAGE", format!("File1 Entry {}", i))
1040 .with_field("ENTRY_ID", format!("file1_{}", i))
1041 })
1042 .collect();
1043
1044 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
1045
1046 // File 2: Empty (0 entries) - created but not indexed since empty files cannot be indexed
1047 let entries_file2: Vec<TestEntry> = vec![];
1048 let _file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
1049
1050 // File 3: 50 entries at t=150..200
1051 let entries_file3: Vec<TestEntry> = (150..200)
1052 .map(|i| {
1053 TestEntry::new(Microseconds(i))
1054 .with_field("MESSAGE", format!("File3 Entry {}", i))
1055 .with_field("ENTRY_ID", format!("file3_{}", i))
1056 })
1057 .collect();
1058
1059 let file3 = create_test_journal(&temp_dir, "file3.journal", entries_file3).unwrap();
1060
1061 // Index files - note that empty file cannot be indexed (returns EmptyHistogramInput error)
1062 // In practice, the system would skip files with no entries
1063 let mut indexer = FileIndexer::default();
1064 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
1065 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
1066
1067 let index1 = indexer
1068 .index(
1069 &file1,
1070 Some(&source_timestamp_field),
1071 &[entry_id_field.clone()],
1072 Seconds(3600),
1073 )
1074 .unwrap();
1075
1076 // Skip empty file - it cannot be indexed
1077 // let index2 = indexer.index(&file2, ...) would fail with EmptyHistogramInput
1078
1079 let index3 = indexer
1080 .index(
1081 &file3,
1082 Some(&source_timestamp_field),
1083 &[entry_id_field],
1084 Seconds(3600),
1085 )
1086 .unwrap();
1087
1088 let file_indexes = vec![index1, index3];
1089
1090 // First page: limit=60, should get 50 from file1 + 10 from file3 (skipping empty file2)
1091 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
1092 .with_limit(60)
1093 .execute_page(None)
1094 .unwrap();
1095
1096 assert_eq!(first_page.len(), 60, "First page should contain 60 entries");
1097
1098 assert_eq!(first_page.first().unwrap().timestamp, 100);
1099 assert_eq!(first_page.last().unwrap().timestamp, 159);
1100
1101 // Second page: remaining 40 from file3
1102 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
1103 .with_limit(60)
1104 .execute_page(Some(&state1))
1105 .unwrap();
1106
1107 assert_eq!(
1108 second_page.len(),
1109 40,
1110 "Second page should contain remaining 40 entries"
1111 );
1112
1113 assert_eq!(second_page.first().unwrap().timestamp, 160);
1114 assert_eq!(second_page.last().unwrap().timestamp, 199);
1115
1116 // Collect all ENTRY_IDs
1117 let mut all_entry_ids = HashSet::new();
1118 for entry in first_page.iter().chain(second_page.iter()) {
1119 for field in &entry.fields {
1120 if field.field() == "ENTRY_ID" {
1121 assert!(
1122 all_entry_ids.insert(field.value().to_string()),
1123 "Found duplicate ENTRY_ID: {}",
1124 field.value()
1125 );
1126 }
1127 }
1128 }
1129
1130 // Verify we got all 100 unique entries (none from empty file)
1131 assert_eq!(
1132 all_entry_ids.len(),
1133 100,
1134 "Should have retrieved all 100 unique entries"
1135 );
1136
1137 let file1_count = all_entry_ids
1138 .iter()
1139 .filter(|id| id.starts_with("file1_"))
1140 .count();
1141 let file3_count = all_entry_ids
1142 .iter()
1143 .filter(|id| id.starts_with("file3_"))
1144 .count();
1145
1146 assert_eq!(file1_count, 50, "Should have 50 entries from file1");
1147 assert_eq!(file3_count, 50, "Should have 50 entries from file3");
1148
1149 // Third page should be empty
1150 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
1151 .with_limit(60)
1152 .execute_page(Some(&state2))
1153 .unwrap();
1154
1155 assert_eq!(
1156 third_page.len(),
1157 0,
1158 "Third page should be empty (no more entries)"
1159 );
1160 }
1161
1162 #[test]
1163 fn test_multi_file_pagination_reverse_file_order() {
1164 // Create temporary directory
1165 let temp_dir = TempDir::new().unwrap();
1166
1167 // File 1: entries at t=100..200 (oldest)
1168 let entries_file1: Vec<TestEntry> = (100..200)
1169 .map(|i| {
1170 TestEntry::new(Microseconds(i))
1171 .with_field("MESSAGE", format!("File1 Entry {}", i))
1172 .with_field("ENTRY_ID", format!("file1_{}", i))
1173 })
1174 .collect();
1175
1176 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
1177
1178 // File 2: entries at t=200..300 (middle)
1179 let entries_file2: Vec<TestEntry> = (200..300)
1180 .map(|i| {
1181 TestEntry::new(Microseconds(i))
1182 .with_field("MESSAGE", format!("File2 Entry {}", i))
1183 .with_field("ENTRY_ID", format!("file2_{}", i))
1184 })
1185 .collect();
1186
1187 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
1188
1189 // File 3: entries at t=300..400 (newest)
1190 let entries_file3: Vec<TestEntry> = (300..400)
1191 .map(|i| {
1192 TestEntry::new(Microseconds(i))
1193 .with_field("MESSAGE", format!("File3 Entry {}", i))
1194 .with_field("ENTRY_ID", format!("file3_{}", i))
1195 })
1196 .collect();
1197
1198 let file3 = create_test_journal(&temp_dir, "file3.journal", entries_file3).unwrap();
1199
1200 // Index all three files
1201 let mut indexer = FileIndexer::default();
1202 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
1203 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
1204
1205 let index1 = indexer
1206 .index(
1207 &file1,
1208 Some(&source_timestamp_field),
1209 &[entry_id_field.clone()],
1210 Seconds(3600),
1211 )
1212 .unwrap();
1213
1214 let index2 = indexer
1215 .index(
1216 &file2,
1217 Some(&source_timestamp_field),
1218 &[entry_id_field.clone()],
1219 Seconds(3600),
1220 )
1221 .unwrap();
1222
1223 let index3 = indexer
1224 .index(
1225 &file3,
1226 Some(&source_timestamp_field),
1227 &[entry_id_field],
1228 Seconds(3600),
1229 )
1230 .unwrap();
1231
1232 // Pass files in REVERSE chronological order (newest first)
1233 // The query should still process them in correct temporal order
1234 let file_indexes = vec![index3, index2, index1];
1235
1236 // Query should still return entries in ascending timestamp order
1237 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
1238 .with_limit(150)
1239 .execute_page(None)
1240 .unwrap();
1241
1242 assert_eq!(first_page.len(), 150);
1243
1244 // First entry should be from file1 (oldest timestamp)
1245 assert_eq!(
1246 first_page.first().unwrap().timestamp,
1247 100,
1248 "First entry should be at timestamp 100 from file1"
1249 );
1250
1251 // Verify ascending order
1252 for i in 1..first_page.len() {
1253 assert!(
1254 first_page[i - 1].timestamp <= first_page[i].timestamp,
1255 "Entries should be in ascending order"
1256 );
1257 }
1258
1259 // Continue pagination
1260 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
1261 .with_limit(150)
1262 .execute_page(Some(&state1))
1263 .unwrap();
1264
1265 assert_eq!(second_page.len(), 150);
1266
1267 for i in 1..second_page.len() {
1268 assert!(
1269 second_page[i - 1].timestamp <= second_page[i].timestamp,
1270 "Entries should be in ascending order"
1271 );
1272 }
1273
1274 // Verify continuity between pages
1275 assert!(
1276 first_page.last().unwrap().timestamp <= second_page.first().unwrap().timestamp,
1277 "Second page should continue from first page"
1278 );
1279
1280 // Collect all ENTRY_IDs
1281 let mut all_entry_ids = HashSet::new();
1282 for entry in first_page.iter().chain(second_page.iter()) {
1283 for field in &entry.fields {
1284 if field.field() == "ENTRY_ID" {
1285 assert!(
1286 all_entry_ids.insert(field.value().to_string()),
1287 "Found duplicate ENTRY_ID: {}",
1288 field.value()
1289 );
1290 }
1291 }
1292 }
1293
1294 // Should have 300 unique entries
1295 assert_eq!(
1296 all_entry_ids.len(),
1297 300,
1298 "Should have retrieved all 300 unique entries"
1299 );
1300
1301 // Third page should be empty
1302 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
1303 .with_limit(150)
1304 .execute_page(Some(&state2))
1305 .unwrap();
1306
1307 assert_eq!(third_page.len(), 0, "Third page should be empty");
1308 }
1309
1310 #[test]
1311 fn test_multi_file_pagination_backward_non_overlapping() {
1312 // Create temporary directory
1313 let temp_dir = TempDir::new().unwrap();
1314
1315 // File 1: entries at t=100..200 (100 entries)
1316 let entries_file1: Vec<TestEntry> = (100..200)
1317 .map(|i| {
1318 TestEntry::new(Microseconds(i))
1319 .with_field("MESSAGE", format!("File1 Entry {}", i))
1320 .with_field("ENTRY_ID", format!("file1_{}", i))
1321 .with_field("FILE", "1")
1322 })
1323 .collect();
1324
1325 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
1326
1327 // File 2: entries at t=200..300 (100 entries)
1328 let entries_file2: Vec<TestEntry> = (200..300)
1329 .map(|i| {
1330 TestEntry::new(Microseconds(i))
1331 .with_field("MESSAGE", format!("File2 Entry {}", i))
1332 .with_field("ENTRY_ID", format!("file2_{}", i))
1333 .with_field("FILE", "2")
1334 })
1335 .collect();
1336
1337 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
1338
1339 // Index both files
1340 let mut indexer = FileIndexer::default();
1341 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
1342 let file_field = FieldName::new("FILE").unwrap();
1343
1344 let index1 = indexer
1345 .index(
1346 &file1,
1347 Some(&source_timestamp_field),
1348 &[file_field.clone()],
1349 Seconds(3600),
1350 )
1351 .unwrap();
1352
1353 let index2 = indexer
1354 .index(
1355 &file2,
1356 Some(&source_timestamp_field),
1357 &[file_field],
1358 Seconds(3600),
1359 )
1360 .unwrap();
1361
1362 let file_indexes = vec![index1, index2];
1363
1364 // First page: limit=150, backward from tail, should get all 100 from file2 + 50 from file1
1365 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
1366 .with_limit(150)
1367 .execute_page(None)
1368 .unwrap();
1369
1370 assert_eq!(
1371 first_page.len(),
1372 150,
1373 "First page should contain exactly 150 entries"
1374 );
1375
1376 // Verify timestamps are in descending order
1377 for i in 1..first_page.len() {
1378 assert!(
1379 first_page[i - 1].timestamp >= first_page[i].timestamp,
1380 "Entries should be in descending timestamp order"
1381 );
1382 }
1383
1384 // First entry should be at timestamp 299 (highest)
1385 assert_eq!(
1386 first_page.first().unwrap().timestamp,
1387 299,
1388 "First entry should be at timestamp 299"
1389 );
1390
1391 // Last entry should be at timestamp 150 (100 from file2: 299-200, then 50 from file1: 199-150)
1392 assert_eq!(
1393 first_page.last().unwrap().timestamp,
1394 150,
1395 "Last entry of first page should be at timestamp 150"
1396 );
1397
1398 // State should track positions for files we read from
1399 assert!(
1400 !state1.file_positions.is_empty(),
1401 "State should track positions"
1402 );
1403
1404 // Second page: use state to get remaining 50 entries
1405 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
1406 .with_limit(150)
1407 .execute_page(Some(&state1))
1408 .unwrap();
1409
1410 assert_eq!(
1411 second_page.len(),
1412 50,
1413 "Second page should contain remaining 50 entries"
1414 );
1415
1416 // Verify timestamps continue in descending order
1417 assert_eq!(
1418 second_page.first().unwrap().timestamp,
1419 149,
1420 "Second page should start at timestamp 149"
1421 );
1422
1423 assert_eq!(
1424 second_page.last().unwrap().timestamp,
1425 100,
1426 "Second page should end at timestamp 100"
1427 );
1428
1429 // Verify no duplicates across both pages
1430 let mut all_timestamps = HashSet::new();
1431 for entry in &first_page {
1432 assert!(
1433 all_timestamps.insert(entry.timestamp),
1434 "Found duplicate timestamp: {}",
1435 entry.timestamp
1436 );
1437 }
1438 for entry in &second_page {
1439 assert!(
1440 all_timestamps.insert(entry.timestamp),
1441 "Found duplicate timestamp: {}",
1442 entry.timestamp
1443 );
1444 }
1445
1446 // Verify we got all 200 unique entries
1447 assert_eq!(
1448 all_timestamps.len(),
1449 200,
1450 "Should have retrieved all 200 unique entries"
1451 );
1452
1453 // Third page should be empty
1454 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
1455 .with_limit(150)
1456 .execute_page(Some(&state2))
1457 .unwrap();
1458
1459 assert_eq!(
1460 third_page.len(),
1461 0,
1462 "Third page should be empty (no more entries)"
1463 );
1464 }
1465
1466 #[test]
1467 fn test_multi_file_pagination_backward_same_timestamps() {
1468 // Create temporary directory
1469 let temp_dir = TempDir::new().unwrap();
1470
1471 // File 1: 150 entries all at timestamp 1000
1472 let entries_file1: Vec<TestEntry> = (0..150)
1473 .map(|i| {
1474 TestEntry::new(Microseconds(1000))
1475 .with_field("MESSAGE", format!("File1 Entry {}", i))
1476 .with_field("ENTRY_ID", format!("file1_{}", i))
1477 .with_field("FILE", "1")
1478 })
1479 .collect();
1480
1481 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
1482
1483 // File 2: 150 entries all at timestamp 1000
1484 let entries_file2: Vec<TestEntry> = (0..150)
1485 .map(|i| {
1486 TestEntry::new(Microseconds(1000))
1487 .with_field("MESSAGE", format!("File2 Entry {}", i))
1488 .with_field("ENTRY_ID", format!("file2_{}", i))
1489 .with_field("FILE", "2")
1490 })
1491 .collect();
1492
1493 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
1494
1495 // Index both files
1496 let mut indexer = FileIndexer::default();
1497 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
1498 let file_field = FieldName::new("FILE").unwrap();
1499 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
1500
1501 let index1 = indexer
1502 .index(
1503 &file1,
1504 Some(&source_timestamp_field),
1505 &[file_field.clone(), entry_id_field.clone()],
1506 Seconds(3600),
1507 )
1508 .unwrap();
1509
1510 let index2 = indexer
1511 .index(
1512 &file2,
1513 Some(&source_timestamp_field),
1514 &[file_field, entry_id_field],
1515 Seconds(3600),
1516 )
1517 .unwrap();
1518
1519 let file_indexes = vec![index1, index2];
1520
1521 // First page: limit=200, backward from tail
1522 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
1523 .with_limit(200)
1524 .execute_page(None)
1525 .unwrap();
1526
1527 assert_eq!(
1528 first_page.len(),
1529 200,
1530 "First page should contain exactly 200 entries"
1531 );
1532
1533 // All timestamps should be 1000
1534 for entry in &first_page {
1535 assert_eq!(
1536 entry.timestamp, 1000,
1537 "All entries should have timestamp 1000"
1538 );
1539 }
1540
1541 // State should track positions for both files
1542 assert_eq!(
1543 state1.file_positions.len(),
1544 2,
1545 "State should track positions for both files"
1546 );
1547
1548 // Second page: use state to get remaining 100 entries
1549 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
1550 .with_limit(200)
1551 .execute_page(Some(&state1))
1552 .unwrap();
1553
1554 assert_eq!(
1555 second_page.len(),
1556 100,
1557 "Second page should contain remaining 100 entries"
1558 );
1559
1560 // All timestamps should still be 1000
1561 for entry in &second_page {
1562 assert_eq!(
1563 entry.timestamp, 1000,
1564 "All entries should have timestamp 1000"
1565 );
1566 }
1567
1568 // Collect all ENTRY_ID values to verify uniqueness
1569 let mut all_entry_ids = HashSet::new();
1570
1571 for entry in &first_page {
1572 for field in &entry.fields {
1573 if field.field() == "ENTRY_ID" {
1574 assert!(
1575 all_entry_ids.insert(field.value().to_string()),
1576 "Found duplicate ENTRY_ID: {}",
1577 field.value()
1578 );
1579 }
1580 }
1581 }
1582
1583 for entry in &second_page {
1584 for field in &entry.fields {
1585 if field.field() == "ENTRY_ID" {
1586 assert!(
1587 all_entry_ids.insert(field.value().to_string()),
1588 "Found duplicate ENTRY_ID: {}",
1589 field.value()
1590 );
1591 }
1592 }
1593 }
1594
1595 // Verify we got all 300 unique entries
1596 assert_eq!(
1597 all_entry_ids.len(),
1598 300,
1599 "Should have retrieved all 300 unique entries"
1600 );
1601
1602 // Verify we have entries from both files
1603 let file1_entries: usize = all_entry_ids
1604 .iter()
1605 .filter(|id| id.starts_with("file1_"))
1606 .count();
1607 let file2_entries: usize = all_entry_ids
1608 .iter()
1609 .filter(|id| id.starts_with("file2_"))
1610 .count();
1611
1612 assert_eq!(file1_entries, 150, "Should have 150 entries from file1");
1613 assert_eq!(file2_entries, 150, "Should have 150 entries from file2");
1614
1615 // Third page should be empty
1616 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
1617 .with_limit(200)
1618 .execute_page(Some(&state2))
1619 .unwrap();
1620
1621 assert_eq!(
1622 third_page.len(),
1623 0,
1624 "Third page should be empty (no more entries)"
1625 );
1626 }
1627
1628 #[test]
1629 fn test_multi_file_pagination_backward_limit_one() {
1630 // Create temporary directory
1631 let temp_dir = TempDir::new().unwrap();
1632
1633 // File 1: 10 entries at t=100..110
1634 let entries_file1: Vec<TestEntry> = (100..110)
1635 .map(|i| {
1636 TestEntry::new(Microseconds(i))
1637 .with_field("MESSAGE", format!("File1 Entry {}", i))
1638 .with_field("ENTRY_ID", format!("file1_{}", i))
1639 })
1640 .collect();
1641
1642 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
1643
1644 // File 2: 10 entries at t=110..120
1645 let entries_file2: Vec<TestEntry> = (110..120)
1646 .map(|i| {
1647 TestEntry::new(Microseconds(i))
1648 .with_field("MESSAGE", format!("File2 Entry {}", i))
1649 .with_field("ENTRY_ID", format!("file2_{}", i))
1650 })
1651 .collect();
1652
1653 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
1654
1655 // Index both files
1656 let mut indexer = FileIndexer::default();
1657 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
1658 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
1659
1660 let index1 = indexer
1661 .index(
1662 &file1,
1663 Some(&source_timestamp_field),
1664 &[entry_id_field.clone()],
1665 Seconds(3600),
1666 )
1667 .unwrap();
1668
1669 let index2 = indexer
1670 .index(
1671 &file2,
1672 Some(&source_timestamp_field),
1673 &[entry_id_field],
1674 Seconds(3600),
1675 )
1676 .unwrap();
1677
1678 let file_indexes = vec![index1, index2];
1679
1680 // Paginate backward with limit=1 (extreme case)
1681 let mut all_entry_ids = HashSet::new();
1682 let mut all_timestamps = Vec::new();
1683 let mut state = None;
1684 let mut page_count = 0;
1685
1686 loop {
1687 let (page, new_state) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
1688 .with_limit(1)
1689 .execute_page(state.as_ref())
1690 .unwrap();
1691
1692 if page.is_empty() {
1693 break;
1694 }
1695
1696 page_count += 1;
1697
1698 // Each page should have exactly 1 entry
1699 assert_eq!(page.len(), 1, "Each page should have exactly 1 entry");
1700
1701 // Collect ENTRY_ID and timestamp
1702 for entry in &page {
1703 all_timestamps.push(entry.timestamp);
1704 for field in &entry.fields {
1705 if field.field() == "ENTRY_ID" {
1706 assert!(
1707 all_entry_ids.insert(field.value().to_string()),
1708 "Found duplicate ENTRY_ID: {}",
1709 field.value()
1710 );
1711 }
1712 }
1713 }
1714
1715 state = Some(new_state);
1716 }
1717
1718 // Should need 20 pages for 20 entries
1719 assert_eq!(page_count, 20, "Should need exactly 20 pages");
1720
1721 // Verify we got all 20 unique entries
1722 assert_eq!(
1723 all_entry_ids.len(),
1724 20,
1725 "Should have retrieved all 20 unique entries"
1726 );
1727
1728 // Verify timestamps are in descending order
1729 for i in 1..all_timestamps.len() {
1730 assert!(
1731 all_timestamps[i - 1] >= all_timestamps[i],
1732 "Timestamps should be in descending order"
1733 );
1734 }
1735
1736 // Verify we got all timestamps from 119 down to 100
1737 let unique_timestamps: HashSet<_> = all_timestamps.into_iter().collect();
1738 assert_eq!(
1739 unique_timestamps.len(),
1740 20,
1741 "Should have 20 unique timestamps"
1742 );
1743 for ts in 100..120 {
1744 assert!(unique_timestamps.contains(&ts), "Missing timestamp: {}", ts);
1745 }
1746 }
1747
1748 #[test]
1749 fn test_multi_file_pagination_anchor_timestamp_forward() {
1750 // Create temporary directory
1751 let temp_dir = TempDir::new().unwrap();
1752
1753 // File 1: entries at t=100..200 (100 entries)
1754 let entries_file1: Vec<TestEntry> = (100..200)
1755 .map(|i| {
1756 TestEntry::new(Microseconds(i))
1757 .with_field("MESSAGE", format!("File1 Entry {}", i))
1758 .with_field("ENTRY_ID", format!("file1_{}", i))
1759 })
1760 .collect();
1761
1762 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
1763
1764 // File 2: entries at t=200..300 (100 entries)
1765 let entries_file2: Vec<TestEntry> = (200..300)
1766 .map(|i| {
1767 TestEntry::new(Microseconds(i))
1768 .with_field("MESSAGE", format!("File2 Entry {}", i))
1769 .with_field("ENTRY_ID", format!("file2_{}", i))
1770 })
1771 .collect();
1772
1773 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
1774
1775 // Index both files
1776 let mut indexer = FileIndexer::default();
1777 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
1778 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
1779
1780 let index1 = indexer
1781 .index(
1782 &file1,
1783 Some(&source_timestamp_field),
1784 &[entry_id_field.clone()],
1785 Seconds(3600),
1786 )
1787 .unwrap();
1788
1789 let index2 = indexer
1790 .index(
1791 &file2,
1792 Some(&source_timestamp_field),
1793 &[entry_id_field],
1794 Seconds(3600),
1795 )
1796 .unwrap();
1797
1798 let file_indexes = vec![index1, index2];
1799
1800 // Start from middle timestamp 150 (in file1), forward direction
1801 let anchor = Anchor::Timestamp(Microseconds(150));
1802
1803 // First page: limit=80, should get 50 from file1 (150-199) + 30 from file2 (200-229)
1804 let (first_page, state1) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
1805 .with_limit(80)
1806 .execute_page(None)
1807 .unwrap();
1808
1809 assert_eq!(
1810 first_page.len(),
1811 80,
1812 "First page should contain exactly 80 entries"
1813 );
1814
1815 // First entry should be at timestamp 150
1816 assert_eq!(
1817 first_page.first().unwrap().timestamp,
1818 150,
1819 "First entry should be at timestamp 150 (anchor)"
1820 );
1821
1822 // Last entry should be at timestamp 229
1823 assert_eq!(
1824 first_page.last().unwrap().timestamp,
1825 229,
1826 "Last entry should be at timestamp 229"
1827 );
1828
1829 // Verify ascending order
1830 for i in 1..first_page.len() {
1831 assert!(
1832 first_page[i - 1].timestamp <= first_page[i].timestamp,
1833 "Entries should be in ascending order"
1834 );
1835 }
1836
1837 // Second page: get remaining entries
1838 let (second_page, state2) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
1839 .with_limit(80)
1840 .execute_page(Some(&state1))
1841 .unwrap();
1842
1843 assert_eq!(
1844 second_page.len(),
1845 70,
1846 "Second page should contain remaining 70 entries (230-299)"
1847 );
1848
1849 assert_eq!(
1850 second_page.first().unwrap().timestamp,
1851 230,
1852 "Second page should start at timestamp 230"
1853 );
1854
1855 assert_eq!(
1856 second_page.last().unwrap().timestamp,
1857 299,
1858 "Second page should end at timestamp 299"
1859 );
1860
1861 // Verify no duplicates
1862 let mut all_entry_ids = HashSet::new();
1863 for entry in first_page.iter().chain(second_page.iter()) {
1864 for field in &entry.fields {
1865 if field.field() == "ENTRY_ID" {
1866 assert!(
1867 all_entry_ids.insert(field.value().to_string()),
1868 "Found duplicate ENTRY_ID: {}",
1869 field.value()
1870 );
1871 }
1872 }
1873 }
1874
1875 // Should have 150 entries total (from 150-299)
1876 assert_eq!(
1877 all_entry_ids.len(),
1878 150,
1879 "Should have retrieved 150 unique entries from timestamp 150 onwards"
1880 );
1881
1882 // Third page should be empty
1883 let (third_page, _state3) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
1884 .with_limit(80)
1885 .execute_page(Some(&state2))
1886 .unwrap();
1887
1888 assert_eq!(
1889 third_page.len(),
1890 0,
1891 "Third page should be empty (no more entries)"
1892 );
1893 }
1894
1895 #[test]
1896 fn test_multi_file_pagination_anchor_timestamp_backward() {
1897 // Create temporary directory
1898 let temp_dir = TempDir::new().unwrap();
1899
1900 // File 1: entries at t=100..200 (100 entries)
1901 let entries_file1: Vec<TestEntry> = (100..200)
1902 .map(|i| {
1903 TestEntry::new(Microseconds(i))
1904 .with_field("MESSAGE", format!("File1 Entry {}", i))
1905 .with_field("ENTRY_ID", format!("file1_{}", i))
1906 })
1907 .collect();
1908
1909 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
1910
1911 // File 2: entries at t=200..300 (100 entries)
1912 let entries_file2: Vec<TestEntry> = (200..300)
1913 .map(|i| {
1914 TestEntry::new(Microseconds(i))
1915 .with_field("MESSAGE", format!("File2 Entry {}", i))
1916 .with_field("ENTRY_ID", format!("file2_{}", i))
1917 })
1918 .collect();
1919
1920 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
1921
1922 // Index both files
1923 let mut indexer = FileIndexer::default();
1924 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
1925 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
1926
1927 let index1 = indexer
1928 .index(
1929 &file1,
1930 Some(&source_timestamp_field),
1931 &[entry_id_field.clone()],
1932 Seconds(3600),
1933 )
1934 .unwrap();
1935
1936 let index2 = indexer
1937 .index(
1938 &file2,
1939 Some(&source_timestamp_field),
1940 &[entry_id_field],
1941 Seconds(3600),
1942 )
1943 .unwrap();
1944
1945 let file_indexes = vec![index1, index2];
1946
1947 // Start from middle timestamp 250 (in file2), backward direction
1948 let anchor = Anchor::Timestamp(Microseconds(250));
1949
1950 // First page: limit=80, should get 51 from file2 (250-200) + 29 from file1 (199-171)
1951 let (first_page, state1) = LogQuery::new(&file_indexes, anchor, Direction::Backward)
1952 .with_limit(80)
1953 .execute_page(None)
1954 .unwrap();
1955
1956 assert_eq!(
1957 first_page.len(),
1958 80,
1959 "First page should contain exactly 80 entries"
1960 );
1961
1962 // First entry should be at timestamp 250
1963 assert_eq!(
1964 first_page.first().unwrap().timestamp,
1965 250,
1966 "First entry should be at timestamp 250 (anchor)"
1967 );
1968
1969 // Last entry should be at timestamp 171
1970 assert_eq!(
1971 first_page.last().unwrap().timestamp,
1972 171,
1973 "Last entry should be at timestamp 171"
1974 );
1975
1976 // Verify descending order
1977 for i in 1..first_page.len() {
1978 assert!(
1979 first_page[i - 1].timestamp >= first_page[i].timestamp,
1980 "Entries should be in descending order"
1981 );
1982 }
1983
1984 // Second page: get remaining entries
1985 let (second_page, state2) = LogQuery::new(&file_indexes, anchor, Direction::Backward)
1986 .with_limit(80)
1987 .execute_page(Some(&state1))
1988 .unwrap();
1989
1990 assert_eq!(
1991 second_page.len(),
1992 71,
1993 "Second page should contain remaining 71 entries (170-100)"
1994 );
1995
1996 assert_eq!(
1997 second_page.first().unwrap().timestamp,
1998 170,
1999 "Second page should start at timestamp 170"
2000 );
2001
2002 assert_eq!(
2003 second_page.last().unwrap().timestamp,
2004 100,
2005 "Second page should end at timestamp 100"
2006 );
2007
2008 // Verify no duplicates
2009 let mut all_entry_ids = HashSet::new();
2010 for entry in first_page.iter().chain(second_page.iter()) {
2011 for field in &entry.fields {
2012 if field.field() == "ENTRY_ID" {
2013 assert!(
2014 all_entry_ids.insert(field.value().to_string()),
2015 "Found duplicate ENTRY_ID: {}",
2016 field.value()
2017 );
2018 }
2019 }
2020 }
2021
2022 // Should have 151 entries total (from 250 down to 100)
2023 assert_eq!(
2024 all_entry_ids.len(),
2025 151,
2026 "Should have retrieved 151 unique entries from timestamp 250 backwards to 100"
2027 );
2028
2029 // Third page should be empty
2030 let (third_page, _state3) = LogQuery::new(&file_indexes, anchor, Direction::Backward)
2031 .with_limit(80)
2032 .execute_page(Some(&state2))
2033 .unwrap();
2034
2035 assert_eq!(
2036 third_page.len(),
2037 0,
2038 "Third page should be empty (no more entries)"
2039 );
2040 }
2041
2042 #[test]
2043 fn test_multi_file_pagination_anchor_timestamp_same_timestamps() {
2044 // Create temporary directory
2045 let temp_dir = TempDir::new().unwrap();
2046
2047 // File 1: 100 entries all at timestamp 150
2048 let entries_file1: Vec<TestEntry> = (0..100)
2049 .map(|i| {
2050 TestEntry::new(Microseconds(150))
2051 .with_field("MESSAGE", format!("File1 Entry {}", i))
2052 .with_field("ENTRY_ID", format!("file1_{}", i))
2053 })
2054 .collect();
2055
2056 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
2057
2058 // File 2: 100 entries all at timestamp 150
2059 let entries_file2: Vec<TestEntry> = (0..100)
2060 .map(|i| {
2061 TestEntry::new(Microseconds(150))
2062 .with_field("MESSAGE", format!("File2 Entry {}", i))
2063 .with_field("ENTRY_ID", format!("file2_{}", i))
2064 })
2065 .collect();
2066
2067 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
2068
2069 // Index both files
2070 let mut indexer = FileIndexer::default();
2071 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
2072 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
2073
2074 let index1 = indexer
2075 .index(
2076 &file1,
2077 Some(&source_timestamp_field),
2078 &[entry_id_field.clone()],
2079 Seconds(3600),
2080 )
2081 .unwrap();
2082
2083 let index2 = indexer
2084 .index(
2085 &file2,
2086 Some(&source_timestamp_field),
2087 &[entry_id_field],
2088 Seconds(3600),
2089 )
2090 .unwrap();
2091
2092 let file_indexes = vec![index1, index2];
2093
2094 // Anchor at timestamp 150 (all entries have this timestamp), forward direction
2095 let anchor = Anchor::Timestamp(Microseconds(150));
2096
2097 // First page: limit=80
2098 let (first_page, state1) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
2099 .with_limit(80)
2100 .execute_page(None)
2101 .unwrap();
2102
2103 assert_eq!(
2104 first_page.len(),
2105 80,
2106 "First page should contain exactly 80 entries"
2107 );
2108
2109 // All timestamps should be 150
2110 for entry in &first_page {
2111 assert_eq!(
2112 entry.timestamp, 150,
2113 "All entries should have timestamp 150"
2114 );
2115 }
2116
2117 // Second page: get remaining entries
2118 let (second_page, state2) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
2119 .with_limit(80)
2120 .execute_page(Some(&state1))
2121 .unwrap();
2122
2123 assert_eq!(
2124 second_page.len(),
2125 80,
2126 "Second page should contain 80 entries"
2127 );
2128
2129 for entry in &second_page {
2130 assert_eq!(
2131 entry.timestamp, 150,
2132 "All entries should have timestamp 150"
2133 );
2134 }
2135
2136 // Third page: remaining entries
2137 let (third_page, state3) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
2138 .with_limit(80)
2139 .execute_page(Some(&state2))
2140 .unwrap();
2141
2142 assert_eq!(
2143 third_page.len(),
2144 40,
2145 "Third page should contain remaining 40 entries"
2146 );
2147
2148 for entry in &third_page {
2149 assert_eq!(
2150 entry.timestamp, 150,
2151 "All entries should have timestamp 150"
2152 );
2153 }
2154
2155 // Verify no duplicates
2156 let mut all_entry_ids = HashSet::new();
2157 for entry in first_page
2158 .iter()
2159 .chain(second_page.iter())
2160 .chain(third_page.iter())
2161 {
2162 for field in &entry.fields {
2163 if field.field() == "ENTRY_ID" {
2164 assert!(
2165 all_entry_ids.insert(field.value().to_string()),
2166 "Found duplicate ENTRY_ID: {}",
2167 field.value()
2168 );
2169 }
2170 }
2171 }
2172
2173 // Should have all 200 entries
2174 assert_eq!(
2175 all_entry_ids.len(),
2176 200,
2177 "Should have retrieved all 200 unique entries"
2178 );
2179
2180 // Fourth page should be empty
2181 let (fourth_page, _state4) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
2182 .with_limit(80)
2183 .execute_page(Some(&state3))
2184 .unwrap();
2185
2186 assert_eq!(
2187 fourth_page.len(),
2188 0,
2189 "Fourth page should be empty (no more entries)"
2190 );
2191 }
2192
2193 #[test]
2194 fn test_multi_file_pagination_forward_with_time_boundaries() {
2195 // Create temporary directory
2196 let temp_dir = TempDir::new().unwrap();
2197
2198 // File 1: entries at t=100..200 (100 entries)
2199 let entries_file1: Vec<TestEntry> = (100..200)
2200 .map(|i| {
2201 TestEntry::new(Microseconds(i))
2202 .with_field("MESSAGE", format!("File1 Entry {}", i))
2203 .with_field("ENTRY_ID", format!("file1_{}", i))
2204 })
2205 .collect();
2206
2207 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
2208
2209 // File 2: entries at t=200..300 (100 entries)
2210 let entries_file2: Vec<TestEntry> = (200..300)
2211 .map(|i| {
2212 TestEntry::new(Microseconds(i))
2213 .with_field("MESSAGE", format!("File2 Entry {}", i))
2214 .with_field("ENTRY_ID", format!("file2_{}", i))
2215 })
2216 .collect();
2217
2218 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
2219
2220 // File 3: entries at t=300..400 (100 entries)
2221 let entries_file3: Vec<TestEntry> = (300..400)
2222 .map(|i| {
2223 TestEntry::new(Microseconds(i))
2224 .with_field("MESSAGE", format!("File3 Entry {}", i))
2225 .with_field("ENTRY_ID", format!("file3_{}", i))
2226 })
2227 .collect();
2228
2229 let file3 = create_test_journal(&temp_dir, "file3.journal", entries_file3).unwrap();
2230
2231 // Index all three files
2232 let mut indexer = FileIndexer::default();
2233 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
2234 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
2235
2236 let index1 = indexer
2237 .index(
2238 &file1,
2239 Some(&source_timestamp_field),
2240 &[entry_id_field.clone()],
2241 Seconds(3600),
2242 )
2243 .unwrap();
2244
2245 let index2 = indexer
2246 .index(
2247 &file2,
2248 Some(&source_timestamp_field),
2249 &[entry_id_field.clone()],
2250 Seconds(3600),
2251 )
2252 .unwrap();
2253
2254 let index3 = indexer
2255 .index(
2256 &file3,
2257 Some(&source_timestamp_field),
2258 &[entry_id_field],
2259 Seconds(3600),
2260 )
2261 .unwrap();
2262
2263 let file_indexes = vec![index1, index2, index3];
2264
2265 // Query with time boundaries: after=150, before=350
2266 // This should return entries from 150-349 (200 entries total)
2267 // File1: 150-199 (50 entries)
2268 // File2: 200-299 (100 entries)
2269 // File3: 300-349 (50 entries)
2270
2271 // First page: limit=80
2272 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
2273 .with_after_usec(150)
2274 .with_before_usec(350)
2275 .with_limit(80)
2276 .execute_page(None)
2277 .unwrap();
2278
2279 assert_eq!(first_page.len(), 80, "First page should contain 80 entries");
2280
2281 // Should start at timestamp 150
2282 assert_eq!(
2283 first_page.first().unwrap().timestamp,
2284 150,
2285 "First entry should be at timestamp 150"
2286 );
2287
2288 // Should end at timestamp 229 (50 from file1 + 30 from file2)
2289 assert_eq!(
2290 first_page.last().unwrap().timestamp,
2291 229,
2292 "Last entry should be at timestamp 229"
2293 );
2294
2295 // Verify all timestamps are within boundaries
2296 for entry in &first_page {
2297 assert!(
2298 entry.timestamp >= 150 && entry.timestamp < 350,
2299 "Entry timestamp {} should be within [150, 350)",
2300 entry.timestamp
2301 );
2302 }
2303
2304 // Second page: limit=80
2305 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
2306 .with_after_usec(150)
2307 .with_before_usec(350)
2308 .with_limit(80)
2309 .execute_page(Some(&state1))
2310 .unwrap();
2311
2312 assert_eq!(
2313 second_page.len(),
2314 80,
2315 "Second page should contain 80 entries"
2316 );
2317
2318 assert_eq!(
2319 second_page.first().unwrap().timestamp,
2320 230,
2321 "Second page should start at timestamp 230"
2322 );
2323
2324 assert_eq!(
2325 second_page.last().unwrap().timestamp,
2326 309,
2327 "Second page should end at timestamp 309"
2328 );
2329
2330 // Verify all timestamps are within boundaries
2331 for entry in &second_page {
2332 assert!(
2333 entry.timestamp >= 150 && entry.timestamp < 350,
2334 "Entry timestamp {} should be within [150, 350)",
2335 entry.timestamp
2336 );
2337 }
2338
2339 // Third page: remaining 40 entries
2340 let (third_page, state3) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
2341 .with_after_usec(150)
2342 .with_before_usec(350)
2343 .with_limit(80)
2344 .execute_page(Some(&state2))
2345 .unwrap();
2346
2347 assert_eq!(
2348 third_page.len(),
2349 40,
2350 "Third page should contain remaining 40 entries"
2351 );
2352
2353 assert_eq!(
2354 third_page.first().unwrap().timestamp,
2355 310,
2356 "Third page should start at timestamp 310"
2357 );
2358
2359 assert_eq!(
2360 third_page.last().unwrap().timestamp,
2361 349,
2362 "Third page should end at timestamp 349 (before boundary)"
2363 );
2364
2365 // Verify all timestamps are within boundaries
2366 for entry in &third_page {
2367 assert!(
2368 entry.timestamp >= 150 && entry.timestamp < 350,
2369 "Entry timestamp {} should be within [150, 350)",
2370 entry.timestamp
2371 );
2372 }
2373
2374 // Verify no duplicates and correct total count
2375 let mut all_entry_ids = HashSet::new();
2376 for entry in first_page
2377 .iter()
2378 .chain(second_page.iter())
2379 .chain(third_page.iter())
2380 {
2381 for field in &entry.fields {
2382 if field.field() == "ENTRY_ID" {
2383 assert!(
2384 all_entry_ids.insert(field.value().to_string()),
2385 "Found duplicate ENTRY_ID: {}",
2386 field.value()
2387 );
2388 }
2389 }
2390 }
2391
2392 // Should have exactly 200 entries (150-349)
2393 assert_eq!(
2394 all_entry_ids.len(),
2395 200,
2396 "Should have retrieved exactly 200 entries within time boundaries"
2397 );
2398
2399 // Fourth page should be empty
2400 let (fourth_page, _state4) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
2401 .with_after_usec(150)
2402 .with_before_usec(350)
2403 .with_limit(80)
2404 .execute_page(Some(&state3))
2405 .unwrap();
2406
2407 assert_eq!(
2408 fourth_page.len(),
2409 0,
2410 "Fourth page should be empty (no more entries)"
2411 );
2412 }
2413
2414 #[test]
2415 fn test_multi_file_pagination_backward_overlapping_timestamps() {
2416 // Create temporary directory
2417 let temp_dir = TempDir::new().unwrap();
2418
2419 // File 1: entries at t=100..200 (100 entries)
2420 let entries_file1: Vec<TestEntry> = (100..200)
2421 .map(|i| {
2422 TestEntry::new(Microseconds(i))
2423 .with_field("MESSAGE", format!("File1 Entry at {}", i))
2424 .with_field("ENTRY_ID", format!("file1_{}", i))
2425 .with_field("FILE", "1")
2426 })
2427 .collect();
2428
2429 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
2430
2431 // File 2: entries at t=150..250 (100 entries) - overlaps with file1 from 150-199
2432 let entries_file2: Vec<TestEntry> = (150..250)
2433 .map(|i| {
2434 TestEntry::new(Microseconds(i))
2435 .with_field("MESSAGE", format!("File2 Entry at {}", i))
2436 .with_field("ENTRY_ID", format!("file2_{}", i))
2437 .with_field("FILE", "2")
2438 })
2439 .collect();
2440
2441 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
2442
2443 // Index both files
2444 let mut indexer = FileIndexer::default();
2445 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
2446 let file_field = FieldName::new("FILE").unwrap();
2447 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
2448
2449 let index1 = indexer
2450 .index(
2451 &file1,
2452 Some(&source_timestamp_field),
2453 &[file_field.clone(), entry_id_field.clone()],
2454 Seconds(3600),
2455 )
2456 .unwrap();
2457
2458 let index2 = indexer
2459 .index(
2460 &file2,
2461 Some(&source_timestamp_field),
2462 &[file_field, entry_id_field],
2463 Seconds(3600),
2464 )
2465 .unwrap();
2466
2467 let file_indexes = vec![index1, index2];
2468
2469 // First page: limit=120, backward from tail
2470 // Expected: from timestamp 249 going backward
2471 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
2472 .with_limit(120)
2473 .execute_page(None)
2474 .unwrap();
2475
2476 assert_eq!(
2477 first_page.len(),
2478 120,
2479 "First page should contain exactly 120 entries"
2480 );
2481
2482 // Verify timestamps are in descending order
2483 for i in 1..first_page.len() {
2484 assert!(
2485 first_page[i - 1].timestamp >= first_page[i].timestamp,
2486 "Entries should be in descending timestamp order"
2487 );
2488 }
2489
2490 // First entry should be at timestamp 249 (highest)
2491 assert_eq!(
2492 first_page.first().unwrap().timestamp,
2493 249,
2494 "First entry should be at timestamp 249"
2495 );
2496
2497 // State should track positions for both files
2498 assert!(
2499 !state1.file_positions.is_empty(),
2500 "State should track positions"
2501 );
2502
2503 // Second page: get remaining entries
2504 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
2505 .with_limit(120)
2506 .execute_page(Some(&state1))
2507 .unwrap();
2508
2509 assert_eq!(
2510 second_page.len(),
2511 80,
2512 "Second page should contain remaining 80 entries"
2513 );
2514
2515 // Verify timestamps continue in descending order
2516 for i in 1..second_page.len() {
2517 assert!(
2518 second_page[i - 1].timestamp >= second_page[i].timestamp,
2519 "Entries should be in descending timestamp order"
2520 );
2521 }
2522
2523 // Verify no timestamp gap between pages
2524 if !first_page.is_empty() && !second_page.is_empty() {
2525 assert!(
2526 first_page.last().unwrap().timestamp >= second_page.first().unwrap().timestamp,
2527 "Second page should continue from first page timestamp"
2528 );
2529 }
2530
2531 // Last entry should be at timestamp 100
2532 assert_eq!(
2533 second_page.last().unwrap().timestamp,
2534 100,
2535 "Last entry should be at timestamp 100"
2536 );
2537
2538 // Collect all ENTRY_ID values to verify uniqueness and completeness
2539 let mut all_entry_ids = HashSet::new();
2540
2541 for entry in &first_page {
2542 for field in &entry.fields {
2543 if field.field() == "ENTRY_ID" {
2544 assert!(
2545 all_entry_ids.insert(field.value().to_string()),
2546 "Found duplicate ENTRY_ID: {}",
2547 field.value()
2548 );
2549 }
2550 }
2551 }
2552
2553 for entry in &second_page {
2554 for field in &entry.fields {
2555 if field.field() == "ENTRY_ID" {
2556 assert!(
2557 all_entry_ids.insert(field.value().to_string()),
2558 "Found duplicate ENTRY_ID: {}",
2559 field.value()
2560 );
2561 }
2562 }
2563 }
2564
2565 // Verify we got all 200 unique entries (100 from each file)
2566 assert_eq!(
2567 all_entry_ids.len(),
2568 200,
2569 "Should have retrieved all 200 unique entries"
2570 );
2571
2572 // Verify we have entries from both files
2573 let file1_entries: usize = all_entry_ids
2574 .iter()
2575 .filter(|id| id.starts_with("file1_"))
2576 .count();
2577 let file2_entries: usize = all_entry_ids
2578 .iter()
2579 .filter(|id| id.starts_with("file2_"))
2580 .count();
2581
2582 assert_eq!(file1_entries, 100, "Should have 100 entries from file1");
2583 assert_eq!(file2_entries, 100, "Should have 100 entries from file2");
2584
2585 // Verify all timestamps from 100-249 are represented
2586 let mut all_timestamps = HashSet::new();
2587 for entry in first_page.iter().chain(second_page.iter()) {
2588 all_timestamps.insert(entry.timestamp);
2589 }
2590
2591 // We should have entries at all timestamps from 100-249 (150 unique timestamps)
2592 for ts in 100..250 {
2593 assert!(all_timestamps.contains(&ts), "Missing timestamp: {}", ts);
2594 }
2595
2596 // Third page should be empty
2597 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
2598 .with_limit(120)
2599 .execute_page(Some(&state2))
2600 .unwrap();
2601
2602 assert_eq!(
2603 third_page.len(),
2604 0,
2605 "Third page should be empty (no more entries)"
2606 );
2607 }
2608
2609 #[test]
2610 fn test_multi_file_pagination_backward_three_files() {
2611 // Create temporary directory
2612 let temp_dir = TempDir::new().unwrap();
2613
2614 // File 1: entries at t=100..200 (100 entries)
2615 let entries_file1: Vec<TestEntry> = (100..200)
2616 .map(|i| {
2617 TestEntry::new(Microseconds(i))
2618 .with_field("MESSAGE", format!("File1 Entry {}", i))
2619 .with_field("ENTRY_ID", format!("file1_{}", i))
2620 .with_field("FILE", "1")
2621 })
2622 .collect();
2623
2624 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
2625
2626 // File 2: entries at t=200..300 (100 entries)
2627 let entries_file2: Vec<TestEntry> = (200..300)
2628 .map(|i| {
2629 TestEntry::new(Microseconds(i))
2630 .with_field("MESSAGE", format!("File2 Entry {}", i))
2631 .with_field("ENTRY_ID", format!("file2_{}", i))
2632 .with_field("FILE", "2")
2633 })
2634 .collect();
2635
2636 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
2637
2638 // File 3: entries at t=300..400 (100 entries)
2639 let entries_file3: Vec<TestEntry> = (300..400)
2640 .map(|i| {
2641 TestEntry::new(Microseconds(i))
2642 .with_field("MESSAGE", format!("File3 Entry {}", i))
2643 .with_field("ENTRY_ID", format!("file3_{}", i))
2644 .with_field("FILE", "3")
2645 })
2646 .collect();
2647
2648 let file3 = create_test_journal(&temp_dir, "file3.journal", entries_file3).unwrap();
2649
2650 // Index all three files
2651 let mut indexer = FileIndexer::default();
2652 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
2653 let file_field = FieldName::new("FILE").unwrap();
2654 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
2655
2656 let index1 = indexer
2657 .index(
2658 &file1,
2659 Some(&source_timestamp_field),
2660 &[file_field.clone(), entry_id_field.clone()],
2661 Seconds(3600),
2662 )
2663 .unwrap();
2664
2665 let index2 = indexer
2666 .index(
2667 &file2,
2668 Some(&source_timestamp_field),
2669 &[file_field.clone(), entry_id_field.clone()],
2670 Seconds(3600),
2671 )
2672 .unwrap();
2673
2674 let index3 = indexer
2675 .index(
2676 &file3,
2677 Some(&source_timestamp_field),
2678 &[file_field, entry_id_field],
2679 Seconds(3600),
2680 )
2681 .unwrap();
2682
2683 let file_indexes = vec![index1, index2, index3];
2684
2685 // First page: limit=125, backward from tail, should get all 100 from file3 + 25 from file2
2686 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
2687 .with_limit(125)
2688 .execute_page(None)
2689 .unwrap();
2690
2691 assert_eq!(
2692 first_page.len(),
2693 125,
2694 "First page should contain exactly 125 entries"
2695 );
2696
2697 // Verify descending order
2698 for i in 1..first_page.len() {
2699 assert!(
2700 first_page[i - 1].timestamp >= first_page[i].timestamp,
2701 "Entries should be in descending order"
2702 );
2703 }
2704
2705 assert_eq!(first_page.first().unwrap().timestamp, 399);
2706 assert_eq!(first_page.last().unwrap().timestamp, 275);
2707
2708 // State should track positions for file3 and file2
2709 assert_eq!(
2710 state1.file_positions.len(),
2711 2,
2712 "State should track positions for 2 files"
2713 );
2714
2715 // Second page: limit=125, should get 75 from file2 + 50 from file1
2716 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
2717 .with_limit(125)
2718 .execute_page(Some(&state1))
2719 .unwrap();
2720
2721 assert_eq!(
2722 second_page.len(),
2723 125,
2724 "Second page should contain exactly 125 entries"
2725 );
2726
2727 // Verify descending order
2728 for i in 1..second_page.len() {
2729 assert!(
2730 second_page[i - 1].timestamp >= second_page[i].timestamp,
2731 "Entries should be in descending order"
2732 );
2733 }
2734
2735 assert_eq!(second_page.first().unwrap().timestamp, 274);
2736 assert_eq!(second_page.last().unwrap().timestamp, 150);
2737
2738 // State should now track all 3 files
2739 assert_eq!(
2740 state2.file_positions.len(),
2741 3,
2742 "State should track positions for all 3 files"
2743 );
2744
2745 // Third page: remaining 50 from file1
2746 let (third_page, state3) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
2747 .with_limit(125)
2748 .execute_page(Some(&state2))
2749 .unwrap();
2750
2751 assert_eq!(
2752 third_page.len(),
2753 50,
2754 "Third page should contain remaining 50 entries"
2755 );
2756
2757 // Verify descending order
2758 for i in 1..third_page.len() {
2759 assert!(
2760 third_page[i - 1].timestamp >= third_page[i].timestamp,
2761 "Entries should be in descending order"
2762 );
2763 }
2764
2765 assert_eq!(third_page.first().unwrap().timestamp, 149);
2766 assert_eq!(third_page.last().unwrap().timestamp, 100);
2767
2768 // Collect all ENTRY_ID values to verify uniqueness
2769 let mut all_entry_ids = HashSet::new();
2770
2771 for entry in first_page
2772 .iter()
2773 .chain(second_page.iter())
2774 .chain(third_page.iter())
2775 {
2776 for field in &entry.fields {
2777 if field.field() == "ENTRY_ID" {
2778 assert!(
2779 all_entry_ids.insert(field.value().to_string()),
2780 "Found duplicate ENTRY_ID: {}",
2781 field.value()
2782 );
2783 }
2784 }
2785 }
2786
2787 // Verify we got all 300 unique entries
2788 assert_eq!(
2789 all_entry_ids.len(),
2790 300,
2791 "Should have retrieved all 300 unique entries"
2792 );
2793
2794 // Verify distribution: 100 from each file
2795 let file1_count = all_entry_ids
2796 .iter()
2797 .filter(|id| id.starts_with("file1_"))
2798 .count();
2799 let file2_count = all_entry_ids
2800 .iter()
2801 .filter(|id| id.starts_with("file2_"))
2802 .count();
2803 let file3_count = all_entry_ids
2804 .iter()
2805 .filter(|id| id.starts_with("file3_"))
2806 .count();
2807
2808 assert_eq!(file1_count, 100, "Should have 100 entries from file1");
2809 assert_eq!(file2_count, 100, "Should have 100 entries from file2");
2810 assert_eq!(file3_count, 100, "Should have 100 entries from file3");
2811
2812 // Fourth page should be empty
2813 let (fourth_page, _state4) = LogQuery::new(&file_indexes, Anchor::Tail, Direction::Backward)
2814 .with_limit(125)
2815 .execute_page(Some(&state3))
2816 .unwrap();
2817
2818 assert_eq!(
2819 fourth_page.len(),
2820 0,
2821 "Fourth page should be empty (no more entries)"
2822 );
2823 }
2824
2825 #[test]
2826 fn test_multi_file_pagination_with_filter() {
2827 // Create temporary directory
2828 let temp_dir = TempDir::new().unwrap();
2829
2830 // File 1: 50 entries at t=100..150 with LEVEL=ERROR
2831 // 50 entries at t=150..200 with LEVEL=INFO
2832 let mut entries_file1: Vec<TestEntry> = (100..150)
2833 .map(|i| {
2834 TestEntry::new(Microseconds(i))
2835 .with_field("MESSAGE", format!("File1 Error {}", i))
2836 .with_field("ENTRY_ID", format!("file1_error_{}", i))
2837 .with_field("LEVEL", "ERROR")
2838 })
2839 .collect();
2840
2841 entries_file1.extend((150..200).map(|i| {
2842 TestEntry::new(Microseconds(i))
2843 .with_field("MESSAGE", format!("File1 Info {}", i))
2844 .with_field("ENTRY_ID", format!("file1_info_{}", i))
2845 .with_field("LEVEL", "INFO")
2846 }));
2847
2848 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
2849
2850 // File 2: 50 entries at t=200..250 with LEVEL=ERROR
2851 // 50 entries at t=250..300 with LEVEL=INFO
2852 let mut entries_file2: Vec<TestEntry> = (200..250)
2853 .map(|i| {
2854 TestEntry::new(Microseconds(i))
2855 .with_field("MESSAGE", format!("File2 Error {}", i))
2856 .with_field("ENTRY_ID", format!("file2_error_{}", i))
2857 .with_field("LEVEL", "ERROR")
2858 })
2859 .collect();
2860
2861 entries_file2.extend((250..300).map(|i| {
2862 TestEntry::new(Microseconds(i))
2863 .with_field("MESSAGE", format!("File2 Info {}", i))
2864 .with_field("ENTRY_ID", format!("file2_info_{}", i))
2865 .with_field("LEVEL", "INFO")
2866 }));
2867
2868 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
2869
2870 // File 3: 50 entries at t=300..350 with LEVEL=ERROR
2871 // 50 entries at t=350..400 with LEVEL=INFO
2872 let mut entries_file3: Vec<TestEntry> = (300..350)
2873 .map(|i| {
2874 TestEntry::new(Microseconds(i))
2875 .with_field("MESSAGE", format!("File3 Error {}", i))
2876 .with_field("ENTRY_ID", format!("file3_error_{}", i))
2877 .with_field("LEVEL", "ERROR")
2878 })
2879 .collect();
2880
2881 entries_file3.extend((350..400).map(|i| {
2882 TestEntry::new(Microseconds(i))
2883 .with_field("MESSAGE", format!("File3 Info {}", i))
2884 .with_field("ENTRY_ID", format!("file3_info_{}", i))
2885 .with_field("LEVEL", "INFO")
2886 }));
2887
2888 let file3 = create_test_journal(&temp_dir, "file3.journal", entries_file3).unwrap();
2889
2890 // Index all three files
2891 let mut indexer = FileIndexer::default();
2892 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
2893 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
2894 let level_field = FieldName::new("LEVEL").unwrap();
2895
2896 let index1 = indexer
2897 .index(
2898 &file1,
2899 Some(&source_timestamp_field),
2900 &[entry_id_field.clone(), level_field.clone()],
2901 Seconds(3600),
2902 )
2903 .unwrap();
2904
2905 let index2 = indexer
2906 .index(
2907 &file2,
2908 Some(&source_timestamp_field),
2909 &[entry_id_field.clone(), level_field.clone()],
2910 Seconds(3600),
2911 )
2912 .unwrap();
2913
2914 let index3 = indexer
2915 .index(
2916 &file3,
2917 Some(&source_timestamp_field),
2918 &[entry_id_field, level_field],
2919 Seconds(3600),
2920 )
2921 .unwrap();
2922
2923 let file_indexes = vec![index1, index2, index3];
2924
2925 // Create filter to match only LEVEL=ERROR entries
2926 // This should match 50 entries per file (150 total)
2927 let filter = Filter::match_field_value_pair(FieldValuePair::parse("LEVEL=ERROR").unwrap());
2928
2929 // First page: limit=80, should get 50 from file1 + 30 from file2
2930 let (first_page, state1) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
2931 .with_filter(filter.clone())
2932 .with_limit(80)
2933 .execute_page(None)
2934 .unwrap();
2935
2936 assert_eq!(
2937 first_page.len(),
2938 80,
2939 "First page should contain 80 ERROR entries"
2940 );
2941
2942 // All entries should have LEVEL=ERROR
2943 for entry in &first_page {
2944 let level_values: Vec<_> = entry
2945 .fields
2946 .iter()
2947 .filter(|f| f.field() == "LEVEL")
2948 .map(|f| f.value())
2949 .collect();
2950 assert_eq!(
2951 level_values,
2952 vec!["ERROR"],
2953 "All entries should have LEVEL=ERROR"
2954 );
2955 }
2956
2957 // First entry should be at timestamp 100
2958 assert_eq!(
2959 first_page.first().unwrap().timestamp,
2960 100,
2961 "First entry should be at timestamp 100"
2962 );
2963
2964 // Last entry should be at timestamp 229
2965 assert_eq!(
2966 first_page.last().unwrap().timestamp,
2967 229,
2968 "Last entry should be at timestamp 229"
2969 );
2970
2971 // Second page: get remaining ERROR entries
2972 let (second_page, state2) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
2973 .with_filter(filter.clone())
2974 .with_limit(80)
2975 .execute_page(Some(&state1))
2976 .unwrap();
2977
2978 assert_eq!(
2979 second_page.len(),
2980 70,
2981 "Second page should contain remaining 70 ERROR entries"
2982 );
2983
2984 // All entries should have LEVEL=ERROR
2985 for entry in &second_page {
2986 let level_values: Vec<_> = entry
2987 .fields
2988 .iter()
2989 .filter(|f| f.field() == "LEVEL")
2990 .map(|f| f.value())
2991 .collect();
2992 assert_eq!(
2993 level_values,
2994 vec!["ERROR"],
2995 "All entries should have LEVEL=ERROR"
2996 );
2997 }
2998
2999 // Should continue from timestamp 230 and go to 349
3000 assert_eq!(
3001 second_page.first().unwrap().timestamp,
3002 230,
3003 "Second page should start at timestamp 230"
3004 );
3005
3006 assert_eq!(
3007 second_page.last().unwrap().timestamp,
3008 349,
3009 "Second page should end at timestamp 349"
3010 );
3011
3012 // Verify no duplicates
3013 let mut all_entry_ids = HashSet::new();
3014 for entry in first_page.iter().chain(second_page.iter()) {
3015 for field in &entry.fields {
3016 if field.field() == "ENTRY_ID" {
3017 assert!(
3018 all_entry_ids.insert(field.value().to_string()),
3019 "Found duplicate ENTRY_ID: {}",
3020 field.value()
3021 );
3022 }
3023 }
3024 }
3025
3026 // Should have exactly 150 ERROR entries (50 from each file)
3027 assert_eq!(
3028 all_entry_ids.len(),
3029 150,
3030 "Should have retrieved exactly 150 ERROR entries"
3031 );
3032
3033 // Verify all are error entries
3034 let error_count = all_entry_ids
3035 .iter()
3036 .filter(|id| id.contains("_error_"))
3037 .count();
3038 assert_eq!(error_count, 150, "All entries should be error entries");
3039
3040 // Verify distribution across files
3041 let file1_count = all_entry_ids
3042 .iter()
3043 .filter(|id| id.starts_with("file1_"))
3044 .count();
3045 let file2_count = all_entry_ids
3046 .iter()
3047 .filter(|id| id.starts_with("file2_"))
3048 .count();
3049 let file3_count = all_entry_ids
3050 .iter()
3051 .filter(|id| id.starts_with("file3_"))
3052 .count();
3053
3054 assert_eq!(file1_count, 50, "Should have 50 ERROR entries from file1");
3055 assert_eq!(file2_count, 50, "Should have 50 ERROR entries from file2");
3056 assert_eq!(file3_count, 50, "Should have 50 ERROR entries from file3");
3057
3058 // Third page should be empty
3059 let (third_page, _state3) = LogQuery::new(&file_indexes, Anchor::Head, Direction::Forward)
3060 .with_filter(filter)
3061 .with_limit(80)
3062 .execute_page(Some(&state2))
3063 .unwrap();
3064
3065 assert_eq!(
3066 third_page.len(),
3067 0,
3068 "Third page should be empty (no more ERROR entries)"
3069 );
3070 }
3071
3072 #[test]
3073 fn test_multi_file_pagination_anchor_at_file_boundary() {
3074 // Create temporary directory
3075 let temp_dir = TempDir::new().unwrap();
3076
3077 // File 1: entries at t=100..200 (100 entries)
3078 let entries_file1: Vec<TestEntry> = (100..200)
3079 .map(|i| {
3080 TestEntry::new(Microseconds(i))
3081 .with_field("MESSAGE", format!("File1 Entry {}", i))
3082 .with_field("ENTRY_ID", format!("file1_{}", i))
3083 })
3084 .collect();
3085
3086 let file1 = create_test_journal(&temp_dir, "file1.journal", entries_file1).unwrap();
3087
3088 // File 2: entries at t=200..300 (100 entries) - starts exactly where file1 ends
3089 let entries_file2: Vec<TestEntry> = (200..300)
3090 .map(|i| {
3091 TestEntry::new(Microseconds(i))
3092 .with_field("MESSAGE", format!("File2 Entry {}", i))
3093 .with_field("ENTRY_ID", format!("file2_{}", i))
3094 })
3095 .collect();
3096
3097 let file2 = create_test_journal(&temp_dir, "file2.journal", entries_file2).unwrap();
3098
3099 // File 3: entries at t=300..400 (100 entries) - starts exactly where file2 ends
3100 let entries_file3: Vec<TestEntry> = (300..400)
3101 .map(|i| {
3102 TestEntry::new(Microseconds(i))
3103 .with_field("MESSAGE", format!("File3 Entry {}", i))
3104 .with_field("ENTRY_ID", format!("file3_{}", i))
3105 })
3106 .collect();
3107
3108 let file3 = create_test_journal(&temp_dir, "file3.journal", entries_file3).unwrap();
3109
3110 // Index all three files
3111 let mut indexer = FileIndexer::default();
3112 let source_timestamp_field = FieldName::new("_SOURCE_REALTIME_TIMESTAMP").unwrap();
3113 let entry_id_field = FieldName::new("ENTRY_ID").unwrap();
3114
3115 let index1 = indexer
3116 .index(
3117 &file1,
3118 Some(&source_timestamp_field),
3119 &[entry_id_field.clone()],
3120 Seconds(3600),
3121 )
3122 .unwrap();
3123
3124 let index2 = indexer
3125 .index(
3126 &file2,
3127 Some(&source_timestamp_field),
3128 &[entry_id_field.clone()],
3129 Seconds(3600),
3130 )
3131 .unwrap();
3132
3133 let index3 = indexer
3134 .index(
3135 &file3,
3136 Some(&source_timestamp_field),
3137 &[entry_id_field],
3138 Seconds(3600),
3139 )
3140 .unwrap();
3141
3142 let file_indexes = vec![index1, index2, index3];
3143
3144 // Test forward from exact boundary timestamp 200 (where file1 ends and file2 starts)
3145 let anchor = Anchor::Timestamp(Microseconds(200));
3146
3147 let (first_page_fwd, state1_fwd) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
3148 .with_limit(80)
3149 .execute_page(None)
3150 .unwrap();
3151
3152 assert_eq!(
3153 first_page_fwd.len(),
3154 80,
3155 "Forward from boundary should return 80 entries"
3156 );
3157
3158 // Should start at timestamp 200 (first entry of file2)
3159 assert_eq!(
3160 first_page_fwd.first().unwrap().timestamp,
3161 200,
3162 "Forward should start at timestamp 200"
3163 );
3164
3165 // Should end at timestamp 279
3166 assert_eq!(
3167 first_page_fwd.last().unwrap().timestamp,
3168 279,
3169 "Forward should end at timestamp 279"
3170 );
3171
3172 // Continue forward pagination
3173 let (second_page_fwd, _state2_fwd) = LogQuery::new(&file_indexes, anchor, Direction::Forward)
3174 .with_limit(80)
3175 .execute_page(Some(&state1_fwd))
3176 .unwrap();
3177
3178 assert_eq!(
3179 second_page_fwd.len(),
3180 80,
3181 "Second forward page should return 80 entries"
3182 );
3183
3184 assert_eq!(
3185 second_page_fwd.first().unwrap().timestamp,
3186 280,
3187 "Second page should start at 280"
3188 );
3189
3190 assert_eq!(
3191 second_page_fwd.last().unwrap().timestamp,
3192 359,
3193 "Second page should end at 359"
3194 );
3195
3196 // Test backward from exact boundary timestamp 200
3197 let (first_page_bwd, state1_bwd) = LogQuery::new(&file_indexes, anchor, Direction::Backward)
3198 .with_limit(80)
3199 .execute_page(None)
3200 .unwrap();
3201
3202 assert_eq!(
3203 first_page_bwd.len(),
3204 80,
3205 "Backward from boundary should return 80 entries"
3206 );
3207
3208 // Should start at timestamp 200 (inclusive for backward)
3209 assert_eq!(
3210 first_page_bwd.first().unwrap().timestamp,
3211 200,
3212 "Backward should start at timestamp 200 (inclusive)"
3213 );
3214
3215 // Should end at timestamp 121
3216 assert_eq!(
3217 first_page_bwd.last().unwrap().timestamp,
3218 121,
3219 "Backward should end at timestamp 121"
3220 );
3221
3222 // Continue backward pagination
3223 let (second_page_bwd, _state2_bwd) = LogQuery::new(&file_indexes, anchor, Direction::Backward)
3224 .with_limit(80)
3225 .execute_page(Some(&state1_bwd))
3226 .unwrap();
3227
3228 assert_eq!(
3229 second_page_bwd.len(),
3230 21,
3231 "Second backward page should return remaining 21 entries"
3232 );
3233
3234 assert_eq!(
3235 second_page_bwd.first().unwrap().timestamp,
3236 120,
3237 "Second backward page should start at 120"
3238 );
3239
3240 assert_eq!(
3241 second_page_bwd.last().unwrap().timestamp,
3242 100,
3243 "Second backward page should end at 100"
3244 );
3245
3246 // Test anchor at boundary 300 (between file2 and file3)
3247 let anchor_300 = Anchor::Timestamp(Microseconds(300));
3248
3249 let (page_fwd_300, _) = LogQuery::new(&file_indexes, anchor_300, Direction::Forward)
3250 .with_limit(50)
3251 .execute_page(None)
3252 .unwrap();
3253
3254 assert_eq!(
3255 page_fwd_300.len(),
3256 50,
3257 "Forward from 300 should return 50 entries"
3258 );
3259
3260 assert_eq!(
3261 page_fwd_300.first().unwrap().timestamp,
3262 300,
3263 "Should start at 300"
3264 );
3265
3266 assert_eq!(
3267 page_fwd_300.last().unwrap().timestamp,
3268 349,
3269 "Should end at 349"
3270 );
3271
3272 let (page_bwd_300, _) = LogQuery::new(&file_indexes, anchor_300, Direction::Backward)
3273 .with_limit(50)
3274 .execute_page(None)
3275 .unwrap();
3276
3277 assert_eq!(
3278 page_bwd_300.len(),
3279 50,
3280 "Backward from 300 should return 50 entries"
3281 );
3282
3283 assert_eq!(
3284 page_bwd_300.first().unwrap().timestamp,
3285 300,
3286 "Should start at 300"
3287 );
3288
3289 assert_eq!(
3290 page_bwd_300.last().unwrap().timestamp,
3291 251,
3292 "Should end at 251"
3293 );
3294
3295 // Verify no duplicates within each query direction from anchor 200
3296 let mut fwd_200_ids = HashSet::new();
3297 for entry in first_page_fwd.iter().chain(second_page_fwd.iter()) {
3298 for field in &entry.fields {
3299 if field.field() == "ENTRY_ID" {
3300 assert!(
3301 fwd_200_ids.insert(field.value().to_string()),
3302 "Found duplicate in forward from 200: {}",
3303 field.value()
3304 );
3305 }
3306 }
3307 }
3308
3309 let mut bwd_200_ids = HashSet::new();
3310 for entry in first_page_bwd.iter().chain(second_page_bwd.iter()) {
3311 for field in &entry.fields {
3312 if field.field() == "ENTRY_ID" {
3313 assert!(
3314 bwd_200_ids.insert(field.value().to_string()),
3315 "Found duplicate in backward from 200: {}",
3316 field.value()
3317 );
3318 }
3319 }
3320 }
3321
3322 // Forward from 200: two pages of 80 = 160 entries (200-359)
3323 assert_eq!(
3324 fwd_200_ids.len(),
3325 160,
3326 "Forward from boundary 200 should return 160 unique entries (2 pages of 80)"
3327 );
3328
3329 // Backward from 200: 80 + 21 = 101 entries (100-200, inclusive)
3330 assert_eq!(
3331 bwd_200_ids.len(),
3332 101,
3333 "Backward from boundary 200 should return 101 unique entries"
3334 );
3335
3336 // The boundary entry (200) should appear in both forward and backward results
3337 assert!(
3338 fwd_200_ids.contains("file2_200"),
3339 "Forward should include boundary entry 200"
3340 );
3341 assert!(
3342 bwd_200_ids.contains("file2_200"),
3343 "Backward should include boundary entry 200"
3344 );
3345 }