feat: remove token _all_

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)
 	}
 }

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) {

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 {

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"}`,

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) {

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

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,39 +115,47 @@ 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)
 	})
 
 	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)
 	})
 
 	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)
 	})
 
 	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)
 	})