fix(thor): Return empty log lines properly for the new engine (#20971)

Author: spiridonov

pkg/engine/compat.go

@@ -134,12 +134,8 @@ func (b *streamsResultBuilder) CollectRecord(rec arrow.RecordBatch) {
 			shortName == types.ColumnNameError || shortName == types.ColumnNameErrorDetails):
 			parsedCol := col.(*array.String)
 
-			// TODO: keep errors if --strict is set
-			// These are reserved column names used to track parsing errors. We are dropping them until
-			// we add support for --strict parsing.
-			if shortName == types.ColumnNameError || shortName == types.ColumnNameErrorDetails {
-				continue
-			}
+			isErrorColumn := ident.ColumnType() == types.ColumnTypeGenerated &&
+				shortName == types.ColumnNameError || shortName == types.ColumnNameErrorDetails
 
 			forEachNotNullRowColValue(numRows, parsedCol, func(rowIdx int) {
 				parsedVal := parsedCol.Value(rowIdx)
@@ -155,7 +151,7 @@ func (b *streamsResultBuilder) CollectRecord(rec arrow.RecordBatch) {
 					b.rowBuilders[rowIdx].metadataBuilder.Del(shortName)
 				}
 				// If the parsed value is empty, the builder won't accept it as it's not a valid Prometheus-style label. We must add it later for LogQL compatibility.
-				if parsedVal == "" {
+				if parsedVal == "" && !isErrorColumn {
 					b.rowBuilders[rowIdx].parsedEmptyKeys = append(b.rowBuilders[rowIdx].parsedEmptyKeys, shortName)
 				}
 			})
@@ -167,8 +163,8 @@ func (b *streamsResultBuilder) CollectRecord(rec arrow.RecordBatch) {
 		lbs := b.rowBuilders[rowIdx].lbsBuilder.Labels()
 		ts := b.rowBuilders[rowIdx].timestamp
 		line := b.rowBuilders[rowIdx].line
-		// Ignore rows that don't have stream labels, log line, or timestamp
-		if line == "" || ts.IsZero() || lbs.IsEmpty() {
+		// Ignore rows that don't have stream labels, or timestamp
+		if ts.IsZero() || lbs.IsEmpty() {
 			b.resetRowBuilder(rowIdx)
 			continue
 		}

pkg/engine/compat_test.go

@@ -30,7 +30,7 @@ func TestStreamsResultBuilder(t *testing.T) {
 		require.NotNil(t, builder.Build(stats.Result{}, md).Data)
 	})
 
-	t.Run("rows without log line, timestamp, or labels are ignored", func(t *testing.T) {
+	t.Run("rows without timestamp, or labels are ignored", func(t *testing.T) {
 		colTs := semconv.ColumnIdentTimestamp
 		colMsg := semconv.ColumnIdentMessage
 		colEnv := semconv.NewIdentifier("env", types.ColumnTypeMetadata, types.Loki.String)
@@ -70,7 +70,7 @@ func TestStreamsResultBuilder(t *testing.T) {
 		err := collectResult(context.Background(), pipeline, builder)
 
 		require.NoError(t, err)
-		require.Equal(t, 0, builder.Len(), "expected no entries to be collected")
+		require.Equal(t, 1, builder.Len(), "expected 1 entry to be collected")
 	})
 
 	t.Run("successful conversion of labels, log line, timestamp, and structured metadata ", func(t *testing.T) {

pkg/logql/bench/generator.go

@@ -381,8 +381,12 @@ func (g *Generator) generateEntriesForStream(meta StreamMetadata) []logproto.Ent
 				}
 			}
 
-			// Generate log line using the application's generators for the selected format
-			line := app.LogGenerator(level, entryTs, faker)
+			line := ""
+			// Leave 3% of logs lines empty
+			if g.rnd.Float32() > 0.03 {
+				// Generate log line using the application's generators for the selected format
+				line = app.LogGenerator(level, entryTs, faker)
+			}
 
 			// Create metadata in a deterministic order
 			var metadata []logproto.LabelAdapter

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

@@ -208,7 +208,7 @@ queries:
       unwrappable_fields:
         - ttl
   - description: HTTP status code distribution
-    query: sum by (level) (count_over_time(${SELECTOR} | json [${RANGE}]))
+    query: sum by (level) (count_over_time(${SELECTOR} | json | __error__="" [${RANGE}]))
     kind: metric
     time_range:
       length: 24h