merge to yq stable YQ added YDB_AUTH_TICKET_HEADER for fq proxy ydb requests

.github/CODEOWNERS

@@ -1,54 +1 @@
-/*.md @ydb-platform/docs
-
-/build @ydb-platform/ci
-/certs @ydb-platform/ci
-/contrib @ydb-platform/ci
-/devtools @ydb-platform/ci
-/library @ydb-platform/ci
-/scripts @ydb-platform/ci
-/tools @ydb-platform/ci
-/util @ydb-platform/ci
-/vendor @ydb-platform/ci
-/yql @ydb-platform/ci
-/yt @ydb-platform/ci
-
-/ydb/core/fq/ @ydb-platform/fq
-/ydb/core/public_http/ @ydb-platform/fq
-
-/ydb/docs/ @ydb-platform/docs
-
-/ydb/library/yql/ @ydb-platform/yql
-/ydb/library/yql/dq @ydb-platform/yql @ydb-platform/qp
-/ydb/library/yql/dq/actors/common @ydb-platform/fq
-/ydb/library/yql/providers/common/http_gateway @ydb-platform/fq
-/ydb/library/yql/providers/common/db_id_async_resolver @ydb-platform/fq
-/ydb/library/yql/providers/common/pushdown @ydb-platform/fq
-/ydb/library/yql/providers/generic @ydb-platform/fq
-/ydb/library/yql/providers/pq @ydb-platform/fq
-/ydb/library/yql/providers/s3 @ydb-platform/fq
-/ydb/library/yql/providers/solomon @ydb-platform/fq
-/ydb/library/yql/tests/sql/solomon @ydb-platform/fq
-/ydb/library/yql/tests/sql/suites/solomon @ydb-platform/fq
-/ydb/library/yql/udfs/common/clickhouse/client @ydb-platform/fq
-
-/ydb/library/yql/yt @Krock21 @Krisha11 @zlobober @gritukan
-
-/ydb/services/fq/ @ydb-platform/fq
-
-/ydb/core/kafka_proxy @ydb-platform/Topics
-/ydb/core/persqueue @ydb-platform/Topics
-/ydb/services/datastreams @ydb-platform/Topics
-/ydb/services/deprecated/persqueue_v0 @ydb-platform/Topics
-/ydb/services/persqueue_v1 @ydb-platform/Topics
-
-/ydb/core/config/ut @ydb-platform/core
-
-/ydb/core/viewer @ydb-platform/ui-backend
-/ydb/core/protos/node_whiteboard.proto @ydb-platform/ui-backend
-
-/ydb/core/formats/arrow @ydb-platform/cs
-/ydb/core/tx/columnshard @ydb-platform/cs
-
-/ydb/apps/ydb @ydb-platform/cli
-/ydb/public/lib/ydb_cli @ydb-platform/cli
-
+* @ydb-platform/fq

ydb/core/driver_lib/run/config_helpers.cpp

@@ -1,5 +1,9 @@
 #include "config_helpers.h"
 
+#include <ydb/core/base/localdb.h>
+#include <ydb/core/protos/bootstrap.pb.h>
+#include <ydb/core/protos/resource_broker.pb.h>
+
 #include <ydb/library/actors/util/affinity.h>
 
 
@@ -114,4 +118,31 @@ NActors::TSchedulerConfig CreateSchedulerConfig(const NKikimrConfig::TActorSyste
 
 }  // namespace NActorSystemConfigHelpers
 
+namespace NKikimrConfigHelpers {
+
+NMemory::TResourceBrokerConfig CreateMemoryControllerResourceBrokerConfig(const NKikimrConfig::TAppConfig& config) {
+    NMemory::TResourceBrokerConfig resourceBrokerSelfConfig; // for backward compatibility
+    auto mergeResourceBrokerConfigs = [&](const NKikimrResourceBroker::TResourceBrokerConfig& resourceBrokerConfig) {
+        if (resourceBrokerConfig.HasResourceLimit() && resourceBrokerConfig.GetResourceLimit().HasMemory()) {
+            resourceBrokerSelfConfig.LimitBytes = resourceBrokerConfig.GetResourceLimit().GetMemory();
+        }
+        for (const auto& queue : resourceBrokerConfig.GetQueues()) {
+            if (queue.GetName() == NLocalDb::KqpResourceManagerQueue) {
+                if (queue.HasLimit() && queue.GetLimit().HasMemory()) {
+                    resourceBrokerSelfConfig.QueryExecutionLimitBytes = queue.GetLimit().GetMemory();
+                }
+            }
+        }
+    };
+    if (config.HasBootstrapConfig() && config.GetBootstrapConfig().HasResourceBroker()) {
+        mergeResourceBrokerConfigs(config.GetBootstrapConfig().GetResourceBroker());
+    }
+    if (config.HasResourceBrokerConfig()) {
+        mergeResourceBrokerConfigs(config.GetResourceBrokerConfig());
+    }
+    return resourceBrokerSelfConfig;
+}
+
+}  // namespace NKikimrConfigHelpers
+
 }  // namespace NKikimr

ydb/core/driver_lib/run/config_helpers.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <ydb/core/memory_controller/memory_controller.h>
 #include <ydb/core/protos/config.pb.h>
 
 #include <ydb/library/actors/core/config.h>
@@ -15,4 +16,10 @@ NActors::TSchedulerConfig CreateSchedulerConfig(const NKikimrConfig::TActorSyste
 
 }  // namespace NActorSystemConfigHelpers
 
+namespace NKikimrConfigHelpers {
+
+NMemory::TResourceBrokerConfig CreateMemoryControllerResourceBrokerConfig(const NKikimrConfig::TAppConfig& config);
+
+}  // namespace NKikimrConfigHelpers
+
 }  // namespace NKikimr

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -2049,28 +2049,8 @@ void TMemoryControllerInitializer::InitializeServices(
     NActors::TActorSystemSetup* setup,
     const NKikimr::TAppData* appData)
 {
-    NMemory::TResourceBrokerConfig resourceBrokerSelfConfig; // for backward compatibility
-    auto mergeResourceBrokerConfigs = [&](const NKikimrResourceBroker::TResourceBrokerConfig& resourceBrokerConfig) {
-        if (resourceBrokerConfig.HasResourceLimit() && resourceBrokerConfig.GetResourceLimit().HasMemory()) {
-            resourceBrokerSelfConfig.LimitBytes = resourceBrokerConfig.GetResourceLimit().GetMemory();
-        }
-        for (const auto& queue : resourceBrokerConfig.GetQueues()) {
-            if (queue.GetName() == NLocalDb::KqpResourceManagerQueue) {
-                if (queue.HasLimit() && queue.GetLimit().HasMemory()) {
-                    resourceBrokerSelfConfig.QueryExecutionLimitBytes = queue.GetLimit().GetMemory();
-                }
-            }
-        }
-    };
-    if (Config.HasBootstrapConfig() && Config.GetBootstrapConfig().HasResourceBroker()) {
-        mergeResourceBrokerConfigs(Config.GetBootstrapConfig().GetResourceBroker());
-    }
-    if (Config.HasResourceBrokerConfig()) {
-        mergeResourceBrokerConfigs(Config.GetResourceBrokerConfig());
-    }
-
     auto* actor = NMemory::CreateMemoryController(TDuration::Seconds(1), ProcessMemoryInfoProvider,
-        Config.GetMemoryControllerConfig(), resourceBrokerSelfConfig,
+        Config.GetMemoryControllerConfig(), NKikimrConfigHelpers::CreateMemoryControllerResourceBrokerConfig(Config),
         appData->Counters);
     setup->LocalServices.emplace_back(
         NMemory::MakeMemoryControllerId(0),

ydb/core/external_sources/s3/ut/s3_aws_credentials_ut.cpp

@@ -251,6 +251,88 @@ Y_UNIT_TEST_SUITE(S3AwsCredentials) {
         }
 
     }
+
+    Y_UNIT_TEST(TestInsertEscaping) {
+        const TString externalDataSourceName = "/Root/external_data_source";
+        auto s3ActorsFactory = NYql::NDq::CreateS3ActorsFactory();
+        auto kikimr = MakeKikimrRunner(true, nullptr, nullptr, std::nullopt, s3ActorsFactory);
+
+        auto tc = kikimr->GetTableClient();
+        auto session = tc.CreateSession().GetValueSync().GetSession();
+        const TString query = fmt::format(R"(
+            CREATE OBJECT id (TYPE SECRET) WITH (value=`minio`);
+            CREATE OBJECT key (TYPE SECRET) WITH (value=`minio123`);
+            CREATE EXTERNAL DATA SOURCE `{external_source}` WITH (
+                SOURCE_TYPE="ObjectStorage",
+                LOCATION="{location}",
+                AUTH_METHOD="AWS",
+                AWS_ACCESS_KEY_ID_SECRET_NAME="id",
+                AWS_SECRET_ACCESS_KEY_SECRET_NAME="key",
+                AWS_REGION="ru-central-1"
+            );
+            )",
+            "external_source"_a = externalDataSourceName,
+            "location"_a = "localhost:" + GetExternalPort("minio", "9000") + "/datalake/"
+            );
+
+        auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+        UNIT_ASSERT_C(result.GetStatus() == NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
+
+        WaitBucket(kikimr, externalDataSourceName);
+
+        auto db = kikimr->GetQueryClient();
+
+        TString path = TStringBuilder() << "exp_folder/some_" << EscapeC(GetSymbolsString(' ', '~', "*?{}`")) << "\\`";
+
+        {
+            // NB: AtomicUploadCommit = "false" because in minio ListMultipartUploads by prefix is not supported
+            auto scriptExecutionOperation = db.ExecuteScript(fmt::format(R"(
+                PRAGMA s3.AtomicUploadCommit = "false";
+                INSERT INTO `{external_source}`.`{path}/` WITH (FORMAT = "csv_with_names")
+                SELECT * FROM `{external_source}`.`/a/` WITH (
+                    format="json_each_row",
+                    schema(
+                        key Utf8 NOT NULL,
+                        value Utf8 NOT NULL
+                    )
+                )
+            )", "external_source"_a = externalDataSourceName, "path"_a = path)).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(scriptExecutionOperation.Status().GetStatus(), EStatus::SUCCESS, scriptExecutionOperation.Status().GetIssues().ToString());
+            UNIT_ASSERT(!scriptExecutionOperation.Metadata().ExecutionId.empty());
+
+            NYdb::NQuery::TScriptExecutionOperation readyOp = WaitScriptExecutionOperation(scriptExecutionOperation.Id(), kikimr->GetDriver());
+            UNIT_ASSERT_EQUAL_C(readyOp.Metadata().ExecStatus, EExecStatus::Completed, readyOp.Status().GetIssues().ToString());
+        }
+
+        {
+            auto scriptExecutionOperation = db.ExecuteScript(fmt::format(R"(
+                SELECT * FROM `{external_source}`.`{path}/` WITH (
+                    format="csv_with_names",
+                    schema(
+                        key Utf8 NOT NULL,
+                        value Utf8 NOT NULL
+                    )
+                )
+            )", "external_source"_a = externalDataSourceName, "path"_a = path)).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(scriptExecutionOperation.Status().GetStatus(), EStatus::SUCCESS, scriptExecutionOperation.Status().GetIssues().ToString());
+            UNIT_ASSERT(!scriptExecutionOperation.Metadata().ExecutionId.empty());
+
+            NYdb::NQuery::TScriptExecutionOperation readyOp = WaitScriptExecutionOperation(scriptExecutionOperation.Id(), kikimr->GetDriver());
+            UNIT_ASSERT_EQUAL_C(readyOp.Metadata().ExecStatus, EExecStatus::Completed, readyOp.Status().GetIssues().ToString());
+            TFetchScriptResultsResult results = db.FetchScriptResults(scriptExecutionOperation.Id(), 0).ExtractValueSync();
+            UNIT_ASSERT_C(results.IsSuccess(), results.GetIssues().ToString());
+
+            TResultSetParser resultSet(results.ExtractResultSet());
+            UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnsCount(), 2);
+            UNIT_ASSERT_VALUES_EQUAL(resultSet.RowsCount(), 2);
+            UNIT_ASSERT(resultSet.TryNextRow());
+            UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(0).GetUtf8(), "1");
+            UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(1).GetUtf8(), "trololo");
+            UNIT_ASSERT(resultSet.TryNextRow());
+            UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(0).GetUtf8(), "2");
+            UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(1).GetUtf8(), "hello world");
+        }
+    }
 }
 
 } // namespace NKikimr::NKqp

ydb/core/fq/libs/actors/clusters_from_connections.cpp

@@ -1,7 +1,7 @@
 #include "clusters_from_connections.h"
 
 #include <yql/essentials/providers/common/provider/yql_provider_names.h>
-#include <ydb/library/yql/providers/generic/connector/api/common/data_source.pb.h>
+#include <yql/essentials/providers/common/proto/gateways_config.pb.h>
 #include <ydb/library/yql/providers/generic/provider/yql_generic_cluster_config.h>
 #include <yql/essentials/utils/url_builder.h>
 #include <ydb/library/actors/http/http.h>
@@ -88,6 +88,19 @@ std::pair<TString, bool> ParseHttpEndpoint(const TString& endpoint) {
     return std::make_pair(ToString(host), scheme != "http");
 }
 
+std::pair<TString, TIpPort> ParseGrpcEndpoint(const TString& endpoint) {
+    TStringBuf scheme;
+    TStringBuf address;
+    TStringBuf uri;
+    NHttp::CrackURL(endpoint, scheme, address, uri);
+
+    TString hostname;
+    TIpPort port;
+    NHttp::CrackAddress(TString(address), hostname, port);
+
+    return {hostname, port};
+}
+
 void FillSolomonClusterConfig(NYql::TSolomonClusterConfig& clusterConfig,
     const TString& name,
     const TString& authToken,
@@ -112,7 +125,7 @@ void FillGenericClusterConfigBase(
     TGenericClusterConfig& clusterCfg,
     const TConnection& connection,
     const TString& connectionName,
-    NConnector::NApi::EDataSourceKind dataSourceKind,
+    NYql::EGenericDataSourceKind dataSourceKind,
     const TString& authToken,
     const THashMap<TString, TString>& accountIdSignatures
 ) {
@@ -128,21 +141,21 @@ void FillGenericClusterConfigBase(
     // In YQv1 we just hardcode the appropriate protocols here.
     // In YQv2 protocol can be configured via `CREATE EXTERNAL DATA SOURCE` params.
     switch (dataSourceKind) {
-        case NYql::NConnector::NApi::CLICKHOUSE:
-            clusterCfg.SetProtocol(common.GetUseNativeProtocolForClickHouse() ? NYql::NConnector::NApi::EProtocol::NATIVE : NYql::NConnector::NApi::EProtocol::HTTP);
+        case NYql::EGenericDataSourceKind::CLICKHOUSE:
+            clusterCfg.SetProtocol(common.GetUseNativeProtocolForClickHouse() ? NYql::EGenericProtocol::NATIVE : NYql::EGenericProtocol::HTTP);
             break;
-        case NYql::NConnector::NApi::GREENPLUM:
-            clusterCfg.SetProtocol(NYql::NConnector::NApi::EProtocol::NATIVE);
+        case NYql::EGenericDataSourceKind::GREENPLUM:
+            clusterCfg.SetProtocol(NYql::EGenericProtocol::NATIVE);
             break;
-        case NYql::NConnector::NApi::MYSQL:
-            clusterCfg.SetProtocol(NYql::NConnector::NApi::EProtocol::NATIVE);
+        case NYql::EGenericDataSourceKind::MYSQL:
+            clusterCfg.SetProtocol(NYql::EGenericProtocol::NATIVE);
             break;
-        case NYql::NConnector::NApi::POSTGRESQL:
-            clusterCfg.SetProtocol(NYql::NConnector::NApi::EProtocol::NATIVE);
+        case NYql::EGenericDataSourceKind::POSTGRESQL:
+            clusterCfg.SetProtocol(NYql::EGenericProtocol::NATIVE);
             break;
         default:
             ythrow yexception() << "Unexpected data source kind: '" 
-                                << NYql::NConnector::NApi::EDataSourceKind_Name(dataSourceKind) << "'";
+                                << NYql::EGenericDataSourceKind_Name(dataSourceKind) << "'";
     }
 
     ValidateGenericClusterConfig(clusterCfg, "NFq::FillGenericClusterFromConfig");
@@ -154,7 +167,7 @@ void FillGenericClusterConfig(
     TGenericClusterConfig& clusterCfg,
     const TConnection& connection,
     const TString& connectionName,
-    NConnector::NApi::EDataSourceKind dataSourceKind,
+    NYql::EGenericDataSourceKind dataSourceKind,
     const TString& authToken,
     const THashMap<TString, TString>& accountIdSignatures
 ) {
@@ -167,7 +180,7 @@ void FillGenericClusterConfig<FederatedQuery::PostgreSQLCluster>(
     TGenericClusterConfig& clusterCfg,
     const FederatedQuery::PostgreSQLCluster& connection,
     const TString& connectionName,
-    NConnector::NApi::EDataSourceKind dataSourceKind,
+    NYql::EGenericDataSourceKind dataSourceKind,
     const TString& authToken,
     const THashMap<TString, TString>& accountIdSignatures
 ){
@@ -227,11 +240,21 @@ void AddClustersFromConnections(
         case FederatedQuery::ConnectionSetting::kYdbDatabase: {
             const auto& db = conn.content().setting().ydb_database();
             auto* clusterCfg = gatewaysConfig.MutableGeneric()->AddClusterMapping();
-            clusterCfg->SetKind(NYql::NConnector::NApi::EDataSourceKind::YDB);
-            clusterCfg->SetProtocol(NYql::NConnector::NApi::EProtocol::NATIVE);
+            clusterCfg->SetKind(NYql::EGenericDataSourceKind::YDB);
+            clusterCfg->SetProtocol(NYql::EGenericProtocol::NATIVE);
             clusterCfg->SetName(connectionName);
-            clusterCfg->SetDatabaseId(db.database_id());
-            clusterCfg->SetUseSsl(!common.GetDisableSslForGenericDataSources());
+            if (const auto& databaseId = db.database_id()) {
+                clusterCfg->SetDatabaseId(databaseId);
+                clusterCfg->SetUseSsl(!common.GetDisableSslForGenericDataSources());
+            } else {
+                const auto& [host, port] = ParseGrpcEndpoint(db.endpoint());
+
+                auto& endpoint = *clusterCfg->MutableEndpoint();
+                endpoint.set_host(host);
+                endpoint.set_port(port);
+                clusterCfg->SetUseSsl(db.secure());
+                clusterCfg->SetDatabaseName(db.database());
+            }
             FillClusterAuth(*clusterCfg, db.auth(), authToken, accountIdSignatures);
             clusters.emplace(connectionName, GenericProviderName);
             break;
@@ -242,7 +265,7 @@ void AddClustersFromConnections(
                 *gatewaysConfig.MutableGeneric()->AddClusterMapping(),
                 conn.content().setting().clickhouse_cluster(),
                 connectionName,
-                NYql::NConnector::NApi::EDataSourceKind::CLICKHOUSE,
+                NYql::EGenericDataSourceKind::CLICKHOUSE,
                 authToken,
                 accountIdSignatures);
             clusters.emplace(connectionName, GenericProviderName);
@@ -275,7 +298,7 @@ void AddClustersFromConnections(
                 *gatewaysConfig.MutableGeneric()->AddClusterMapping(),
                 conn.content().setting().postgresql_cluster(),
                 connectionName,
-                NYql::NConnector::NApi::EDataSourceKind::POSTGRESQL,
+                NYql::EGenericDataSourceKind::POSTGRESQL,
                 authToken,
                 accountIdSignatures);
             clusters.emplace(connectionName, GenericProviderName);
@@ -287,7 +310,7 @@ void AddClustersFromConnections(
                 *gatewaysConfig.MutableGeneric()->AddClusterMapping(),
                 conn.content().setting().greenplum_cluster(),
                 connectionName,
-                NYql::NConnector::NApi::EDataSourceKind::GREENPLUM,
+                NYql::EGenericDataSourceKind::GREENPLUM,
                 authToken,
                 accountIdSignatures);
             clusters.emplace(connectionName, GenericProviderName);
@@ -299,7 +322,7 @@ void AddClustersFromConnections(
                 *gatewaysConfig.MutableGeneric()->AddClusterMapping(),
                 conn.content().setting().mysql_cluster(),
                 connectionName,
-                NYql::NConnector::NApi::EDataSourceKind::MYSQL,
+                NYql::EGenericDataSourceKind::MYSQL,
                 authToken,
                 accountIdSignatures);
             clusters.emplace(connectionName, GenericProviderName);
@@ -308,7 +331,7 @@ void AddClustersFromConnections(
         case FederatedQuery::ConnectionSetting::kLogging: {
             const auto& connection = conn.content().setting().logging();
             auto* clusterCfg = gatewaysConfig.MutableGeneric()->AddClusterMapping();
-            clusterCfg->SetKind(NYql::NConnector::NApi::EDataSourceKind::LOGGING);
+            clusterCfg->SetKind(NYql::EGenericDataSourceKind::LOGGING);
             clusterCfg->SetName(connectionName);
             clusterCfg->mutable_datasourceoptions()->insert({"folder_id", connection.folder_id()});
             FillClusterAuth(*clusterCfg, connection.auth(), authToken, accountIdSignatures);