perf: implement batch execution AND/OR/NAND

cmd/seq-db/seq-db.go

@@ -285,6 +285,12 @@ func startStore(
 						MaxGroupTokens:     cfg.Limits.Aggregation.GroupTokens,
 						MaxTIDsPerFraction: cfg.Limits.Aggregation.FractionTokens,
 					},
+					QueryOptimization: frac.QueryOptimizationConfig{
+						BatchExecution: frac.BatchExecutionConfig{
+							Enabled:       cfg.QueryOptimization.BatchExecution.Enabled,
+							CostThreshold: cfg.QueryOptimization.BatchExecution.CostThreshold,
+						},
+					},
 				},
 				SkipSortDocs: !cfg.DocsSorting.Enabled,
 				KeepWalFile:  false,

config/config.go

@@ -176,6 +176,15 @@ type Config struct {
 		} `config:"aggregation"`
 	} `config:"limits"`
 
+	QueryOptimization struct {
+		BatchExecution struct {
+			Enabled bool `config:"enabled"`
+			// CostThreshold is the minimum estimated non-batched execution cost required to enable batch-at-a-time query
+			// evaluation. Suggestion is to use value which is greater than 3 x LID block size.
+			CostThreshold int `config:"cost_threshold" default:"150000"`
+		} `config:"batch_execution"`
+	} `config:"query_optimization"`
+
 	CircuitBreaker struct {
 		Bulk struct {
 			// Checkout [CircuitBreaker] for more information.

frac/active_index.go

@@ -110,6 +110,9 @@ func (dp *activeDataProvider) Search(params processor.SearchParams) (*seq.QPR, e
 	params.To = min(params.To, dp.info.To)
 
 	aggLimits := processor.AggLimits(dp.config.Search.AggLimits)
+	queryOpt := processor.QueryOptimizationConfig{
+		BatchExecution: processor.BatchExecutionConfig(dp.config.Search.QueryOptimization.BatchExecution),
+	}
 
 	sw := stopwatch.New()
 
@@ -132,7 +135,7 @@ func (dp *activeDataProvider) Search(params processor.SearchParams) (*seq.QPR, e
 	qprs := make([]*seq.QPR, 0, len(indexes))
 
 	for _, si := range indexes {
-		qpr, err := processor.IndexSearch(dp.ctx, params, &si, aggLimits, sw)
+		qpr, err := processor.IndexSearch(dp.ctx, dp.info.BinaryDataVer, params, &si, aggLimits, queryOpt, sw)
 		if err != nil {
 			return nil, err
 		}
@@ -252,6 +255,10 @@ func (si *activeTokenIndex) GetTIDsByTokenExpr(t parser.Token) ([]uint32, error)
 	return si.tokenList.FindPattern(si.ctx, t)
 }
 
+func (si *activeTokenIndex) GetFreqsByTIDs(tids []uint32, field string) []uint32 {
+	return make([]uint32, len(tids))
+}
+
 func (si *activeTokenIndex) GetLIDsFromTIDs(tids []uint32, _ lids.Counter, minLID, maxLID uint32, order seq.DocsOrder) []node.Node {
 	nodes := make([]node.Node, 0, len(tids))
 	for _, tid := range tids {
@@ -263,6 +270,17 @@ func (si *activeTokenIndex) GetLIDsFromTIDs(tids []uint32, _ lids.Counter, minLI
 	return nodes
 }
 
+func (si *activeTokenIndex) GetBatchedLIDsFromTIDs(tids []uint32, _ lids.Counter, minLID, maxLID uint32, order seq.DocsOrder) []node.BatchedNode {
+	nodes := make([]node.BatchedNode, 0, len(tids))
+	for _, tid := range tids {
+		tlids := si.tokenList.Provide(tid)
+		unmapped := tlids.GetLIDs(si.mids, si.rids)
+		inverse := inverseLIDs(unmapped, si.inverser, minLID, maxLID)
+		nodes = append(nodes, node.NewStaticBatched(inverse, order.IsReverse()))
+	}
+	return nodes
+}
+
 func inverseLIDs(unmapped []uint32, inv *inverser, minLID, maxLID uint32) []uint32 {
 	result := make([]uint32, 0, len(unmapped))
 	for _, v := range unmapped {

frac/config.go

@@ -8,7 +8,8 @@ type Config struct {
 }
 
 type SearchConfig struct {
-	AggLimits AggLimits
+	AggLimits         AggLimits
+	QueryOptimization QueryOptimizationConfig
 }
 
 type AggLimits struct {
@@ -17,3 +18,14 @@ type AggLimits struct {
 	MaxGroupTokens     int // MaxGroupTokens max AggQuery.GroupBy unique values.
 	MaxTIDsPerFraction int // MaxTIDsPerFraction max number of tokens per fraction.
 }
+
+type QueryOptimizationConfig struct {
+	BatchExecution BatchExecutionConfig
+}
+
+type BatchExecutionConfig struct {
+	Enabled bool
+	// CostThreshold is the minimum estimated non-batched iteration
+	// cost required to enable batch-at-a-time query evaluation.
+	CostThreshold int
+}

frac/fraction_concurrency_test.go

@@ -58,7 +58,7 @@ func TestConcurrentAppendAndQuery(t *testing.T) {
 	const numMessagesPerWriter = 5000
 	const bulkSize = 100
 
-	docs, bulks, fromTime, toTime := generatesMessages(numWriters*numMessagesPerWriter, bulkSize)
+	docs, bulks, fromTime, toTime := generatesMessages(numWriters*numMessagesPerWriter, bulkSize, false)
 
 	tmpDir := testcommon.CreateTempDir()
 	fracPath := filepath.Join(tmpDir, "test_fraction")
@@ -179,7 +179,7 @@ func TestConcurrentColdQueriesSealedFrac(t *testing.T) {
 	const bulkSize = 100
 	const numIterations = 100
 
-	docs, bulks, _, toTime := generatesMessages(numWriters*numMessagesPerWriter, bulkSize)
+	docs, bulks, _, toTime := generatesMessages(numWriters*numMessagesPerWriter, bulkSize, false)
 
 	tmpDir := testcommon.CreateTempDir()
 	fracPath := filepath.Join(tmpDir, "test_fraction")
@@ -476,7 +476,7 @@ func readTest(t *testing.T, fraction frac.Fraction, numReaders, numQueries int,
 	assert.NoError(t, err, "concurrent queries should complete without errors")
 }
 
-func generatesMessages(numMessages, bulkSize int) ([]*testDoc, [][]string, time.Time, time.Time) {
+func generatesMessages(numMessages, bulkSize int, nestedIndexes bool) ([]*testDoc, [][]string, time.Time, time.Time) {
 	services := []string{gateway, proxy, scheduler, database, bus, kafka}
 	messages := []string{
 		"request started", "request completed", "processing timed out",
@@ -502,6 +502,17 @@ func generatesMessages(numMessages, bulkSize int) ([]*testDoc, [][]string, time.
 			message += fmt.Sprintf(" %d", rand.IntN(10000000))
 		}
 
+		var spansJson string
+
+		if nestedIndexes {
+			numSpans := 1 + rand.IntN(5)
+			spans := make([]string, numSpans)
+			for j := 0; j < numSpans; j++ {
+				spans[j] = fmt.Sprintf(`{"span_id":"span-%d"}`, rand.IntN(5000))
+			}
+			spansJson = fmt.Sprintf(`, "spans":[%s]`, strings.Join(spans, ","))
+		}
+
 		level := rand.IntN(6)
 		timestamp := fromTime.Add(time.Duration(i) * time.Millisecond)
 		id := fmt.Sprintf("id-%d", i)
@@ -512,8 +523,8 @@ func generatesMessages(numMessages, bulkSize int) ([]*testDoc, [][]string, time.
 			toTime = timestamp
 		}
 
-		json := fmt.Sprintf(`{"timestamp":%q,"id": %q, "service":%q,"pod":%q,"client_ip":%q,"message":%q,"trace_id": %q,"level":"%d"}`,
-			timestamp.Format(time.RFC3339Nano), id, service, pod, clientIp, message, traceId, level)
+		json := fmt.Sprintf(`{"timestamp":%q,"id": %q, "service":%q,"pod":%q,"client_ip":%q,"message":%q,"trace_id": %q,"level":"%d"%s}`,
+			timestamp.Format(time.RFC3339Nano), id, service, pod, clientIp, message, traceId, level, spansJson)
 
 		docs = append(docs, &testDoc{
 			json:      json,