chore: Fix column name regression (#19468)

The temporary streamID column from the dataobj scan node caused an error when the expression evaluator tried to parse the FQN, which was not valid any more after the recent change that implemented the semantic naming conventions.

Signed-off-by: Christian Haudum <christian.haudum@gmail.com>

Author: chaudum

pkg/engine/internal/executor/dataobjscan.go

@@ -221,16 +221,9 @@ func (s *dataobjScan) initLogs() error {
 		return fmt.Errorf("logs.Reader returned schema with %d fields, expected %d", got, want)
 	}
 
+	// Convert the logs columns to engine-compatible fields.
 	var desiredFields []arrow.Field
-	for i, col := range columnsToRead {
-		if col.Type == logs.ColumnTypeStreamID {
-			// The stream ID field should be left as-is for use with the streams
-			// injector.
-			desiredFields = append(desiredFields, origSchema.Field(i))
-			continue
-		}
-
-		// Convert the logs column to an engine-compatible field.
+	for _, col := range columnsToRead {
 		field, err := logsColumnToEngineField(col)
 		if err != nil {
 			return err
@@ -255,6 +248,9 @@ func makeScalars[S ~[]E, E any](s S) []scalar.Scalar {
 // engine.
 func logsColumnToEngineField(col *logs.Column) (arrow.Field, error) {
 	switch col.Type {
+	case logs.ColumnTypeStreamID:
+		return semconv.FieldFromIdent(streamInjectorColumnIdent, true), nil
+
 	case logs.ColumnTypeTimestamp:
 		return semconv.FieldFromIdent(semconv.ColumnIdentTimestamp, true), nil
 

pkg/engine/internal/executor/stream_injector.go

@@ -14,7 +14,10 @@ import (
 	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 )
 
-var streamInjectorColumnName = "stream_id.int64"
+var (
+	streamInjectorColumnName  = "int64.generated.stream_id"
+	streamInjectorColumnIdent = semconv.MustParseFQN(streamInjectorColumnName)
+)
 
 // streamInjector injects stream labels into a logs Arrow record, replacing the
 // streams ID column with columns for the labels composing those streams.
@@ -40,14 +43,14 @@ func newStreamInjector(alloc memory.Allocator, view *streamsView) *streamInjecto
 //
 // The returned record must be Release()d by the caller when no longer needed.
 func (si *streamInjector) Inject(ctx context.Context, in arrow.Record) (arrow.Record, error) {
-	streamIDCol, streamIDIndex, err := columnForFQN(streamInjectorColumnName, in)
+	streamIDCol, streamIDIndex, err := columnForIdent(streamInjectorColumnIdent, in)
 	if err != nil {
 		return nil, err
 	}
 
 	streamIDValues, ok := streamIDCol.(*array.Int64)
 	if !ok {
-		return nil, fmt.Errorf("column %s must be of type int64, got %s", streamInjectorColumnName, in.Schema().Field(streamIDIndex))
+		return nil, fmt.Errorf("column %s must be of type int64, got %s", streamInjectorColumnIdent.FQN(), in.Schema().Field(streamIDIndex))
 	}
 
 	type labelColumn struct {