Skip to content

YDB FQ: move ListSplits call from the request execution to the request optimization phase #14497

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 23 commits into from
Feb 18, 2025
Merged

YDB FQ: move ListSplits call from the request execution to the request optimization phase #14497

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
a7c6c50
Extract RunDescribeTableRequest
vitalyisaev2 Feb 7, 2025
cb7cf5f
Impl ListSplits call (not operational)
vitalyisaev2 Feb 7, 2025
9945dd7
Impl ListSplits call pt. 2 (not operational)
vitalyisaev2 Feb 10, 2025
871db4f
Minor changes
vitalyisaev2 Feb 10, 2025
16cec79
Chain async calls
vitalyisaev2 Feb 10, 2025
08dcb4b
Minor changes pt. 2
vitalyisaev2 Feb 12, 2025
3dfbd00
Validate splits in type annotation
vitalyisaev2 Feb 12, 2025
ce48e5d
Splits are passed via AST
vitalyisaev2 Feb 12, 2025
326dc2d
Splits are passed via state
vitalyisaev2 Feb 12, 2025
fab9313
Remove splits listing from ReadActor
vitalyisaev2 Feb 12, 2025
d259ce0
Minor changes
vitalyisaev2 Feb 12, 2025
b3cf7d7
Fix V1 (use TaaskParams alongside with ReadRanges)
vitalyisaev2 Feb 13, 2025
bb16582
Stream drainer
vitalyisaev2 Feb 13, 2025
872390d
Stream drainer debugged
vitalyisaev2 Feb 14, 2025
c4961ef
Review fixes
vitalyisaev2 Feb 14, 2025
d6c86ed
Review fixes 2
vitalyisaev2 Feb 14, 2025
042585e
Use TPartition for TSplit batching and distribution
vitalyisaev2 Feb 14, 2025
0ccd982
Review fix pt. 3
vitalyisaev2 Feb 17, 2025
e5c10bf
Fix trouble with TPartitions deserialization
vitalyisaev2 Feb 17, 2025
cd056f3
Ya style
vitalyisaev2 Feb 17, 2025
8efc4b6
Fix pushdown ut
vitalyisaev2 Feb 17, 2025
7feb1f9
Fix pushdown ut pt. 2
vitalyisaev2 Feb 17, 2025
433625f
Review fix
vitalyisaev2 Feb 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 8 additions & 13 deletions ydb/core/kqp/ut/federated_query/generic_ut/kqp_generic_provider_ut.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,9 +110,6 @@ namespace NKikimr::NKqp {
clientMock->ExpectListSplits()
.Select()
.DataSourceInstance(dataSourceInstance)
.What()
.Column("col1", Ydb::Type::UINT16)
.Done()
.Done()
.Result()
.AddResponse(NewSuccess())
Expand Down Expand Up @@ -209,9 +206,6 @@ namespace NKikimr::NKqp {
clientMock->ExpectListSplits()
.Select()
.DataSourceInstance(dataSourceInstance)
.What()
// Empty
.Done()
.Done()
.Result()
.AddResponse(NewSuccess())
Expand Down Expand Up @@ -302,9 +296,6 @@ namespace NKikimr::NKqp {
clientMock->ExpectListSplits()
.Select()
.DataSourceInstance(dataSourceInstance)
.What()
// Empty
.Done()
.Done()
.Result()
.AddResponse(NewSuccess())
Expand Down Expand Up @@ -375,8 +366,12 @@ namespace NKikimr::NKqp {
auto clientMock = std::make_shared<TConnectorClientMock>();

const NYql::TGenericDataSourceInstance dataSourceInstance = MakeDataSourceInstance(providerType);

// clang-format off
const NApi::TSelect select = TConnectorClientMock::TSelectBuilder<>()
const NApi::TSelect selectInListSplits = TConnectorClientMock::TSelectBuilder<>()
.DataSourceInstance(dataSourceInstance).GetResult();

const NApi::TSelect selectInReadSplits = TConnectorClientMock::TSelectBuilder<>()
.DataSourceInstance(dataSourceInstance)
.What()
.NullableColumn("data_column", Ydb::Type::STRING)
Expand Down Expand Up @@ -406,11 +401,11 @@ namespace NKikimr::NKqp {
// step 2: ListSplits
// clang-format off
clientMock->ExpectListSplits()
.Select(select)
.Select(selectInListSplits)
.Result()
.AddResponse(NewSuccess())
.Description("some binary description")
.Select(select);
.Select(selectInReadSplits);
// clang-format on

// step 3: ReadSplits
Expand All @@ -424,7 +419,7 @@ namespace NKikimr::NKqp {
.Filtering(NYql::NConnector::NApi::TReadSplitsRequest::FILTERING_OPTIONAL)
.Split()
.Description("some binary description")
.Select(select)
.Select(selectInReadSplits)
.Done()
.Result()
.AddResponse(MakeRecordBatch(
Expand Down
4 changes: 2 additions & 2 deletions ydb/library/yql/providers/generic/actors/ut/yql_generic_lookup_actor_ut.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,7 @@ Y_UNIT_TEST_SUITE(GenericProviderLookupActor) {
;
// clang-format on

NYql::Generic::TLookupSource lookupSourceSettings;
NYql::NGeneric::TLookupSource lookupSourceSettings;
*lookupSourceSettings.mutable_data_source_instance() = dsi;
lookupSourceSettings.Settable("lookup_test");
lookupSourceSettings.SetServiceAccountId("testsaid");
Expand Down Expand Up @@ -351,7 +351,7 @@ Y_UNIT_TEST_SUITE(GenericProviderLookupActor) {
}
// clang-format on

NYql::Generic::TLookupSource lookupSourceSettings;
NYql::NGeneric::TLookupSource lookupSourceSettings;
*lookupSourceSettings.mutable_data_source_instance() = dsi;
lookupSourceSettings.Settable("lookup_test");
lookupSourceSettings.SetServiceAccountId("testsaid");
Expand Down
7 changes: 3 additions & 4 deletions ydb/library/yql/providers/generic/actors/yql_generic_lookup_actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
#include <ydb/library/yql/providers/generic/proto/source.pb.h>
#include <ydb/library/yql/providers/generic/connector/libcpp/error.h>
#include <ydb/library/yql/providers/generic/connector/libcpp/utils.h>
#include <ydb/library/yql/providers/generic/proto/range.pb.h>
#include <yql/essentials/providers/common/provider/yql_provider_names.h>
#include <yql/essentials/public/udf/arrow/util.h>
#include <yql/essentials/utils/log/log.h>
Expand Down Expand Up @@ -69,7 +68,7 @@ namespace NYql::NDq {
::NMonitoring::TDynamicCounterPtr taskCounters,
std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> alloc,
std::shared_ptr<IDqAsyncLookupSource::TKeyTypeHelper> keyTypeHelper,
NYql::Generic::TLookupSource&& lookupSource,
NYql::NGeneric::TLookupSource&& lookupSource,
const NKikimr::NMiniKQL::TStructType* keyType,
const NKikimr::NMiniKQL::TStructType* payloadType,
const NKikimr::NMiniKQL::TTypeEnvironment& typeEnv,
Expand Down Expand Up @@ -500,7 +499,7 @@ namespace NYql::NDq {
const NActors::TActorId ParentId;
std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> Alloc;
std::shared_ptr<TKeyTypeHelper> KeyTypeHelper;
const NYql::Generic::TLookupSource LookupSource;
const NYql::NGeneric::TLookupSource LookupSource;
const NKikimr::NMiniKQL::TStructType* const KeyType;
const NKikimr::NMiniKQL::TStructType* const PayloadType;
const NKikimr::NMiniKQL::TStructType* const SelectResultType; // columns from KeyType + PayloadType
Expand Down Expand Up @@ -529,7 +528,7 @@ namespace NYql::NDq {
::NMonitoring::TDynamicCounterPtr taskCounters,
std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> alloc,
std::shared_ptr<IDqAsyncLookupSource::TKeyTypeHelper> keyTypeHelper,
NYql::Generic::TLookupSource&& lookupSource,
NYql::NGeneric::TLookupSource&& lookupSource,
const NKikimr::NMiniKQL::TStructType* keyType,
const NKikimr::NMiniKQL::TStructType* payloadType,
const NKikimr::NMiniKQL::TTypeEnvironment& typeEnv,
Expand Down
2 changes: 1 addition & 1 deletion ydb/library/yql/providers/generic/actors/yql_generic_lookup_actor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ namespace NYql::NDq {
::NMonitoring::TDynamicCounterPtr taskCounters,
std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> alloc,
std::shared_ptr<IDqAsyncLookupSource::TKeyTypeHelper> keyTypeHelper,
NYql::Generic::TLookupSource&& lookupSource,
NYql::NGeneric::TLookupSource&& lookupSource,
const NKikimr::NMiniKQL::TStructType* keyType,
const NKikimr::NMiniKQL::TStructType* payloadType,
const NKikimr::NMiniKQL::TTypeEnvironment& typeEnv,
Expand Down
39 changes: 24 additions & 15 deletions ydb/library/yql/providers/generic/actors/yql_generic_provider_factories.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,23 @@ namespace NYql::NDq {
ISecuredServiceAccountCredentialsFactory::TPtr credentialsFactory,
NYql::NConnector::IClient::TPtr genericClient) {
auto readActorFactory = [credentialsFactory, genericClient](
Generic::TSource&& settings,
NGeneric::TSource&& settings,
IDqAsyncIoFactory::TSourceArguments&& args) {
return CreateGenericReadActor(genericClient, std::move(settings), args.InputIndex, args.StatsLevel,
args.SecureParams, args.TaskParams, args.ComputeActorId, credentialsFactory, args.HolderFactory);
return CreateGenericReadActor(
genericClient,
std::move(settings),
args.InputIndex,
args.StatsLevel,
args.SecureParams,
args.TaskId,
args.TaskParams,
args.ReadRanges,
args.ComputeActorId,
credentialsFactory,
args.HolderFactory);
};

auto lookupActorFactory = [credentialsFactory, genericClient](NYql::Generic::TLookupSource&& lookupSource, IDqAsyncIoFactory::TLookupSourceArguments&& args) {
auto lookupActorFactory = [credentialsFactory, genericClient](NYql::NGeneric::TLookupSource&& lookupSource, IDqAsyncIoFactory::TLookupSourceArguments&& args) {
return CreateGenericLookupActor(
genericClient,
credentialsFactory,
Expand All @@ -34,17 +44,16 @@ namespace NYql::NDq {
};

for (auto& name : {
"ClickHouseGeneric",
"PostgreSqlGeneric",
"YdbGeneric",
"MySqlGeneric",
"GreenplumGeneric",
"MsSQLServerGeneric",
"OracleGeneric",
"LoggingGeneric"}
) {
factory.RegisterSource<Generic::TSource>(name, readActorFactory);
factory.RegisterLookupSource<Generic::TLookupSource>(name, lookupActorFactory);
"ClickHouseGeneric",
"PostgreSqlGeneric",
"YdbGeneric",
"MySqlGeneric",
"GreenplumGeneric",
"MsSQLServerGeneric",
"OracleGeneric",
"LoggingGeneric"}) {
factory.RegisterSource<NGeneric::TSource>(name, readActorFactory);
factory.RegisterLookupSource<NGeneric::TLookupSource>(name, lookupActorFactory);
}
}

Expand Down
Loading
Loading