Apply TablePathPrefix to topic. (#4741)

Author: Tony-Romanov

ydb/library/yql/sql/v1/query.cpp

@@ -120,8 +120,8 @@ class TTopicKey: public ITableKeys {
         return Name.GetLiteral() ? &Full : nullptr;
     }
 
-    TNodePtr BuildKeys(TContext&, ITableKeys::EBuildKeysMode) override {
-        auto path = Name.Build(); // ToDo Need some prefix here?
+    TNodePtr BuildKeys(TContext& ctx, ITableKeys::EBuildKeysMode) override {
+        const auto path = ctx.GetPrefixedPath(Service, Cluster, Name);
         if (!path) {
             return nullptr;
         }

ydb/library/yql/sql/v1/sql_ut.cpp

@@ -6515,6 +6515,20 @@ Y_UNIT_TEST_SUITE(TopicsDDL) {
                 ALTER CONSUMER consumer3 SET (read_from = 2);
         )", false);
     }
+
+    Y_UNIT_TEST(TopicWithPrefix) {
+        NYql::TAstParseResult res = SqlToYql(R"(
+            USE plato;
+            PRAGMA TablePathPrefix = '/database/path/to/tables';
+            ALTER TOPIC `my_table/my_feed` ADD CONSUMER `my_consumer`;
+        )");
+        UNIT_ASSERT(res.Root);
+
+        TWordCountHive elementStat = {{TString("/database/path/to/tables/my_table/my_feed"), 0}, {"topic", 0}};
+        VerifyProgram(res, elementStat);
+        UNIT_ASSERT_VALUES_EQUAL(1, elementStat["topic"]);
+        UNIT_ASSERT_VALUES_EQUAL(1, elementStat["/database/path/to/tables/my_table/my_feed"]);
+    }
 }
 
 Y_UNIT_TEST_SUITE(BlockEnginePragma) {