Skip to content

Commit 00bc3bd

Browse files
GrigoriyPAGrigoriyPA
authored
YQ RD enable LLVM in purecalc filters (#11442)
1 parent ce7eba6 commit 00bc3bd

File tree

15 files changed

+121
-35
lines changed

15 files changed

+121
-35
lines changed

‎ydb/core/fq/libs/row_dispatcher/actors_factory.cpp

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@
22

33
#include <ydb/core/fq/libs/row_dispatcher/topic_session.h>
44

5-
#include <ydb/library/yql/public/purecalc/common/interface.h>
6-
75
namespace NFq::NRowDispatcher {
86

97

@@ -19,7 +17,7 @@ struct TActorFactory : public IActorFactory {
1917
ui32 partitionId,
2018
NYdb::TDriver driver,
2119
std::shared_ptr<NYdb::ICredentialsProviderFactory> credentialsProviderFactory,
22-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory,
20+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
2321
const ::NMonitoring::TDynamicCounterPtr& counters,
2422
const NYql::IPqGateway::TPtr& pqGateway) const override {
2523

‎ydb/core/fq/libs/row_dispatcher/actors_factory.h

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,12 @@
11
#pragma once
22

3+
#include "common.h"
4+
35
#include <ydb/core/fq/libs/config/protos/row_dispatcher.pb.h>
46
#include <util/generic/ptr.h>
57
#include <ydb/library/actors/core/actor.h>
68
#include <ydb/public/sdk/cpp/client/ydb_driver/driver.h>
79
#include <ydb/library/yql/providers/pq/provider/yql_pq_gateway.h>
8-
#include <ydb/library/yql/public/purecalc/common/fwd.h>
910

1011
namespace NFq::NRowDispatcher {
1112

@@ -21,7 +22,7 @@ struct IActorFactory : public TThrRefBase {
2122
ui32 partitionId,
2223
NYdb::TDriver driver,
2324
std::shared_ptr<NYdb::ICredentialsProviderFactory> credentialsProviderFactory,
24-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory,
25+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
2526
const ::NMonitoring::TDynamicCounterPtr& counters,
2627
const NYql::IPqGateway::TPtr& pqGateway) const = 0;
2728
};

‎ydb/core/fq/libs/row_dispatcher/common.cpp

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
#include "common.h"
2+
3+
#include <util/system/mutex.h>
4+
5+
#include <ydb/library/yql/public/purecalc/common/interface.h>
6+
7+
namespace NFq {
8+
9+
namespace {
10+
11+
class TPureCalcProgramFactory : public IPureCalcProgramFactory {
12+
public:
13+
TPureCalcProgramFactory() {
14+
CreateFactory({.EnabledLLVM = false});
15+
CreateFactory({.EnabledLLVM = true});
16+
}
17+
18+
NYql::NPureCalc::IProgramFactoryPtr GetFactory(const TSettings& settings) const override {
19+
const auto it = ProgramFactories.find(settings);
20+
Y_ENSURE(it != ProgramFactories.end());
21+
return it->second;
22+
}
23+
24+
private:
25+
void CreateFactory(const TSettings& settings) {
26+
ProgramFactories.insert({settings, NYql::NPureCalc::MakeProgramFactory(
27+
NYql::NPureCalc::TProgramFactoryOptions()
28+
.SetLLVMSettings(settings.EnabledLLVM ? "ON" : "OFF")
29+
)});
30+
}
31+
32+
private:
33+
std::map<TSettings, NYql::NPureCalc::IProgramFactoryPtr> ProgramFactories;
34+
};
35+
36+
} // anonymous namespace
37+
38+
IPureCalcProgramFactory::TPtr CreatePureCalcProgramFactory() {
39+
return MakeIntrusive<TPureCalcProgramFactory>();
40+
}
41+
42+
} // namespace NFq

‎ydb/core/fq/libs/row_dispatcher/common.h

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
#pragma once
2+
3+
#include <util/generic/ptr.h>
4+
5+
#include <ydb/library/yql/public/purecalc/common/fwd.h>
6+
7+
namespace NFq {
8+
9+
class IPureCalcProgramFactory : public TThrRefBase {
10+
public:
11+
using TPtr = TIntrusivePtr<IPureCalcProgramFactory>;
12+
13+
struct TSettings {
14+
bool EnabledLLVM = false;
15+
16+
std::strong_ordering operator<=>(const TSettings& other) const = default;
17+
};
18+
19+
public:
20+
virtual NYql::NPureCalc::IProgramFactoryPtr GetFactory(const TSettings& settings) const = 0;
21+
};
22+
23+
IPureCalcProgramFactory::TPtr CreatePureCalcProgramFactory();
24+
25+
} // namespace NFq

‎ydb/core/fq/libs/row_dispatcher/json_filter.cpp

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -264,14 +264,15 @@ class TJsonFilter::TImpl {
264264
const TVector<TString>& types,
265265
const TString& whereFilter,
266266
TCallback callback,
267-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory)
268-
: Sql(GenerateSql(whereFilter)) {
267+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
268+
const IPureCalcProgramFactory::TSettings& factorySettings)
269+
: Sql(GenerateSql(whereFilter, factorySettings)) {
269270
Y_ENSURE(columns.size() == types.size(), "Number of columns and types should by equal");
270271

271272
// Program should be stateless because input values
272273
// allocated on another allocator and should be released
273274
LOG_ROW_DISPATCHER_DEBUG("Creating program...");
274-
Program = pureCalcProgramFactory->MakePushStreamProgram(
275+
Program = pureCalcProgramFactory->GetFactory(factorySettings)->MakePushStreamProgram(
275276
TFilterInputSpec(MakeInputSchema(columns, types)),
276277
TFilterOutputSpec(MakeOutputSchema()),
277278
Sql,
@@ -291,8 +292,9 @@ class TJsonFilter::TImpl {
291292
}
292293

293294
private:
294-
TString GenerateSql(const TString& whereFilter) {
295+
TString GenerateSql(const TString& whereFilter, const IPureCalcProgramFactory::TSettings& factorySettings) {
295296
TStringStream str;
297+
str << "PRAGMA config.flags(\"LLVM\", \"" << (factorySettings.EnabledLLVM ? "ON" : "OFF") << "\");\n";
296298
str << "$filtered = SELECT * FROM Input " << whereFilter << ";\n";
297299

298300
str << "SELECT " << OffsetFieldName << ", Unwrap(Json::SerializeJson(Yson::From(RemoveMembers(TableRow(), [\"" << OffsetFieldName;
@@ -312,8 +314,9 @@ TJsonFilter::TJsonFilter(
312314
const TVector<TString>& types,
313315
const TString& whereFilter,
314316
TCallback callback,
315-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory)
316-
: Impl(std::make_unique<TJsonFilter::TImpl>(columns, types, whereFilter, callback, pureCalcProgramFactory)) {
317+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
318+
const IPureCalcProgramFactory::TSettings& factorySettings)
319+
: Impl(std::make_unique<TJsonFilter::TImpl>(columns, types, whereFilter, callback, pureCalcProgramFactory, factorySettings)) {
317320
}
318321

319322
TJsonFilter::~TJsonFilter() {
@@ -332,8 +335,9 @@ std::unique_ptr<TJsonFilter> NewJsonFilter(
332335
const TVector<TString>& types,
333336
const TString& whereFilter,
334337
TCallback callback,
335-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory) {
336-
return std::unique_ptr<TJsonFilter>(new TJsonFilter(columns, types, whereFilter, callback, pureCalcProgramFactory));
338+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
339+
const IPureCalcProgramFactory::TSettings& factorySettings) {
340+
return std::unique_ptr<TJsonFilter>(new TJsonFilter(columns, types, whereFilter, callback, pureCalcProgramFactory, factorySettings));
337341
}
338342

339343
} // namespace NFq

‎ydb/core/fq/libs/row_dispatcher/json_filter.h

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
#pragma once
22

3+
#include "common.h"
4+
35
#include <yql/essentials/minikql/computation/mkql_computation_node_holders.h>
4-
#include <ydb/library/yql/public/purecalc/common/fwd.h>
56
#include <yql/essentials/public/udf/udf_data_type.h>
67
#include <yql/essentials/public/udf/udf_value.h>
78

@@ -17,7 +18,8 @@ class TJsonFilter {
1718
const TVector<TString>& types,
1819
const TString& whereFilter,
1920
TCallback callback,
20-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory);
21+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
22+
const IPureCalcProgramFactory::TSettings& factorySettings);
2123

2224
~TJsonFilter();
2325

@@ -34,6 +36,7 @@ std::unique_ptr<TJsonFilter> NewJsonFilter(
3436
const TVector<TString>& types,
3537
const TString& whereFilter,
3638
TJsonFilter::TCallback callback,
37-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory);
39+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
40+
const IPureCalcProgramFactory::TSettings& factorySettings);
3841

3942
} // namespace NFq

‎ydb/core/fq/libs/row_dispatcher/row_dispatcher.cpp

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
#include "row_dispatcher.h"
2+
#include "common.h"
23
#include "coordinator.h"
34

45
#include <ydb/library/actors/core/actorid.h>
@@ -214,7 +215,7 @@ class TRowDispatcher : public TActorBootstrapped<TRowDispatcher> {
214215

215216
NConfig::TRowDispatcherConfig Config;
216217
NKikimr::TYdbCredentialsProviderFactory CredentialsProviderFactory;
217-
NYql::NPureCalc::IProgramFactoryPtr PureCalcProgramFactory;
218+
IPureCalcProgramFactory::TPtr PureCalcProgramFactory;
218219
TYqSharedResources::TPtr YqSharedResources;
219220
TMaybe<TActorId> CoordinatorActorId;
220221
TSet<TActorId> CoordinatorChangedSubscribers;
@@ -362,7 +363,7 @@ TRowDispatcher::TRowDispatcher(
362363
const NYql::IPqGateway::TPtr& pqGateway)
363364
: Config(config)
364365
, CredentialsProviderFactory(credentialsProviderFactory)
365-
, PureCalcProgramFactory(NYql::NPureCalc::MakeProgramFactory(NYql::NPureCalc::TProgramFactoryOptions()))
366+
, PureCalcProgramFactory(CreatePureCalcProgramFactory())
366367
, YqSharedResources(yqSharedResources)
367368
, CredentialsFactory(credentialsFactory)
368369
, LogPrefix("RowDispatcher: ")

‎ydb/core/fq/libs/row_dispatcher/topic_session.cpp

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -158,7 +158,7 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
158158
ui32 PartitionId;
159159
NYdb::TDriver Driver;
160160
std::shared_ptr<NYdb::ICredentialsProviderFactory> CredentialsProviderFactory;
161-
NYql::NPureCalc::IProgramFactoryPtr PureCalcProgramFactory;
161+
IPureCalcProgramFactory::TPtr PureCalcProgramFactory;
162162
NYql::ITopicClient::TPtr TopicClient;
163163
std::shared_ptr<NYdb::NTopic::IReadSession> ReadSession;
164164
const i64 BufferSize;
@@ -190,7 +190,7 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
190190
ui32 partitionId,
191191
NYdb::TDriver driver,
192192
std::shared_ptr<NYdb::ICredentialsProviderFactory> credentialsProviderFactory,
193-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory,
193+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
194194
const ::NMonitoring::TDynamicCounterPtr& counters,
195195
const NYql::IPqGateway::TPtr& pqGateway);
196196

@@ -281,7 +281,7 @@ TTopicSession::TTopicSession(
281281
ui32 partitionId,
282282
NYdb::TDriver driver,
283283
std::shared_ptr<NYdb::ICredentialsProviderFactory> credentialsProviderFactory,
284-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory,
284+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
285285
const ::NMonitoring::TDynamicCounterPtr& counters,
286286
const NYql::IPqGateway::TPtr& pqGateway)
287287
: TopicPath(topicPath)
@@ -737,10 +737,11 @@ void TTopicSession::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
737737
std::forward_as_tuple(ev)).first->second;
738738
UpdateFieldsIds(clientInfo);
739739

740-
TString predicate = clientInfo.Settings.GetSource().GetPredicate();
740+
const auto& source = clientInfo.Settings.GetSource();
741+
TString predicate = source.GetPredicate();
741742

742743
// TODO: remove this when the re-parsing is removed from pq read actor
743-
if (predicate.empty() && HasJsonColumns(clientInfo.Settings.GetSource())) {
744+
if (predicate.empty() && HasJsonColumns(source)) {
744745
predicate = "WHERE TRUE";
745746
}
746747

@@ -752,7 +753,9 @@ void TTopicSession::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
752753
[&, actorId = clientInfo.ReadActorId](ui64 offset, const TString& json){
753754
Send(SelfId(), new NFq::TEvPrivate::TEvDataAfterFilteration(offset, json, actorId));
754755
},
755-
PureCalcProgramFactory);
756+
PureCalcProgramFactory,
757+
{.EnabledLLVM = source.GetEnabledLLVM()}
758+
);
756759
} else {
757760
ClientsWithoutPredicate.insert(ev->Sender);
758761
}
@@ -993,7 +996,7 @@ std::unique_ptr<NActors::IActor> NewTopicSession(
993996
ui32 partitionId,
994997
NYdb::TDriver driver,
995998
std::shared_ptr<NYdb::ICredentialsProviderFactory> credentialsProviderFactory,
996-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory,
999+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
9971000
const ::NMonitoring::TDynamicCounterPtr& counters,
9981001
const NYql::IPqGateway::TPtr& pqGateway) {
9991002
return std::unique_ptr<NActors::IActor>(new TTopicSession(topicPath, endpoint, database, config, rowDispatcherActorId, partitionId, std::move(driver), credentialsProviderFactory, pureCalcProgramFactory, counters, pqGateway));

‎ydb/core/fq/libs/row_dispatcher/topic_session.h

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
#pragma once
22

3+
#include "common.h"
4+
35
#include <ydb/core/fq/libs/config/protos/row_dispatcher.pb.h>
46
#include <ydb/core/fq/libs/config/protos/common.pb.h>
57
#include <ydb/core/fq/libs/shared_resources/shared_resources.h>
@@ -8,7 +10,6 @@
810

911
#include <ydb/library/yql/providers/pq/proto/dq_io.pb.h>
1012
#include <ydb/library/yql/providers/pq/provider/yql_pq_gateway.h>
11-
#include <ydb/library/yql/public/purecalc/common/fwd.h>
1213

1314
#include <ydb/library/actors/core/actor.h>
1415

@@ -25,7 +26,7 @@ std::unique_ptr<NActors::IActor> NewTopicSession(
2526
ui32 partitionId,
2627
NYdb::TDriver driver,
2728
std::shared_ptr<NYdb::ICredentialsProviderFactory> credentialsProviderFactory,
28-
NYql::NPureCalc::IProgramFactoryPtr pureCalcProgramFactory,
29+
IPureCalcProgramFactory::TPtr pureCalcProgramFactory,
2930
const ::NMonitoring::TDynamicCounterPtr& counters,
3031
const NYql::IPqGateway::TPtr& pqGateway);
3132

‎ydb/core/fq/libs/row_dispatcher/ut/json_filter_ut.cpp

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,14 +3,14 @@
33
#include <ydb/core/fq/libs/ydb/ydb.h>
44
#include <ydb/core/fq/libs/events/events.h>
55

6+
#include <ydb/core/fq/libs/row_dispatcher/common.h>
67
#include <ydb/core/fq/libs/row_dispatcher/json_filter.h>
78

89
#include <ydb/core/testlib/actors/test_runtime.h>
910
#include <ydb/core/testlib/basics/helpers.h>
1011
#include <ydb/core/testlib/actor_helpers.h>
1112

1213
#include <yql/essentials/minikql/mkql_string_util.h>
13-
#include <ydb/library/yql/public/purecalc/common/interface.h>
1414

1515
#include <library/cpp/testing/unittest/registar.h>
1616

@@ -23,7 +23,7 @@ class TFixture : public NUnitTest::TBaseFixture {
2323

2424
public:
2525
TFixture()
26-
: PureCalcProgramFactory(NYql::NPureCalc::MakeProgramFactory(NYql::NPureCalc::TProgramFactoryOptions()))
26+
: PureCalcProgramFactory(CreatePureCalcProgramFactory())
2727
, Runtime(true)
2828
, Alloc(__LOCATION__, NKikimr::TAlignedPagePoolCounters(), true, false)
2929
{}
@@ -65,7 +65,8 @@ class TFixture : public NUnitTest::TBaseFixture {
6565
types,
6666
whereFilter,
6767
callback,
68-
PureCalcProgramFactory);
68+
PureCalcProgramFactory,
69+
{.EnabledLLVM = false});
6970
}
7071

7172
const NKikimr::NMiniKQL::TUnboxedValueVector* MakeVector(size_t size, std::function<NYql::NUdf::TUnboxedValuePod(size_t)> valueCreator) {
@@ -100,7 +101,7 @@ class TFixture : public NUnitTest::TBaseFixture {
100101
});
101102
}
102103

103-
NYql::NPureCalc::IProgramFactoryPtr PureCalcProgramFactory;
104+
IPureCalcProgramFactory::TPtr PureCalcProgramFactory;
104105
NActors::TTestActorRuntime Runtime;
105106
TActorSystemStub ActorSystemStub;
106107
std::unique_ptr<NFq::TJsonFilter> Filter;

‎ydb/core/fq/libs/row_dispatcher/ut/row_dispatcher_ut.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ struct TTestActorFactory : public NFq::NRowDispatcher::IActorFactory {
3535
ui32 /*partitionId*/,
3636
NYdb::TDriver /*driver*/,
3737
std::shared_ptr<NYdb::ICredentialsProviderFactory> /*credentialsProviderFactory*/,
38-
NYql::NPureCalc::IProgramFactoryPtr /*pureCalcProgramFactory*/,
38+
IPureCalcProgramFactory::TPtr /*pureCalcProgramFactory*/,
3939
const ::NMonitoring::TDynamicCounterPtr& /*counters*/,
4040
const NYql::IPqGateway::TPtr& /*pqGateway*/) const override {
4141
auto actorId = Runtime.AllocateEdgeActor();

‎ydb/core/fq/libs/row_dispatcher/ut/topic_session_ut.cpp

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,6 @@
1313
#include <ydb/tests/fq/pq_async_io/ut_helpers.h>
1414

1515
#include <ydb/library/yql/providers/pq/gateway/native/yql_pq_gateway.h>
16-
#include <ydb/library/yql/public/purecalc/common/interface.h>
1716

1817
namespace {
1918

@@ -27,7 +26,7 @@ const ui64 GrabTimeoutSec = 4 * TimeoutBeforeStartSessionSec;
2726
class TFixture : public NUnitTest::TBaseFixture {
2827
public:
2928
TFixture()
30-
: PureCalcProgramFactory(NYql::NPureCalc::MakeProgramFactory(NYql::NPureCalc::TProgramFactoryOptions()))
29+
: PureCalcProgramFactory(CreatePureCalcProgramFactory())
3130
, Runtime(true)
3231
{}
3332

@@ -158,7 +157,7 @@ class TFixture : public NUnitTest::TBaseFixture {
158157
return eventHolder->Get()->Record.MessagesSize();
159158
}
160159

161-
NYql::NPureCalc::IProgramFactoryPtr PureCalcProgramFactory;
160+
IPureCalcProgramFactory::TPtr PureCalcProgramFactory;
162161
NActors::TTestActorRuntime Runtime;
163162
TActorSystemStub ActorSystemStub;
164163
NActors::TActorId TopicSession;

‎ydb/core/fq/libs/row_dispatcher/ya.make

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

33
SRCS(
44
actors_factory.cpp
5+
common.cpp
56
coordinator.cpp
67
json_filter.cpp
78
json_parser.cpp

‎ydb/library/yql/providers/pq/proto/dq_io.proto

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ message TDqPqTopicSource {
3838
string Predicate = 14;
3939
bool SharedReading = 15;
4040
string ReconnectPeriod = 16; // disabled by default, example of a parameter: 5m
41+
bool EnabledLLVM = 17;
4142
}
4243

4344
message TDqPqTopicSink {

0 commit comments

Comments
 (0)