Skip to content

Commit 90d332a

Browse files
yeya24yeya24
committed
add projection pushdown to binary expression
Signed-off-by: yeya24 <benye@amazon.com>
1 parent dd52237 commit 90d332a

8 files changed

Lines changed: 803 additions & 14 deletions

File tree

‎engine/projection_binary_bench_test.go‎

Lines changed: 293 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,293 @@
1+
// Copyright (c) The Thanos Community Authors.
2+
// Licensed under the Apache License 2.0.
3+
4+
package engine
5+
6+
import (
7+
"context"
8+
"fmt"
9+
"testing"
10+
"time"
11+
12+
"github.com/thanos-io/promql-engine/execution/binary"
13+
"github.com/thanos-io/promql-engine/execution/model"
14+
"github.com/thanos-io/promql-engine/logicalplan"
15+
"github.com/thanos-io/promql-engine/query"
16+
17+
"github.com/efficientgo/core/testutil"
18+
"github.com/prometheus/prometheus/model/labels"
19+
"github.com/prometheus/prometheus/promql/parser"
20+
)
21+
22+
func BenchmarkBinaryProjectionPushdown(b *testing.B) {
23+
// Create mock series with many labels
24+
numSeries := 10000
25+
lhsSeries := make([]labels.Labels, numSeries)
26+
rhsSeries := make([]labels.Labels, 100)
27+
28+
// LHS: 10000 series with 20 labels each
29+
for i := range numSeries {
30+
lhsSeries[i] = labels.FromStrings(
31+
"__name__", "kube_pod_info",
32+
"cluster", "c1",
33+
"node", fmt.Sprintf("n%d", i%100),
34+
"namespace", fmt.Sprintf("ns%d", i),
35+
"pod", fmt.Sprintf("p%d", i),
36+
"label_a", fmt.Sprintf("a%d", i),
37+
"label_b", fmt.Sprintf("b%d", i),
38+
"label_c", fmt.Sprintf("c%d", i),
39+
"label_d", fmt.Sprintf("d%d", i),
40+
"label_e", fmt.Sprintf("e%d", i),
41+
"label_f", fmt.Sprintf("f%d", i),
42+
"label_g", fmt.Sprintf("g%d", i),
43+
"label_h", fmt.Sprintf("h%d", i),
44+
"label_i", fmt.Sprintf("i%d", i),
45+
"label_j", fmt.Sprintf("j%d", i),
46+
"label_k", fmt.Sprintf("k%d", i),
47+
"label_l", fmt.Sprintf("l%d", i),
48+
"label_m", fmt.Sprintf("m%d", i),
49+
"label_n", fmt.Sprintf("n%d", i),
50+
"label_o", fmt.Sprintf("o%d", i),
51+
)
52+
}
53+
54+
// RHS: 100 series with 3 labels each
55+
for i := range 100 {
56+
rhsSeries[i] = labels.FromStrings(
57+
"__name__", "kube_node_labels",
58+
"cluster", "c1",
59+
"node", fmt.Sprintf("n%d", i),
60+
"instance_type", fmt.Sprintf("t%d", i),
61+
)
62+
}
63+
64+
opts := &query.Options{
65+
Start: time.Unix(0, 0),
66+
End: time.Unix(3000, 0),
67+
Step: 30 * time.Second,
68+
}
69+
70+
b.Run("without_projection", func(b *testing.B) {
71+
b.ReportAllocs()
72+
for i := 0; b.Loop(); i++ {
73+
op, err := binary.NewVectorOperator(
74+
&mockOperator{series: lhsSeries},
75+
&mockOperator{series: rhsSeries},
76+
&parser.VectorMatching{
77+
Card: parser.CardManyToOne,
78+
MatchingLabels: []string{"cluster", "node"},
79+
On: true,
80+
Include: []string{"instance_type"},
81+
},
82+
parser.MUL,
83+
false,
84+
nil, // No projection
85+
opts,
86+
)
87+
testutil.Ok(b, err)
88+
series, _ := op.Series(context.Background())
89+
if i == 0 {
90+
b.Logf("Result series count: %d", len(series))
91+
if len(series) > 0 {
92+
b.Logf("First series labels: %v", series[0])
93+
b.Logf("First series label count: %d", series[0].Len())
94+
}
95+
}
96+
}
97+
})
98+
99+
b.Run("with_projection", func(b *testing.B) {
100+
projection := &logicalplan.Projection{
101+
Labels: []string{"namespace", "instance_type"},
102+
Include: true,
103+
}
104+
105+
b.ReportAllocs()
106+
for i := 0; b.Loop(); i++ {
107+
op, err := binary.NewVectorOperator(
108+
&mockOperator{series: lhsSeries},
109+
&mockOperator{series: rhsSeries},
110+
&parser.VectorMatching{
111+
Card: parser.CardManyToOne,
112+
MatchingLabels: []string{"cluster", "node"},
113+
On: true,
114+
Include: []string{"instance_type"},
115+
},
116+
parser.MUL,
117+
false,
118+
projection,
119+
opts,
120+
)
121+
testutil.Ok(b, err)
122+
series, _ := op.Series(context.Background())
123+
if i == 0 {
124+
b.Logf("Result series count: %d", len(series))
125+
if len(series) > 0 {
126+
b.Logf("First series labels: %v", series[0])
127+
b.Logf("First series label count: %d", series[0].Len())
128+
}
129+
}
130+
}
131+
})
132+
}
133+
134+
func BenchmarkBinaryProjectionPushdownOneToOne(b *testing.B) {
135+
numSeries := 1000
136+
lhsSeries := make([]labels.Labels, numSeries)
137+
rhsSeries := make([]labels.Labels, numSeries)
138+
for i := range numSeries {
139+
lhsSeries[i] = labels.FromStrings(
140+
"__name__", "metric_a",
141+
"cluster", "c1",
142+
"node", fmt.Sprintf("n%d", i),
143+
"namespace", fmt.Sprintf("ns%d", i),
144+
"pod", fmt.Sprintf("p%d", i),
145+
)
146+
rhsSeries[i] = labels.FromStrings(
147+
"__name__", "metric_b",
148+
"cluster", "c1",
149+
"node", fmt.Sprintf("n%d", i),
150+
"env", fmt.Sprintf("e%d", i),
151+
)
152+
}
153+
opts := &query.Options{
154+
Start: time.Unix(0, 0),
155+
End: time.Unix(3000, 0),
156+
Step: 30 * time.Second,
157+
}
158+
matching := &parser.VectorMatching{
159+
Card: parser.CardOneToOne,
160+
MatchingLabels: []string{"cluster", "node"},
161+
On: true,
162+
}
163+
164+
b.Run("without_projection", func(b *testing.B) {
165+
b.ReportAllocs()
166+
for b.Loop() {
167+
op, err := binary.NewVectorOperator(
168+
&mockOperator{series: lhsSeries},
169+
&mockOperator{series: rhsSeries},
170+
matching,
171+
parser.MUL,
172+
false,
173+
nil,
174+
opts,
175+
)
176+
testutil.Ok(b, err)
177+
_, _ = op.Series(context.Background())
178+
}
179+
})
180+
b.Run("with_projection", func(b *testing.B) {
181+
projection := &logicalplan.Projection{
182+
Labels: []string{"cluster", "node"},
183+
Include: true,
184+
}
185+
b.ReportAllocs()
186+
for b.Loop() {
187+
op, err := binary.NewVectorOperator(
188+
&mockOperator{series: lhsSeries},
189+
&mockOperator{series: rhsSeries},
190+
matching,
191+
parser.MUL,
192+
false,
193+
projection,
194+
opts,
195+
)
196+
testutil.Ok(b, err)
197+
_, _ = op.Series(context.Background())
198+
}
199+
})
200+
}
201+
202+
func BenchmarkBinaryProjectionPushdownOneToMany(b *testing.B) {
203+
// One-to-many: LHS 100, RHS 10000 (RHS is high-card after swap in execution).
204+
lhsNum, rhsNum := 100, 10000
205+
lhsSeries := make([]labels.Labels, lhsNum)
206+
rhsSeries := make([]labels.Labels, rhsNum)
207+
for i := range lhsNum {
208+
lhsSeries[i] = labels.FromStrings(
209+
"__name__", "kube_node_labels",
210+
"cluster", "c1",
211+
"node", fmt.Sprintf("n%d", i),
212+
"instance_type", fmt.Sprintf("t%d", i),
213+
)
214+
}
215+
for i := range rhsNum {
216+
rhsSeries[i] = labels.FromStrings(
217+
"__name__", "kube_pod_info",
218+
"cluster", "c1",
219+
"node", fmt.Sprintf("n%d", i%lhsNum),
220+
"namespace", fmt.Sprintf("ns%d", i),
221+
"pod", fmt.Sprintf("p%d", i),
222+
)
223+
}
224+
opts := &query.Options{
225+
Start: time.Unix(0, 0),
226+
End: time.Unix(3000, 0),
227+
Step: 30 * time.Second,
228+
}
229+
// group_right(namespace): RHS many, include namespace from RHS.
230+
matching := &parser.VectorMatching{
231+
Card: parser.CardOneToMany,
232+
MatchingLabels: []string{"cluster", "node"},
233+
On: true,
234+
Include: []string{"namespace"},
235+
}
236+
237+
b.Run("without_projection", func(b *testing.B) {
238+
b.ReportAllocs()
239+
for b.Loop() {
240+
op, err := binary.NewVectorOperator(
241+
&mockOperator{series: lhsSeries},
242+
&mockOperator{series: rhsSeries},
243+
matching,
244+
parser.MUL,
245+
false,
246+
nil,
247+
opts,
248+
)
249+
testutil.Ok(b, err)
250+
_, _ = op.Series(context.Background())
251+
}
252+
})
253+
b.Run("with_projection", func(b *testing.B) {
254+
projection := &logicalplan.Projection{
255+
Labels: []string{"cluster", "node", "namespace"},
256+
Include: true,
257+
}
258+
b.ReportAllocs()
259+
for b.Loop() {
260+
op, err := binary.NewVectorOperator(
261+
&mockOperator{series: lhsSeries},
262+
&mockOperator{series: rhsSeries},
263+
matching,
264+
parser.MUL,
265+
false,
266+
projection,
267+
opts,
268+
)
269+
testutil.Ok(b, err)
270+
_, _ = op.Series(context.Background())
271+
}
272+
})
273+
}
274+
275+
type mockOperator struct {
276+
series []labels.Labels
277+
}
278+
279+
func (m *mockOperator) Series(context.Context) ([]labels.Labels, error) {
280+
return m.series, nil
281+
}
282+
283+
func (m *mockOperator) Next(context.Context, []model.StepVector) (int, error) {
284+
return 0, nil
285+
}
286+
287+
func (m *mockOperator) Explain() []model.VectorOperator {
288+
return nil
289+
}
290+
291+
func (m *mockOperator) String() string {
292+
return "mock"
293+
}

0 commit comments

Comments
 (0)