Implement safe compaction. (#16744)

Author: Tony-Romanov

Diff for: ydb/apps/etcd_proxy/readme.txt

@@ -12,3 +12,4 @@ And other todo's:
 - Implement retries on "Transaction lock invslideted" error.
 - Implement compaction with control of a requested revision.
 - Implement the watches for ranges. (Now the watches work only with a single key or a prefix.)
+- Add unit tests for watches.

Diff for: ydb/apps/etcd_proxy/service/etcd_impl.cpp

@@ -1067,13 +1067,23 @@ class TCompactRequest
     }
 
     void MakeQueryWithParams(std::ostream& sql, NYdb::TParamsBuilder& params) final {
-        sql << "delete from `content` where `modified` < " << AddParam("Revision", params, KeyRevision) << ';' << std::endl;
+        sql << "$Trash = select c.key as key, c.modified as modified from `content` as c inner join (" << std::endl;
+        sql << "select max_by((`key`, `modified`), `modified`) as pair from `content`" << std::endl;
+        sql << "where `modified` < " << AddParam("Revision", params, KeyRevision) << " and 0L = `version` group by `key`" << std::endl;
+        sql << ") as keys on keys.pair.0 = c.key where c.modified <= keys.pair.1;" << std::endl;
+        sql << "select count(*) from $Trash;" << std::endl;
+        sql << "delete from `content` on select * from $Trash;" << std::endl;
     }
 
-    void ReplyWith(const NYdb::TResultSets&, const TActorContext& ctx) final {
+    void ReplyWith(const NYdb::TResultSets& results, const TActorContext& ctx) final {
+        auto parser = NYdb::TResultSetParser(results.front());
+        const auto erased = parser.TryNextRow() ? NYdb::TValueParser(parser.GetValue(0)).GetUint64() : 0ULL;
+        if (!erased)
+            TryToRollbackRevision();
+
         etcdserverpb::CompactionResponse response;
         response.mutable_header()->set_revision(Revision);
-        Dump(std::cout) << std::endl;
+        Dump(std::cout) << '=' << erased << std::endl;
         return Reply(response, ctx);
     }
 

Diff for: ydb/apps/etcd_proxy/service/ut/etcd_service_ut.cpp

@@ -1020,6 +1020,60 @@ Y_UNIT_TEST_SUITE(Etcd_KV) {
             }
         });
     }
+
+    Y_UNIT_TEST(Compact) {
+        MakeSimpleTest([](const std::unique_ptr<etcdserverpb::KV::Stub> &etcd) {
+            Put("key0", "value0", etcd);
+            Put("key3", "value1", etcd);
+            Put("key2", "value2", etcd);
+            Put("key0", "value3", etcd);
+            Put("key1", "value4", etcd);
+            Delete("key2", etcd);
+            const auto revForCompact = Put("key3", "value5", etcd);
+            Delete("key1", etcd);
+            const auto revForRequest = Put("key3", "value6", etcd);
+            Delete("key0", etcd);
+            Delete("key3", etcd);
+
+            {
+                grpc::ClientContext readRangeCtx;
+                etcdserverpb::RangeRequest rangeRequest;
+                rangeRequest.set_key("key");
+                rangeRequest.set_range_end("kez");
+                rangeRequest.set_keys_only(true);
+                etcdserverpb::RangeResponse rangeResponse;
+                UNIT_ASSERT(etcd->Range(&readRangeCtx, rangeRequest, &rangeResponse).ok());
+                UNIT_ASSERT_VALUES_EQUAL(rangeResponse.count(), 0LL);
+            }
+
+            {
+                grpc::ClientContext compactCtx;
+                etcdserverpb::CompactionRequest compactionRequest;
+                compactionRequest.set_revision(revForCompact);
+                etcdserverpb::CompactionResponse compactionResponse;
+                UNIT_ASSERT(etcd->Compact(&compactCtx, compactionRequest, &compactionResponse).ok());
+            }
+
+            {
+                grpc::ClientContext readRangeCtx;
+                etcdserverpb::RangeRequest rangeRequest;
+                rangeRequest.set_key("key");
+                rangeRequest.set_range_end("kez");
+                rangeRequest.set_revision(revForRequest);
+                rangeRequest.set_sort_target(etcdserverpb::RangeRequest_SortTarget_VALUE);
+                rangeRequest.set_sort_order(etcdserverpb::RangeRequest_SortOrder_ASCEND);
+                etcdserverpb::RangeResponse rangeResponse;
+                UNIT_ASSERT(etcd->Range(&readRangeCtx, rangeRequest, &rangeResponse).ok());
+                UNIT_ASSERT_VALUES_EQUAL(rangeResponse.count(), 2LL);
+                UNIT_ASSERT_VALUES_EQUAL(rangeResponse.kvs().size(), 2U);
+                UNIT_ASSERT_VALUES_EQUAL(rangeResponse.kvs(0).key(), "key0");
+                UNIT_ASSERT_VALUES_EQUAL(rangeResponse.kvs(1).key(), "key3");
+                UNIT_ASSERT_VALUES_EQUAL(rangeResponse.kvs(0).value(), "value3");
+                UNIT_ASSERT_VALUES_EQUAL(rangeResponse.kvs(1).value(), "value6");
+            }
+
+        });
+    }
 } // Y_UNIT_TEST_SUITE(Etcd_KV)
 
 Y_UNIT_TEST_SUITE(Etcd_Lease) {