Add TablePathPrefix. (#16894)

Author: Tony-Romanov

ydb/apps/etcd_proxy/proxy.cpp

@@ -61,7 +61,7 @@ int TProxy::Discovery() {
 }
 
 int TProxy::StartServer() {
-    if (const auto res = Stuff->Client->ExecuteQuery(NEtcd::GetLastRevisionSQL(), NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync(); res.IsSuccess()) {
+    if (const auto res = Stuff->Client->ExecuteQuery(NEtcd::GetLastRevisionSQL(Stuff->TablePrefix), NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync(); res.IsSuccess()) {
         if (auto result = res.GetResultSetParser(0); result.TryNextRow()) {
             Stuff->Revision.store(NYdb::TValueParser(result.GetValue(0)).GetInt64());
         } else {
@@ -129,7 +129,7 @@ int TProxy::Run() {
 }
 
 int TProxy::InitDatabase() {
-    if (const auto res = Stuff->Client->ExecuteQuery(NEtcd::GetCreateTablesSQL(), NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync(); res.IsSuccess()) {
+    if (const auto res = Stuff->Client->ExecuteQuery(NEtcd::GetCreateTablesSQL(Stuff->TablePrefix), NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync(); res.IsSuccess()) {
         std::cout << "Database " << Database << " on " << Endpoint << " was initialized." << std::endl;
         return 0;
     } else {
@@ -210,7 +210,7 @@ int TProxy::ImportDatabase() {
     const auto driver = NYdb::TDriver(config);
     auto client = NYdb::NTable::TTableClient(driver);
 
-    if (const auto res = client.BulkUpsert(Database + "/content", std::move(value)).ExtractValueSync(); !res.IsSuccess()) {
+    if (const auto res = client.BulkUpsert(Database + Folder + "/content", std::move(value)).ExtractValueSync(); !res.IsSuccess()) {
         std::cout << res.GetIssues().ToString() << std::endl;
         return 1;
     }
@@ -235,6 +235,7 @@ TProxy::TProxy(int argc, char** argv)
 
     opts.AddLongOption("database", "YDB etcd databse").Required().RequiredArgument("DATABASE").StoreResult(&Database);
     opts.AddLongOption("endpoint", "YDB endpoint to connect").Required().RequiredArgument("ENDPOINT").StoreResult(&Endpoint);
+    opts.AddLongOption("folder", "YDB etcd root folder").Required().RequiredArgument("FOLDER").StoreResult(&Folder);
     opts.AddLongOption("token", "YDB token for connection").Optional().RequiredArgument("TOKEN").StoreResult(&Token);
     opts.AddLongOption("ydbca", "YDB CA for connection").Optional().RequiredArgument("CA").StoreResult(&CA);
 
@@ -256,6 +257,12 @@ TProxy::TProxy(int argc, char** argv)
         LockAllMemory(LockCurrentMemory);
     }
 
+    if (!Folder.empty()) {
+        std::ostringstream prefix;
+        prefix << "pragma TablePathPrefix = '" << Database << Folder << "';" << std::endl;
+        Stuff->TablePrefix = prefix.str();
+    }
+
     THolder<NActors::TActorSystemSetup> actorSystemSetup = BuildActorSystemSetup();
 
     TIntrusivePtr<NActors::NLog::TSettings> loggerSettings = BuildLoggerSettings();

ydb/apps/etcd_proxy/proxy.h

@@ -45,7 +45,7 @@ class TProxy {
     std::unique_ptr<NYdbGrpc::TGRpcServer> GRpcServer;
 
     bool Initialize_ = false;
-    std::string Database, Endpoint, Token, CA;
+    std::string Database, Endpoint, Token, CA, Folder;
     ui16 ListeningPort = 2379U;
     std::string Root, Cert, Key;
     std::string ImportFrom_, ImportPrefix_;

ydb/apps/etcd_proxy/service/etcd_base_init.cpp

@@ -4,12 +4,12 @@
 
 namespace NEtcd {
 
-std::string GetCreateTablesSQL() {
-    return NResource::Find("create.sql"sv);
+std::string GetCreateTablesSQL(const std::string& prefix) {
+    return prefix + NResource::Find("create.sql"sv);
 }
 
-std::string GetLastRevisionSQL() {
-    return "select nvl(max(`modified`), 1L) from `content`; select nvl(max(`id`), 1L) from `leases`;";
+std::string GetLastRevisionSQL(const std::string& prefix) {
+    return prefix + "select nvl(max(`modified`), 1L) from `content`; select nvl(max(`id`), 1L) from `leases`;";
 }
 
 }

ydb/apps/etcd_proxy/service/etcd_base_init.h

@@ -4,9 +4,9 @@
 
 namespace NEtcd {
 
-std::string GetCreateTablesSQL();
+std::string GetCreateTablesSQL(const std::string& prefix);
 
-std::string GetLastRevisionSQL();
+std::string GetLastRevisionSQL(const std::string& prefix);
 
 }
 

ydb/apps/etcd_proxy/service/etcd_impl.cpp

@@ -816,6 +816,7 @@ class TEtcdRequestGrpc
         std::ostringstream sql;
         NYdb::TParamsBuilder params;
         sql << "-- " << GetRequestName() << " >>>>" << std::endl;
+        sql << Stuff->TablePrefix;
         this->MakeQueryWithParams(sql, params);
         sql << "-- " << GetRequestName() << " <<<<" << std::endl;
 //      std::cout << std::endl << sql.view() << std::endl;

ydb/apps/etcd_proxy/service/etcd_shared.h

@@ -17,6 +17,7 @@ struct TSharedStuff {
     std::unique_ptr<NYdb::NQuery::TQueryClient> Client;
     std::atomic<i64> Revision = 0LL, Lease = 0LL;
     NActors::TActorId Watchtower;
+    std::string TablePrefix;
 };
 
 std::string IncrementKey(std::string key);

ydb/apps/etcd_proxy/service/etcd_watch.cpp

@@ -65,6 +65,7 @@ class TKeysKeeper : public TActorBootstrapped<TKeysKeeper> {
         const auto& leasePraramName = AddParam<i64>("Lease", params, ev->Get()->Record.id());
 
         std::ostringstream sql;
+        sql << Stuff->TablePrefix;
         sql << "update `leases` set `updated` = CurrentUtcDatetime(`id`) where " << leasePraramName << " = `id`;" << std::endl;
         sql << "select `id`, `ttl` - unwrap(cast(CurrentUtcDatetime(`id`) - `updated` as Int64) / 1000000L) as `granted` from `leases` where " << leasePraramName << " = `id`;" << std::endl;
 
@@ -165,6 +166,7 @@ class TWatch : public TActorBootstrapped<TWatch> {
         MakeSimplePredicate(Key, RangeEnd, where, params);
 
         std::ostringstream sql;
+        sql << Stuff->TablePrefix;
         if (WithPrevious) {
             sql << "select * from (select max_by(TableRow(), `modified`) from `content` where " << revName << " > `modified` and " << where.view() << " group by `key`) flatten columns union all" << std::endl;
         }
@@ -586,6 +588,7 @@ class TWatchtower : public TActorBootstrapped<TWatchtower> {
         Revision = Stuff->Revision.fetch_add(1LL) + 1LL;
 
         std::ostringstream sql;
+        sql << Stuff->TablePrefix;
         NYdb::TParamsBuilder params;
         const auto& revName = AddParam("Revision", params, Revision);
 

ydb/apps/etcd_proxy/service/ut/etcd_service_ut.cpp

@@ -105,7 +105,7 @@ void MakeTables(auto &channel) {
     const auto stub = Ydb::Query::V1::QueryService::NewStub(channel);
     Ydb::Query::ExecuteQueryRequest request;
     request.set_exec_mode(Ydb::Query::EXEC_MODE_EXECUTE);
-    request.mutable_query_content()->set_text(std::string("PRAGMA TablePathPrefix='/Root';\n") + NEtcd::GetCreateTablesSQL());
+    request.mutable_query_content()->set_text(NEtcd::GetCreateTablesSQL(std::string("PRAGMA TablePathPrefix='/Root';\n")));
 
     grpc::ClientContext executeCtx;
     Ydb::Query::ExecuteQueryResponsePart response;