Skip to content

Commit ed0f141

Browse files
uzhastikuzhastik
authored
Read solomon tests (#5618)
1 parent 5854f64 commit ed0f141

File tree

62 files changed

+1488
-45
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

62 files changed

+1488
-45
lines changed

‎ydb/library/yql/cfg/tests/gateways.conf

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -216,3 +216,10 @@ Fs {
216216
TargetUrl: "arc:/$1"
217217
}
218218
}
219+
220+
Solomon {
221+
DefaultSettings {
222+
Name: "_EnableReading"
223+
Value: "1"
224+
}
225+
}

‎ydb/library/yql/providers/solomon/expr_nodes/yql_solomon_expr_nodes.json

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -48,25 +48,27 @@
4848
"Base": "TCallable",
4949
"Match": {"Type": "Callable", "Name": "SoSourceSettings"},
5050
"Children": [
51-
{"Index": 0, "Name": "Token", "Type": "TCoSecureParam"},
52-
{"Index": 1, "Name": "RowType", "Type": "TExprBase"},
53-
{"Index": 2, "Name": "SystemColumns", "Type": "TCoAtomList"},
54-
{"Index": 3, "Name": "LabelNames", "Type": "TCoAtomList"},
55-
{"Index": 4, "Name": "From", "Type": "TCoAtom"},
56-
{"Index": 5, "Name": "To", "Type": "TCoAtom"},
57-
{"Index": 6, "Name": "Program", "Type": "TCoAtom"},
58-
{"Index": 7, "Name": "DownsamplingDisabled", "Type": "TCoBool"},
59-
{"Index": 8, "Name": "DownsamplingAggregation", "Type": "TCoAtom"},
60-
{"Index": 9, "Name": "DownsamplingFill", "Type": "TCoAtom"},
61-
{"Index": 10, "Name": "DownsamplingGridSec", "Type": "TCoUint32"}
51+
{"Index": 0, "Name": "Project", "Type": "TCoAtom"},
52+
{"Index": 1, "Name": "Token", "Type": "TCoSecureParam"},
53+
{"Index": 2, "Name": "RowType", "Type": "TExprBase"},
54+
{"Index": 3, "Name": "SystemColumns", "Type": "TCoAtomList"},
55+
{"Index": 4, "Name": "LabelNames", "Type": "TCoAtomList"},
56+
{"Index": 5, "Name": "From", "Type": "TCoAtom"},
57+
{"Index": 6, "Name": "To", "Type": "TCoAtom"},
58+
{"Index": 7, "Name": "Program", "Type": "TCoAtom"},
59+
{"Index": 8, "Name": "DownsamplingDisabled", "Type": "TCoBool"},
60+
{"Index": 9, "Name": "DownsamplingAggregation", "Type": "TCoAtom"},
61+
{"Index": 10, "Name": "DownsamplingFill", "Type": "TCoAtom"},
62+
{"Index": 11, "Name": "DownsamplingGridSec", "Type": "TCoUint32"}
6263
]
6364
},
6465
{
6566
"Name": "TSoObject",
6667
"Base": "TCallable",
6768
"Match": {"Type": "Callable", "Name": "SoObject"},
6869
"Children": [
69-
{"Index": 0, "Name": "Settings", "Type": "TExprBase"}
70+
{"Index": 0, "Name": "Project", "Type": "TCoAtom"},
71+
{"Index": 1, "Name": "Settings", "Type": "TExprBase"}
7072
]
7173
},
7274
{

‎ydb/library/yql/providers/solomon/provider/yql_solomon_datasource_type_ann.cpp

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,12 @@ class TSolomonDataSourceTypeAnnotationTransformer : public TVisitorTransformerBa
3434
}
3535

