feat: implement unwrap as a projection

Author: trevorwhitney

pkg/engine/compat.go

@@ -124,7 +124,7 @@ func (b *streamsResultBuilder) collectRow(rec arrow.Record, i int) (labels.Label
 				// 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 colName == types.ColumnNameParsedError || colName == types.ColumnNameParsedErrorDetails {
+				if colName == types.ColumnNameError || colName == types.ColumnNameErrorDetails {
 					continue
 				}
 

pkg/engine/internal/datatype/arrow.go

@@ -12,6 +12,7 @@ var (
 		Timestamp DataType
 		Duration  DataType
 		Bytes     DataType
+    Struct    DataType
 	}{
 		Null:      tNull{},
 		Bool:      tBool{},
@@ -21,6 +22,7 @@ var (
 		Timestamp: tTimestamp{},
 		Duration:  tDuration{},
 		Bytes:     tBytes{},
+    Struct:    tStruct{},
 	}
 
 	Arrow = struct {

pkg/engine/internal/datatype/types.go

@@ -18,6 +18,7 @@ const (
 	STRING  = Type(arrow.STRING)
 	INT64   = Type(arrow.INT64)
 	FLOAT64 = Type(arrow.FLOAT64)
+	STRUCT  = Type(arrow.STRUCT)
 )
 
 func (t Type) String() string {
@@ -32,6 +33,8 @@ func (t Type) String() string {
 		return "INT64"
 	case FLOAT64:
 		return "FLOAT64"
+	case STRUCT:
+		return "STRUCT"
 	default:
 		return "INVALID"
 	}
@@ -91,6 +94,19 @@ func (tBytes) ID() Type                  { return INT64 }
 func (tBytes) String() string            { return "bytes" }
 func (tBytes) ArrowType() arrow.DataType { return Arrow.Integer }
 
+type tStruct struct {
+	arrowType *arrow.StructType
+}
+
+func (t tStruct) ID() Type                  { return STRUCT }
+func (t tStruct) String() string            { return "struct" }
+func (t tStruct) ArrowType() arrow.DataType { return t.arrowType }
+
+// NewStructType creates a DataType from an Arrow StructType
+func NewStructType(arrowType *arrow.StructType) DataType {
+	return tStruct{arrowType: arrowType}
+}
+
 var (
 	names = map[string]DataType{
 		Loki.Null.String():      Loki.Null,
@@ -101,6 +117,7 @@ var (
 		Loki.Timestamp.String(): Loki.Timestamp,
 		Loki.Duration.String():  Loki.Duration,
 		Loki.Bytes.String():     Loki.Bytes,
+		Loki.Struct.String():    Loki.Struct,
 	}
 )
 

pkg/engine/internal/executor/expressions.go

@@ -13,7 +13,19 @@ import (
 	"github.com/grafana/loki/v3/pkg/engine/internal/planner/physical"
 )
 
-type expressionEvaluator struct{}
+type expressionEvaluator struct {
+	allocator memory.Allocator
+}
+
+func newExpressionEvaluator(allocator memory.Allocator) expressionEvaluator {
+	if allocator == nil {
+		allocator = memory.DefaultAllocator
+	}
+
+	return expressionEvaluator{
+		allocator: allocator,
+	}
+}
 
 func (e expressionEvaluator) eval(expr physical.Expression, input arrow.Record) (ColumnVector, error) {
 	switch expr := expr.(type) {
@@ -135,6 +147,17 @@ func (e expressionEvaluator) eval(expr physical.Expression, input arrow.Record)
 			return nil, fmt.Errorf("failed to lookup binary function for signature %v(%v,%v): %w", expr.Op, lhs.Type().ArrowType(), rhs.Type().ArrowType(), err)
 		}
 		return fn.Evaluate(lhs, rhs)
+	case *physical.UnwrapExpr:
+		arr, err := unwrap(expr.Op, expr.Ref.Column, input, e.allocator)
+		if err != nil {
+			return nil, err
+		}
+
+		return &ArrayStruct{
+			array: arr,
+			ct:    types.ColumnTypeGenerated,
+			rows:  input.NumRows(),
+		}, nil
 	}
 
 	return nil, fmt.Errorf("unknown expression: %v", expr)
@@ -279,6 +302,91 @@ func (a *Array) Len() int64 {
 	return int64(a.array.Len())
 }
 
+// ArrayStruct represents multiple columns of data, stored as an [array.Struct].
+// It implements the ColumnVector interface while preserving access to individual fields.
+type ArrayStruct struct {
+	array *array.Struct
+	ct    types.ColumnType
+	rows  int64
+}
+
+var _ ColumnVector = (*ArrayStruct)(nil)
+
+// NewArrayStruct creates a new ArrayStruct from an array.Struct
+func NewArrayStruct(arr *array.Struct, ct types.ColumnType) *ArrayStruct {
+	return &ArrayStruct{
+		array: arr,
+		ct:    ct,
+		rows:  int64(arr.Len()),
+	}
+}
+
+// ToArray implements ColumnVector.
+// Returns the underlying struct array.
+func (a *ArrayStruct) ToArray() arrow.Array {
+	return a.array
+}
+
+// Value implements ColumnVector.
+// Returns a map of field names to values at the specified index.
+func (a *ArrayStruct) Value(i int) any {
+	if a.array.IsNull(i) || !a.array.IsValid(i) {
+		return nil
+	}
+
+	// Return a map representing the struct's fields at this index
+	result := make(map[string]any)
+	structType := a.array.DataType().(*arrow.StructType)
+
+	for fieldIdx := 0; fieldIdx < a.array.NumField(); fieldIdx++ {
+		field := a.array.Field(fieldIdx)
+		if field.IsNull(i) {
+			continue
+		}
+		fieldMeta := structType.Field(fieldIdx)
+
+		// Extract value from the field array at index i
+		var value any
+		switch arr := field.(type) {
+		case *array.Boolean:
+			value = arr.Value(i)
+		case *array.String:
+			value = arr.Value(i)
+		case *array.Int64:
+			value = arr.Value(i)
+		case *array.Uint64:
+			value = arr.Value(i)
+		case *array.Float64:
+			value = arr.Value(i)
+		default:
+			value = nil
+		}
+		result[fieldMeta.Name] = value
+	}
+
+	return result
+}
+
+// Type implements ColumnVector.
+func (a *ArrayStruct) Type() datatype.DataType {
+	dt := a.array.DataType()
+	st, ok := dt.(*arrow.StructType)
+	if !ok {
+		return nil
+	}
+	return datatype.NewStructType(st)
+}
+
+// ColumnType implements ColumnVector.
+func (a *ArrayStruct) ColumnType() types.ColumnType {
+	return a.ct
+}
+
+// Len implements ColumnVector.
+func (a *ArrayStruct) Len() int64 {
+	return a.rows
+}
+
 // CoalesceVector represents multiple columns with the same name but different [types.ColumnType]
 // Vectors are ordered by precedence (highest precedence first).
 type CoalesceVector struct {