Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #371 +/- ##
==========================================
- Coverage 71.29% 70.23% -1.06%
==========================================
Files 206 210 +4
Lines 15156 15581 +425
==========================================
+ Hits 10805 10943 +138
- Misses 3565 3837 +272
- Partials 786 801 +15 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
🔴 Performance DegradationSome benchmarks have degraded compared to the previous run. Show table
|
🔴 Performance DegradationSome benchmarks have degraded compared to the previous run. Show table
|
| Filtering Filter `config:"filtering"` | ||
| DocsFilter struct { | ||
| DataDir string `config:"data_dir"` | ||
| Concurrency int `config:"concurrency"` |
There was a problem hiding this comment.
nit: It's not possible to understand what concurrency means without reading code.
Besides, other props are named like reader_workers, replay_workers, etc
There was a problem hiding this comment.
nit: filter_workers? workers?
|
|
||
| for _, p := range params { | ||
| f := NewFilter(p) | ||
| filtersMap[string(f.Hash())] = f |
There was a problem hiding this comment.
nit: cast to string is not needed?
|
|
||
| type Params struct { | ||
| Query string | ||
| From int64 |
There was a problem hiding this comment.
nit: when migrating to nanos, I learned the hard way that variables with name From/To/Interval and type i64 are source of potential errors since you don't know if it's millis or nanos or etc. Not to say about additional cognitive load you have when look at them.
So, if nothing prevents it, the faster you convert to seq.MID the better.
| type lidsBlockHeader struct { | ||
| Codec lidsCodec | ||
| Length uint32 // Number of LIDs in block | ||
| MinLID uint32 |
There was a problem hiding this comment.
nit: haven't looked at other PR yet, but do we really need to export all fields here?
There was a problem hiding this comment.
Do you mean we could avoid storing min here and instead calculate it as the max value from the previous block?
| "github.com/ozontech/seq-db/seq" | ||
| ) | ||
|
|
||
| type ActiveLIDs struct { |
There was a problem hiding this comment.
nit: file named active_lids_map.go
| for i := range numberOfBlocks { | ||
| dst, src, err = unmarshalLIDsBlock(dst, src, headers[i]) | ||
| if err != nil { | ||
| return dst, src, err |
There was a problem hiding this comment.
nit: I noticed you return other variables even if err is not nil. Do we have a reason to do this in Go?
| CacheSizeLimit uint64 | ||
| } | ||
|
|
||
| type DocsFilter struct { |
There was a problem hiding this comment.
nit: I would personally named it FilterManager (same like we have FracManager). I think it took me ~10 minutes of scrolling through code to realize there are Filter and DocsFilter structs. They are not the same and have different scope of responsibility.
I mean, this struct manages and updates filters, but it doesn't filter anything itself.
| // RefreshFrac replaces frac's tombstone files with newly found results. Used after active frac is sealed. | ||
| func (df *DocsFilter) RefreshFrac(fraction frac.Fraction) { | ||
| df.fracsMu.RLock() | ||
| fracsFiles, has := df.fracs[fraction.Info().Name()] |
There was a problem hiding this comment.
nit: let's say I added a filter and started seq-db. Now, there might be two concurrent processes:
- running and updating a filter for active frac
- sealing same frac. Techincally, sealing is slow, but we don't know for sure. After sealing, I assume, this method
RefreshFracwill be called and update filter again.
It seems, those processes do not have any synchronization between them. Problems will probably never arise, but, it seems, to me, we need some kind of way to serialize filter update for same frac (within same filter of course).
There was a problem hiding this comment.
Oh, this is actually a design issue in the current MVP. With the active faction in place, we’re facing a number of problems, and we’ll need to figure out how to address this technical debt.
| config Config | ||
| filters map[string]*Filter | ||
|
|
||
| fracs map[string][]string |
There was a problem hiding this comment.
nit: could you explain why we have a single collections of fracs across multiple Filter? It would be intuitive for me to have such kind of collection for every Filter.
forshev
force-pushed
the
290-docs-filter
branch
from
0506ad3 to
98257ad
Compare
| diskUsage = promauto.NewGauge(prometheus.GaugeOpts{ | ||
| Namespace: "seq_db_store", | ||
| Subsystem: "filters", | ||
| Name: "disk_usage_bytes", |
There was a problem hiding this comment.
If I’m not mistaken, the Help field must be specified
| return nil | ||
| } | ||
|
|
||
| // createDataDir creates dir data lazily to avoid creating extra folders. |
There was a problem hiding this comment.
It looks like we’re creating the directory immediately at startup — not lazily, and without concurrency. In this case, using sync.Once is unnecessary.
| } | ||
| } | ||
|
|
||
| func MustWriteFileAtomic(fpath string, data []byte, tmpFileExt string) { |
There was a problem hiding this comment.
nit: It appears that the atomicWrite function in the faction manager performs the same operations. If so, should we merge these functions?
| } | ||
|
|
||
| // buildQueue creates a directory for each of unprocessed filters and creates .queue files | ||
| func (df *DocsFilter) buildQueue(fracs fracmanager.List) error { |
There was a problem hiding this comment.
This queue implementation is not fault‑tolerant: if the server crashes mid‑process, we’ll end up with an incomplete queue.
Let’s try the following approach:
- write all data to a temporary directory within the same folder;
- then atomically rename (with fsync of parent dir) the directory once the operation is complete.
| return nil | ||
| } | ||
|
|
||
| func (df *DocsFilter) maintenance() { |
There was a problem hiding this comment.
I’d use a context and a WaitGroup here to ensure all goroutines are properly terminated
See:
seq-db/fracmanager/fracmanager.go
Line 157 in 5115f7b
| defer lidsBlockBufPool.Put(b) | ||
|
|
||
| numberOfBlocks := (len(in) + maxLIDsBlockLen - 1) / maxLIDsBlockLen | ||
| dst = binary.BigEndian.AppendUint32(dst, uint32(numberOfBlocks)) |
There was a problem hiding this comment.
Historically, we’ve been using Little‑Endian across the codebase. As a result, this file is the only one that uses a different encoding, which breaks consistency in the code
| Length uint32 // Number of LIDs in block | ||
| MinLID uint32 | ||
| MaxLID uint32 | ||
| Size uint32 // Size of ids block in bytes. |
There was a problem hiding this comment.
nit: We could also store the size of the uncompressed block — this would allow more efficient memory allocation during decompression.
|
|
||
| dst[0] = byte(h.Codec) | ||
| dst = dst[1:] | ||
| binary.BigEndian.PutUint32(dst, h.Length) |
There was a problem hiding this comment.
nit: Using AppendUint32 will make the code more compact
| type lidsBlockHeader struct { | ||
| Codec lidsCodec | ||
| Length uint32 // Number of LIDs in block | ||
| MinLID uint32 |
There was a problem hiding this comment.
Do you mean we could avoid storing min here and instead calculate it as the max value from the previous block?
| // RefreshFrac replaces frac's tombstone files with newly found results. Used after active frac is sealed. | ||
| func (df *DocsFilter) RefreshFrac(fraction frac.Fraction) { | ||
| df.fracsMu.RLock() | ||
| fracsFiles, has := df.fracs[fraction.Info().Name()] |
There was a problem hiding this comment.
Oh, this is actually a design issue in the current MVP. With the active faction in place, we’re facing a number of problems, and we’ll need to figure out how to address this technical debt.
4 participants
Description
Added DocsFilter struct. Not in use yet
If you have used LLM/AI assistance please provide model name and full prompt: