Skip to content

YDB FQ: Add MySQL to fq proxy #6298

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 4 commits into from
Jul 8, 2024
Merged

YDB FQ: Add MySQL to fq proxy #6298

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a6924c2
lol
skywalker-jpg Jul 3, 2024
8ccd5c3
[add] mysql to fq proxy
skywalker-jpg Jul 4, 2024
a0fc4ef
Revert "lol"
skywalker-jpg Jul 4, 2024
c6ecfbb
[refactor]
skywalker-jpg Jul 5, 2024
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
12 changes: 12 additions & 0 deletions ydb/core/fq/libs/actors/clusters_from_connections.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,6 +283,18 @@ void AddClustersFromConnections(
clusters.emplace(connectionName, GenericProviderName);
break;
}
case FederatedQuery::ConnectionSetting::kMysqlCluster: {
FillGenericClusterConfig(
common,
*gatewaysConfig.MutableGeneric()->AddClusterMapping(),
conn.content().setting().greenplum_cluster(),
connectionName,
NYql::NConnector::NApi::EDataSourceKind::GREENPLUM,
authToken,
accountIdSignatures);
clusters.emplace(connectionName, GenericProviderName);
break;
}

// Do not replace with default. Adding a new connection should cause a compilation error
case FederatedQuery::ConnectionSetting::CONNECTION_NOT_SET:
Expand Down
11 changes: 11 additions & 0 deletions ydb/core/fq/libs/common/util.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,9 @@ TString ExtractServiceAccountId(const FederatedQuery::ConnectionSetting& setting
case FederatedQuery::ConnectionSetting::kGreenplumCluster: {
return GetServiceAccountId(setting.greenplum_cluster().auth());
}
case FederatedQuery::ConnectionSetting::kMysqlCluster: {
return GetServiceAccountId(setting.mysql_cluster().auth());
}
// Do not replace with default. Adding a new connection should cause a compilation error
case FederatedQuery::ConnectionSetting::CONNECTION_NOT_SET:
break;
Expand Down Expand Up @@ -162,6 +165,8 @@ TMaybe<TString> GetLogin(const FederatedQuery::ConnectionSetting& setting) {
return setting.postgresql_cluster().login();
case FederatedQuery::ConnectionSetting::kGreenplumCluster:
return setting.greenplum_cluster().login();
case FederatedQuery::ConnectionSetting::kMysqlCluster:
return setting.greenplum_cluster().login();
}
}

Expand All @@ -183,6 +188,8 @@ TMaybe<TString> GetPassword(const FederatedQuery::ConnectionSetting& setting) {
return setting.postgresql_cluster().password();
case FederatedQuery::ConnectionSetting::kGreenplumCluster:
return setting.greenplum_cluster().password();
case FederatedQuery::ConnectionSetting::kMysqlCluster:
return setting.mysql_cluster().password();
}
}

Expand All @@ -204,6 +211,8 @@ EYdbComputeAuth GetYdbComputeAuthMethod(const FederatedQuery::ConnectionSetting&
return GetBasicAuthMethod(setting.postgresql_cluster().auth());
case FederatedQuery::ConnectionSetting::kGreenplumCluster:
return GetBasicAuthMethod(setting.greenplum_cluster().auth());
case FederatedQuery::ConnectionSetting::kMysqlCluster:
return GetBasicAuthMethod(setting.mysql_cluster().auth());
}
}

