chore(engine): Column naming conventions (#19396)

In the new engine, we need fully qualified column names, since columns from different sources can have the same name.

Right now, the distinction between columns with the same name is implemented using the `Metadata` field on the `arrow.Field`. However, it is quite cumbersome to parse the column type and data type from this generic map.

This PR introduces package with naming conventions for columns, defined by name, data type, and column type. So, this information can be encoded into the `Name` field of the `arrow.Field`. The convention is defined as 

```
[DATA_TYPE].[COLUMN_TYPE].[COLUMN_NAME]
```

#### Examples:
* `utf8.label.service_name`
* `timestamp_ns.builtin.timestamp`

The column type can easily be converted into a `Scope`, which is defined by an origin and type.

The mapping is as follows:

```
ColumnTypeBuiltin   -> Scope{Record, Attribute}
ColumnTypeMetadata  -> Scope{Record, Builtin}
ColumnTypeLabel     -> Scope{Resource, Attribute}
ColumnTypeParsed    -> Scope{Generated, Attribute}
ColumnTypeGenerated -> Scope{Generated, Builtin}
ColumnTypeAmbiguous -> Scope{Unscoped, Attribute}
```

---
Signed-off-by: Christian Haudum <[email protected]>

Author: chaudum

pkg/engine/compat.go

@@ -9,7 +9,6 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/promql"
 
-	"github.com/grafana/loki/v3/pkg/engine/internal/datatype"
 	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/logproto"
 	"github.com/grafana/loki/v3/pkg/logqlmodel"
@@ -311,7 +310,7 @@ func collectSamplesFromRow(builder *labels.Builder, rec arrow.Record, i int) (pr
 			sample.F = col.Value(i)
 		default:
 			// allow any string columns
-			if colDataType == datatype.Loki.String.String() {
+			if colDataType == types.Loki.String.String() {
 				builder.Set(colName, col.(*array.String).Value(i))
 			}
 		}

pkg/engine/compat_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/stretchr/testify/require"
 
-	"github.com/grafana/loki/v3/pkg/engine/internal/datatype"
 	"github.com/grafana/loki/v3/pkg/engine/internal/executor"
 	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/logproto"
@@ -28,8 +27,8 @@ func TestStreamsResultBuilder(t *testing.T) {
 	alloc := memory.NewCheckedAllocator(memory.DefaultAllocator)
 	defer alloc.AssertSize(t, 0)
 
-	mdTypeLabel := datatype.ColumnMetadata(types.ColumnTypeLabel, datatype.Loki.String)
-	mdTypeMetadata := datatype.ColumnMetadata(types.ColumnTypeMetadata, datatype.Loki.String)
+	mdTypeLabel := types.ColumnMetadata(types.ColumnTypeLabel, types.Loki.String)
+	mdTypeMetadata := types.ColumnMetadata(types.ColumnTypeMetadata, types.Loki.String)
 
 	t.Run("empty builder returns non-nil result", func(t *testing.T) {
 		builder := newStreamsResultBuilder()
@@ -40,8 +39,8 @@ func TestStreamsResultBuilder(t *testing.T) {
 	t.Run("rows without log line, timestamp, or labels are ignored", func(t *testing.T) {
 		schema := arrow.NewSchema(
 			[]arrow.Field{
-				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp},
-				{Name: types.ColumnNameBuiltinMessage, Type: arrow.BinaryTypes.String, Metadata: datatype.ColumnMetadataBuiltinMessage},
+				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp},
+				{Name: types.ColumnNameBuiltinMessage, Type: arrow.BinaryTypes.String, Metadata: types.ColumnMetadataBuiltinMessage},
 				{Name: "env", Type: arrow.BinaryTypes.String, Metadata: mdTypeLabel},
 			},
 			nil,
@@ -97,8 +96,8 @@ func TestStreamsResultBuilder(t *testing.T) {
 	t.Run("successful conversion of labels, log line, timestamp, and structured metadata ", func(t *testing.T) {
 		schema := arrow.NewSchema(
 			[]arrow.Field{
-				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp},
-				{Name: types.ColumnNameBuiltinMessage, Type: arrow.BinaryTypes.String, Metadata: datatype.ColumnMetadataBuiltinMessage},
+				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp},
+				{Name: types.ColumnNameBuiltinMessage, Type: arrow.BinaryTypes.String, Metadata: types.ColumnMetadataBuiltinMessage},
 				{Name: "env", Type: arrow.BinaryTypes.String, Metadata: mdTypeLabel},
 				{Name: "namespace", Type: arrow.BinaryTypes.String, Metadata: mdTypeLabel},
 				{Name: "traceID", Type: arrow.BinaryTypes.String, Metadata: mdTypeMetadata},
@@ -167,7 +166,7 @@ func TestStreamsResultBuilder(t *testing.T) {
 }
 
 func TestVectorResultBuilder(t *testing.T) {
-	mdTypeString := datatype.ColumnMetadata(types.ColumnTypeAmbiguous, datatype.Loki.String)
+	mdTypeString := types.ColumnMetadata(types.ColumnTypeAmbiguous, types.Loki.String)
 	alloc := memory.NewCheckedAllocator(memory.DefaultAllocator)
 	defer alloc.AssertSize(t, 0)
 
@@ -180,8 +179,8 @@ func TestVectorResultBuilder(t *testing.T) {
 	t.Run("successful conversion of vector data", func(t *testing.T) {
 		schema := arrow.NewSchema(
 			[]arrow.Field{
-				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp},
-				{Name: types.ColumnNameGeneratedValue, Type: arrow.PrimitiveTypes.Float64, Metadata: datatype.ColumnMetadata(types.ColumnTypeGenerated, datatype.Loki.Float)},
+				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp},
+				{Name: types.ColumnNameGeneratedValue, Type: arrow.PrimitiveTypes.Float64, Metadata: types.ColumnMetadata(types.ColumnTypeGenerated, types.Loki.Float)},
 				{Name: "instance", Type: arrow.BinaryTypes.String, Metadata: mdTypeString},
 				{Name: "job", Type: arrow.BinaryTypes.String, Metadata: mdTypeString},
 			},
@@ -235,8 +234,8 @@ func TestVectorResultBuilder(t *testing.T) {
 	t.Run("rows without timestamp or value are ignored", func(t *testing.T) {
 		schema := arrow.NewSchema(
 			[]arrow.Field{
-				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp},
-				{Name: types.ColumnNameGeneratedValue, Type: arrow.PrimitiveTypes.Float64, Metadata: datatype.ColumnMetadata(types.ColumnTypeGenerated, datatype.Loki.Float)},
+				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp},
+				{Name: types.ColumnNameGeneratedValue, Type: arrow.PrimitiveTypes.Float64, Metadata: types.ColumnMetadata(types.ColumnTypeGenerated, types.Loki.Float)},
 				{Name: "instance", Type: arrow.BinaryTypes.String, Metadata: mdTypeString},
 			},
 			nil,
@@ -262,7 +261,7 @@ func TestVectorResultBuilder(t *testing.T) {
 }
 
 func TestMatrixResultBuilder(t *testing.T) {
-	mdTypeString := datatype.ColumnMetadata(types.ColumnTypeAmbiguous, datatype.Loki.String)
+	mdTypeString := types.ColumnMetadata(types.ColumnTypeAmbiguous, types.Loki.String)
 	alloc := memory.NewCheckedAllocator(memory.DefaultAllocator)
 	defer alloc.AssertSize(t, 0)
 
@@ -275,8 +274,8 @@ func TestMatrixResultBuilder(t *testing.T) {
 	t.Run("successful conversion of matrix data", func(t *testing.T) {
 		schema := arrow.NewSchema(
 			[]arrow.Field{
-				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp},
-				{Name: types.ColumnNameGeneratedValue, Type: arrow.PrimitiveTypes.Float64, Metadata: datatype.ColumnMetadata(types.ColumnTypeGenerated, datatype.Loki.Float)},
+				{Name: types.ColumnNameBuiltinTimestamp, Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp},
+				{Name: types.ColumnNameGeneratedValue, Type: arrow.PrimitiveTypes.Float64, Metadata: types.ColumnMetadata(types.ColumnTypeGenerated, types.Loki.Float)},
 				{Name: "instance", Type: arrow.BinaryTypes.String, Metadata: mdTypeString},
 				{Name: "job", Type: arrow.BinaryTypes.String, Metadata: mdTypeString},
 			},

pkg/engine/internal/datatype/util.go

@@ -12,9 +12,8 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/memory"
 	"github.com/cespare/xxhash/v2"
 
-	"github.com/grafana/loki/v3/pkg/engine/internal/datatype"
-	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/engine/internal/planner/physical"
+	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 )
 
 type groupState struct {
@@ -125,15 +124,15 @@ func (a *aggregator) BuildRecord() (arrow.Record, error) {
 	fields = append(fields,
 		arrow.Field{
 			Name:     types.ColumnNameBuiltinTimestamp,
-			Type:     datatype.Arrow.Timestamp,
+			Type:     types.Arrow.Timestamp,
 			Nullable: false,
-			Metadata: datatype.ColumnMetadataBuiltinTimestamp,
+			Metadata: types.ColumnMetadataBuiltinTimestamp,
 		},
 		arrow.Field{
 			Name:     types.ColumnNameGeneratedValue,
-			Type:     datatype.Arrow.Float,
+			Type:     types.Arrow.Float,
 			Nullable: false,
-			Metadata: datatype.ColumnMetadata(types.ColumnTypeGenerated, datatype.Loki.Float),
+			Metadata: types.ColumnMetadata(types.ColumnTypeGenerated, types.Loki.Float),
 		},
 	)
 
@@ -145,9 +144,9 @@ func (a *aggregator) BuildRecord() (arrow.Record, error) {
 
 		fields = append(fields, arrow.Field{
 			Name:     colExpr.Ref.Column,
-			Type:     datatype.Arrow.String,
+			Type:     types.Arrow.String,
 			Nullable: true,
-			Metadata: datatype.ColumnMetadata(colExpr.Ref.Type, datatype.Loki.String),
+			Metadata: types.ColumnMetadata(colExpr.Ref.Type, types.Loki.String),
 		})
 	}
 

pkg/engine/internal/executor/aggregator.go

@@ -7,8 +7,8 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/memory"
 	"github.com/stretchr/testify/require"
 
-	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/engine/internal/planner/physical"
+	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/util/arrowtest"
 )
 

pkg/engine/internal/executor/aggregator_test.go

@@ -16,9 +16,8 @@ import (
 
 	"github.com/grafana/loki/v3/pkg/dataobj/sections/logs"
 	"github.com/grafana/loki/v3/pkg/dataobj/sections/streams"
-	"github.com/grafana/loki/v3/pkg/engine/internal/datatype"
-	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/engine/internal/planner/physical"
+	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 )
 
 type dataobjScanOptions struct {
@@ -258,25 +257,25 @@ func logsColumnToEngineField(col *logs.Column) (arrow.Field, error) {
 	case logs.ColumnTypeTimestamp:
 		return arrow.Field{
 			Name:     types.ColumnNameBuiltinTimestamp,
-			Type:     datatype.Arrow.Timestamp,
+			Type:     types.Arrow.Timestamp,
 			Nullable: true,
-			Metadata: datatype.ColumnMetadata(types.ColumnTypeBuiltin, datatype.Loki.Timestamp),
+			Metadata: types.ColumnMetadata(types.ColumnTypeBuiltin, types.Loki.Timestamp),
 		}, nil
 
 	case logs.ColumnTypeMessage:
 		return arrow.Field{
 			Name:     types.ColumnNameBuiltinMessage,
-			Type:     datatype.Arrow.String,
+			Type:     types.Arrow.String,
 			Nullable: true,
-			Metadata: datatype.ColumnMetadata(types.ColumnTypeBuiltin, datatype.Loki.String),
+			Metadata: types.ColumnMetadata(types.ColumnTypeBuiltin, types.Loki.String),
 		}, nil
 
 	case logs.ColumnTypeMetadata:
 		return arrow.Field{
 			Name:     col.Name,
-			Type:     datatype.Arrow.String,
+			Type:     types.Arrow.String,
 			Nullable: true,
-			Metadata: datatype.ColumnMetadata(types.ColumnTypeMetadata, datatype.Loki.String),
+			Metadata: types.ColumnMetadata(types.ColumnTypeMetadata, types.Loki.String),
 		}, nil
 	}
 

pkg/engine/internal/executor/dataobjscan.go

@@ -10,9 +10,8 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/scalar"
 
 	"github.com/grafana/loki/v3/pkg/dataobj/sections/logs"
-	"github.com/grafana/loki/v3/pkg/engine/internal/datatype"
-	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/engine/internal/planner/physical"
+	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 )
 
 // buildLogsPredicate builds a [logs.Predicate] from an expr. The columns slice
@@ -38,8 +37,8 @@ func buildLogsPredicate(expr physical.Expression, columns []*logs.Column) (logs.
 		return buildLogsBinaryPredicate(expr, columns)
 
 	case *physical.LiteralExpr:
-		if expr.Literal.Type() == datatype.Loki.Bool {
-			val := expr.Literal.(datatype.TypedLiteral[bool]).Value()
+		if expr.Literal.Type() == types.Loki.Bool {
+			val := expr.Literal.(types.TypedLiteral[bool]).Value()
 			if val {
 				return logs.TruePredicate{}, nil
 			}
@@ -265,12 +264,12 @@ func findColumn(ref types.ColumnRef, columns []*logs.Column) (*logs.Column, erro
 }
 
 // buildDataobjScalar builds a dataobj-compatible [scalar.Scalar] from a
-// [datatype.Literal].
-func buildDataobjScalar(lit datatype.Literal) (scalar.Scalar, error) {
+// [types.Literal].
+func buildDataobjScalar(lit types.Literal) (scalar.Scalar, error) {
 	// [logs.ReaderOptions.Validate] specifies that all scalars must be one of
 	// the given types:
 	//
-	// * [scalar.Null] (of any datatype)
+	// * [scalar.Null] (of any types)
 	// * [scalar.Int64]
 	// * [scalar.Uint64]
 	// * [scalar.Timestamp] (nanosecond precision)
@@ -279,18 +278,18 @@ func buildDataobjScalar(lit datatype.Literal) (scalar.Scalar, error) {
 	// All of our mappings below evaluate to one of the above types.
 
 	switch lit := lit.(type) {
-	case datatype.NullLiteral:
+	case types.NullLiteral:
 		return scalar.ScalarNull, nil
-	case datatype.IntegerLiteral:
+	case types.IntegerLiteral:
 		return scalar.NewInt64Scalar(lit.Value()), nil
-	case datatype.BytesLiteral:
-		// [datatype.BytesLiteral] refers to byte sizes, not binary data.
+	case types.BytesLiteral:
+		// [types.BytesLiteral] refers to byte sizes, not binary data.
 		return scalar.NewInt64Scalar(int64(lit.Value())), nil
-	case datatype.TimestampLiteral:
+	case types.TimestampLiteral:
 		ts := arrow.Timestamp(lit.Value())
 		tsType := arrow.FixedWidthTypes.Timestamp_ns
 		return scalar.NewTimestampScalar(ts, tsType), nil
-	case datatype.StringLiteral:
+	case types.StringLiteral:
 		buf := memory.NewBufferBytes([]byte(lit.Value()))
 		return scalar.NewBinaryScalar(buf, arrow.BinaryTypes.Binary), nil
 	}

pkg/engine/internal/executor/dataobjscan_predicate.go

@@ -9,9 +9,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/grafana/loki/v3/pkg/dataobj/sections/logs"
-	"github.com/grafana/loki/v3/pkg/engine/internal/datatype"
-	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/engine/internal/planner/physical"
+	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 )
 
 func Test_buildLogsPredicate(t *testing.T) {
@@ -156,7 +155,7 @@ func Test_buildLogsPredicate(t *testing.T) {
 			expr: &physical.BinaryExpr{
 				Op:    types.BinaryOpEq,
 				Left:  columnRef(types.ColumnTypeBuiltin, types.ColumnNameBuiltinTimestamp),
-				Right: physical.NewLiteral(datatype.Bytes(1024)),
+				Right: physical.NewLiteral(types.Bytes(1024)),
 			},
 			expect: logs.EqualPredicate{
 				Column: timestampColumn,
@@ -168,7 +167,7 @@ func Test_buildLogsPredicate(t *testing.T) {
 			expr: &physical.BinaryExpr{
 				Op:    types.BinaryOpEq,
 				Left:  columnRef(types.ColumnTypeBuiltin, types.ColumnNameBuiltinTimestamp),
-				Right: physical.NewLiteral(datatype.Timestamp(1609459200000000000)), // 2021-01-01 00:00:00 UTC in nanoseconds
+				Right: physical.NewLiteral(types.Timestamp(1609459200000000000)), // 2021-01-01 00:00:00 UTC in nanoseconds
 			},
 			expect: logs.EqualPredicate{
 				Column: timestampColumn,

pkg/engine/internal/executor/dataobjscan_predicate_test.go

@@ -13,17 +13,16 @@ import (
 	"github.com/grafana/loki/v3/pkg/dataobj/consumer/logsobj"
 	"github.com/grafana/loki/v3/pkg/dataobj/sections/logs"
 	"github.com/grafana/loki/v3/pkg/dataobj/sections/streams"
-	"github.com/grafana/loki/v3/pkg/engine/internal/datatype"
-	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/engine/internal/planner/physical"
+	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/logproto"
 
 	"github.com/grafana/loki/pkg/push"
 )
 
 var (
-	labelMD    = datatype.ColumnMetadata(types.ColumnTypeLabel, datatype.Loki.String)
-	metadataMD = datatype.ColumnMetadata(types.ColumnTypeMetadata, datatype.Loki.String)
+	labelMD    = types.ColumnMetadata(types.ColumnTypeLabel, types.Loki.String)
+	metadataMD = types.ColumnMetadata(types.ColumnTypeMetadata, types.Loki.String)
 )
 
 func Test_dataobjScan(t *testing.T) {
@@ -94,8 +93,8 @@ func Test_dataobjScan(t *testing.T) {
 			{Name: "service", Type: arrow.BinaryTypes.String, Metadata: labelMD, Nullable: true},
 			{Name: "guid", Type: arrow.BinaryTypes.String, Metadata: metadataMD, Nullable: true},
 			{Name: "pod", Type: arrow.BinaryTypes.String, Metadata: metadataMD, Nullable: true},
-			{Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp, Nullable: true},
-			{Name: "message", Type: arrow.BinaryTypes.String, Metadata: datatype.ColumnMetadataBuiltinMessage, Nullable: true},
+			{Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp, Nullable: true},
+			{Name: "message", Type: arrow.BinaryTypes.String, Metadata: types.ColumnMetadataBuiltinMessage, Nullable: true},
 		}
 
 		expectCSV := `prod,loki,eeee-ffff-aaaa-bbbb,NULL,1970-01-01 00:00:10,goodbye world
@@ -125,7 +124,7 @@ prod,notloki,NULL,notloki-pod-1,1970-01-01 00:00:02,hello world`
 
 		expectFields := []arrow.Field{
 			{Name: "env", Type: arrow.BinaryTypes.String, Metadata: labelMD, Nullable: true},
-			{Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp, Nullable: true},
+			{Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp, Nullable: true},
 		}
 
 		expectCSV := `prod,1970-01-01 00:00:10
@@ -155,8 +154,8 @@ prod,1970-01-01 00:00:02`
 			{Name: "service", Type: arrow.BinaryTypes.String, Metadata: labelMD, Nullable: true},
 			{Name: "guid", Type: arrow.BinaryTypes.String, Metadata: metadataMD, Nullable: true},
 			{Name: "pod", Type: arrow.BinaryTypes.String, Metadata: metadataMD, Nullable: true},
-			{Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp, Nullable: true},
-			{Name: "message", Type: arrow.BinaryTypes.String, Metadata: datatype.ColumnMetadataBuiltinMessage, Nullable: true},
+			{Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp, Nullable: true},
+			{Name: "message", Type: arrow.BinaryTypes.String, Metadata: types.ColumnMetadataBuiltinMessage, Nullable: true},
 		}
 
 		expectCSV := `prod,notloki,NULL,notloki-pod-1,1970-01-01 00:00:03,goodbye world
@@ -271,8 +270,8 @@ func Test_dataobjScan_DuplicateColumns(t *testing.T) {
 			{Name: "namespace", Type: arrow.BinaryTypes.String, Metadata: metadataMD, Nullable: true},
 			{Name: "pod", Type: arrow.BinaryTypes.String, Metadata: metadataMD, Nullable: true},
 
-			{Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: datatype.ColumnMetadataBuiltinTimestamp, Nullable: true},
-			{Name: "message", Type: arrow.BinaryTypes.String, Metadata: datatype.ColumnMetadataBuiltinMessage, Nullable: true},
+			{Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ns, Metadata: types.ColumnMetadataBuiltinTimestamp, Nullable: true},
+			{Name: "message", Type: arrow.BinaryTypes.String, Metadata: types.ColumnMetadataBuiltinMessage, Nullable: true},
 		}
 
 		expectCSV := `prod,namespace-2,NULL,loki,NULL,NULL,1970-01-01 00:00:03,message 3

pkg/engine/internal/executor/dataobjscan_test.go

@@ -17,8 +17,8 @@ import (
 	"github.com/grafana/loki/v3/pkg/dataobj"
 	"github.com/grafana/loki/v3/pkg/dataobj/sections/logs"
 	"github.com/grafana/loki/v3/pkg/dataobj/sections/streams"
-	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 	"github.com/grafana/loki/v3/pkg/engine/internal/planner/physical"
+	"github.com/grafana/loki/v3/pkg/engine/internal/types"
 )
 
 var tracer = otel.Tracer("pkg/engine/internal/executor")