Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
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
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
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
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
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