table path prefix has been fixed

При чтении из external data source c указанием TablePathPrefix:
```
PRAGMA TablePathPrefix("/Root");
SELECT * FROM `s3_tpc`.`*` WITH  ...
```

Эта конструкция разворачивалась в:
```
PRAGMA TablePathPrefix("/Root");
SELECT * FROM `s3_tpc`.`/Root/*` WITH ...
```
В этом ревью это поправлено и теперь она разворачивается в:
```
PRAGMA TablePathPrefix("/Root");
SELECT * FROM `/Root/s3_tpc`.`*` WITH ...
```

3 files changed, 71 insertions, 1 deletions

diff --git a/ydb/core/kqp/ut/federated_query/kqp_federated_query_ut.cpp b/ydb/core/kqp/ut/federated_query/kqp_federated_query_ut.cpp
index 55a02670d22..36fc003d62c 100644
--- a/ydb/core/kqp/ut/federated_query/kqp_federated_query_ut.cpp
+++ b/ydb/core/kqp/ut/federated_query/kqp_federated_query_ut.cpp
@@ -486,6 +486,57 @@ Y_UNIT_TEST_SUITE(KqpFederatedQuery) {
         UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(1).GetUtf8(), "hello world");
         UNIT_ASSERT_VALUES_EQUAL(*resultSet.ColumnParser(2).GetOptionalUtf8(), "two");
     }
+
+    Y_UNIT_TEST(ExecuteScriptWithDataSourceAndTablePathPrefix) {
+        using namespace fmt::literals;
+        const TString externalDataSourceName = "external_data_source";
+        const TString bucket = "test_bucket7";
+
+        CreateBucketWithObject(bucket, "test_object", TEST_CONTENT);
+
+        auto kikimr = DefaultKikimrRunner();
+        kikimr.GetTestServer().GetRuntime()->GetAppData(0).FeatureFlags.SetEnableExternalDataSources(true);
+        auto tc = kikimr.GetTableClient();
+        auto session = tc.CreateSession().GetValueSync().GetSession();
+        const TString query = fmt::format(R"(
+            CREATE EXTERNAL DATA SOURCE `{external_source}` WITH (
+                SOURCE_TYPE="ObjectStorage",
+                LOCATION="{location}",
+                AUTH_METHOD="NONE"
+            );)",
+            "external_source"_a = externalDataSourceName,
+            "location"_a = GetEnv("S3_ENDPOINT") + "/" + bucket + "/"
+            );
+        auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+        UNIT_ASSERT_C(result.GetStatus() == NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
+        auto db = kikimr.GetQueryClient();
+        auto scriptExecutionOperation = db.ExecuteScript(fmt::format(R"(
+            SELECT * FROM `{external_source}`.`*` WITH (
+                format="json_each_row",
+                schema(
+                    key Utf8 NOT NULL,
+                    value Utf8 NOT NULL
+                )
+            )
+        )", "external_source"_a = externalDataSourceName)).ExtractValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(scriptExecutionOperation.Status().GetStatus(), EStatus::SUCCESS, scriptExecutionOperation.Status().GetIssues().ToString());
+        UNIT_ASSERT(scriptExecutionOperation.Metadata().ExecutionId);
+
+        NYdb::NQuery::TScriptExecutionOperation readyOp = WaitScriptExecutionOperation(scriptExecutionOperation.Id(), kikimr.GetDriver());
+        UNIT_ASSERT_EQUAL(readyOp.Metadata().ExecStatus, EExecStatus::Completed);
+        TFetchScriptResultsResult results = db.FetchScriptResults(scriptExecutionOperation.Metadata().ExecutionId).ExtractValueSync();
+        UNIT_ASSERT_C(results.IsSuccess(), results.GetIssues().ToString());
+
+        TResultSetParser resultSet(results.ExtractResultSet());
+        UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnsCount(), 2);
+        UNIT_ASSERT_VALUES_EQUAL(resultSet.RowsCount(), 2);
+        UNIT_ASSERT(resultSet.TryNextRow());
+        UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(0).GetUtf8(), "1");
+        UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(1).GetUtf8(), "trololo");
+        UNIT_ASSERT(resultSet.TryNextRow());
+        UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(0).GetUtf8(), "2");
+        UNIT_ASSERT_VALUES_EQUAL(resultSet.ColumnParser(1).GetUtf8(), "hello world");
+    }
 }
 
 } // namespace NKqp
diff --git a/ydb/library/yql/sql/v1/context.cpp b/ydb/library/yql/sql/v1/context.cpp
index c7f95e4863c..e097ce174bd 100644
--- a/ydb/library/yql/sql/v1/context.cpp
+++ b/ydb/library/yql/sql/v1/context.cpp
@@ -241,6 +241,21 @@ IOutputStream& TContext::MakeIssue(ESeverity severity, TIssueCode code, NYql::TP
     return *IssueMsgHolder;
 }
 
+bool TContext::IsDynamicCluster(const TDeferredAtom& cluster) const {
+    const TString* clusterPtr = cluster.GetLiteral();
+    if (!clusterPtr) {
+        return false;
+    }
+    TString unused;
+    if (ClusterMapping.GetClusterProvider(*clusterPtr, unused)) {
+        return false;
+    }
+    if (Settings.AssumeYdbOnClusterWithSlash && clusterPtr->StartsWith('/')) {
+        return false;
+    }
+    return !Settings.DynamicClusterProvider.Empty();
+}
+
 bool TContext::SetPathPrefix(const TString& value, TMaybe<TString> arg) {
     if (arg.Defined()) {
         if (*arg == YtProviderName
@@ -276,6 +291,9 @@ TNodePtr TContext::GetPrefixedPath(const TString& service, const TDeferredAtom&
 }
 
 TStringBuf TContext::GetPrefixPath(const TString& service, const TDeferredAtom& cluster) const {
+    if (IsDynamicCluster(cluster)) {
+        return {};
+    }
     auto* clusterPrefix = cluster.GetLiteral()
                             ? ClusterPathPrefixes.FindPtr(*cluster.GetLiteral())
                             : nullptr;
diff --git a/ydb/library/yql/sql/v1/context.h b/ydb/library/yql/sql/v1/context.h
index 773f4763fda..033996ed8aa 100644
--- a/ydb/library/yql/sql/v1/context.h
+++ b/ydb/library/yql/sql/v1/context.h
@@ -139,7 +139,7 @@ namespace NSQLTranslationV1 {
                     return TString(NYql::KikimrProviderName);
                 }
                 if (Settings.DynamicClusterProvider) {
-                    normalizedClusterName = cluster;
+                    normalizedClusterName = cluster.StartsWith('/') ? cluster : Settings.PathPrefix + "/" + cluster;
                     return Settings.DynamicClusterProvider;
                 }
                 return Nothing();
@@ -148,6 +148,7 @@ namespace NSQLTranslationV1 {
             return provider;
         }
 
+        bool IsDynamicCluster(const TDeferredAtom& cluster) const;
         bool HasNonYtProvider(const ISource& source) const;
         bool UseUnordered(const ISource& source) const;
         bool UseUnordered(const TTableRef& table) const;