99 "iter"
1010 "time"
1111
12+ "github.com/kelindar/roaring"
1213 "github.com/kelindar/tales/internal/codec"
1314 "github.com/kelindar/tales/internal/seq"
14- "github.com/weaviate/sroar"
1515)
1616
1717// queryWarm queries the in-memory buffer for entries.
@@ -60,12 +60,13 @@ func (l *Service) queryDay(ctx context.Context, actors []uint32, day time.Time,
6060
6161 // For each chunk, load all relevant bitmaps and compute intersection
6262 for _ , chunk := range meta .Chunks {
63- // Skip chunks that don't overlap with query time range
6463 if ! chunk .Between (fromSec , toSec ) {
6564 continue
6665 }
66+
6767 chunkKey := keyOfChunk (seq .FormatDate (day ), chunk .Offset ())
68- var index * sroar.Bitmap
68+
69+ var index * roaring.Bitmap
6970 for _ , a := range actors {
7071 idx , ok := chunk .Actors [a ]
7172 if ! ok || uint32 (idx [0 ]) < t0 || uint32 (idx [0 ]) > t1 {
@@ -82,14 +83,14 @@ func (l *Service) queryDay(ctx context.Context, actors []uint32, day time.Time,
8283 switch {
8384 case index == nil :
8485 index = bitmap
85- case ! index .IsEmpty () :
86+ case index .Count () > 0 :
8687 index .And (bitmap )
8788 }
8889 }
8990
9091 // Query log section with intersection bitmap
91- if index != nil && ! index .IsEmpty () {
92- if ! l .queryChunk (ctx , chunkKey , chunk , * index , day , from , to , yield ) {
92+ if index != nil && index .Count () > 0 {
93+ if ! l .queryChunk (ctx , chunkKey , chunk , index , day , from , to , yield ) {
9394 return false
9495 }
9596 }
@@ -99,7 +100,7 @@ func (l *Service) queryDay(ctx context.Context, actors []uint32, day time.Time,
99100}
100101
101102// loadBitmap downloads and decodes a single bitmap for a given index entry.
102- func (l * Service ) loadBitmap (ctx context.Context , key string , entry codec.IndexEntry ) (* sroar .Bitmap , error ) {
103+ func (l * Service ) loadBitmap (ctx context.Context , key string , entry codec.IndexEntry ) (* roaring .Bitmap , error ) {
103104 i0 := int64 (entry [1 ])
104105 i1 := i0 + int64 (entry [2 ]) - 1
105106
@@ -109,27 +110,26 @@ func (l *Service) loadBitmap(ctx context.Context, key string, entry codec.IndexE
109110 }
110111
111112 // Bitmaps are stored uncompressed, so just deserialize
112- bm := sroar . FromBuffer (data )
113+ bm := roaring . FromBytes (data )
113114 return bm , nil
114115}
115116
116117// queryChunk queries a specific log chunk for sequence IDs using an optimized bitmap iterator.
117118// This function efficiently filters log entries by leveraging the sorted nature of both the log entries
118119// and the bitmap, avoiding unnecessary bitmap lookups for entries that don't match.
119- func (l * Service ) queryChunk (ctx context.Context , chunkKey string , chunk codec.ChunkEntry , sids sroar .Bitmap , day , from , to time.Time , yield func (time.Time , string ) bool ) bool {
120- if sids .IsEmpty () {
120+ func (l * Service ) queryChunk (ctx context.Context , chunkKey string , chunk codec.ChunkEntry , sids * roaring .Bitmap , day , from , to time.Time , yield func (time.Time , string ) bool ) bool {
121+ if sids .Count () == 0 {
121122 return true
122123 }
123124
124- entries , err := l .rangeChunks (ctx , chunkKey , chunk , & sids )
125+ entries , err := l .rangeChunks (ctx , chunkKey , chunk , sids )
125126 if err != nil {
126127 return true // Skip chunks that fail to process
127128 }
128129
129130 // Process only the filtered entries
130131 for entry := range entries {
131- id := entry .ID ()
132- ts := seq .TimeOf (id , day )
132+ ts := entry .Time (day )
133133 if ! ts .Before (from ) && ! ts .After (to ) && ! yield (ts , entry .Text ()) {
134134 return false // Stop iteration
135135 }
@@ -140,8 +140,8 @@ func (l *Service) queryChunk(ctx context.Context, chunkKey string, chunk codec.C
140140
141141// rangeChunks downloads the log section from a chunk file, decompresses it, and returns
142142// an iterator over log entries that are filtered using an optimized bitmap iterator merge algorithm.
143- func (l * Service ) rangeChunks (ctx context.Context , chunkKey string , chunk codec.ChunkEntry , sids * sroar .Bitmap ) (iter.Seq [codec.LogEntry ], error ) {
144- if chunk .DataSize () == 0 || sids .IsEmpty () {
143+ func (l * Service ) rangeChunks (ctx context.Context , chunkKey string , chunk codec.ChunkEntry , sids * roaring .Bitmap ) (iter.Seq [codec.LogEntry ], error ) {
144+ if chunk .DataSize () == 0 || sids .Count () == 0 {
145145 return func (yield func (codec.LogEntry ) bool ) {}, nil // Empty iterator
146146 }
147147
@@ -159,36 +159,34 @@ func (l *Service) rangeChunks(ctx context.Context, chunkKey string, chunk codec.
159159 }
160160
161161 return func (yield func (codec.LogEntry ) bool ) {
162- iter := sids .NewIterator ()
163- idx := iter .Next ()
164- for len (buffer ) > 4 && idx != 0 {
165- entry := codec .LogEntry (buffer )
166- size := entry .Size ()
167- if size == 0 || uint32 (len (buffer )) < size {
168- return // Invalid size or not enough data, stop iteration
169- }
170-
171- // Advance bitmap iterator until we find a target >= current entry ID
172- entryID := uint64 (entry .ID ())
173- for idx != 0 && idx < entryID {
174- idx = iter .Next ()
175- }
162+ sids .Range (func (sidToFind uint32 ) bool {
163+ for len (buffer ) > 4 {
164+ entry := codec .LogEntry (buffer )
165+ size := entry .Size ()
166+ if size == 0 || uint32 (len (buffer )) < size {
167+ return false // Invalid size or not enough data, stop iteration
168+ }
176169
177- // If we've exhausted all targets, we're done
178- if idx == 0 {
179- return
180- }
170+ entryID := entry .ID ()
171+ if entryID < sidToFind {
172+ buffer = buffer [size :] // Advance buffer
173+ continue // Continue scanning buffer for current sid
174+ }
181175
182- // If current entry matches current target, yield it
183- if entryID == idx {
184- if ! yield (entry [:size ]) {
185- return // Stop iteration if yield returns false
176+ // If we found the entry, yield it. If we overshot, the entry is not
177+ // in the buffer, so we just move to the next sid.
178+ if entryID == sidToFind {
179+ if ! yield (entry [:size ]) {
180+ return false // Stop iteration if yield returns false
181+ }
182+ buffer = buffer [size :] // Advance buffer
186183 }
187- idx = iter .Next ()
188- }
189184
190- // Always advance buffer after processing each entry
191- buffer = buffer [size :]
192- }
185+ // We either found the entry and yielded it, or we've overshot.
186+ // In both cases, we are done with this sid and should get the next one.
187+ return true // Continue to next sid
188+ }
189+ return false // Buffer is exhausted, stop iteration
190+ })
193191 }, nil
194192}
0 commit comments