Skip to content

Commit 8f494ce

Browse files
draganmdraganm
committed
cache bitmaps throughout the batch processing
1 parent df296df commit 8f494ce

File tree

4 files changed

+89
-219
lines changed

4 files changed

+89
-219
lines changed

‎bitmap_cache.go‎

Lines changed: 70 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ func (c *bitmapCache) AddToStringBitmap(ctx context.Context, name string, value
3535
bitmap, err = c.st.GetStringAttributeValueBitmap(ctx, store.GetStringAttributeValueBitmapParams{Name: name, Value: value})
3636

3737
if err != nil && err != sql.ErrNoRows {
38-
return fmt.Errorf("failed to get numeric attribute %q value %q bitmap: %w", name, value, err)
38+
return fmt.Errorf("failed to get string attribute %q value %q bitmap: %w", name, value, err)
3939
}
4040

4141
if bitmap == nil {
@@ -51,6 +51,29 @@ func (c *bitmapCache) AddToStringBitmap(ctx context.Context, name string, value
5151

5252
}
5353

54+
func (c *bitmapCache) RemoveFromStringBitmap(ctx context.Context, name string, value string, id uint64) (err error) {
55+
k := nameValue[string]{name: name, value: value}
56+
bitmap, ok := c.stringBitmaps[k]
57+
if !ok {
58+
bitmap, err = c.st.GetStringAttributeValueBitmap(ctx, store.GetStringAttributeValueBitmapParams{Name: name, Value: value})
59+
60+
if err != nil && err != sql.ErrNoRows {
61+
return fmt.Errorf("failed to get string attribute %q value %q bitmap: %w", name, value, err)
62+
}
63+
64+
if bitmap == nil {
65+
bitmap = store.NewBitmap()
66+
}
67+
68+
c.stringBitmaps[k] = bitmap
69+
}
70+
71+
bitmap.Remove(id)
72+
73+
return nil
74+
75+
}
76+
5477
func (c *bitmapCache) AddToNumericBitmap(ctx context.Context, name string, value uint64, id uint64) (err error) {
5578
k := nameValue[uint64]{name: name, value: value}
5679
bitmap, ok := c.numericBitmaps[k]
@@ -74,14 +97,60 @@ func (c *bitmapCache) AddToNumericBitmap(ctx context.Context, name string, value
7497

7598
}
7699

100+
func (c *bitmapCache) RemoveFromNumericBitmap(ctx context.Context, name string, value uint64, id uint64) (err error) {
101+
k := nameValue[uint64]{name: name, value: value}
102+
bitmap, ok := c.numericBitmaps[k]
103+
if !ok {
104+
bitmap, err = c.st.GetNumericAttributeValueBitmap(ctx, store.GetNumericAttributeValueBitmapParams{Name: name, Value: value})
105+
106+
if err != nil && err != sql.ErrNoRows {
107+
return fmt.Errorf("failed to get numeric attribute %q value %q bitmap: %w", name, value, err)
108+
}
109+
110+
if bitmap == nil {
111+
bitmap = store.NewBitmap()
112+
}
113+
114+
c.numericBitmaps[k] = bitmap
115+
}
116+
117+
bitmap.Remove(id)
118+
119+
return nil
120+
121+
}
122+
77123
func (c *bitmapCache) Flush(ctx context.Context) (err error) {
78124
for k, bitmap := range c.stringBitmaps {
125+
126+
if bitmap.IsEmpty() {
127+
err = c.st.DeleteStringAttributeValueBitmap(ctx, store.DeleteStringAttributeValueBitmapParams{Name: k.name, Value: k.value})
128+
if err != nil {
129+
return fmt.Errorf("failed to delete string attribute %q value %q bitmap: %w", k.name, k.value, err)
130+
}
131+
continue
132+
}
133+
134+
bitmap.RunOptimize()
135+
79136
err = c.st.UpsertStringAttributeValueBitmap(ctx, store.UpsertStringAttributeValueBitmapParams{Name: k.name, Value: k.value, Bitmap: bitmap})
80137
if err != nil {
81138
return fmt.Errorf("failed to upsert string attribute %q value %q bitmap: %w", k.name, k.value, err)
82139
}
83140
}
141+
84142
for k, bitmap := range c.numericBitmaps {
143+
144+
if bitmap.IsEmpty() {
145+
err = c.st.DeleteNumericAttributeValueBitmap(ctx, store.DeleteNumericAttributeValueBitmapParams{Name: k.name, Value: k.value})
146+
if err != nil {
147+
return fmt.Errorf("failed to delete numeric attribute %q value %q bitmap: %w", k.name, k.value, err)
148+
}
149+
continue
150+
}
151+
152+
bitmap.RunOptimize()
153+
85154
err = c.st.UpsertNumericAttributeValueBitmap(ctx, store.UpsertNumericAttributeValueBitmapParams{Name: k.name, Value: k.value, Bitmap: bitmap})
86155
if err != nil {
87156
return fmt.Errorf("failed to upsert numeric attribute %q value %q bitmap: %w", k.name, k.value, err)

‎numeric_bitmap_ops.go‎

Lines changed: 0 additions & 95 deletions
This file was deleted.

‎sqlitestore.go‎

Lines changed: 19 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,8 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
127127
return fmt.Errorf("failed to get last block from database: %w", err)
128128
}
129129

130+
cache := newBitmapCache(st)
131+
130132
mainLoop:
131133
for _, block := range batch.Batch.Blocks {
132134

@@ -193,17 +195,13 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
193195
return fmt.Errorf("failed to insert payload %s at block %d txIndex %d opIndex %d: %w", key.Hex(), block.Number, operation.TxIndex, operation.OpIndex, err)
194196
}
195197

196-
sbo := newStringBitmapOps(st)
197-
198198
for k, v := range stringAttributes {
199-
err = sbo.Add(ctx, k, v, id)
199+
err = cache.AddToStringBitmap(ctx, k, v, id)
200200
if err != nil {
201201
return fmt.Errorf("failed to add string attribute value bitmap: %w", err)
202202
}
203203
}
204204

205-
nbo := newNumericBitmapOps(st)
206-
207205
for k, v := range numericAttributes {
208206

209207
// skip txIndex and opIndex because they are not used for querying
@@ -212,7 +210,7 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
212210
continue
213211
}
214212

215-
err = nbo.Add(ctx, k, v, id)
213+
err = cache.AddToNumericBitmap(ctx, k, v, id)
216214
if err != nil {
217215
return fmt.Errorf("failed to add numeric attribute value bitmap: %w", err)
218216
}
@@ -268,25 +266,21 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
268266
return fmt.Errorf("failed to insert payload 0x%x at block %d txIndex %d opIndex %d: %w", key, block.Number, operation.TxIndex, operation.OpIndex, err)
269267
}
270268

271-
sbo := newStringBitmapOps(st)
272-
273269
for k, v := range oldStringAttributes.Values {
274-
err = sbo.Remove(ctx, k, v, id)
270+
err = cache.RemoveFromStringBitmap(ctx, k, v, id)
275271
if err != nil {
276272
return fmt.Errorf("failed to remove string attribute value bitmap: %w", err)
277273
}
278274
}
279275

280-
nbo := newNumericBitmapOps(st)
281-
282276
for k, v := range oldNumericAttributes.Values {
283277
// skip txIndex and opIndex because they are not used for querying
284278
switch k {
285279
case "$txIndex", "$opIndex":
286280
continue
287281
}
288282

289-
err = nbo.Remove(ctx, k, v, id)
283+
err = cache.RemoveFromNumericBitmap(ctx, k, v, id)
290284
if err != nil {
291285
return fmt.Errorf("failed to remove numeric attribute value bitmap: %w", err)
292286
}
@@ -295,7 +289,7 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
295289
// TODO: delete entity from the indexes
296290

297291
for k, v := range stringAttributes {
298-
err = sbo.Add(ctx, k, v, id)
292+
err = cache.AddToStringBitmap(ctx, k, v, id)
299293
if err != nil {
300294
return fmt.Errorf("failed to add string attribute value bitmap: %w", err)
301295
}
@@ -308,7 +302,7 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
308302
continue
309303
}
310304

311-
err = nbo.Add(ctx, k, v, id)
305+
err = cache.AddToNumericBitmap(ctx, k, v, id)
312306
if err != nil {
313307
return fmt.Errorf("failed to add numeric attribute value bitmap: %w", err)
314308
}
@@ -333,25 +327,21 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
333327

334328
oldNumericAttributes := latestPayload.NumericAttributes
335329

336-
sbo := newStringBitmapOps(st)
337-
338330
for k, v := range oldStringAttributes.Values {
339-
err = sbo.Remove(ctx, k, v, latestPayload.ID)
331+
err = cache.RemoveFromStringBitmap(ctx, k, v, latestPayload.ID)
340332
if err != nil {
341333
return fmt.Errorf("failed to remove string attribute value bitmap: %w", err)
342334
}
343335
}
344336

345-
nbo := newNumericBitmapOps(st)
346-
347337
for k, v := range oldNumericAttributes.Values {
348338
// skip txIndex and opIndex because they are not used for querying
349339
switch k {
350340
case "$txIndex", "$opIndex":
351341
continue
352342
}
353343

354-
err = nbo.Remove(ctx, k, v, latestPayload.ID)
344+
err = cache.RemoveFromNumericBitmap(ctx, k, v, latestPayload.ID)
355345
if err != nil {
356346
return fmt.Errorf("failed to remove numeric attribute value bitmap: %w", err)
357347
}
@@ -393,14 +383,12 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
393383
return fmt.Errorf("failed to insert payload at block %d txIndex %d opIndex %d: %w", block.Number, operation.TxIndex, operation.OpIndex, err)
394384
}
395385

396-
nbo := newNumericBitmapOps(st)
397-
398-
err = nbo.Remove(ctx, "$expiration", oldExpiration, id)
386+
err = cache.RemoveFromNumericBitmap(ctx, "$expiration", oldExpiration, id)
399387
if err != nil {
400388
return fmt.Errorf("failed to remove numeric attribute value bitmap: %w", err)
401389
}
402390

403-
err = nbo.Add(ctx, "$expiration", newToBlock, id)
391+
err = cache.AddToNumericBitmap(ctx, "$expiration", newToBlock, id)
404392
if err != nil {
405393
return fmt.Errorf("failed to add numeric attribute value bitmap: %w", err)
406394
}
@@ -436,14 +424,12 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
436424
return fmt.Errorf("failed to insert payload at block %d txIndex %d opIndex %d: %w", block.Number, operation.TxIndex, operation.OpIndex, err)
437425
}
438426

439-
sbo := newStringBitmapOps(st)
440-
441-
err = sbo.Remove(ctx, "$owner", oldOwner, id)
427+
err = cache.RemoveFromStringBitmap(ctx, "$owner", oldOwner, id)
442428
if err != nil {
443429
return fmt.Errorf("failed to remove string attribute value bitmap for owner: %w", err)
444430
}
445431

446-
err = sbo.Add(ctx, "$owner", newOwner, id)
432+
err = cache.AddToStringBitmap(ctx, "$owner", newOwner, id)
447433
if err != nil {
448434
return fmt.Errorf("failed to add string attribute value bitmap for owner: %w", err)
449435
}
@@ -463,6 +449,11 @@ func (s *SQLiteStore) FollowEvents(ctx context.Context, iterator arkivevents.Bat
463449
return fmt.Errorf("failed to upsert last block: %w", err)
464450
}
465451

452+
err = cache.Flush(ctx)
453+
if err != nil {
454+
return fmt.Errorf("failed to flush bitmap cache: %w", err)
455+
}
456+
466457
err = tx.Commit()
467458
if err != nil {
468459
return fmt.Errorf("failed to commit transaction: %w", err)

0 commit comments

Comments
 (0)