feat: use DocsFilter

cache/cache.go

@@ -126,6 +126,20 @@ func (c *Cache[V]) Cleanup() uint64 {
 	return totalFreed
 }
 
+func (c *Cache[V]) Evict(key uint32) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	e, ok := c.payload[key]
+	if !ok {
+		// no key in cache
+		return
+	}
+
+	delete(c.payload, key)
+	e.deleted = true
+}
+
 // Recreates the payload map. If len is too small, fraction is probably out of date and useless
 func (c *Cache[V]) recreatePayload() {
 	if c.maxPayloadSize < recreateThreshold { // not large enough

cmd/seq-db/seq-db.go

@@ -33,6 +33,8 @@ import (
 	"github.com/ozontech/seq-db/proxy/search"
 	"github.com/ozontech/seq-db/proxy/stores"
 	"github.com/ozontech/seq-db/proxyapi"
+	"github.com/ozontech/seq-db/seq"
+	"github.com/ozontech/seq-db/skipmaskmanager"
 	"github.com/ozontech/seq-db/storage/s3"
 	"github.com/ozontech/seq-db/storeapi"
 	"github.com/ozontech/seq-db/tracing"
@@ -316,10 +318,15 @@ func startStore(
 				From:  cfg.Filtering.From,
 			},
 		},
+		SkipMaskManagerConfig: skipmaskmanager.Config{
+			DataDir:        cfg.SkipMaskManager.DataDir,
+			Workers:        cfg.SkipMaskManager.Workers,
+			CacheSizeLimit: uint64(cfg.SkipMaskManager.CacheSize),
+		},
 	}
 
 	s3cli := initS3Client(cfg)
-	store, err := storeapi.NewStore(ctx, sconfig, s3cli, mp)
+	store, err := storeapi.NewStore(ctx, sconfig, s3cli, mp, skipMaskParamsFromCfg(cfg.SkipMaskManager.SkipMasks))
 	if err != nil {
 		logger.Fatal("initializing store", zap.Error(err))
 	}
@@ -361,3 +368,15 @@ func initS3Client(cfg config.Config) *s3.Client {
 func enableIndexingForAllFields(mappingPath string) bool {
 	return mappingPath == "auto"
 }
+
+func skipMaskParamsFromCfg(in []config.SkipMaskParams) []skipmaskmanager.SkipMaskParams {
+	out := make([]skipmaskmanager.SkipMaskParams, 0, len(in))
+	for _, f := range in {
+		out = append(out, skipmaskmanager.SkipMaskParams{
+			Query: f.Query,
+			From:  seq.TimeToMID(f.From),
+			To:    seq.TimeToMID(f.To),
+		})
+	}
+	return out
+}

config/config.go

@@ -277,13 +277,14 @@ type Config struct {
 	} `config:"tracing"`
 
 	// Additional filtering options
-	Filtering struct {
-		// If a search query time range overlaps with the [from; to] range
-		// the search query will be `AND`-ed with an additional predicate with the provided query expression
-		Query string    `config:"query"`
-		From  time.Time `config:"from"`
-		To    time.Time `config:"to"`
-	} `config:"filtering"`
+	Filtering SkipMaskParams `config:"filtering"`
+
+	SkipMaskManager struct {
+		DataDir   string           `config:"data_dir"`
+		Workers   int              `config:"workers" default:"1"`
+		SkipMasks []SkipMaskParams `config:"skip_masks"`
+		CacheSize Bytes            `config:"cache_size" default:"100MiB"`
+	} `config:"skip_mask_manager"`
 
 	// Experimental provides flags
 	// For configuring experimental features.
@@ -295,6 +296,12 @@ type Config struct {
 	} `config:"experimental"`
 }
 
+type SkipMaskParams struct {
+	Query string    `config:"query"`
+	From  time.Time `config:"from"`
+	To    time.Time `config:"to"`
+}
+
 type Bytes units.Base2Bytes
 
 func (b *Bytes) UnmarshalString(s string) error {