3636
TStatus HandleSoSourceSettings(const TExprNode::TPtr& input, TExprContext& ctx) {
37-
if (!EnsureArgsCount(*input, 11U, ctx)) {
37+
if (!EnsureArgsCount(*input, 12U, ctx)) {
38+
return TStatus::Error;
39+
}
40+
41+
auto& project = *input->Child(TSoSourceSettings::idx_Project);
42+
if (!EnsureAtom(project, ctx)) {
3843
return TStatus::Error;
3944
}
4045

@@ -59,12 +64,12 @@ class TSolomonDataSourceTypeAnnotationTransformer : public TVisitorTransformerBa
5964
}
6065

6166
auto& from = *input->Child(TSoSourceSettings::idx_From);
62-
if (!EnsureAtom(from, ctx) || !ValidateDatetimeFormat("from"sv, from, ctx)) {
67+
if (!EnsureAtom(from, ctx) || !ValidateDatetimeFormat("from", from, ctx)) {
6368
return TStatus::Error;
6469
}
6570

6671
auto& to = *input->Child(TSoSourceSettings::idx_To);
67-
if (!EnsureAtom(to, ctx) || !ValidateDatetimeFormat("to"sv, to, ctx)) {
72+
if (!EnsureAtom(to, ctx) || !ValidateDatetimeFormat("to", to, ctx)) {
6873
return TStatus::Error;
6974
}
7075

@@ -106,11 +111,15 @@ class TSolomonDataSourceTypeAnnotationTransformer : public TVisitorTransformerBa
106111
}
107112

108113
TStatus HandleSoObject(const TExprNode::TPtr& input, TExprContext& ctx) {
109-
if (!EnsureArgsCount(*input, 1U, ctx)) {
114+
if (!EnsureArgsCount(*input, 2U, ctx)) {
115+
return TStatus::Error;
116+
}
117+
118+
auto& project = *input->Child(TSoObject::idx_Project);
119+
if (!EnsureAtom(project, ctx)) {
110120
return TStatus::Error;
111121
}
112122

113-
// todo: check settings
114123
input->SetTypeAnn(ctx.MakeType<TUnitExprType>());
115124
return TStatus::Ok;
116125
}

‎ydb/library/yql/providers/solomon/provider/yql_solomon_dq_integration.cpp

Lines changed: 19 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -131,7 +131,7 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
131131
for (auto i = 0U; i < settingsRef.ChildrenSize(); ++i) {
132132
if (settingsRef.Child(i)->Head().IsAtom("from"sv)) {
133133
TStringBuf value;
134-
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), "from"sv, ctx, value)) {
134+
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), settingsRef.Child(i)->Head().Content(), ctx, value)) {
135135
return {};
136136
}
137137

@@ -140,7 +140,7 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
140140
}
141141
if (settingsRef.Child(i)->Head().IsAtom("to"sv)) {
142142
TStringBuf value;
143-
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), "to"sv, ctx, value)) {
143+
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), settingsRef.Child(i)->Head().Content(), ctx, value)) {
144144
return {};
145145
}
146146

@@ -149,7 +149,7 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
149149
}
150150
if (settingsRef.Child(i)->Head().IsAtom("program"sv)) {
151151
TStringBuf value;
152-
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), "program"sv, ctx, value)) {
152+
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), settingsRef.Child(i)->Head().Content(), ctx, value)) {
153153
return {};
154154
}
155155

