YQ-4844 supported recursive drop of directory with stream queries (#28059)

Author: GrigoriyPA

ydb/core/kqp/ut/common/kqp_ut_common.h

@@ -180,12 +180,15 @@ class TKikimrRunner {
     NYdb::TDriver* GetDriverMut() { return Driver.Get(); }
     const TString& GetEndpoint() const { return Endpoint; }
     const NYdb::TDriver& GetDriver() const { return *Driver; }
-    NYdb::NScheme::TSchemeClient GetSchemeClient() const { return NYdb::NScheme::TSchemeClient(*Driver); }
     Tests::TClient& GetTestClient() const { return *Client; }
     Tests::TServer& GetTestServer() const { return *Server; }
 
     NYdb::TDriverConfig GetDriverConfig() const { return DriverConfig; }
 
+    NYdb::NScheme::TSchemeClient GetSchemeClient(NYdb::TCommonClientSettings settings = NYdb::TCommonClientSettings()) const {
+        return NYdb::NScheme::TSchemeClient(*Driver, settings);
+    }
+
     NYdb::NTable::TTableClient GetTableClient(
         NYdb::NTable::TClientSettings settings = NYdb::NTable::TClientSettings()) const {
         return NYdb::NTable::TTableClient(*Driver, settings.UseQueryCache(false));

ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp

@@ -12809,6 +12809,16 @@ END DO)",
             });
         }
 
+        {
+            auto schemeClient = kikimr->GetSchemeClient(TCommonClientSettings().AuthToken(BUILTIN_ACL_ROOT));
+            const auto result = schemeClient.DescribePath("/Root/MyFolder/MyStreamingQuery").ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToOneLineString());
+            const auto& entry = result.GetEntry();
+            UNIT_ASSERT_VALUES_EQUAL(entry.Name, "MyStreamingQuery");
+            UNIT_ASSERT_VALUES_EQUAL(entry.Owner, BUILTIN_ACL_ROOT);
+            UNIT_ASSERT_VALUES_EQUAL(entry.Type, NYdb::NScheme::ESchemeEntryType::StreamingQuery);
+        }
+
         {
             const auto result = db.ExecuteQuery(R"(
                 CREATE OR REPLACE STREAMING QUERY `MyFolder/MyStreamingQuery` WITH (

ydb/public/api/protos/ydb_scheme.proto

@@ -68,7 +68,7 @@ message Entry {
         RESOURCE_POOL = 21;
         TRANSFER = 23;
         SYS_VIEW = 24;
-        STREAMING_QUERY = 25;
+        STREAMING_QUERY = 26;
     }
 
     // Name of scheme entry (dir2 of /dir1/dir2)

ydb/public/lib/ydb_cli/common/recursive_remove.cpp

@@ -13,6 +13,8 @@ using namespace NScheme;
 using namespace NTable;
 using namespace NTopic;
 
+namespace {
+
 TStatus RemoveDirectory(TSchemeClient& client, const TString& path, const TRemoveDirectorySettings& settings) {
     return RetryFunction([&]() -> TStatus {
         return client.RemoveDirectory(path, settings).ExtractValueSync();
@@ -83,12 +85,18 @@ TStatus RemoveTransfer(NQuery::TQueryClient& client, const TString& path, const
     return DropSchemeObject("TRANSFER", client, path, settings);
 }
 
+TStatus RemoveStreamingQuery(NQuery::TQueryClient& client, const TString& path, const TRemoveDirectorySettings& settings) {
+    return DropSchemeObject("STREAMING QUERY", client, path, settings);
+}
+
 NYdb::NIssue::TIssues MakeIssues(const TString& error) {
     NYdb::NIssue::TIssues issues;
     issues.AddIssue(NYdb::NIssue::TIssue(error));
     return issues;
 }
 
+} // anonymous namespace
+
 bool Prompt(const TString& path, ESchemeEntryType type) {
     Cout << "Remove " << to_lower(ToString(type)) << " '" << path << "' (y/n)? ";
     return AskYesOrNo();
@@ -173,6 +181,8 @@ TStatus Remove(
         return Remove(&RemoveReplication, schemeClient, queryClient, type, path, prompt, settings);
     case ESchemeEntryType::Transfer:
         return Remove(&RemoveTransfer, schemeClient, queryClient, type, path, prompt, settings);
+    case ESchemeEntryType::StreamingQuery:
+        return Remove(&RemoveStreamingQuery, schemeClient, queryClient, type, path, prompt, settings);
 
     default:
         return TStatus(EStatus::UNSUPPORTED, MakeIssues(TStringBuilder()