diff --git a/cmd/fraction/decoding_suites_test.go b/cmd/fraction/decoding_suites_test.go index cf456395a..d914151c1 100644 --- a/cmd/fraction/decoding_suites_test.go +++ b/cmd/fraction/decoding_suites_test.go @@ -28,6 +28,8 @@ type FractionDecoderTestSuite struct { fracName string expectedVer config.BinaryDataVersion + // sealedWithAllToken is true for fractions sealed before token _all_ removal + sealedWithAllToken bool } func (s *FractionDecoderTestSuite) sealDocs(docsPerDocBlock int) { @@ -40,6 +42,7 @@ func (s *FractionDecoderTestSuite) sealDocs(docsPerDocBlock int) { ) s.Require().NoError(err) s.expectedVer = config.CurrentFracVersion + s.sealedWithAllToken = false } func (s *FractionDecoderTestSuite) sealDocsVersion(v config.BinaryDataVersion) { @@ -51,6 +54,14 @@ func (s *FractionDecoderTestSuite) sealDocsVersion(v config.BinaryDataVersion) { s.Require().NoError(err, "seal-fraction.sh failed:\n%s", out) s.expectedVer = v + s.sealedWithAllToken = true +} + +func expectedTokens(withAllToken bool) []tokenRecord { + if !withAllToken { + return testTokens + } + return append([]tokenRecord{{Field: "_all_", Token: "", Freq: 3}}, testTokens...) } func (s *FractionDecoderTestSuite) decodeAll() *collectedContent { @@ -101,14 +112,15 @@ func (s *FractionDecoderTestSuite) checkIDs() { func (s *FractionDecoderTestSuite) checkTokens() { content := s.decodeAll() + expected := expectedTokens(s.sealedWithAllToken) - s.Require().Len(content.Tokens, len(testTokens)) + s.Require().Len(content.Tokens, len(expected)) for i, tok := range content.Tokens { - expected := testTokens[i] - expected.TID = uint32(i + 1) - expected.Kind = kindToken - expected.LIDs = tok.LIDs - s.Equal(expected, tok) + want := expected[i] + want.TID = uint32(i + 1) + want.Kind = kindToken + want.LIDs = tok.LIDs + s.Equal(want, tok) s.Len(tok.LIDs, int(tok.Freq)) for _, lid := range tok.LIDs { @@ -123,10 +135,11 @@ func (s *FractionDecoderTestSuite) checkTokensWithoutPostings() { err := decodeFraction(s.fracName, kindToken, content) s.Require().NoError(err) - s.Require().Len(content.Tokens, len(testTokens)) + expected := expectedTokens(s.sealedWithAllToken) + s.Require().Len(content.Tokens, len(expected)) for i, tok := range content.Tokens { - s.Equal(testTokens[i].Token, tok.Token) - s.Equal(testTokens[i].Freq, tok.Freq) + s.Equal(expected[i].Token, tok.Token) + s.Equal(expected[i].Freq, tok.Freq) s.Empty(tok.LIDs) } } diff --git a/cmd/fraction/sealing_test.go b/cmd/fraction/sealing_test.go index ae33f2ed7..d38a9c754 100644 --- a/cmd/fraction/sealing_test.go +++ b/cmd/fraction/sealing_test.go @@ -120,7 +120,6 @@ const testDocs = ` ` var testTokens = []tokenRecord{ - {Field: "_all_", Token: "", Freq: 3}, {Field: "_exists_", Token: "level", Freq: 3}, {Field: "_exists_", Token: "message", Freq: 3}, {Field: "_exists_", Token: "service", Freq: 3}, @@ -218,14 +217,13 @@ func TestSealCustomMapping(t *testing.T) { `{"timestamp":"2024-01-01T10:00:00.000Z","request_uri":"/api/v1/users"}`, ) - require.Len(t, content.Tokens, 5) - assert.Equal(t, "_all_", content.Tokens[0].Field) - assert.Equal(t, "_exists_", content.Tokens[1].Field) - assert.Equal(t, "request_uri", content.Tokens[1].Token) - assert.Equal(t, "request_uri", content.Tokens[2].Field) - assert.Equal(t, "/api", content.Tokens[2].Token) - assert.Equal(t, "/api/v1", content.Tokens[3].Token) - assert.Equal(t, "/api/v1/users", content.Tokens[4].Token) + require.Len(t, content.Tokens, 4) + assert.Equal(t, "_exists_", content.Tokens[0].Field) + assert.Equal(t, "request_uri", content.Tokens[0].Token) + assert.Equal(t, "request_uri", content.Tokens[1].Field) + assert.Equal(t, "/api", content.Tokens[1].Token) + assert.Equal(t, "/api/v1", content.Tokens[2].Token) + assert.Equal(t, "/api/v1/users", content.Tokens[3].Token) } func TestSealMissingMapping(t *testing.T) { diff --git a/frac/active_indexer.go b/frac/active_indexer.go index b60a63313..c02026fac 100644 --- a/frac/active_indexer.go +++ b/frac/active_indexer.go @@ -34,6 +34,8 @@ type mergeTask struct { tokenLIDs *TokenLIDs } +const minMergeQueue = 10000 + func NewActiveIndexer(workerCount, chLen int) (*ActiveIndexer, func()) { idx := ActiveIndexer{ ch: make(chan *indexTask, chLen), @@ -176,6 +178,12 @@ func (ai *ActiveIndexer) appendWorker(index int) { m = sw.Start("put_lids_queue") tokensToMerge := addLIDsToTokens(tokenLIDsPlaces, groups) + // _all_ no longer comes in document metas, so it has no group in the + // collector; instead its posting list is fed directly with the LIDs + // just assigned to this batch, since it matches every document. + if allLIDs := active.TokenList.GetAllTokenLIDs(); allLIDs.PutLIDsInQueue(lids) > minMergeQueue { + tokensToMerge = append(tokensToMerge, allLIDs) + } ai.sendTokensToMergeWorkers(active, tokensToMerge) m.Stop() @@ -202,8 +210,6 @@ func (ai *ActiveIndexer) sendTokensToMergeWorkers(frac *Active, tokens []*TokenL } func addLIDsToTokens(tlids []*TokenLIDs, lids [][]uint32) []*TokenLIDs { - const minMergeQueue = 10000 - needMerge := make([]*TokenLIDs, 0, len(tlids)) for i, tl := range tlids { if l := tl.PutLIDsInQueue(lids[i]); l > minMergeQueue { diff --git a/frac/fraction_test.go b/frac/fraction_test.go index afe22bf36..4f6152d04 100644 --- a/frac/fraction_test.go +++ b/frac/fraction_test.go @@ -193,6 +193,62 @@ func (s *FractionTestSuite) TestSearchKeyword() { s.AssertSearch("_exists_:level", docs, []int{3, 2, 1, 0}) } +func (s *FractionTestSuite) TestSearchMatchAll() { + docs := []string{ + `{"timestamp":"2000-01-01T13:00:25Z","message":"first","level":"1"}`, + `{"timestamp":"2000-01-01T13:00:32Z","message":"second"}`, + `{"timestamp":"2000-01-01T13:00:43Z","message":"third","level":"2"}`, + `{"timestamp":"2000-01-01T13:00:55Z","not_in_mapping":"value"}`, + } + + s.insertDocuments(docs) + + s.AssertSearch("", docs, []int{3, 2, 1, 0}) + s.AssertSearch("*", docs, []int{3, 2, 1, 0}) + s.AssertSearch("_all_:*", docs, []int{3, 2, 1, 0}) + s.AssertSearch(`_all_:""`, docs, []int{3, 2, 1, 0}) + s.AssertSearch("level:1", docs, []int{0}) + s.AssertSearch("NOT message:second", docs, []int{3, 2, 0}) + + searchParams := s.query("*", withAggQuery(processor.AggQuery{ + GroupBy: aggField(seq.TokenAll), + Func: seq.AggFuncCount, + })) + s.AssertAggregation(searchParams, seq.AggregateArgs{Func: seq.AggFuncCount}, []seq.AggregationBucket{ + {Name: "", Value: float64(len(docs))}, + }) + + searchParams = s.query("*", withAggQuery(processor.AggQuery{ + GroupBy: aggField("level"), + Func: seq.AggFuncCount, + })) + s.AssertAggregation(searchParams, seq.AggregateArgs{Func: seq.AggFuncCount}, []seq.AggregationBucket{ + {Name: "1", Value: 1}, + {Name: "2", Value: 1}, + // Documents 2 and 3 have no level token in their metas at all. + {Name: "_not_exists", Value: 2}, + }) +} + +func (s *FractionTestSuite) TestSearchMatchAllDegenerateWindows() { + docs := []string{ + `{"timestamp":"2000-01-01T13:00:25Z","message":"only one"}`, + } + + s.insertDocuments(docs) + + s.AssertSearch("*", docs, []int{0}) + + s.AssertSearchWithSearchParams( + s.query("*", withFrom("2000-01-01T12:00:00Z"), withTo("2000-01-01T12:30:00Z")), + docs, []int{}, nil, + ) + s.AssertSearchWithSearchParams( + s.query("*", withFrom("2000-01-01T14:00:00Z"), withTo("2000-01-01T14:30:00Z")), + docs, []int{}, nil, + ) +} + func (s *FractionTestSuite) TestSearchNot() { docs := []string{ /*0*/ `{"timestamp":"2000-01-01T13:00:25Z","message":"bad","level":"1","service":"srv_1","status":"ok"}`, diff --git a/frac/meta_data_collector.go b/frac/meta_data_collector.go index 490f551ea..3b5caef6c 100644 --- a/frac/meta_data_collector.go +++ b/frac/meta_data_collector.go @@ -75,13 +75,11 @@ func (c *metaDataCollector) AppendMeta(m indexer.MetaData) { } c.IDs = append(c.IDs, m.ID) - c.tokensInDocs = append(c.tokensInDocs, uint32(len(m.Tokens))) + c.tokensInDocs = append(c.tokensInDocs, uint32(c.extractTokens(m.Tokens))) c.Positions = append(c.Positions, pos) c.DocsCounter++ c.SizeCounter += uint64(m.Size) - - c.extractTokens(m.Tokens) } // Filter keeps only documents at the given ascending indexes. @@ -126,9 +124,13 @@ func (c *metaDataCollector) Filter(appended []int) { c.tokensIndex = tokensIndex } -func (c *metaDataCollector) extractTokens(tokens []tokenizer.MetaToken) { +func (c *metaDataCollector) extractTokens(tokens []tokenizer.MetaToken) int { + count := 0 for _, token := range tokens { key, value := token.Key, token.Value + if string(key) == seq.TokenAll { + continue + } pos := len(c.tokensBuf) c.tokensBuf = append(c.tokensBuf, key...) c.tokensBuf = append(c.tokensBuf, ':') @@ -145,7 +147,9 @@ func (c *metaDataCollector) extractTokens(tokens []tokenizer.MetaToken) { c.FieldsLengths = append(c.FieldsLengths, len(key)) } c.tokensIndex = append(c.tokensIndex, index) + count++ } + return count } func (c *metaDataCollector) Init(blockIndex uint32) { diff --git a/frac/processor/batch_eval_tree.go b/frac/processor/batch_eval_tree.go index 421160f2c..e5613acc7 100644 --- a/frac/processor/batch_eval_tree.go +++ b/frac/processor/batch_eval_tree.go @@ -44,7 +44,7 @@ func tryBuildBatchEvalTree( } cache := make(leafTIDsCache) - cost, err := calculateQueryIterationCost(root, ti, cache) + cost, err := calculateQueryIterationCost(root, ti, cache, minLID, maxLID) if err != nil { return nil, err } @@ -68,6 +68,9 @@ func astSupportsBatching(root *parser.ASTNode) bool { switch v := root.Value.(type) { case *parser.Literal: + if parser.IsMatchAll(v) { + return true + } return len(v.Terms) == 1 && v.Terms[0].Kind == parser.TermText case *parser.Range: return true @@ -86,23 +89,34 @@ func astSupportsBatching(root *parser.ASTNode) bool { } } -func calculateQueryIterationCost(root *parser.ASTNode, ti tokenIndex, cache leafTIDsCache) (uint64, error) { +func calculateQueryIterationCost( + root *parser.ASTNode, + ti tokenIndex, + cache leafTIDsCache, + minLID, maxLID uint32, +) (uint64, error) { if root == nil { return 0, fmt.Errorf("empty AST") } switch token := root.Value.(type) { case *parser.Literal: - return leafIterationCost(ti, token.Field, token, cache) + return leafIterationCost(ti, token.Field, token, cache, minLID, maxLID) case *parser.Range: - return leafIterationCost(ti, token.Field, token, cache) + return leafIterationCost(ti, token.Field, token, cache, minLID, maxLID) case *parser.Logical: if len(root.Children) == 0 { return 0, nil } childCosts := make([]uint64, len(root.Children)) for i, child := range root.Children { - c, err := calculateQueryIterationCost(child, ti, cache) + c, err := calculateQueryIterationCost( + child, + ti, + cache, + minLID, + maxLID, + ) if err != nil { return 0, err } @@ -126,7 +140,17 @@ func calculateQueryIterationCost(root *parser.ASTNode, ti tokenIndex, cache leaf } } -func leafIterationCost(ti tokenIndex, field string, token parser.Token, cache leafTIDsCache) (uint64, error) { +func leafIterationCost( + ti tokenIndex, + field string, + token parser.Token, + cache leafTIDsCache, + minLID, maxLID uint32, +) (uint64, error) { + if parser.IsMatchAll(token) { + return uint64(maxLID - minLID + 1), nil + } + tids, err := ti.GetTIDsByTokenExpr(token) if err != nil { return 0, err @@ -203,6 +227,11 @@ func evalBatchLeaf( ) (node.BatchedNode, error) { stats.LeavesTotal++ + if parser.IsMatchAll(token) { + stats.NodesTotal++ + return node.NewRangeBatched(minLID, maxLID, order.IsDesc()), nil + } + tids, ok := cache[token] if !ok { var err error diff --git a/frac/processor/batch_eval_tree_test.go b/frac/processor/batch_eval_tree_test.go index 77b15dc6a..1a79ae776 100644 --- a/frac/processor/batch_eval_tree_test.go +++ b/frac/processor/batch_eval_tree_test.go @@ -21,6 +21,14 @@ func TestASTSupportsBatching(t *testing.T) { assert.True(t, astSupportsBatching(q.Root)) }) + for _, query := range []string{`*`, `_all_:*`, `_all_:""`} { + t.Run("match all "+query, func(t *testing.T) { + q, err := parser.ParseSeqQL(query, nil) + require.NoError(t, err) + assert.True(t, astSupportsBatching(q.Root)) + }) + } + t.Run("and of fields", func(t *testing.T) { q, err := parser.ParseSeqQL(`service:"foo" AND level:"error"`, nil) require.NoError(t, err) @@ -107,15 +115,23 @@ func TestQueryIterationCost(t *testing.T) { t.Run("leaf", func(t *testing.T) { q, err := parser.ParseSeqQL(`a:1`, nil) require.NoError(t, err) - cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache)) + cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache), 1, 1) require.NoError(t, err) assert.Equal(t, uint64(80_000), cost) }) + t.Run("match all", func(t *testing.T) { + q, err := parser.ParseSeqQL(`_all_:""`, nil) + require.NoError(t, err) + cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache), 1, 3) + require.NoError(t, err) + assert.Equal(t, uint64(3), cost) + }) + t.Run("and", func(t *testing.T) { q, err := parser.ParseSeqQL(`a:1 AND b:2`, nil) require.NoError(t, err) - cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache)) + cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache), 1, 2) require.NoError(t, err) assert.Equal(t, uint64(80_000), cost) }) @@ -123,7 +139,7 @@ func TestQueryIterationCost(t *testing.T) { t.Run("or", func(t *testing.T) { q, err := parser.ParseSeqQL(`a:1 OR b:2`, nil) require.NoError(t, err) - cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache)) + cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache), 1, 2) require.NoError(t, err) assert.Equal(t, uint64(200_000), cost) }) @@ -131,7 +147,7 @@ func TestQueryIterationCost(t *testing.T) { t.Run("and not", func(t *testing.T) { q, err := parser.ParseSeqQL(`a:1 AND NOT b:2`, nil) require.NoError(t, err) - cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache)) + cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache), 1, 2) require.NoError(t, err) assert.Equal(t, uint64(200_000), cost) }) @@ -139,7 +155,7 @@ func TestQueryIterationCost(t *testing.T) { t.Run("nested and-or", func(t *testing.T) { q, err := parser.ParseSeqQL(`a:1 AND (b:2 OR c:3)`, nil) require.NoError(t, err) - cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache)) + cost, err := calculateQueryIterationCost(q.Root, index, make(leafTIDsCache), 1, 2) require.NoError(t, err) assert.Equal(t, uint64(80_000), cost) }) diff --git a/frac/processor/eval_test.go b/frac/processor/eval_test.go index dee779f07..f153e242b 100644 --- a/frac/processor/eval_test.go +++ b/frac/processor/eval_test.go @@ -8,8 +8,11 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/ozontech/seq-db/frac/sealed/lids" + "github.com/ozontech/seq-db/metric/stopwatch" "github.com/ozontech/seq-db/node" "github.com/ozontech/seq-db/parser" + "github.com/ozontech/seq-db/seq" ) type staticProvider struct { @@ -93,3 +96,99 @@ func TestEval(t *testing.T) { assert.Equal(t, []uint32{1, 2, 4, 5, 7, 8, 11, 12}, readAll(root)) }) } + +func TestMatchAllNode(t *testing.T) { + all := func(minLID, maxLID uint32, asc bool) []uint32 { + return readAll(matchAllNode(minLID, maxLID, asc)) + } + + // In node package convention asc means LID numbers ascend, which is + // desc document order: LID 1 is the newest document. + t.Run("asc lids", func(t *testing.T) { + assert.Equal(t, []uint32{3, 4, 5}, all(3, 5, true)) + }) + t.Run("desc lids", func(t *testing.T) { + assert.Equal(t, []uint32{5, 4, 3}, all(3, 5, false)) + }) + t.Run("single lid window", func(t *testing.T) { + assert.Equal(t, []uint32{2}, all(2, 2, false)) + }) + t.Run("degenerate window", func(t *testing.T) { + // maxLID = minLID-1 happens when the query range misses the fraction. + assert.Empty(t, all(4, 3, false)) + }) +} + +type recordingProvider struct { + tokenIndex + calls []string +} + +func (p *recordingProvider) GetTIDsByTokenExpr(token parser.Token) ([]uint32, error) { + p.calls = append(p.calls, "GetTIDsByTokenExpr") + return nil, nil +} + +func (p *recordingProvider) GetTIDsByField(field string) ([]uint32, error) { + p.calls = append(p.calls, "GetTIDsByField") + return nil, nil +} + +func (p *recordingProvider) GetLIDsFromTIDs(tids []uint32, stats lids.Counter, minLID, maxLID uint32, order seq.DocsOrder) []node.Node { + p.calls = append(p.calls, "GetLIDsFromTIDs") + return nil +} + +func TestMatchAllDoesNotTouchIndex(t *testing.T) { + p := &recordingProvider{} + + token := &parser.Literal{ + Field: seq.TokenAll, + Terms: []parser.Term{{Kind: parser.TermSymbol, Data: "*"}}, + } + leaf, err := evalLeaf(p, token, stopwatch.New(), &searchStats{}, 2, 7, seq.DocsOrderDesc) + require.NoError(t, err) + assert.Equal(t, []uint32{2, 3, 4, 5, 6, 7}, readAll(leaf)) + + it, err := iteratorFromLiteral( + p, token, stopwatch.New(), &searchStats{}, 2, 7, + 0, iteratorLimit{}, seq.DocsOrderDesc, QueryStats{}, QueryOptimizationConfig{}, + ) + require.NoError(t, err) + assert.Equal(t, []uint32{0}, it.tids) + + assert.Empty(t, p.calls, "match-all filter must not query the index") +} + +func TestIteratorFromLiteralMatchAllAggregation(t *testing.T) { + token := &parser.Literal{ + Field: seq.TokenAll, + Terms: []parser.Term{{Kind: parser.TermSymbol, Data: "*"}}, + } + + for _, order := range []seq.DocsOrder{seq.DocsOrderDesc, seq.DocsOrderAsc} { + it, err := iteratorFromLiteral( + &MockTokenIndex{}, token, stopwatch.New(), &searchStats{}, 1, 4, + 0, iteratorLimit{}, order, QueryStats{}, QueryOptimizationConfig{}, + ) + require.NoError(t, err) + + agg := NewSingleSourceCountAggregator(it, provideExtractTimeFunc(nil, nil, 0), false) + for lid := uint32(1); lid <= 4; lid++ { + docLID := lid + if order.IsReverse() { + docLID = 5 - lid + } + require.NoError(t, agg.Next([]node.LID{node.NewLID(docLID, order.IsDesc())})) + } + + result, err := agg.Aggregate() + require.NoError(t, err) + + assert.Len(t, result.SamplesByBin, 1) + for bin, samples := range result.SamplesByBin { + assert.Equal(t, "0", bin.Token) + assert.Equal(t, int64(4), samples.Total) + } + } +} diff --git a/frac/processor/eval_tree.go b/frac/processor/eval_tree.go index 5703abd00..ced54208b 100644 --- a/frac/processor/eval_tree.go +++ b/frac/processor/eval_tree.go @@ -60,6 +60,22 @@ func buildEvalTree(root *parser.ASTNode, minVal, maxVal uint32, stats *searchSta return nil, fmt.Errorf("unknown token type") } +// matchAllNode yields every LID in [minLID, maxLID] without touching the index. +// A sealed fraction's _all_ postings are exactly 1..DocsTotal, because sealing +// derives the new LID numbering from the _all_ list itself; an active fraction's +// are 1..N in the inverser-mapped space for the same reason. Either way the +// window computed by getLIDsBorders is already the answer. +// +// The asc flag follows the node package convention: it means "LID numbers +// ascend", which is desc document order, since LID 1 is the newest document. +func matchAllNode(minLID, maxLID uint32, asc bool) node.Node { + if asc { + return node.NewRange(node.NewAscLID(minLID), node.NewAscLID(maxLID)) + } + + return node.NewRange(node.NewDescLID(maxLID), node.NewDescLID(minLID)) +} + // evalLeaf finds suitable matching fraction tokens and returns Node that generate corresponding tokens LIDs func evalLeaf( ti tokenIndex, @@ -69,6 +85,12 @@ func evalLeaf( minLID, maxLID uint32, order seq.DocsOrder, ) (node.Node, error) { + if parser.IsMatchAll(token) { + stats.LeavesTotal++ + stats.NodesTotal++ + return matchAllNode(minLID, maxLID, order.IsDesc()), nil + } + m := sw.Start("get_tids_by_token_expr") tids, err := ti.GetTIDsByTokenExpr(token) m.Stop() @@ -226,6 +248,11 @@ func iteratorFromLiteral( queryStats QueryStats, queryOpt QueryOptimizationConfig, ) (*SourcedNodeIterator, error) { + if parser.IsMatchAll(literal) { + sourcedNode := node.NewSourcedNodeWrapper(matchAllNode(minLID, maxLID, order.IsDesc()), 0) + return NewSourcedNodeIterator(sourcedNode, ti, []uint32{0}, literal.Field, iteratorLimit), nil + } + m := sw.Start("get_tids_by_field") // For aggregations we can receive the first and the last TID for field, // because we build tree over *all* token values. diff --git a/frac/processor/search.go b/frac/processor/search.go index 97d1fb2d7..d5305843f 100644 --- a/frac/processor/search.go +++ b/frac/processor/search.go @@ -99,6 +99,13 @@ func IndexSearch( minLID, maxLID := getLIDsBorders(params, index) m.Stop() + if minLID > maxLID || minLID == 0 { + return &seq.QPR{ + Histogram: make(map[seq.MID]uint64), + Aggs: make([]seq.AggregatableSamples, len(params.AggQ)), + }, nil + } + m = sw.Start("get_skip_lids") skipLIDs, hasSkipLIDs, release, err := index.GetSkipLIDs(minLID, maxLID, params.Order.IsReverse()) defer func() { diff --git a/frac/processor/search_test.go b/frac/processor/search_test.go index 85ce76162..3d7e049a5 100644 --- a/frac/processor/search_test.go +++ b/frac/processor/search_test.go @@ -12,7 +12,8 @@ import ( ) type stubIDsIndex struct { - ids []seq.ID + ids []seq.ID + lidsRead int } func (s *stubIDsIndex) LessOrEqual(lid seq.LID, id seq.ID) bool { panic("not used") } @@ -21,6 +22,7 @@ func (s *stubIDsIndex) GetRID(lid seq.LID) seq.RID { return s.ids[l func (s *stubIDsIndex) Len() int { return len(s.ids) } func (s *stubIDsIndex) GetMIDs(lids []node.LID, out []seq.MID) []seq.MID { + s.lidsRead += len(lids) for _, lid := range lids { out = append(out, s.ids[lid.Unpack()].MID) } @@ -65,3 +67,76 @@ func TestIterateEvalTreeDuplicateIDsAtBatchBoundary(t *testing.T) { require.Equal(t, []seq.MID{600, 500, 400}, got) require.Equal(t, 4, total) // 4 LIDs scanned to produce 3 distinct IDs } + +func TestIterateEvalTreeBatchTotal(t *testing.T) { + tests := []struct { + name string + build func(bool) node.BatchedNode + lids []uint32 + }{ + { + name: "range", + build: func(asc bool) node.BatchedNode { + return node.NewRangeBatched(1, 6, asc) + }, + lids: []uint32{1, 2, 3, 4, 5, 6}, + }, + { + name: "slice", + build: func(asc bool) node.BatchedNode { + return node.NewStaticBatched([]uint32{1, 2, 3, 4, 5, 6}, asc) + }, + lids: []uint32{1, 2, 3, 4, 5, 6}, + }, + { + name: "and", + build: func(asc bool) node.BatchedNode { + return node.NewAndBatched(node.NewRangeBatched(1, 6, asc), + node.NewStaticBatched([]uint32{2, 4, 5}, asc), asc) + }, + lids: []uint32{2, 4, 5}, + }, + { + name: "or", + build: func(asc bool) node.BatchedNode { + return node.NewOrBatched(node.NewRangeBatched(2, 5, asc), + node.NewStaticBatched([]uint32{1, 3, 6}, asc), asc) + }, + lids: []uint32{1, 2, 3, 4, 5, 6}, + }, + } + + for _, tt := range tests { + for _, order := range []seq.DocsOrder{seq.DocsOrderDesc, seq.DocsOrderAsc} { + t.Run(tt.name+"/"+order.String(), func(t *testing.T) { + idx := &stubIDsIndex{ids: []seq.ID{ + 1: {MID: 600}, 2: {MID: 500}, 3: {MID: 400}, + 4: {MID: 300}, 5: {MID: 200}, 6: {MID: 100}, + }} + params := SearchParams{Limit: 1, WithTotal: true, Order: order} + total, ids, _, _, err := iterateEvalTree(context.Background(), params, idx, + tt.build(order.IsDesc()), nil, 1, 6, stopwatch.New()) + require.NoError(t, err) + require.Equal(t, len(tt.lids), total) + require.Len(t, ids, 1) + wantLID := tt.lids[0] + if order.IsReverse() { + wantLID = tt.lids[len(tt.lids)-1] + } + require.Equal(t, idx.ids[wantLID], ids[0].ID) + require.Equal(t, 1, idx.lidsRead) + }) + } + } +} + +func TestIterateEvalTreeCountOnly(t *testing.T) { + for _, downsample := range []uint32{0, 1, 2} { + params := SearchParams{WithTotal: true, Downsample: downsample} + total, ids, _, _, err := iterateEvalTree(context.Background(), params, &stubIDsIndex{}, + node.NewRangeBatched(1, 6, true), nil, 1, 6, stopwatch.New()) + require.NoError(t, err) + require.Equal(t, 6, total) + require.Empty(t, ids) + } +} diff --git a/indexer/indexer.go b/indexer/indexer.go index fd774a808..ee2ceec03 100644 --- a/indexer/indexer.go +++ b/indexer/indexer.go @@ -22,7 +22,7 @@ type indexer struct { // Index returns a list of metadata of the given json node. // Return at least one metadata. May return more if there are nested fields in the mapping. -// Each metadata has special token _all_, we use to find stored documents. +// A metadata without indexed fields has no tokens at all. // Each indexed field has special token _exists_, // we use it to find documents that have this field in search and aggregate requests. // @@ -38,8 +38,7 @@ func (i *indexer) Index(node *insaneJSON.Node, id seq.ID, size uint32) { m := i.metas parent := m[0] for j := 1; j < len(m); j++ { - // Copy tokens from the parent, except of the _all_ token. - m[j].Tokens = append(m[j].Tokens, parent.Tokens[1:]...) + m[j].Tokens = append(m[j].Tokens, parent.Tokens...) } } @@ -150,11 +149,6 @@ func (i *indexer) appendMeta(id seq.ID, size uint32) { i.metas[n].ID = id i.metas[n].Size = size - - i.metas[n].Tokens = append(i.metas[n].Tokens, tokenizer.MetaToken{ - Key: seq.AllTokenName, - Value: []byte{}, - }) } func (i *indexer) appendNestedMeta() { diff --git a/indexwriter/blocks.go b/indexwriter/blocks.go index bb6cc5a59..043a0ad80 100644 --- a/indexwriter/blocks.go +++ b/indexwriter/blocks.go @@ -9,6 +9,7 @@ import ( "github.com/ozontech/seq-db/frac/sealed/lids" "github.com/ozontech/seq-db/frac/sealed/seqids" "github.com/ozontech/seq-db/frac/sealed/token" + "github.com/ozontech/seq-db/seq" "github.com/ozontech/seq-db/util" ) @@ -105,6 +106,10 @@ func tokenBlock( block.ext.minTID = 1 for field, tokIt := range it { + if field == seq.TokenAll { + continue + } + emitFieldEntry() fieldName = field diff --git a/node/batch_ops.go b/node/batch_ops.go index c8694a56a..39ad99d65 100644 --- a/node/batch_ops.go +++ b/node/batch_ops.go @@ -180,18 +180,18 @@ func toBitmapBatch(b LIDBatch) *bitmapBatch { if b.IsEmpty() { panic("empty batch is not allowed to be cast to bitmap batch") } - if bb, ok := b.(*bitmapBatch); ok { - return bb - } - slice, ok := b.(*sliceBatch) - if !ok { + switch batchTyped := b.(type) { + case *bitmapBatch: + return batchTyped + case *sliceBatch: + bm := roaring.NewBitmap() + bm.AddMany(batchTyped.lids) + return &bitmapBatch{ + bm: bm, + min: batchTyped.Min(), + max: batchTyped.Max(), + } + default: panic("unsupported batch type") } - bm := roaring.NewBitmap() - bm.AddMany(slice.lids) - return &bitmapBatch{ - bm: bm, - min: slice.Min(), - max: slice.Max(), - } } diff --git a/node/batch_ops_test.go b/node/batch_ops_test.go index 7fe4b5b42..142ca7229 100644 --- a/node/batch_ops_test.go +++ b/node/batch_ops_test.go @@ -83,6 +83,15 @@ func TestLIDBatch_And(t *testing.T) { wantLeftRes: []uint32{1, 2, 3}, wantRightRes: nil, }, + { + name: "desc overlap right extends below left", + left: []uint32{3, 4, 5, 6, 7, 8, 9, 10}, + right: []uint32{1, 5, 8, 10, 12}, + asc: false, + wantResult: []uint32{5, 8, 10}, + wantLeftRes: nil, + wantRightRes: []uint32{1, 5, 8, 10, 12}, + }, { name: "identical inputs have no residuals", left: []uint32{2, 4, 9}, @@ -177,6 +186,15 @@ func TestLIDBatch_Or(t *testing.T) { wantLeftRes: []uint32{1, 2, 3}, wantRightRes: nil, }, + { + name: "desc overlap right extends below left", + left: []uint32{3, 4, 5, 6, 7, 8, 9, 10}, + right: []uint32{1, 8, 12}, + asc: false, + wantResult: []uint32{3, 4, 5, 6, 7, 8, 9, 10, 12}, + wantLeftRes: nil, + wantRightRes: []uint32{1}, + }, { name: "empty left", left: nil, diff --git a/node/node_range.go b/node/node_range.go index bf31e4a52..a54b445a7 100644 --- a/node/node_range.go +++ b/node/node_range.go @@ -1,5 +1,7 @@ package node +import "github.com/RoaringBitmap/roaring/v2" + type nodeRange struct { maxID LID curID LID @@ -29,3 +31,44 @@ func (n *nodeRange) NextGeq(nextID LID) LID { n.curID = Max(n.curID, nextID) return n.Next() } + +type nodeRangeBatched struct { + min, max uint32 + asc bool + done bool +} + +func NewRangeBatched(minLID, maxLID uint32, asc bool) BatchedNode { + return &nodeRangeBatched{min: minLID, max: maxLID, asc: asc} +} + +func (n *nodeRangeBatched) String() string { + return "(RANGE_BATCHED)" +} + +func (n *nodeRangeBatched) NextBatch() LIDBatch { + if n.asc { + return n.NextBatchGeq(NewAscZeroLID()) + } + return n.NextBatchGeq(NewDescZeroLID()) +} + +func (n *nodeRangeBatched) NextBatchGeq(nextID LID) LIDBatch { + if n.done || n.min > n.max { + return EmptyBatch() + } + + raw := nextID.Unpack() + n.done = true + if n.asc { + return newBitmapBatchFromRange(max(n.min, raw), n.max) + } + return newBitmapBatchFromRange(n.min, min(n.max, raw)) +} + +func newBitmapBatchFromRange(minLID, maxLID uint32) LIDBatch { + bm := roaring.NewBitmap() + bm.AddRange(uint64(minLID), uint64(maxLID)+1) + + return NewBitmapBatch(bm) +} diff --git a/node/node_range_test.go b/node/node_range_test.go index 6141f6877..e2d671799 100644 --- a/node/node_range_test.go +++ b/node/node_range_test.go @@ -1,9 +1,13 @@ package node import ( + "math" "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/ozontech/seq-db/consts" ) func TestNodeRange_NextGeq_JumpToLast(t *testing.T) { @@ -54,3 +58,111 @@ func TestNodeRange_NextGeq_Reverse(t *testing.T) { id = node.NextGeq(NewDescLID(3)) assert.True(t, id.IsNull()) } + +func collectBatched(t *testing.T, n BatchedNode) []uint32 { + t.Helper() + var got []uint32 + for { + b := n.NextBatch() + if b.IsEmpty() { + return got + } + it := b.Iter() + for v, ok := it.Next(); ok; v, ok = it.Next() { + got = append(got, v) + } + } +} + +func rangeLIDs(from, to int) []uint32 { + var out []uint32 + for v := from; v <= to; v++ { + out = append(out, uint32(v)) + } + return out +} + +func TestRangeBatched(t *testing.T) { + tests := []struct { + name string + min, max uint32 + asc bool + want []uint32 + }{ + {name: "asc", min: 3, max: 10, asc: true, want: rangeLIDs(3, 10)}, + {name: "desc", min: 3, max: 10, want: rangeLIDs(3, 10)}, + {name: "empty asc", min: 10, max: 3, asc: true}, + {name: "empty desc", min: 10, max: 3}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := collectBatched(t, NewRangeBatched(tt.min, tt.max, tt.asc)) + require.Equal(t, tt.want, got) + }) + } +} + +func TestRangeBatchedNextGeq(t *testing.T) { + tests := []struct { + name string + asc bool + next LID + wantMin, wantMax uint32 + }{ + {name: "asc", asc: true, next: NewAscLID(5), wantMin: 5, wantMax: 10}, + {name: "desc", next: NewDescLID(7), wantMin: 3, wantMax: 7}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := NewRangeBatched(3, 10, tt.asc).NextBatchGeq(tt.next) + require.False(t, b.IsEmpty()) + require.Equal(t, tt.wantMin, b.Min()) + require.Equal(t, tt.wantMax, b.Max()) + }) + } +} + +func TestRangeBatchedWholeWindowInOneBatch(t *testing.T) { + n := NewRangeBatched(1, uint32(consts.DefaultLIDBlockCap)+10, true) + b := n.NextBatch() + require.False(t, b.IsEmpty()) + require.Equal(t, consts.DefaultLIDBlockCap+10, b.Len()) + require.True(t, n.NextBatch().IsEmpty()) +} + +func TestRangeBatchedUint32Bounds(t *testing.T) { + tests := []struct { + name string + value uint32 + asc bool + }{ + {name: "max asc", value: math.MaxUint32, asc: true}, + {name: "zero desc"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + n := NewRangeBatched(tt.value, tt.value, tt.asc) + b := n.NextBatch() + it := b.ManyIter(tt.asc) + buf := make([]uint32, 1) + require.Equal(t, 1, it.CopyRawLIDs(buf)) + require.Equal(t, tt.value, buf[0]) + require.Zero(t, it.CopyRawLIDs(buf)) + require.True(t, n.NextBatch().IsEmpty()) + }) + } +} + +func TestRangeBatchFullUint32Range(t *testing.T) { + b := newBitmapBatchFromRange(0, math.MaxUint32) + require.Equal(t, uint64(math.MaxUint32)+1, uint64(b.Len())) + + buf := make([]uint32, 1) + require.Equal(t, 1, b.ManyIter(true).CopyRawLIDs(buf)) + require.Equal(t, uint32(0), buf[0]) + require.Equal(t, 1, b.ManyIter(false).CopyRawLIDs(buf)) + require.Equal(t, uint32(math.MaxUint32), buf[0]) +} diff --git a/parser/token_literal.go b/parser/token_literal.go index 62b4ca721..785779531 100644 --- a/parser/token_literal.go +++ b/parser/token_literal.go @@ -12,8 +12,17 @@ type Literal struct { Terms []Term } +func IsMatchAll(t Token) bool { + l, ok := t.(*Literal) + if !ok || l.Field != seq.TokenAll || len(l.Terms) != 1 { + return false + } + term := l.Terms[0] + return term.IsWildcard() || (term.Kind == TermText && term.Data == "") +} + func (n *Literal) DumpSeqQL(o *strings.Builder) { - if n.Field == seq.TokenAll && len(n.Terms) == 1 && n.Terms[0].IsWildcard() { + if IsMatchAll(n) { o.WriteString("*") return } diff --git a/parser/token_literal_test.go b/parser/token_literal_test.go new file mode 100644 index 000000000..f2b690e5c --- /dev/null +++ b/parser/token_literal_test.go @@ -0,0 +1,46 @@ +package parser + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/ozontech/seq-db/seq" +) + +func TestIsMatchAll(t *testing.T) { + mustParse := func(t *testing.T, query string) Token { + t.Helper() + q, err := ParseSeqQL(query, nil) + require.NoError(t, err) + require.NotNil(t, q.Root) + return q.Root.Value + } + + wildcard := []Term{{Kind: TermSymbol, Data: "*"}} + emptyText := []Term{{Kind: TermText, Data: ""}} + + tests := []struct { + name string + token Token + want bool + }{ + {"nil", nil, false}, + {"bare asterisk query", mustParse(t, "*"), true}, + {"empty query", mustParse(t, ""), true}, + {"explicit _all_:*", &Literal{Field: seq.TokenAll, Terms: wildcard}, true}, + {"_all_ with empty text", &Literal{Field: seq.TokenAll, Terms: emptyText}, true}, + {"_all_ with several terms", &Literal{Field: seq.TokenAll, Terms: append(wildcard, emptyText...)}, false}, + {"_all_ with non-empty text", &Literal{Field: seq.TokenAll, Terms: []Term{{Kind: TermText, Data: "foo"}}}, false}, + {"other field wildcard", &Literal{Field: "level", Terms: wildcard}, false}, + {"other field empty text", &Literal{Field: "level", Terms: emptyText}, false}, + {"_exists_ wildcard", &Literal{Field: seq.TokenExists, Terms: wildcard}, false}, + {"non-literal token", &Range{}, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, IsMatchAll(tt.token)) + }) + } +} diff --git a/proxy/bulk/ingestor_test.go b/proxy/bulk/ingestor_test.go index aff7383e9..f082a734a 100644 --- a/proxy/bulk/ingestor_test.go +++ b/proxy/bulk/ingestor_test.go @@ -101,7 +101,6 @@ func TestProcessDocuments(t *testing.T) { } var ( - all = newToken(seq.TokenAll, "") existsMsg = newToken(seq.TokenExists, "message") ) @@ -115,7 +114,7 @@ func TestProcessDocuments(t *testing.T) { ExpMeta: []indexer.MetaData{{ ID: id, Size: 2, - Tokens: []tokenizer.MetaToken{newToken(seq.TokenAll, "")}, + Tokens: nil, }}, } }, @@ -141,14 +140,14 @@ func TestProcessDocuments(t *testing.T) { }, ExpDocs: nil, ExpMeta: []indexer.MetaData{ - {ID: id, Size: 30, Tokens: []tokenizer.MetaToken{all, tk("*prefix_asterisk"), existsMsg}}, - {ID: id, Size: 31, Tokens: []tokenizer.MetaToken{all, tk("*"), tk("prefix_asterisk"), existsMsg}}, - {ID: id, Size: 28, Tokens: []tokenizer.MetaToken{all, tk("infix*asterisk"), existsMsg}}, - {ID: id, Size: 30, Tokens: []tokenizer.MetaToken{all, tk("infix"), tk("*"), tk("asterisk"), existsMsg}}, - {ID: id, Size: 29, Tokens: []tokenizer.MetaToken{all, tk("infix"), tk("*asterisk"), existsMsg}}, - {ID: id, Size: 29, Tokens: []tokenizer.MetaToken{all, tk("infix*"), tk("asterisk"), existsMsg}}, - {ID: id, Size: 31, Tokens: []tokenizer.MetaToken{all, tk("postfix"), tk("asterisk*"), existsMsg}}, - {ID: id, Size: 32, Tokens: []tokenizer.MetaToken{all, tk("postfix"), tk("asterisk"), tk("*"), existsMsg}}, + {ID: id, Size: 30, Tokens: []tokenizer.MetaToken{tk("*prefix_asterisk"), existsMsg}}, + {ID: id, Size: 31, Tokens: []tokenizer.MetaToken{tk("*"), tk("prefix_asterisk"), existsMsg}}, + {ID: id, Size: 28, Tokens: []tokenizer.MetaToken{tk("infix*asterisk"), existsMsg}}, + {ID: id, Size: 30, Tokens: []tokenizer.MetaToken{tk("infix"), tk("*"), tk("asterisk"), existsMsg}}, + {ID: id, Size: 29, Tokens: []tokenizer.MetaToken{tk("infix"), tk("*asterisk"), existsMsg}}, + {ID: id, Size: 29, Tokens: []tokenizer.MetaToken{tk("infix*"), tk("asterisk"), existsMsg}}, + {ID: id, Size: 31, Tokens: []tokenizer.MetaToken{tk("postfix"), tk("asterisk*"), existsMsg}}, + {ID: id, Size: 32, Tokens: []tokenizer.MetaToken{tk("postfix"), tk("asterisk"), tk("*"), existsMsg}}, }, } }, @@ -164,7 +163,6 @@ func TestProcessDocuments(t *testing.T) { ID: id, Size: 22, Tokens: []tokenizer.MetaToken{ - newToken(seq.TokenAll, ""), newToken(seq.TokenExists, "exists_only"), }, }}, @@ -180,7 +178,6 @@ func TestProcessDocuments(t *testing.T) { ExpDocs: []string{doc}, ExpMeta: []indexer.MetaData{ {ID: id, Size: uint32(len(doc)), Tokens: []tokenizer.MetaToken{ - newToken(seq.TokenAll, ""), newToken("_exists_", "tags.level"), newToken("_exists_", "tags.message"), newToken("_exists_", "tags.path"), @@ -199,9 +196,9 @@ func TestProcessDocuments(t *testing.T) { InDocs: []string{doc1, doc2, doc3}, ExpDocs: nil, ExpMeta: []indexer.MetaData{ - {ID: id, Size: uint32(len(doc1)), Tokens: []tokenizer.MetaToken{newToken(seq.TokenAll, ""), newToken(seq.TokenExists, "level")}}, - {ID: id, Size: uint32(len(doc2)), Tokens: []tokenizer.MetaToken{newToken(seq.TokenAll, ""), newToken(seq.TokenExists, "message")}}, - {ID: id, Size: uint32(len(doc3)), Tokens: []tokenizer.MetaToken{newToken(seq.TokenAll, ""), newToken(seq.TokenExists, "path")}}, + {ID: id, Size: uint32(len(doc1)), Tokens: []tokenizer.MetaToken{newToken(seq.TokenExists, "level")}}, + {ID: id, Size: uint32(len(doc2)), Tokens: []tokenizer.MetaToken{newToken(seq.TokenExists, "message")}}, + {ID: id, Size: uint32(len(doc3)), Tokens: []tokenizer.MetaToken{newToken(seq.TokenExists, "path")}}, }, } }, @@ -211,7 +208,6 @@ func TestProcessDocuments(t *testing.T) { Payload: func() TestPayload { const doc = `{"level":"error", "message":" request 🫦 failed! ", "error": "context cancelled", "shard": "1", "path":"http://localhost:8080/example"}` meta := indexer.MetaData{ID: id, Size: uint32(len(doc)), Tokens: []tokenizer.MetaToken{ - newToken(seq.TokenAll, ""), newToken("level", "error"), newToken(seq.TokenExists, "level"), newToken("message", "request"), @@ -538,8 +534,6 @@ func newToken(k, v string) tokenizer.MetaToken { func buildKeywordTokens(kvs ...string) []tokenizer.MetaToken { var tokens []tokenizer.MetaToken - tokens = append(tokens, newToken("_all_", "")) - for i := 0; i < len(kvs); i += 2 { k := kvs[i] v := kvs[i+1]