@@ -158,39 +158,42 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
158158
}
159159
if (settingsRef.Child(i)->Head().IsAtom("downsampling.disabled"sv)) {
160160
TStringBuf value;
161-
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), "downsampling.disabled"sv, ctx, value)) {
161+
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), settingsRef.Child(i)->Head().Content(), ctx, value)) {
162+
return {};
163+
}
164+
if (!TryFromString<bool>(value, downsamplingDisabled)) {
165+
ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), TStringBuilder() << "downsampling.disabled must be true or false, but has " << value));
162166
return {};
163167
}
164-
downsamplingDisabled = FromString<bool>(value);
165168
continue;
166169
}
167-
if (settingsRef.Child(i)->Head().IsAtom("downsampling.gridaggregation"sv)) {
170+
if (settingsRef.Child(i)->Head().IsAtom("downsampling.aggregation"sv)) {
168171
TStringBuf value;
169-
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), "downsampling.gridaggregation"sv, ctx, value)) {
172+
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), settingsRef.Child(i)->Head().Content(), ctx, value)) {
170173
return {};
171174
}
172175
if (!IsIn({ "AVG"sv, "COUNT"sv, "DEFAULT_AGGREGATION"sv, "LAST"sv, "MAX"sv, "MIN"sv, "SUM"sv }, value)) {
173-
ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), TStringBuilder() << "downsampling.grid_aggregation must be one of AVG, COUNT, DEFAULT_AGGREGATION, LAST, MAX, MIN, SUM, but has " << value));
176+
ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), TStringBuilder() << "downsampling.aggregation must be one of AVG, COUNT, DEFAULT_AGGREGATION, LAST, MAX, MIN, SUM, but has " << value));
174177
return {};
175178
}
176179
downsamplingAggregation = value;
177180
continue;
178181
}
179182
if (settingsRef.Child(i)->Head().IsAtom("downsampling.fill"sv)) {
180183
TStringBuf value;
181-
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), "downsampling.fill"sv, ctx, value)) {
184+
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), settingsRef.Child(i)->Head().Content(), ctx, value)) {
182185
return {};
183186
}
184187
if (!IsIn({ "NONE"sv, "NULL"sv, "PREVIOUS"sv }, value)) {
185-
ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), TStringBuilder() << "downsampling.grid_fill must be one of NONE, NULL, PREVIOUS, but has " << value));
188+
ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), TStringBuilder() << "downsampling.fill must be one of NONE, NULL, PREVIOUS, but has " << value));
186189
return {};
187190
}
188191
downsamplingFill = value;
189192
continue;
190193
}
191194
if (settingsRef.Child(i)->Head().IsAtom("downsampling.gridinterval"sv)) {
192195
TStringBuf value;
193-
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), "downsampling.gridinterval"sv, ctx, value)) {
196+
if (!ExtractSettingValue(settingsRef.Child(i)->Tail(), "downsampling.grid_interval"sv, ctx, value)) {
194197
return {};
195198
}
196199
ui32 intValue = 0;
@@ -201,10 +204,14 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
201204
downsamplingGridSec = intValue;
202205
continue;
203206
}
207+
208+
ctx.AddError(TIssue(ctx.GetPosition(settingsRef.Child(i)->Head().Pos()), TStringBuilder() << "Unknown setting " << settingsRef.Child(i)->Head().Content()));
209+
return {};
204210
}
205211

206212
return Build<TDqSourceWrap>(ctx, read->Pos())
207213
.Input<TSoSourceSettings>()
214+
.Project(soReadObject.Object().Project())
208215
.Token<TCoSecureParam>()
209216
.Name().Build(token)
210217
.Build()
@@ -244,7 +251,7 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
244251
YQL_ENSURE(clusterDesc, "Unknown cluster " << cluster);
245252
NSo::NProto::TDqSolomonSource source;
246253
source.SetEndpoint(clusterDesc->GetCluster());
247-
source.SetProject("yq");
254+
source.SetProject(settings.Project().StringValue());
248255

249256
source.SetClusterType(MapClusterType(clusterDesc->GetClusterType()));
250257
source.SetUseSsl(clusterDesc->GetUseSsl());

‎ydb/library/yql/providers/solomon/provider/yql_solomon_io_discovery.cpp