Expand All @@ -223,6 +232,8 @@ FederatedQuery::IamAuth GetAuth(const FederatedQuery::Connection& connection) {
return connection.content().setting().postgresql_cluster().auth();
case FederatedQuery::ConnectionSetting::kGreenplumCluster:
return connection.content().setting().greenplum_cluster().auth();
case FederatedQuery::ConnectionSetting::kMysqlCluster:
return connection.content().setting().mysql_cluster().auth();
case FederatedQuery::ConnectionSetting::CONNECTION_NOT_SET:
return FederatedQuery::IamAuth{};
}
Expand Down
1 change: 1 addition & 0 deletions ydb/core/fq/libs/compute/common/config.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,7 @@ class TComputeConfig {
case FederatedQuery::ConnectionSetting::kClickhouseCluster:
case FederatedQuery::ConnectionSetting::kPostgresqlCluster:
case FederatedQuery::ConnectionSetting::kGreenplumCluster:
case FederatedQuery::ConnectionSetting::kMysqlCluster:
case FederatedQuery::ConnectionSetting::kYdbDatabase:
return true;
case FederatedQuery::ConnectionSetting::kDataStreams:
Expand Down
14 changes: 14 additions & 0 deletions ydb/core/fq/libs/control_plane_proxy/actors/query_utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,20 @@ TString MakeCreateExternalDataSourceQuery(
"use_tls"_a = common.GetDisableSslForGenericDataSources() ? "false" : "true",
"schema"_a = gpschema ? ", SCHEMA=" + EncloseAndEscapeString(gpschema, '"') : TString{});

}
case FederatedQuery::ConnectionSetting::kMysqlCluster: {
properties = fmt::format(
R"(
SOURCE_TYPE="MySQL",
MDB_CLUSTER_ID={mdb_cluster_id},
DATABASE_NAME={database_name},
USE_TLS="{use_tls}"
{schema}
)",
"mdb_cluster_id"_a = EncloseAndEscapeString(connectionContent.setting().mysql_cluster().database_id(), '"'),
"database_name"_a = EncloseAndEscapeString(connectionContent.setting().mysql_cluster().database_name(), '"'),
"use_tls"_a = common.GetDisableSslForGenericDataSources() ? "false" : "true");

}
break;
}
Expand Down
3 changes: 3 additions & 0 deletions ydb/core/fq/libs/control_plane_proxy/utils/utils.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,9 @@ TString ExtractServiceAccountIdWithConnection(const T& setting) {
case FederatedQuery::ConnectionSetting::kGreenplumCluster: {
return GetServiceAccountId(setting.greenplum_cluster().auth());
}
case FederatedQuery::ConnectionSetting::kMysqlCluster: {
return GetServiceAccountId(setting.greenplum_cluster().auth());
}
// Do not replace with default. Adding a new connection should cause a compilation error
case FederatedQuery::ConnectionSetting::CONNECTION_NOT_SET:
break;
Expand Down
15 changes: 15 additions & 0 deletions ydb/core/fq/libs/control_plane_storage/request_validators.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,21 @@ NYql::TIssues ValidateConnectionSetting(
}
break;
}
case FederatedQuery::ConnectionSetting::kMysqlCluster: {
const FederatedQuery::GreenplumCluster database = setting.greenplum_cluster();
if (!database.has_auth() || database.auth().identity_case() == FederatedQuery::IamAuth::IDENTITY_NOT_SET) {
issues.AddIssue(MakeErrorIssue(TIssuesIds::BAD_REQUEST, "content.setting.greenplum_database.auth field is not specified"));
}

if (database.auth().identity_case() == FederatedQuery::IamAuth::kCurrentIam && disableCurrentIam) {
issues.AddIssue(MakeErrorIssue(TIssuesIds::BAD_REQUEST, "current iam authorization is disabled"));
}

if (!database.database_id() && !database.database_name()) {
issues.AddIssue(MakeErrorIssue(TIssuesIds::BAD_REQUEST, "content.setting.greenplum_database.{database_id or database_name} field is not specified"));
}
break;
}
case FederatedQuery::ConnectionSetting::kObjectStorage: {
const FederatedQuery::ObjectStorageConnection objectStorage = setting.object_storage();
if (!objectStorage.has_auth() || objectStorage.auth().identity_case() == FederatedQuery::IamAuth::IDENTITY_NOT_SET) {
Expand Down
2 changes: 2 additions & 0 deletions ydb/core/fq/libs/control_plane_storage/ydb_control_plane_storage_queries.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@ FederatedQuery::IamAuth::IdentityCase GetIamAuth(const FederatedQuery::Connectio
return setting.postgresql_cluster().auth().identity_case();
case FederatedQuery::ConnectionSetting::kGreenplumCluster:
return setting.greenplum_cluster().auth().identity_case();
case FederatedQuery::ConnectionSetting::kMysqlCluster:
return setting.mysql_cluster().auth().identity_case();
case FederatedQuery::ConnectionSetting::CONNECTION_NOT_SET:
return FederatedQuery::IamAuth::IDENTITY_NOT_SET;
}
Expand Down
10 changes: 10 additions & 0 deletions ydb/public/api/protos/draft/fq.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -501,6 +501,14 @@ message GreenplumCluster {
IamAuth auth = 6;
}

message MySQLCluster {
string database_id = 1 [(Ydb.length).le = 1024];
string database_name = 2 [(Ydb.length).le = 1024];
string login = 3 [(Ydb.length).le = 1024, (Ydb.sensitive) = true];
string password = 4 [(Ydb.length).le = 1024, (Ydb.sensitive) = true];
IamAuth auth = 5;
}

message ConnectionSetting {
enum ConnectionType {
CONNECTION_TYPE_UNSPECIFIED = 0;
Expand All @@ -511,6 +519,7 @@ message ConnectionSetting {
MONITORING = 5;
POSTGRESQL_CLUSTER = 6;
GREENPLUM_CLUSTER = 7;
MYSQL_CLUSTER = 8;
}

oneof connection {
Expand All @@ -521,6 +530,7 @@ message ConnectionSetting {
Monitoring monitoring = 5;
PostgreSQLCluster postgresql_cluster = 6;
GreenplumCluster greenplum_cluster = 7;
MySQLCluster mysql_cluster = 8;
}
}

Expand Down