chore(engine): Refactor streams result builder to use column-oriented processsing

[ivkalita]

What this PR does / why we need it:

Change streamsResultBuilder.CollectRecord() from row-by-row to column-by-column iteration. This better aligns with Arrow's columnar memory layout, improving CPU cache locality by reading contiguous memory regions.

I also remove IsValid checks given that IsNull checks are present. IsValid seems to be an exact opposite of IsNull (based on this code).

The slices and label builders for intermediate rows data are reused between CollectRecord.

old vs latest version

goos: darwin
goarch: arm64
pkg: github.com/grafana/loki/v3/pkg/engine
cpu: Apple M4
                                           │   old.txt   │      new.row-struct-final.txt       │
                                           │   sec/op    │   sec/op     vs base                │
StreamsResultBuilder/records_equal_size-10   8.798m ± 1%   6.495m ± 1%  -26.18% (p=0.000 n=10)
StreamsResultBuilder/record_two_bigger-10    14.05m ± 6%   10.12m ± 0%  -27.93% (p=0.000 n=10)
StreamsResultBuilder/record_two_smaller-10   6.889m ± 0%   5.114m ± 0%  -25.76% (p=0.000 n=10)
geomean                                      9.478m        6.954m       -26.63%

                                           │    old.txt    │       new.row-struct-final.txt       │
                                           │     B/op      │     B/op      vs base                │
StreamsResultBuilder/records_equal_size-10   13.751Mi ± 0%   7.637Mi ± 0%  -44.46% (p=0.000 n=10)
StreamsResultBuilder/record_two_bigger-10     20.80Mi ± 0%   13.47Mi ± 0%  -35.25% (p=0.000 n=10)
StreamsResultBuilder/record_two_smaller-10   10.402Mi ± 0%   6.706Mi ± 0%  -35.53% (p=0.000 n=10)
geomean                                       14.38Mi        8.836Mi       -38.57%

                                           │   old.txt    │      new.row-struct-final.txt       │
                                           │  allocs/op   │  allocs/op   vs base                │
StreamsResultBuilder/records_equal_size-10    82.23k ± 0%   25.28k ± 0%  -69.26% (p=0.000 n=10)
StreamsResultBuilder/record_two_bigger-10    123.26k ± 0%   44.31k ± 0%  -64.05% (p=0.000 n=10)
StreamsResultBuilder/record_two_smaller-10    61.73k ± 0%   22.28k ± 0%  -63.91% (p=0.000 n=10)
geomean                                       85.53k        29.22k       -65.83%

Special notes for your reviewer:

Checklist

• Reviewed the CONTRIBUTING.md guide (required)
• Documentation added
• Tests updated
• Title matches the required conventional commits format, see here
  • Note that Promtail is considered to be feature complete, and future development for logs collection will be in Grafana Alloy. As such, feat PRs are unlikely to be accepted unless a case can be made for the feature actually being a bug fix to existing behavior.
• Changes that require user attention or interaction to upgrade are documented in docs/sources/setup/upgrade/_index.md
• If the change is deprecating or removing a configuration option, update the deprecated-config.yaml and deleted-config.yaml files respectively in the tools/deprecated-config-checker directory. Example PR

[CLAassistant]

All committers have signed the CLA.

[chaudum]

Generally LGTM.
Left a suggestion how to reduce allocations. Also, feel free to add the benchmark code and run the benchmarks for old and new version and post the output of benchstat old.txt new.txt

[chaudum]

The benchmarks look great!

[ivkalita]

@chaudum I updated the implementation to use a slice of structs for rows buffer. Benchmark shows there is no major difference: here's benchstat between the previously reviewed implementation and the latest one (diff):

goos: darwin
goarch: arm64
pkg: github.com/grafana/loki/v3/pkg/engine
cpu: Apple M4
                                           │ new.base.txt │      new.row-struct-final.txt      │
                                           │    sec/op    │   sec/op     vs base               │
StreamsResultBuilder/records_equal_size-10    6.464m ± 1%   6.495m ± 1%       ~ (p=0.190 n=10)
StreamsResultBuilder/record_two_bigger-10     10.10m ± 0%   10.12m ± 0%       ~ (p=0.165 n=10)
StreamsResultBuilder/record_two_smaller-10    5.114m ± 0%   5.114m ± 0%       ~ (p=0.529 n=10)
geomean                                       6.936m        6.954m       +0.25%

                                           │ new.base.txt │      new.row-struct-final.txt       │
                                           │     B/op     │     B/op      vs base               │
StreamsResultBuilder/records_equal_size-10   7.637Mi ± 0%   7.637Mi ± 0%  -0.00% (p=0.000 n=10)
StreamsResultBuilder/record_two_bigger-10    13.47Mi ± 0%   13.47Mi ± 0%  -0.00% (p=0.000 n=10)
StreamsResultBuilder/record_two_smaller-10   6.706Mi ± 0%   6.706Mi ± 0%  -0.00% (p=0.000 n=10)
geomean                                      8.836Mi        8.836Mi       -0.00%

                                           │ new.base.txt │      new.row-struct-final.txt      │
                                           │  allocs/op   │  allocs/op   vs base               │
StreamsResultBuilder/records_equal_size-10    25.28k ± 0%   25.28k ± 0%  -0.02% (p=0.000 n=10)
StreamsResultBuilder/record_two_bigger-10     44.32k ± 0%   44.31k ± 0%  -0.02% (p=0.000 n=10)
StreamsResultBuilder/record_two_smaller-10    22.28k ± 0%   22.28k ± 0%  -0.02% (p=0.000 n=10)
geomean                                       29.23k        29.22k       -0.02%

I also tried to use log.LabelsBuilder instead of 3 different builders but it didn't work out. log.LabelsBuilder does not support a scenario when we start with an empty set of labels and then gradually add more and more labels. .Set / .Add only works for overriding existing base labels (which are empty in our case). E.g.:

func TestLabelsBuilder_Playground(t *testing.T) {
	lbs := labels.EmptyLabels()
	b := NewBaseLabelsBuilder().ForLabels(lbs, labels.StableHash(lbs))
        // b.Set(ParsedLabel, "a", "b") // will be returned as part of result.Labels()
	b.Set(StreamLabel, "a", "b") // won't be returned as result.Labels()
	result := b.LabelsResult()

	require.Equal(t, "{a=\"b\"}", result.Labels().String())
}

I'm trying to add this support but it takes time and increases the size of the PR so I'd like to have a different PR for it, wdyt?