Lines changed: 32 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,17 +12,18 @@ namespace NYql {
1212
namespace {
1313
using namespace NNodes;
1414

15-
std::array<TExprNode::TPtr, 2U> GetSchema(const TExprNode::TListType& settings) {
15+
std::array<TExprNode::TPtr, 2U> ExtractSchema(TExprNode::TListType& settings) {
1616
for (auto it = settings.cbegin(); settings.cend() != it; ++it) {
1717
if (const auto item = *it; item->Head().IsAtom("userschema")) {
18+
settings.erase(it);
1819
return {item->ChildPtr(1), item->ChildrenSize() > 2 ? item->TailPtr() : TExprNode::TPtr()};
1920
}
2021
}
2122

2223
return {};
2324
}
2425

25-
TVector<TCoAtom> GetUserLabels(TPositionHandle pos, TExprContext& ctx, const TExprNode::TListType& settings) {
26+
TVector<TCoAtom> ExtractUserLabels(TPositionHandle pos, TExprContext& ctx, TExprNode::TListType& settings) {
2627
for (auto it = settings.cbegin(); settings.cend() != it; ++it) {
2728
if (const auto item = *it; item->Head().IsAtom("labels")) {
2829
TVector<TCoAtom> result;
@@ -32,6 +33,7 @@ TVector<TCoAtom> GetUserLabels(TPositionHandle pos, TExprContext& ctx, const TEx
3233
auto v = Build<TCoAtom>(ctx, pos).Value(StripString(label)).Done();
3334
result.emplace_back(std::move(v));
3435
}
36+
settings.erase(it);
3537
return result;
3638
}
3739
}
@@ -68,7 +70,7 @@ const TStructExprType* BuildScheme(TPositionHandle pos, const TVector<TCoAtom>&
6870
}
6971

7072
for (const auto& label : userLabels) {
71-
if (IsIn({ SOLOMON_SCHEME_TS, SOLOMON_SCHEME_KIND, SOLOMON_SCHEME_TYPE, SOLOMON_SCHEME_LABELS, SOLOMON_SCHEME_VALUE }, label.Value())) {
73+
if (IsIn(allSystemColumns, label.Value())) {
7274
// tmp constraint
7375
ctx.AddError(TIssue(ctx.GetPosition(pos), TStringBuilder() << "System column should not be used as label name: " << label.Value()));
7476
return nullptr;
@@ -124,9 +126,26 @@ class TSolomonIODiscoveryTransformer : public TSyncTransformerBase {
124126
return node;
125127
}
126128

127-
const auto& object = read.Arg(2).Ref();
128-
YQL_ENSURE(object.IsCallable("MrTableConcat"));
129+
auto& object = read.Arg(2).Ref();
130+
if (!object.IsCallable("MrTableConcat")) {
131+
ctx.AddError(TIssue(ctx.GetPosition(object.Pos()), TStringBuilder() << "Expected MrTableConcat"));
132+
return {};
133+
}
134+
135+
const auto maybeKey = TMaybeNode<TCoKey>(&object.Head());
136+
if (!maybeKey) {
137+
ctx.AddError(TIssue(ctx.GetPosition(object.Pos()), TStringBuilder() << "Expected Key"));
138+
return {};
139+
}
129140

141+
const auto& keyArg = maybeKey.Cast().Ref().Head();
142+
if (!keyArg.IsList() || keyArg.ChildrenSize() != 2U
143+
|| !keyArg.Head().IsAtom("table") || !keyArg.Tail().IsCallable(TCoString::CallableName())) {
144+
ctx.AddError(TIssue(ctx.GetPosition(keyArg.Pos()), TStringBuilder() << "Expected single table name"));
145+
return {};
146+
}
147+
148+
const auto project = TString(keyArg.Tail().Head().Content());
130149
auto cluster = read.DataSource().Cluster().StringValue();
131150
if (!this->State_->Configuration->_EnableReading.Get().GetOrElse(false)) {
132151
ctx.AddError(TIssue(ctx.GetPosition(object.Pos()), TStringBuilder() << "Reading is disabled for monitoring cluster " << cluster));
@@ -135,11 +154,16 @@ class TSolomonIODiscoveryTransformer : public TSyncTransformerBase {
135154

136155
auto settings = read.Ref().Child(4);
137156
auto settingsList = read.Ref().Child(4)->ChildrenList();
138-
auto userSchema = GetSchema(settingsList);
139-
TVector<TCoAtom> userLabels = GetUserLabels(settings->Pos(), ctx, settingsList);
157+
auto userSchema = ExtractSchema(settingsList);
158+
TVector<TCoAtom> userLabels = ExtractUserLabels(settings->Pos(), ctx, settingsList);
159+
160+
auto newSettings = Build<TCoNameValueTupleList>(ctx, settings->Pos())
161+
.Add(settingsList)
162+
.Done();
140163

141164
auto soObject = Build<TSoObject>(ctx, read.Pos())
142-
.Settings(settings)
165+
.Project<TCoAtom>().Build(project)
166+
.Settings(newSettings)
143167
.Done();
144168

145169
TVector<TCoAtom> systemColumns;

‎ydb/library/yql/sql/v1/sql_ut.cpp

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -4599,14 +4599,6 @@ select FormatType($f());
45994599
UNIT_ASSERT_NO_DIFF(Err2Str(res), "<main>:1:17: Error: Named subquery can not be used as a top level statement in libraries\n");
46004600
}
46014601

4602-
Y_UNIT_TEST(SelectingFromMonitoringIsNotAllowed) {
4603-
NSQLTranslation::TTranslationSettings settings;
4604-
auto res = SqlToYqlWithSettings("select * from mon.zzz;", settings);
4605-
UNIT_ASSERT(!res.Root);
4606-
UNIT_ASSERT_NO_DIFF(Err2Str(res),
4607-
"<main>:1:15: Error: Selecting data from monitoring source is not supported\n");
4608-
}
4609-
46104602
Y_UNIT_TEST(SessionStartAndSessionStateShouldSurviveSessionWindowArgsError){
46114603
TString query = R"(
46124604
$init = ($_row) -> (min(1, 2)); -- error: aggregation func min() can not be used here

‎ydb/library/yql/tests/common/test_framework/conftest.py

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,12 @@
33
except ImportError:
44
yql_http_file_server = None
55

6+
try:
7+
from solomon_runner import solomon
8+
except ImportError:
9+
solomon = None
10+
611
# bunch of useless statements for linter happiness
712
# (otherwise it complains about unused names)
813
assert yql_http_file_server is yql_http_file_server
14+
assert solomon is solomon

‎ydb/library/yql/tests/common/test_framework/solomon_runner.py

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
import os
2+
import pytest
3+
import requests
4+
5+
6+
class SolomonWrapper(object):
7+
def __init__(self, url, endpoint):
8+
self._url = url
9+
self._endpoint = endpoint
10+
self.table_prefix = ""
11+
12+
def is_valid(self):
13+
return self._url is not None
14+
15+
def cleanup(self):
16+
res = requests.post(self._url + "/cleanup")
17+
res.raise_for_status()
18+
19+
def get_metrics(self):
20+
res = requests.get(self._url + "/metrics?project=my_project&cluster=my_cluster&service=my_service")
21+
res.raise_for_status()
22+
return res.text
23+
24+
def prepare_program(self, program, program_file, res_dir, lang='sql'):
25+
return program, program_file
26+
27+
@property
28+
def url(self):
29+
return self._url
30+
31+
@property
32+
def endpoint(self):
33+
return self._endpoint
34+
35+
36+
@pytest.fixture(scope='module')
37+
def solomon(request):
38+
solomon_url = os.environ.get("SOLOMON_URL")
39+
solomon_endpoint = os.environ.get("SOLOMON_ENDPOINT")
40+
return SolomonWrapper(solomon_url, solomon_endpoint)

‎ydb/library/yql/tests/common/test_framework/ya.make

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ PY23_LIBRARY()
22

33
PY_SRCS(
44
TOP_LEVEL
5+
solomon_runner.py
56
yql_utils.py
67
yql_ports.py
78
yqlrun.py
@@ -14,6 +15,7 @@ PY_SRCS(
1415
)
1516

1617
PEERDIR(
18+
contrib/python/requests
1719
contrib/python/six
1820
contrib/python/urllib3
1921
library/python/cyson

0 commit comments

Comments
 (0)