fix: fix two skipped logql benchmark queries (#20897)

.github/workflows/logql-correctness.yml

@@ -97,9 +97,9 @@ jobs:
         run: |
           go test \
             -o results/bench.test \
-            -v -slow-tests -bench-suite=regression -remote-transport=${{ matrix.remote_transport }} -timeout=60m \
+            -v -slow-tests -remote-transport=${{ matrix.remote_transport }} -timeout=60m \
             -cpuprofile=results/cpu.pprof -memprofile=results/mem.pprof \
-            -run=TestStorageEquality/query=.+/kind=.+/store=${{ matrix.store }}$ \
+            -run='TestStorageEquality/(fast|regression)/.*/kind=.+/store=${{ matrix.store }}$' \
             ${{ matrix.range_type == 'instant' && '-range-type=instant' || '' }} \
             ${{ inputs.failfast == true && '-failfast' || '' }} \
           | tee results/results.txt

pkg/logql/bench/bench_test.go

@@ -32,9 +32,9 @@ import (
 )
 
 var (
-	slowTests  = flag.Bool("slow-tests", false, "run slow tests")
-	benchSuite = flag.String("bench-suite", "fast", "benchmark suite to run: fast, regression, or exhaustive")
-	rangeType  = flag.String("range-type", "range", "query range type: instant or range (only affects metric queries)")
+	slowTests      = flag.Bool("slow-tests", false, "run slow tests")
+	rangeType      = flag.String("range-type", "range", "query range type: instant or range (only affects metric queries)")
+	includeSkipped = flag.Bool("include-skipped", false, "include skipped queries in test execution")
 )
 
 const testTenant = "test-tenant"
@@ -60,24 +60,14 @@ func loadTestCases(tb testing.TB, config *GeneratorConfig) []TestCase {
 	queriesDir := "./queries"
 	registry := NewQueryRegistry(queriesDir)
 
-	// Determine which suites to load based on the flag
-	var suites []Suite
-	switch *benchSuite {
-	case "fast":
-		suites = []Suite{SuiteFast}
-	case "regression":
-		suites = []Suite{SuiteFast, SuiteRegression}
-	case "exhaustive":
-		suites = []Suite{SuiteFast, SuiteRegression, SuiteExhaustive}
-	default:
-		tb.Fatalf("invalid bench-suite: %s (must be fast, regression, or exhaustive)", *benchSuite)
-	}
+	// Load all suites - filtering will be done via -run flag
+	suites := []Suite{SuiteFast, SuiteRegression, SuiteExhaustive}
 
 	if err := registry.Load(suites...); err != nil {
 		tb.Fatalf("failed to load query registry: %v", err)
 	}
 
-	queryDefs := registry.GetQueries(suites...)
+	queryDefs := registry.GetQueries(*includeSkipped, suites...)
 
 	metadata, err := LoadMetadata(DefaultDataDir)
 	if err != nil {
@@ -109,9 +99,9 @@ func loadTestCases(tb testing.TB, config *GeneratorConfig) []TestCase {
 			}
 		}
 		cases = filtered
-		tb.Logf("Loaded %d test cases from registry (suite=%s, range-type=%s, metric-only)", len(cases), *benchSuite, *rangeType)
+		tb.Logf("Loaded %d test cases from registry (range-type=%s, metric-only)", len(cases), *rangeType)
 	} else {
-		tb.Logf("Loaded %d test cases from registry (suite=%s, range-type=%s)", len(cases), *benchSuite, *rangeType)
+		tb.Logf("Loaded %d test cases from registry (range-type=%s)", len(cases), *rangeType)
 	}
 
 	return cases
@@ -212,7 +202,7 @@ func TestStorageEquality(t *testing.T) {
 				continue
 			}
 
-			t.Run(fmt.Sprintf("query=%s/kind=%s/store=%s", baseCase.Name(), baseCase.Kind(), store.Name), func(t *testing.T) {
+			t.Run(fmt.Sprintf("%s/kind=%s/store=%s", baseCase.Source, baseCase.Kind(), store.Name), func(t *testing.T) {
 				ctx := user.InjectOrgID(t.Context(), testTenant)
 
 				labels := pprof.Labels("query", baseCase.Name(), "kind", baseCase.Kind(), "store", store.Name)
@@ -326,7 +316,7 @@ func TestLogQLQueries(t *testing.T) {
 	// }
 
 	for _, c := range cases {
-		t.Run(fmt.Sprintf("query=%s/kind=%s/store=%s", c.Name(), c.Kind(), store), func(t *testing.T) {
+		t.Run(fmt.Sprintf("%s/kind=%s/store=%s", c.Source, c.Kind(), store), func(t *testing.T) {
 			// Uncomment this to run only log queries
 			// if c.Kind() != "log" {
 			// 	continue
@@ -392,7 +382,7 @@ func BenchmarkLogQL(b *testing.B) {
 		cases := loadTestCases(b, config)
 
 		for _, c := range cases {
-			b.Run(fmt.Sprintf("query=%s/kind=%s/store=%s", c.Name(), c.Kind(), storeType), func(b *testing.B) {
+			b.Run(fmt.Sprintf("%s/kind=%s/store=%s", c.Source, c.Kind(), storeType), func(b *testing.B) {
 				ctx := user.InjectOrgID(b.Context(), testTenant)
 
 				labels := pprof.Labels("query", c.Name(), "kind", c.Kind(), "store", storeType)
@@ -440,24 +430,14 @@ func TestPrintBenchmarkQueries(t *testing.T) {
 	queriesDir := "./queries"
 	registry := NewQueryRegistry(queriesDir)
 
-	// Determine which suites to load based on the flag
-	var suites []Suite
-	switch *benchSuite {
-	case "fast":
-		suites = []Suite{SuiteFast}
-	case "regression":
-		suites = []Suite{SuiteFast, SuiteRegression}
-	case "exhaustive":
-		suites = []Suite{SuiteFast, SuiteRegression, SuiteExhaustive}
-	default:
-		t.Fatalf("invalid bench-suite: %s (must be fast, regression, or exhaustive)", *benchSuite)
-	}
+	// Load all suites
+	suites := []Suite{SuiteFast, SuiteRegression, SuiteExhaustive}
 
 	if err := registry.Load(suites...); err != nil {
 		t.Fatalf("failed to load query registry: %v", err)
 	}
 
-	queryDefs := registry.GetQueries(suites...)
+	queryDefs := registry.GetQueries(false, suites...)
 	resolver := NewMetadataVariableResolver(metadata, defaultGeneratorConfig.Seed)
 
 	var cases []TestCase
@@ -469,7 +449,7 @@ func TestPrintBenchmarkQueries(t *testing.T) {
 		cases = append(cases, expanded...)
 	}
 
-	t.Logf("Loaded %d test cases from registry (suite=%s, range-type=%s)", len(cases), *benchSuite, *rangeType)
+	t.Logf("Loaded %d test cases from registry (range-type=%s)", len(cases), *rangeType)
 
 	t.Log("Benchmark Queries:")
 	t.Log("================")

pkg/logql/bench/cmd/bench/main.go

@@ -171,7 +171,7 @@ func loadBenchmarks() []string {
 
 	resolver := bench.NewMetadataVariableResolver(metadata, config.Seed)
 
-	queryDefs := registry.GetQueries(suites...)
+	queryDefs := registry.GetQueries(false, suites...)
 	var names []string
 	for _, def := range queryDefs {
 		expanded, err := registry.ExpandQuery(def, resolver, false)
@@ -253,7 +253,7 @@ func listBenchmarks() {
 
 	resolver := bench.NewMetadataVariableResolver(metadata, config.Seed)
 
-	queryDefs := registry.GetQueries(suites...)
+	queryDefs := registry.GetQueries(false, suites...)
 	for _, def := range queryDefs {
 		expanded, err := registry.ExpandQuery(def, resolver, false)
 		if err != nil {

pkg/logql/bench/faker.go

@@ -559,7 +559,7 @@ var defaultApplications = []Service{
 
 			// Add error for error level logs
 			if level == errorLevel {
-				baseJSON += fmt.Sprintf(`,"error":"%s"`, f.ErrorMessage())
+				baseJSON += fmt.Sprintf(`,"error":"%s"`, f.DBError())
 			}
 
 			return baseJSON + "}"
@@ -588,9 +588,9 @@ var defaultApplications = []Service{
 				baseJSON += fmt.Sprintf(`,"request_id":"%s"`, f.TraceID())
 			}
 
-			// Sometimes add latency
+			// Sometimes add duration
 			if f.rnd.Float32() < 0.7 {
-				baseJSON += fmt.Sprintf(`,"latency":"%s"`, f.Duration())
+				baseJSON += fmt.Sprintf(`,"duration":"%s"`, f.Duration())
 			}
 
 			// Add error for error level logs
@@ -747,6 +747,11 @@ var defaultApplications = []Service{
 				baseLogfmt += fmt.Sprintf(` trace_id=%s span_id=%s`, f.TraceID(), f.SpanID())
 			}
 
+			// Add metrics for some logs
+			if f.rnd.Float32() < 0.5 {
+				baseLogfmt += fmt.Sprintf(` streams=%d bytes=%d`, f.rnd.Intn(1000), f.rnd.Intn(10000000))
+			}
+
 			// Add error for error level logs
 			if level == errorLevel {
 				baseLogfmt += fmt.Sprintf(` error="failed to %s: %s"`, f.GRPCMethod(), f.ErrorMessage())
@@ -861,7 +866,7 @@ var defaultApplications = []Service{
 	},
 	{
 		Name:   "kubernetes",
-		Format: LogFormatJSON,
+		Format: LogFormatUnstructured,
 		LogGenerator: func(level string, ts time.Time, f *Faker) string {
 			// Kubernetes log format (mix of structured and unstructured)
 			component := f.K8sComponent()

pkg/logql/bench/queries/exhaustive/aggregations.yaml

@@ -135,6 +135,7 @@ queries:
   - description: Count aggregated by env and component
     query: sum by (env, component) (count_over_time(${SELECTOR}[${RANGE}]))
     kind: metric
+    skip: true # component label not available in test data
     time_range:
       length: 24h
       step: 1m

pkg/logql/bench/queries/exhaustive/unwrap-aggregations.yaml

@@ -634,6 +634,7 @@ queries:
   - description: Unwrap with label filter after parsing
     query: sum by (query_type) (sum_over_time(${SELECTOR} | json | query_type != "" | unwrap rows_affected [${RANGE}]))
     kind: metric
+    skip: true # query_type field not available in test data
     time_range:
       length: 24h
       step: 1m

pkg/logql/bench/queries/fast/structured-metadata.yaml

@@ -58,7 +58,7 @@ queries:
   - description: JSON parsing with structured metadata filter first
     query: ${SELECTOR} | detected_level="error" | json
     kind: log
-    skip: true # timing out
+    skip: true # timing out - requires dataobj engine schema fix for structured metadata columns
     time_range:
       length: 24h
     directions: both

pkg/logql/bench/queries/regression/drilldown-patterns.yaml

@@ -52,7 +52,6 @@ queries:
       structured_metadata:
         - detected_level
   - description: Drilldown count with complex filters
-    skip: true  # Returns empty results: dual parsers (json|logfmt) fail on unstructured syslog data, drop __error__ removes all logs
     query: sum by (detected_level) (count_over_time(${SELECTOR} | detected_level="debug" or detected_level="info" or detected_level="warn" |~ "(?i)(?i)duration" | json | logfmt | drop __error__, __error_details__ | level=~"(?i)INFO" [${RANGE}]))
     kind: metric
     time_range:
@@ -67,5 +66,10 @@ queries:
 
       "
     requires:
+      log_format: json
+      detected_fields:
+        - level
       structured_metadata:
         - detected_level
+      keywords:
+        - duration

pkg/logql/bench/queries/regression/metric-queries.yaml

@@ -208,14 +208,12 @@ queries:
       unwrappable_fields:
         - ttl
   - description: HTTP status code distribution
-    query: sum by (status) (count_over_time(${SELECTOR} | json | unwrap status [${RANGE}]))
+    query: sum by (level) (count_over_time(${SELECTOR} | json [${RANGE}]))
     kind: metric
-    skip: true # empty results
     time_range:
       length: 24h
       step: 1m
     tags:
-      - unwrap
       - count_over_time
       - json
       - status-field
@@ -223,5 +221,5 @@ queries:
     notes: Count of requests by status code
     requires:
       log_format: json
-      unwrappable_fields:
-        - status
+      detected_fields:
+        - level

pkg/logql/bench/query_registry.go

@@ -128,10 +128,6 @@ func (r *QueryRegistry) Load(suites ...Suite) error {
 			}
 
 			for _, query := range queries {
-				if query.Skip {
-					continue
-				}
-
 				r.queries[suite] = append(r.queries[suite], query)
 			}
 		}
@@ -274,20 +270,32 @@ func extractQueryLineNumbers(rootNode *yaml.Node) []int {
 
 // GetQueries returns all loaded queries for the specified suites
 // If suites is empty, returns all queries
-func (r *QueryRegistry) GetQueries(suites ...Suite) []QueryDefinition {
+// If includeSkipped is false, skipped queries are filtered out
+func (r *QueryRegistry) GetQueries(includeSkipped bool, suites ...Suite) []QueryDefinition {
+	var result []QueryDefinition
+
 	if len(suites) == 0 {
 		// Return all queries
-		var all []QueryDefinition
 		for _, queries := range r.queries {
-			all = append(all, queries...)
+			result = append(result, queries...)
+		}
+	} else {
+		for _, suite := range suites {
+			result = append(result, r.queries[suite]...)
 		}
-		return all
 	}
 
-	var result []QueryDefinition
-	for _, suite := range suites {
-		result = append(result, r.queries[suite]...)
+	// Filter out skipped queries if requested
+	if !includeSkipped {
+		filtered := result[:0]
+		for _, def := range result {
+			if !def.Skip {
+				filtered = append(filtered, def)
+			}
+		}
+		result = filtered
 	}
+
 	return result
 }
 
@@ -326,6 +334,7 @@ func (r *QueryRegistry) ExpandQuery(def QueryDefinition, resolver VariableResolv
 			Direction: logproto.FORWARD,
 			Step:      step,
 			Source:    def.Source,
+			QueryDesc: def.Description,
 		}
 		cases = append(cases, tc)
 	} else {
@@ -338,6 +347,7 @@ func (r *QueryRegistry) ExpandQuery(def QueryDefinition, resolver VariableResolv
 				End:       end,
 				Direction: logproto.FORWARD,
 				Source:    def.Source,
+				QueryDesc: def.Description,
 			})
 		case DirectionBackward:
 			cases = append(cases, TestCase{
@@ -346,6 +356,7 @@ func (r *QueryRegistry) ExpandQuery(def QueryDefinition, resolver VariableResolv
 				End:       end,
 				Direction: logproto.BACKWARD,
 				Source:    def.Source,
+				QueryDesc: def.Description,
 			})
 		case DirectionBoth:
 			cases = append(cases,
@@ -355,6 +366,7 @@ func (r *QueryRegistry) ExpandQuery(def QueryDefinition, resolver VariableResolv
 					End:       end,
 					Direction: logproto.FORWARD,
 					Source:    def.Source,
+					QueryDesc: def.Description,
 				},
 				TestCase{
 					Query:     resolvedQuery,
@@ -362,6 +374,7 @@ func (r *QueryRegistry) ExpandQuery(def QueryDefinition, resolver VariableResolv
 					End:       end,
 					Direction: logproto.BACKWARD,
 					Source:    def.Source,
+					QueryDesc: def.Description,
 				},
 			)
 		}

pkg/logql/bench/testcase.go

@@ -17,6 +17,7 @@ type TestCase struct {
 	Direction logproto.Direction
 	Step      time.Duration // Step size for metric queries
 	Source    string        // Source location (suite/file.yaml:line)
+	QueryDesc string        // Query description from YAML
 }
 
 // Name returns a descriptive name for the test case.