-
Notifications
You must be signed in to change notification settings - Fork 145
Expand file tree
/
Copy pathtest_aggregate_join.py
More file actions
111 lines (104 loc) · 2.69 KB
/
Copy pathtest_aggregate_join.py
File metadata and controls
111 lines (104 loc) · 2.69 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import os
import unittest
from string import Template
from feldera.runtime_config import Resources
from feldera.testutils import run_workload, ViewSpec, unique_pipeline_name
tables = {
"t1": Template("""
create table t1(
id bigint not null primary key,
group_id bigint,
s string
) with (
'materialized' = 'true',
'connectors' = '[{
"transport": {
"name": "datagen",
"config": {
"plan": [{
"limit": $limit,
"fields": {
"group_id": { "range": [1, 10000] },
"s": { "strategy": "word" }
}
}]
}
}
}]');
""").substitute(limit=os.environ.get("ROW_LIMIT", "1000000")),
"t2": Template("""
create table t2(
id bigint not null primary key,
group_id bigint,
s string
) with (
'materialized' = 'true',
'connectors' = '[{
"transport": {
"name": "datagen",
"config": {
"plan": [{
"limit": $limit,
"fields": {
"group_id": { "range": [1, 10000] },
"s": { "strategy": "word" }
}
}]
}
}
}]');
""").substitute(limit=os.environ.get("ROW_LIMIT", "1000000")),
}
views = [
ViewSpec(
"t1_aggregate",
"""
select
group_id,
count(*) as cnt,
SORT_ARRAY(array_agg(s)) as arr
from t1
group by group_id
""",
),
ViewSpec(
"t2_aggregate",
"""
select
group_id,
count(*) as cnt,
SORT_ARRAY(array_agg(s)) as arr
from t2
group by group_id
""",
),
ViewSpec(
"result",
"""
select
t1_aggregate.group_id,
t1_aggregate.cnt as cnt1,
t1_aggregate.arr as arr1,
t2_aggregate.cnt as cnt2,
t2_aggregate.arr as arr2
from
t1_aggregate join t2_aggregate
on
t1_aggregate.group_id = t2_aggregate.group_id
""",
),
]
class TestAggregateJoin(unittest.TestCase):
def test_aggregate_joins(self):
run_workload(
unique_pipeline_name("aggregate-join-test"),
tables,
views,
# Peaks above 5 GB; an honest request keeps k8s from
# OOM-killing the pipeline mid-test. No memory_mb_max: it
# would cap the DataFusion pool at 5% and the big ad-hoc
# validation queries exhaust that.
resources=Resources(memory_mb_min=6000),
)
if __name__ == "__main__":
unittest.main()