Merge branch 'main' into merge-libs-250627-0050

Author: naspirato

.github/config/muted_ya.txt

@@ -19,7 +19,7 @@ ydb/core/blobstorage/pdisk/ut TPDiskTest.PDiskSlotSizeInUnits
 ydb/core/blobstorage/ut_blobstorage GroupReconfiguration.BsControllerDoesNotDisableGroupNoRequestsToNodesWVDisks
 ydb/core/blobstorage/ut_blobstorage NodeDisconnected.BsQueueRetries
 ydb/core/blobstorage/ut_blobstorage/ut_group_reconfiguration GroupReconfigurationRace.Test_block42
-ydb/core/blobstorage/ut_blobstorage/ut_group_reconfiguration unittest.[*/*] chunk
+ydb/core/blobstorage/ut_blobstorage/ut_group_reconfiguration *.[*/*] chunk
 ydb/core/blobstorage/ut_blobstorage/ut_huge HugeBlobOnlineSizeChange.Compaction
 ydb/core/blobstorage/ut_blobstorage/ut_huge unittest.[*/*] chunk
 ydb/core/blobstorage/ut_blobstorage/ut_scrub BlobScrubbing.block42
@@ -36,7 +36,7 @@ ydb/core/keyvalue/ut_trace TKeyValueTracingTest.ReadHuge
 ydb/core/keyvalue/ut_trace TKeyValueTracingTest.ReadSmall
 ydb/core/keyvalue/ut_trace TKeyValueTracingTest.WriteHuge
 ydb/core/keyvalue/ut_trace TKeyValueTracingTest.WriteSmall
-ydb/core/kqp/ut/batch_operations unittest.[*/*] chunk
+ydb/core/kqp/ut/batch_operations *.[*/*] chunk
 ydb/core/kqp/ut/cost KqpCost.VectorIndexLookup+useSink
 ydb/core/kqp/ut/cost KqpCost.VectorIndexLookup-useSink
 ydb/core/kqp/ut/data_integrity KqpDataIntegrityTrails.UpsertEvWriteQueryService-isOlap+useOltpSink

.github/workflows/update_changelog.yml

@@ -13,11 +13,11 @@ on:
         description: "Date '2021-01-01' or tag 'v1.0.0' or commit 'sha'"
         required: true
       changelog_path:
-        description: "The value associated with the type."
+        description: "Changelog path"
         required: true
         default: "CHANGELOG.md"
       prefix:
-        description: "Prefix for the changelog update"
+        description: "Prefix for the branch with PR"
         required: true
         default: "changelog-for-"
 

ROADMAP.md

@@ -126,8 +126,8 @@ We use the following symbols as abbreviations:
 ## Embedded UI
 Detailed roadmap could be found at [YDB Embedded UI repo](https://github.com/ydb-platform/ydb-embedded-ui/blob/main/ROADMAP.md).
 
-## Command Line Utility
-1. 🚧 ㉓ Use a **single `ydb yql`** instead of `ydb table query` or `ydb scripting`
+## Command Line Interface
+1. ✅ ㉔ Use a **single `ydb sql` command** instead of `ydb yql`, `ydb table query` or `ydb scripting` to execute any query
 1. ✅ ㉓ Interactive CLI
 
 ## Tests and Benchmarks

ydb/apps/etcd_proxy/proxy.cpp

@@ -11,6 +11,8 @@
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/discovery/discovery.h>
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/table/table.h>
 #include <ydb/apps/etcd_proxy/service/etcd_base_init.h>
+#include <ydb/apps/etcd_proxy/service/etcd_gate.h>
+#include <ydb/apps/etcd_proxy/service/etcd_lease.h>
 #include <ydb/apps/etcd_proxy/service/etcd_watch.h>
 #include <ydb/apps/etcd_proxy/service/etcd_grpc.h>
 #include <ydb/core/grpc_services/base/base.h>
@@ -61,25 +63,6 @@ int TProxy::Discovery() {
 }
 
 int TProxy::StartServer() {
-    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 {
-            std::cout << "Unexpected result of get last revision." << std::endl;
-            return 1;
-        }
-        if (auto result = res.GetResultSetParser(1); result.TryNextRow()) {
-            Stuff->Lease.store(NYdb::TValueParser(result.GetValue(0)).GetInt64());
-        } else {
-            std::cout << "Unexpected result of get last lease." << std::endl;
-            return 1;
-        }
-        std::cout << "The last revision is " << Stuff->Revision.load() << ", the last lease is " << Stuff->Lease.load() << '.' << std::endl;
-    } else {
-        std::cout << res.GetIssues().ToString() << std::endl;
-        return 1;
-    }
-
     NYdbGrpc::TServerOptions opts;
     opts.SetPort(ListeningPort);
 
@@ -94,11 +77,13 @@ int TProxy::StartServer() {
     }
 
     const auto watchtower = ActorSystem->Register(NEtcd::BuildWatchtower(Counters, Stuff));
+    const auto holderhouse = ActorSystem->Register(NEtcd::BuildHolderHouse(Counters, Stuff));
+    ActorSystem->Register(NEtcd::BuildMainGate(Counters, Stuff));
 
     GRpcServer = std::make_unique<NYdbGrpc::TGRpcServer>(opts, Counters);
-    GRpcServer->AddService(new NEtcd::TEtcdKVService(ActorSystem.get(), Counters, watchtower, Stuff));
+    GRpcServer->AddService(new NEtcd::TEtcdKVService(ActorSystem.get(), Counters, {}, Stuff));
     GRpcServer->AddService(new NEtcd::TEtcdWatchService(ActorSystem.get(), Counters, watchtower, Stuff));
-    GRpcServer->AddService(new NEtcd::TEtcdLeaseService(ActorSystem.get(), Counters, watchtower, Stuff));
+    GRpcServer->AddService(new NEtcd::TEtcdLeaseService(ActorSystem.get(), Counters, holderhouse, Stuff));
     GRpcServer->Start();
     std::cout << "Etcd service over " << Database << " on " << Endpoint << " was started." << std::endl;
     return 0;
@@ -135,13 +120,18 @@ int TProxy::Run() {
 }
 
 int TProxy::InitDatabase() {
-    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 {
+    if (const auto res = Stuff->Client->ExecuteQuery(NEtcd::GetCreateTablesSQL(Stuff->TablePrefix), NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync(); !res.IsSuccess()) {
         std::cout << res.GetIssues().ToString() << std::endl;
         return 1;
     }
+    if (ImportPrefix_.empty()) {
+        if (const auto res = Stuff->Client->ExecuteQuery(NEtcd::GetInitializeTablesSQL(Stuff->TablePrefix), NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync(); !res.IsSuccess()) {
+            std::cout << res.GetIssues().ToString() << std::endl;
+            return 1;
+        }
+    }
+    std::cout << "Database " << Database << " on " << Endpoint << " was initialized." << std::endl;
+    return 0;
 }
 
 int TProxy::ImportDatabase() {
@@ -216,13 +206,28 @@ int TProxy::ImportDatabase() {
     const auto driver = NYdb::TDriver(config);
     auto client = NYdb::NTable::TTableClient(driver);
 
+    if (const auto res = Stuff->Client->ExecuteQuery(Stuff->TablePrefix + "ALTER TABLE `current` DROP INDEX `lease`;", NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync(); !res.IsSuccess()) {
+        std::cout << res.GetIssues().ToString() << std::endl;
+        return 1;
+    }
+
     if (const auto res = client.BulkUpsert(Database + Folder + "/current", std::move(value)).ExtractValueSync(); !res.IsSuccess()) {
         std::cout << res.GetIssues().ToString() << std::endl;
         return 1;
     }
 
+    if (const auto res = Stuff->Client->ExecuteQuery(Stuff->TablePrefix + "ALTER TABLE `current` ADD INDEX `lease` GLOBAL ON (`lease`);", NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync(); !res.IsSuccess()) {
+        std::cout << res.GetIssues().ToString() << std::endl;
+        return 1;
+    }
+
     const auto& param = NYdb::TParamsBuilder().AddParam("$Prefix").String(ImportPrefix_).Build().Build();
-    if (const auto res = Stuff->Client->ExecuteQuery("insert into `history` select * from `current` where startswith(`key`,$Prefix);", NYdb::NQuery::TTxControl::NoTx(), param).ExtractValueSync(); !res.IsSuccess()) {
+    if (const auto res = Stuff->Client->ExecuteQuery(Stuff->TablePrefix + R"(
+insert into `history` select * from `current` where startswith(`key`,$Prefix);
+insert into `revision` (`stub`,`revision`,`timestamp`) values (true,0L,CurrentUtcTimestamp());
+insert into `revision` select false as `stub`, nvl(max(`modified`), 0L) as `revision`, CurrentUtcTimestamp(max(`modified`)) as `timestamp` from `history`;
+insert into `commited` select `revision`, `timestamp` from `revision` where not `stub`;
+    )", NYdb::NQuery::TTxControl::NoTx(), param).ExtractValueSync(); !res.IsSuccess()) {
         std::cout << res.GetIssues().ToString() << std::endl;
         return 1;
     }

ydb/apps/etcd_proxy/readme.txt

@@ -1,10 +1,7 @@
 Etcd over YDB. Zero version.
 
 Main restriction: Can work only in single instance mode.
-
-To remove the restriction two tasks must be done:
-- First, get and update a current revision number on YDB side trought a separate request before any logical query.
-- Second, use CDC for implement the watches instead of self-notifications.
+To remove the restriction we have to use CDC for implement the watches instead of self-notifications.
 
 And other todo's:
 - Add merics.

ydb/apps/etcd_proxy/service/create.sql

@@ -1,3 +1,19 @@
+CREATE TABLE revision
+(
+    `stub` Bool NOT NULL,
+    `revision` Int64 NOT NULL,
+    `timestamp` Timestamp NOT NULL,
+    PRIMARY KEY (`stub`)
+) WITH (PARTITION_AT_KEYS=(false,true));
+
+CREATE TABLE commited
+(
+    `revision` Int64 NOT NULL,
+    `timestamp` Timestamp NOT NULL,
+    PRIMARY KEY (`revision`)
+)
+WITH (AUTO_PARTITIONING_BY_LOAD = ENABLED, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = 23, AUTO_PARTITIONING_PARTITION_SIZE_MB = 11);
+
 CREATE TABLE current
 (
     `key` Bytes NOT NULL,
@@ -8,7 +24,6 @@ CREATE TABLE current
     `lease` Int64 NOT NULL,
     PRIMARY KEY (`key`),
     INDEX `lease` GLOBAL ON (`lease`)
-
 )
 WITH (AUTO_PARTITIONING_BY_LOAD = ENABLED, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = 23, AUTO_PARTITIONING_PARTITION_SIZE_MB = 11);
 
@@ -34,4 +49,4 @@ CREATE TABLE leases
 )
 WITH (AUTO_PARTITIONING_BY_LOAD = ENABLED, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = 23, AUTO_PARTITIONING_PARTITION_SIZE_MB = 11);
 
-ALTER TABLE history ADD CHANGEFEED changes WITH (format="JSON", mode="UPDATES");
+ALTER TABLE `current` ADD CHANGEFEED changes WITH (format="JSON", mode="NEW_AND_OLD_IMAGES");

ydb/apps/etcd_proxy/service/etcd_base_init.cpp

@@ -8,8 +8,8 @@ std::string GetCreateTablesSQL(const std::string& prefix) {
     return prefix + NResource::Find("create.sql"sv);
 }
 
-std::string GetLastRevisionSQL(const std::string& prefix) {
-    return prefix + "select nvl(max(`modified`), 1L) from `history`; select nvl(max(`id`), 1L) from `leases`;";
+std::string GetInitializeTablesSQL(const std::string& prefix) {
+    return prefix + "insert into `revision` (`stub`,`revision`,`timestamp`) values (false,0L,CurrentUtcTimestamp()),(true,0L,CurrentUtcTimestamp());";
 }
 
 }

ydb/apps/etcd_proxy/service/etcd_base_init.h

@@ -6,7 +6,7 @@ namespace NEtcd {
 
 std::string GetCreateTablesSQL(const std::string& prefix);
 
-std::string GetLastRevisionSQL(const std::string& prefix);
+std::string GetInitializeTablesSQL(const std::string& prefix);
 
 }
 

ydb/apps/etcd_proxy/service/etcd_events.h

@@ -38,6 +38,9 @@ enum Ev : ui32 {
     Changes,
     Cancel,
 
+    RequestRevision,
+    ReturnRevision,
+
     End
 };
 
@@ -111,6 +114,25 @@ class TEtcdRequestStreamWrapper
     const TIntrusivePtr<IStreamCtx> Ctx_;
 };
 
+using TKeysSet = std::set<std::pair<std::string, std::string>>;
+
+struct TEvRequestRevision : public NActors::TEventLocal<TEvRequestRevision, Ev::RequestRevision> {
+    explicit TEvRequestRevision(TKeysSet&& keysSet = {})
+        : KeysSet(std::move(keysSet))
+    {}
+
+    TKeysSet KeysSet;
+};
+
+using TGuard = std::shared_ptr<void>;
+
+struct TEvReturnRevision : public NActors::TEventLocal<TEvReturnRevision, Ev::ReturnRevision> {
+    explicit TEvReturnRevision(const i64 revision, const TGuard& guard = {}) : Revision(revision), Guard(std::move(guard)) {}
+
+    const i64 Revision;
+    const TGuard Guard;
+};
+
 using TEvWatchRequest = TEtcdRequestStreamWrapper<Ev::Watch, etcdserverpb::WatchRequest, etcdserverpb::WatchResponse>;
 using TEvLeaseKeepAliveRequest = TEtcdRequestStreamWrapper<Ev::LeaseKeepAlive, etcdserverpb::LeaseKeepAliveRequest, etcdserverpb::LeaseKeepAliveResponse>;