sorted_test.mx raw
1 package sorted
2
3 import (
4 "bytes"
5 "os"
6 "sort"
7 "testing"
8 )
9
10 // Records are 8 bytes: a 4-byte key followed by a 4-byte payload. cmpLen is 4,
11 // so the payload is invisible to ordering but must survive a round trip.
12 const (
13 tRecLen = 8
14 tCmpLen = 4
15 )
16
17 type sortedSink struct {
18 recs [][]byte
19 n int32
20 max int32 // 0 = unlimited; otherwise stop after max records
21 }
22
23 func newSortedSink() (s *sortedSink) {
24 return &sortedSink{recs: [][]byte{:64}}
25 }
26
27 func (s *sortedSink) add(rec []byte) (ok bool) {
28 if s.max > 0 && s.n >= s.max {
29 return false
30 }
31 if s.n >= int32(len(s.recs)) {
32 return false
33 }
34 cp := []byte{:len(rec)}
35 copy(cp, rec)
36 s.recs[s.n] = cp
37 s.n++
38 return true
39 }
40
41 func srRec(key byte, payload byte) (r []byte) {
42 r = []byte{:tRecLen}
43 r[0], r[1], r[2], r[3] = key, key, key, key
44 r[4], r[5], r[6], r[7] = payload, payload, payload, payload
45 return
46 }
47
48 // srBound is a scan boundary: exactly cmpLen bytes, as Scan's comparisons are
49 // on the full key slice it is handed, not on key[:cmpLen].
50 func srBound(key byte) (r []byte) {
51 r = []byte{:tCmpLen}
52 r[0], r[1], r[2], r[3] = key, key, key, key
53 return
54 }
55
56 func srKey(k0, k1, k2, k3 byte, payload byte) (r []byte) {
57 r = []byte{:tRecLen}
58 r[0], r[1], r[2], r[3] = k0, k1, k2, k3
59 r[4], r[5], r[6], r[7] = payload, payload, payload, payload
60 return
61 }
62
63 func srTmp(t *testing.T) (dir string, ok bool) {
64 t.Helper()
65 d, err := os.MkdirTemp("", "sorted-*")
66 if err != nil {
67 t.Fatal(err)
68 return "", false
69 }
70 return d, true
71 }
72
73 func srOpen(t *testing.T, path string) (f *File, ok bool) {
74 t.Helper()
75 g, err := Open(path, tRecLen, tCmpLen)
76 if err != nil {
77 t.Fatal(err)
78 return nil, false
79 }
80 return g, true
81 }
82
83 func TestEmptyFile(t *testing.T) {
84 dir, ok := srTmp(t)
85 if !ok {
86 return
87 }
88 defer os.RemoveAll(dir)
89
90 f, ok2 := srOpen(t, dir|"/idx.dat")
91 if !ok2 {
92 return
93 }
94 defer f.Close()
95
96 if f.Count() != 0 {
97 t.Fatalf("Count = %d", f.Count())
98 }
99 if f.Dirty() {
100 t.Fatal("a fresh file is not dirty")
101 }
102 key := srRec(1, 0)
103 if _, got := f.Get(key); got {
104 t.Fatal("Get on an empty file")
105 }
106 if _, got := f.GetPrefix(key[:2]); got {
107 t.Fatal("GetPrefix on an empty file")
108 }
109 if _, got := f.Last(); got {
110 t.Fatal("Last on an empty file")
111 }
112 if f.LastFlushedSer() != 0 {
113 t.Fatal("LastFlushedSer on a fresh file")
114 }
115 sink := newSortedSink()
116 f.Scan(srBound(0), srBound(255), sink.add)
117 if sink.n != 0 {
118 t.Fatalf("Scan on an empty file yielded %d", sink.n)
119 }
120 if err := f.Flush(); err != nil {
121 t.Fatal(err)
122 return
123 }
124 if err := f.Clear(); err != nil {
125 t.Fatal(err)
126 return
127 }
128 }
129
130 func TestPutGetScan(t *testing.T) {
131 dir, ok := srTmp(t)
132 if !ok {
133 return
134 }
135 defer os.RemoveAll(dir)
136
137 f, ok2 := srOpen(t, dir|"/idx.dat")
138 if !ok2 {
139 return
140 }
141 defer f.Close()
142
143 f.Put(srRec(3, 0x33))
144 f.Put(srRec(1, 0x11))
145 f.Put(srRec(2, 0x22))
146 if f.Count() != 3 {
147 t.Fatalf("Count = %d", f.Count())
148 }
149 if !f.Dirty() {
150 t.Fatal("unflushed records must make the file dirty")
151 }
152
153 rec, found := f.Get(srRec(2, 0x99))
154 if !found {
155 t.Fatal("Get key 2")
156 }
157 if rec[4] != 0x22 {
158 t.Fatalf("Get returned payload %d", rec[4])
159 }
160 if _, got := f.Get(srRec(9, 0)); got {
161 t.Fatal("Get of an absent key")
162 }
163
164 last, lok := f.Last()
165 if !lok || last[0] != 3 {
166 t.Fatal("Last must be the largest key")
167 }
168
169 sink := newSortedSink()
170 f.Scan(srBound(0), srBound(255), sink.add)
171 if sink.n != 3 {
172 t.Fatalf("Scan count = %d", sink.n)
173 }
174 if sink.recs[0][0] != 1 || sink.recs[1][0] != 2 || sink.recs[2][0] != 3 {
175 t.Fatal("Scan must return ascending keys")
176 }
177 if sink.recs[1][4] != 0x22 {
178 t.Fatal("Scan lost a payload")
179 }
180
181 if err := f.Flush(); err != nil {
182 t.Fatal(err)
183 return
184 }
185 if f.Dirty() {
186 t.Fatal("file still dirty after Flush")
187 }
188 if f.Count() != 3 {
189 t.Fatalf("Count after Flush = %d", f.Count())
190 }
191 diskRec, dfound := f.Get(srRec(2, 0))
192 if !dfound || diskRec[4] != 0x22 {
193 t.Fatal("Get from disk after Flush")
194 }
195 }
196
197 func TestReopenPersists(t *testing.T) {
198 dir, ok := srTmp(t)
199 if !ok {
200 return
201 }
202 defer os.RemoveAll(dir)
203
204 f1, ok2 := srOpen(t, dir|"/idx.dat")
205 if !ok2 {
206 return
207 }
208 f1.Put(srRec(1, 0x0A))
209 f1.Put(srRec(2, 0x0B))
210 if err := f1.Flush(); err != nil {
211 t.Fatal(err)
212 return
213 }
214 if err := f1.Close(); err != nil {
215 t.Fatal(err)
216 return
217 }
218
219 f2, ok3 := srOpen(t, dir|"/idx.dat")
220 if !ok3 {
221 return
222 }
223 defer f2.Close()
224 if f2.Count() != 2 {
225 t.Fatalf("reopened Count = %d", f2.Count())
226 }
227 rec, found := f2.Get(srRec(1, 0))
228 if !found || rec[4] != 0x0A {
229 t.Fatal("reopened Get key 1")
230 }
231 last, lok := f2.Last()
232 if !lok || last[0] != 2 {
233 t.Fatal("reopened Last")
234 }
235 sink := newSortedSink()
236 f2.Scan(srBound(0), srBound(255), sink.add)
237 if sink.n != 2 {
238 t.Fatalf("reopened Scan count = %d", sink.n)
239 }
240 }
241
242 // TestSidecarAcrossHandles exercises the .buf sidecar path: one handle flushes
243 // records that a second handle opened earlier can only see by re-reading the
244 // sidecar size.
245 func TestSidecarAcrossHandles(t *testing.T) {
246 dir, ok := srTmp(t)
247 if !ok {
248 return
249 }
250 defer os.RemoveAll(dir)
251
252 f1, ok2 := srOpen(t, dir|"/idx.dat")
253 if !ok2 {
254 return
255 }
256 f2, ok3 := srOpen(t, dir|"/idx.dat")
257 if !ok3 {
258 f1.Close()
259 return
260 }
261 defer f1.Close()
262 defer f2.Close()
263
264 f1.Put(srRec(5, 0x55))
265 needs, qerr := f1.QuickFlush(1, false)
266 if qerr != nil {
267 t.Fatal(qerr)
268 return
269 }
270 if needs {
271 t.Fatal("a small sidecar must not ask for a merge")
272 }
273 if f1.LastFlushedSer() != 1 {
274 t.Fatalf("LastFlushedSer = %d", f1.LastFlushedSer())
275 }
276 if f1.Count() != 1 {
277 t.Fatalf("writer Count = %d", f1.Count())
278 }
279 // The reader sees the sidecar record; Get refreshes the cached sidecar
280 // size, so Count is accurate after it.
281 rec, found := f2.Get(srRec(5, 0))
282 if !found || rec[4] != 0x55 {
283 t.Fatal("reader Get from sidecar")
284 }
285 if f2.Count() != 1 {
286 t.Fatalf("reader Count = %d", f2.Count())
287 }
288 sink := newSortedSink()
289 f2.Scan(srBound(0), srBound(255), sink.add)
290 if sink.n != 1 || sink.recs[0][0] != 5 {
291 t.Fatalf("reader Scan count = %d", sink.n)
292 }
293
294 // A second flush grows the sidecar; the reader's cached size refreshes.
295 f1.Put(srRec(6, 0x66))
296 if _, qerr2 := f1.QuickFlush(2, false); qerr2 != nil {
297 t.Fatal(qerr2)
298 return
299 }
300 rec2, found2 := f2.Get(srRec(6, 0))
301 if !found2 || rec2[4] != 0x66 {
302 t.Fatal("reader Get after a second sidecar flush")
303 }
304 sink2 := newSortedSink()
305 f2.Scan(srBound(0), srBound(255), sink2.add)
306 if sink2.n != 2 || sink2.recs[0][0] != 5 || sink2.recs[1][0] != 6 {
307 t.Fatalf("reader Scan after a second flush = %d", sink2.n)
308 }
309 }
310
311 // TestMixedSourcesDedup has one record on disk, one in the sidecar and two in
312 // memory, with a duplicate key in two sources. Scan must merge all three
313 // sources in order and emit a duplicated key once.
314 func TestMixedSourcesDedup(t *testing.T) {
315 dir, ok := srTmp(t)
316 if !ok {
317 return
318 }
319 defer os.RemoveAll(dir)
320
321 f, ok2 := srOpen(t, dir|"/idx.dat")
322 if !ok2 {
323 return
324 }
325 defer f.Close()
326
327 f.Put(srRec(1, 0x01))
328 if err := f.Flush(); err != nil {
329 t.Fatal(err)
330 return
331 }
332 f.Put(srRec(2, 0x02))
333 if _, qerr := f.QuickFlush(2, false); qerr != nil {
334 t.Fatal(qerr)
335 return
336 }
337 f.Put(srRec(3, 0x03))
338 f.Put(srRec(1, 0x99)) // duplicate key now in the memory buffer
339
340 sink := newSortedSink()
341 f.Scan(srBound(0), srBound(255), sink.add)
342 if sink.n != 3 {
343 t.Fatalf("merged Scan count = %d", sink.n)
344 }
345 if sink.recs[0][0] != 1 || sink.recs[1][0] != 2 || sink.recs[2][0] != 3 {
346 t.Fatal("merged Scan not ascending")
347 }
348
349 if err := f.Flush(); err != nil {
350 t.Fatal(err)
351 return
352 }
353 if f.Count() != 3 {
354 t.Fatalf("Count after merge = %d", f.Count())
355 }
356 }
357
358 func TestDelete(t *testing.T) {
359 dir, ok := srTmp(t)
360 if !ok {
361 return
362 }
363 defer os.RemoveAll(dir)
364
365 f, ok2 := srOpen(t, dir|"/idx.dat")
366 if !ok2 {
367 return
368 }
369 defer f.Close()
370
371 f.Put(srRec(1, 0x01))
372 f.Put(srRec(2, 0x02))
373 f.Put(srRec(3, 0x03))
374 if err := f.Flush(); err != nil {
375 t.Fatal(err)
376 return
377 }
378 delKey := srRec(2, 0)
379 f.Delete(delKey)
380 if !f.Dirty() {
381 t.Fatal("a pending delete must make the file dirty")
382 }
383 sink := newSortedSink()
384 f.Scan(srBound(0), srBound(255), sink.add)
385 if sink.n != 2 || sink.recs[0][0] != 1 || sink.recs[1][0] != 3 {
386 t.Fatalf("Scan after Delete count = %d", sink.n)
387 }
388
389 // Before the Flush compacts the file, the tombstone has to hide the record
390 // from a point lookup as well: Get read the on-disk record and never
391 // consulted the delete set, so a deleted key was still readable here while
392 // Scan already hid it.
393 if _, stillThere := f.Get(delKey); stillThere {
394 t.Fatal("deleted key still readable before Flush")
395 }
396 if _, keep := f.Get(srRec(1, 0)); !keep {
397 t.Fatal("Get lost a live key while skipping a tombstone")
398 }
399
400 if err := f.Flush(); err != nil {
401 t.Fatal(err)
402 return
403 }
404 if _, found := f.Get(delKey); found {
405 t.Fatal("deleted key still readable after Flush")
406 }
407 sink2 := newSortedSink()
408 f.Scan(srBound(0), srBound(255), sink2.add)
409 if sink2.n != 2 {
410 t.Fatalf("Scan after delete Flush count = %d", sink2.n)
411 }
412 last, lok := f.Last()
413 if !lok || last[0] != 3 {
414 t.Fatal("Last after delete")
415 }
416 }
417
418 func TestQuickFlushEmptyAndThreshold(t *testing.T) {
419 dir, ok := srTmp(t)
420 if !ok {
421 return
422 }
423 defer os.RemoveAll(dir)
424
425 f, ok2 := srOpen(t, dir|"/idx.dat")
426 if !ok2 {
427 return
428 }
429 defer f.Close()
430
431 needs, err := f.QuickFlush(7, false)
432 if err != nil {
433 t.Fatal(err)
434 return
435 }
436 if needs {
437 t.Fatal("an empty flush must not ask for a merge")
438 }
439 if f.LastFlushedSer() != 7 {
440 t.Fatalf("LastFlushedSer = %d", f.LastFlushedSer())
441 }
442 f.Put(srRec(1, 0x01))
443 needs2, err2 := f.QuickFlush(9, true)
444 if err2 != nil {
445 t.Fatal(err2)
446 return
447 }
448 if needs2 {
449 t.Fatal("below the 1MB threshold no merge is needed")
450 }
451 if f.LastFlushedSer() != 9 {
452 t.Fatalf("LastFlushedSer after flush = %d", f.LastFlushedSer())
453 }
454 if f.Count() != 1 {
455 t.Fatalf("Count after sidecar flush = %d", f.Count())
456 }
457 }
458
459 func TestScanBoundsInclusive(t *testing.T) {
460 dir, ok := srTmp(t)
461 if !ok {
462 return
463 }
464 defer os.RemoveAll(dir)
465
466 f, ok2 := srOpen(t, dir|"/idx.dat")
467 if !ok2 {
468 return
469 }
470 defer f.Close()
471
472 for k := 1; k <= 5; k++ {
473 f.Put(srRec(byte(k), byte(k)))
474 }
475 if err := f.Flush(); err != nil {
476 t.Fatal(err)
477 return
478 }
479
480 sink := newSortedSink()
481 f.Scan(srBound(2), srBound(4), sink.add)
482 if sink.n != 3 || sink.recs[0][0] != 2 || sink.recs[2][0] != 4 {
483 t.Fatalf("bounded Scan count = %d", sink.n)
484 }
485
486 sink2 := newSortedSink()
487 f.Scan(srBound(5), srBound(5), sink2.add)
488 if sink2.n != 1 || sink2.recs[0][0] != 5 {
489 t.Fatalf("single-key Scan count = %d", sink2.n)
490 }
491
492 sink3 := newSortedSink()
493 f.Scan(srBound(6), srBound(9), sink3.add)
494 if sink3.n != 0 {
495 t.Fatalf("out-of-range Scan count = %d", sink3.n)
496 }
497
498 sink4 := newSortedSink()
499 f.Scan(srBound(0), srBound(0), sink4.add)
500 if sink4.n != 0 {
501 t.Fatalf("below-range Scan count = %d", sink4.n)
502 }
503 }
504
505 func TestScanStopsOnFalse(t *testing.T) {
506 dir, ok := srTmp(t)
507 if !ok {
508 return
509 }
510 defer os.RemoveAll(dir)
511
512 f, ok2 := srOpen(t, dir|"/idx.dat")
513 if !ok2 {
514 return
515 }
516 defer f.Close()
517
518 for k := 1; k <= 4; k++ {
519 f.Put(srRec(byte(k), byte(k)))
520 }
521 if err := f.Flush(); err != nil {
522 t.Fatal(err)
523 return
524 }
525 stop := newSortedSink()
526 stop.max = 2
527 f.Scan(srBound(0), srBound(255), stop.add)
528 if stop.n != 2 || stop.recs[0][0] != 1 || stop.recs[1][0] != 2 {
529 t.Fatalf("stopped Scan count = %d", stop.n)
530 }
531 }
532
533 func TestGetPrefix(t *testing.T) {
534 dir, ok := srTmp(t)
535 if !ok {
536 return
537 }
538 defer os.RemoveAll(dir)
539
540 f, ok2 := srOpen(t, dir|"/idx.dat")
541 if !ok2 {
542 return
543 }
544 defer f.Close()
545
546 f.Put(srKey(1, 1, 0, 0, 0x0A))
547 f.Put(srKey(1, 1, 0, 1, 0x0B))
548 f.Put(srKey(1, 2, 0, 0, 0x0C))
549
550 first, found := f.GetPrefix([]byte{1, 1})
551 if !found {
552 t.Fatal("GetPrefix 1,1 in memory")
553 }
554 if first[0] != 1 || first[1] != 1 {
555 t.Fatal("GetPrefix returned a non-matching prefix")
556 }
557 second, found2 := f.GetPrefix([]byte{1, 2})
558 if !found2 || second[2] != 0 {
559 t.Fatal("GetPrefix 1,2")
560 }
561 if _, found3 := f.GetPrefix([]byte{9}); found3 {
562 t.Fatal("GetPrefix of an absent prefix")
563 }
564
565 if err := f.Flush(); err != nil {
566 t.Fatal(err)
567 return
568 }
569 diskFirst, dfound := f.GetPrefix([]byte{1, 1})
570 if !dfound || diskFirst[0] != 1 || diskFirst[1] != 1 {
571 t.Fatal("GetPrefix from disk")
572 }
573 }
574
575 func TestClearAndSkipFlush(t *testing.T) {
576 dir, ok := srTmp(t)
577 if !ok {
578 return
579 }
580 defer os.RemoveAll(dir)
581
582 f, ok2 := srOpen(t, dir|"/idx.dat")
583 if !ok2 {
584 return
585 }
586 defer f.Close()
587
588 f.Put(srRec(1, 0x01))
589 f.Put(srRec(2, 0x02))
590 if _, qerr := f.QuickFlush(3, false); qerr != nil {
591 t.Fatal(qerr)
592 return
593 }
594 if f.Count() != 2 {
595 t.Fatalf("Count before Clear = %d", f.Count())
596 }
597 if err := f.Clear(); err != nil {
598 t.Fatal(err)
599 return
600 }
601 if f.Count() != 0 || f.Dirty() {
602 t.Fatalf("Clear left count = %d", f.Count())
603 }
604 if f.Dirty() {
605 t.Fatal("Clear left the file dirty")
606 }
607 if _, found := f.Get(srRec(1, 0)); found {
608 t.Fatal("Get after Clear")
609 }
610 if _, lok := f.Last(); lok {
611 t.Fatal("Last after Clear")
612 }
613
614 f.Put(srRec(4, 0x04))
615 if !f.Dirty() {
616 t.Fatal("Put after Clear must be dirty")
617 }
618 f.SkipFlush()
619 if f.Dirty() {
620 t.Fatal("SkipFlush must drop the pending buffer")
621 }
622 }
623
624 func TestRecSorter(t *testing.T) {
625 data := []byte{}
626 data = data | srRec(3, 0x03)
627 data = data | srRec(1, 0x01)
628 data = data | srRec(2, 0x02)
629 tmp := []byte{:tRecLen}
630 rs := &recSorter{data: data, recLen: tRecLen, cmpLen: tCmpLen, tmp: tmp}
631 if rs.Len() != 3 {
632 t.Fatalf("recSorter Len = %d", rs.Len())
633 }
634 if !rs.Less(1, 0) {
635 t.Fatal("recSorter Less")
636 }
637 sort.Sort(rs)
638 if rs.data[0] != 1 || rs.data[8] != 2 || rs.data[16] != 3 {
639 t.Fatal("recSorter sort order")
640 }
641 if rs.data[4] != 0x01 || rs.data[12] != 0x02 || rs.data[20] != 0x03 {
642 t.Fatal("recSorter moved payloads with their keys")
643 }
644 }
645
646 func TestOpenMissingDirectory(t *testing.T) {
647 dir, ok := srTmp(t)
648 if !ok {
649 return
650 }
651 defer os.RemoveAll(dir)
652 if _, err := Open(dir|"/nope/idx.dat", tRecLen, tCmpLen); err == nil {
653 t.Fatal("Open in a missing directory must fail")
654 }
655 }
656
657 // TestKeyByteOrder pins that ordering is byte-lexicographic on cmpLen bytes and
658 // ignores the trailing payload.
659 func TestKeyByteOrder(t *testing.T) {
660 dir, ok := srTmp(t)
661 if !ok {
662 return
663 }
664 defer os.RemoveAll(dir)
665
666 f, ok2 := srOpen(t, dir|"/idx.dat")
667 if !ok2 {
668 return
669 }
670 defer f.Close()
671
672 f.Put(srKey(0x01, 0x00, 0x00, 0x00, 0xFF))
673 f.Put(srKey(0x00, 0xFF, 0xFF, 0xFF, 0x00))
674 if err := f.Flush(); err != nil {
675 t.Fatal(err)
676 return
677 }
678 sink := newSortedSink()
679 f.Scan(srBound(0), srBound(255), sink.add)
680 if sink.n != 2 {
681 t.Fatalf("order Scan count = %d", sink.n)
682 }
683 if sink.recs[0][0] != 0x00 || sink.recs[1][0] != 0x01 {
684 t.Fatal("byte-lexicographic order violated")
685 }
686 if !bytes.Equal(sink.recs[0], srKey(0x00, 0xFF, 0xFF, 0xFF, 0x00)) {
687 t.Fatal("payload not carried with the record")
688 }
689 }
690