CTAS mkdir (#23807)

nikvas0 · web-flow · commit 5dc55194389f · 2025-09-16T17:24:07.000+03:00
diff --git a/ydb/core/kqp/executer_actor/kqp_executer.h b/ydb/core/kqp/executer_actor/kqp_executer.h
@@ -160,7 +160,7 @@ IActor* CreateKqpSchemeExecuter(
     TKqpPhyTxHolder::TConstPtr phyTx, NKikimrKqp::EQueryType queryType, const TActorId& target,
     const TMaybe<TString>& requestType, const TString& database,
     TIntrusiveConstPtr<NACLib::TUserToken> userToken, const TString& clientAddress,
-    bool temporary, TString SessionId, TIntrusivePtr<TUserRequestContext> ctx,
+    bool temporary, bool isCreateTableAs, TString SessionId, TIntrusivePtr<TUserRequestContext> ctx,
     const TActorId& kqpTempTablesAgentActor = TActorId());
 
 std::unique_ptr<TEvKqpExecuter::TEvTxResponse> ExecuteLiteral(
diff --git a/ydb/core/kqp/executer_actor/kqp_scheme_executer.cpp b/ydb/core/kqp/executer_actor/kqp_scheme_executer.cpp
@@ -49,6 +49,7 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
             EvResult = EventSpaceBegin(TEvents::ES_PRIVATE),
             EvMakeTempDirResult,
             EvMakeSessionDirResult,
+            EvMakeCTASDirResult,
         };
 
         struct TEvResult : public TEventLocal<TEvResult, EEv::EvResult> {
@@ -62,6 +63,10 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
         struct TEvMakeSessionDirResult : public TEventLocal<TEvMakeSessionDirResult, EEv::EvMakeSessionDirResult> {
             IKqpGateway::TGenericResult Result;
         };
+
+        struct TEvMakeCTASDirResult : public TEventLocal<TEvMakeCTASDirResult, EEv::EvMakeCTASDirResult> {
+            IKqpGateway::TGenericResult Result;
+        };
     };
 public:
     static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
@@ -71,7 +76,7 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
     TKqpSchemeExecuter(
         TKqpPhyTxHolder::TConstPtr phyTx, NKikimrKqp::EQueryType queryType, const TActorId& target, const TMaybe<TString>& requestType,
         const TString& database, TIntrusiveConstPtr<NACLib::TUserToken> userToken, const TString& clientAddress,
-        bool temporary, TString sessionId, TIntrusivePtr<TUserRequestContext> ctx,
+        bool temporary, bool isCreateTableAs, TString sessionId, TIntrusivePtr<TUserRequestContext> ctx,
         const TActorId& kqpTempTablesAgentActor)
         : PhyTx(phyTx)
         , QueryType(queryType)
@@ -80,6 +85,7 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
         , UserToken(userToken)
         , ClientAddress(clientAddress)
         , Temporary(temporary)
+        , IsCreateTableAs(isCreateTableAs)
         , SessionId(sessionId)
         , RequestContext(std::move(ctx))
         , RequestType(requestType)
@@ -175,6 +181,128 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
         });
     }
 
+    void FindWorkingDirForCTAS() {
+        const auto& schemeOp = PhyTx->GetSchemeOperation();
+
+        AFL_ENSURE(schemeOp.GetOperationCase() == NKqpProto::TKqpSchemeOperation::kAlterTable);
+        const auto& alterTableModifyScheme = schemeOp.GetAlterTable();
+        AFL_ENSURE(alterTableModifyScheme.GetOperationType() == NKikimrSchemeOp::ESchemeOpMoveTable);
+
+        const auto dirPath = SplitPath(alterTableModifyScheme.GetMoveTable().GetDstPath());
+        AFL_ENSURE(dirPath.size() >= 2);
+
+        auto request = MakeHolder<NSchemeCache::TSchemeCacheNavigate>();
+        TVector<TString> path;
+
+        for (const auto& part : dirPath) {
+            path.emplace_back(part);
+
+            NSchemeCache::TSchemeCacheNavigate::TEntry entry;
+            entry.Path = path;
+            entry.Operation = NSchemeCache::TSchemeCacheNavigate::EOp::OpPath;
+            entry.SyncVersion = true;
+            entry.RedirectRequired = false;
+            request->ResultSet.emplace_back(entry);
+        }
+        request->ResultSet.pop_back();
+
+        auto ev = std::make_unique<TEvTxProxySchemeCache::TEvNavigateKeySet>(request);
+
+        Send(MakeSchemeCacheID(), ev.release());
+        Become(&TKqpSchemeExecuter::ExecuteState);
+    }
+
+    void HandleCTASWorkingDir(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
+        const auto& resultSet = ev->Get()->Request->ResultSet;
+
+        const TVector<TString>* workingDir = nullptr;
+        for (auto it = resultSet.rbegin(); it != resultSet.rend(); ++it) {
+            if (it->Status == NSchemeCache::TSchemeCacheNavigate::EStatus::Ok) {
+                workingDir = &it->Path;
+                break;
+            }
+        }
+
+        const auto& schemeOp = PhyTx->GetSchemeOperation();
+        AFL_ENSURE(schemeOp.GetOperationCase() == NKqpProto::TKqpSchemeOperation::kAlterTable);
+        const auto& alterTableModifyScheme = schemeOp.GetAlterTable();
+        AFL_ENSURE(alterTableModifyScheme.GetOperationType() == NKikimrSchemeOp::ESchemeOpMoveTable);
+        const auto dirPath = SplitPath(alterTableModifyScheme.GetMoveTable().GetDstPath());
+
+        if (!workingDir) {
+            const auto errText = TStringBuilder()
+                << "Cannot resolve working dir."
+                << " path# " << JoinPath(dirPath);
+            LOG_D(errText);
+
+            const auto issue = MakeIssue(NKikimrIssues::TIssuesIds::RESOLVE_LOOKUP_ERROR, errText);
+            return ReplyErrorAndDie(Ydb::StatusIds::BAD_REQUEST, issue);
+        }
+        AFL_ENSURE(!workingDir->empty() && workingDir->size() < dirPath.size());
+        AFL_ENSURE(std::equal(
+            workingDir->begin(),
+            workingDir->end(),
+            dirPath.begin(),
+            dirPath.begin() + workingDir->size()));
+
+        CreateCTASDirectory(
+            TConstArrayRef<TString>(
+                workingDir->begin(),
+                workingDir->end()),
+            TConstArrayRef<TString>(
+                dirPath.begin() + workingDir->size(),
+                dirPath.end()));
+    }
+
+    void CreateCTASDirectory(const TConstArrayRef<TString> workingDir, const TConstArrayRef<TString> dirPath) {
+        AFL_ENSURE(!dirPath.empty());
+        AFL_ENSURE(!workingDir.empty());
+
+        auto ev = MakeHolder<TEvTxUserProxy::TEvProposeTransaction>();
+        auto& record = ev->Record;
+
+        auto actorSystem = TActivationContext::ActorSystem();
+        auto selfId = SelfId();
+
+        record.SetDatabaseName(Database);
+        if (UserToken) {
+            record.SetUserToken(UserToken->GetSerializedToken());
+        }
+        record.SetPeerName(ClientAddress);
+
+        const auto& schemeOp = PhyTx->GetSchemeOperation();
+
+        AFL_ENSURE(schemeOp.GetOperationCase() == NKqpProto::TKqpSchemeOperation::kAlterTable);
+        const auto& alterTableModifyScheme = schemeOp.GetAlterTable();
+        AFL_ENSURE(alterTableModifyScheme.GetOperationType() == NKikimrSchemeOp::ESchemeOpMoveTable);
+
+        if (dirPath.size() == 1) {
+            auto ev = MakeHolder<TEvPrivate::TEvMakeCTASDirResult>();
+            ev->Result.SetSuccess();
+            actorSystem->Send(selfId, ev.Release());
+            Become(&TKqpSchemeExecuter::ExecuteState);
+            return;
+        }
+
+        auto* modifyScheme = record.MutableTransaction()->MutableModifyScheme();
+        modifyScheme->SetWorkingDir(CombinePath(workingDir.begin(), workingDir.end()));
+        modifyScheme->SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpMkDir);
+
+        auto* makeDir = modifyScheme->MutableMkDir();
+        makeDir->SetName(CombinePath(dirPath.begin(), std::prev(dirPath.end()), false));
+
+        auto promise = NewPromise<IKqpGateway::TGenericResult>();
+        IActor* requestHandler = new TSchemeOpRequestHandler(ev.Release(), promise, false);
+        RegisterWithSameMailbox(requestHandler);
+
+        promise.GetFuture().Subscribe([actorSystem, selfId](const TFuture<IKqpGateway::TGenericResult>& future) {
+            auto ev = MakeHolder<TEvPrivate::TEvMakeCTASDirResult>();
+            ev->Result = future.GetValue();
+            actorSystem->Send(selfId, ev.Release());
+        });
+        Become(&TKqpSchemeExecuter::ExecuteState);
+    }
+
     void MakeSchemeOperationRequest() {
         using TRequest = TEvTxUserProxy::TEvProposeTransaction;
 
@@ -604,6 +732,8 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
         const auto& schemeOp = PhyTx->GetSchemeOperation();
         if (schemeOp.GetObjectType()) {
             MakeObjectRequest();
+        } else if (IsCreateTableAs && schemeOp.GetOperationCase() == NKqpProto::TKqpSchemeOperation::kAlterTable) {
+            FindWorkingDirForCTAS();
         } else {
             if (Temporary) {
                 CreateTmpDirectory();
@@ -620,6 +750,7 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
                 hFunc(TEvPrivate::TEvResult, HandleExecute);
                 hFunc(TEvPrivate::TEvMakeTempDirResult, Handle);
                 hFunc(TEvPrivate::TEvMakeSessionDirResult, Handle);
+                hFunc(TEvPrivate::TEvMakeCTASDirResult, Handle);
                 hFunc(TEvKqp::TEvAbortExecution, HandleAbortExecution);
                 hFunc(TEvTxUserProxy::TEvAllocateTxIdResult, Handle);
                 hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
@@ -669,6 +800,15 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
         MakeSchemeOperationRequest();
     }
 
+    void Handle(TEvPrivate::TEvMakeCTASDirResult::TPtr& result) {
+        if (!result->Get()->Result.Success()) {
+            InternalError(TStringBuilder()
+                << "Error creating directory:"
+                << result->Get()->Result.Issues().ToString(true));
+        }
+        MakeSchemeOperationRequest();
+    }
+
     void Handle(TEvTxUserProxy::TEvAllocateTxIdResult::TPtr& ev) {
         const auto* msg = ev->Get();
         TxId = msg->TxId;
@@ -713,6 +853,14 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
 
         NSchemeCache::TSchemeCacheNavigate* resp = ev->Get()->Request.Get();
 
+        if (IsCreateTableAs) {
+            AFL_ENSURE(std::all_of(resp->ResultSet.begin(), resp->ResultSet.end(), [](const auto& entry) {
+                return entry.Operation == NSchemeCache::TSchemeCacheNavigate::OpPath;
+            }));
+            HandleCTASWorkingDir(ev);
+            return;
+        }
+
         if (resp->ErrorCount > 0 || resp->ResultSet.empty()) {
             TStringBuilder builder;
             builder << "Unable to navigate:";
@@ -728,6 +876,8 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
             return ReplyErrorAndDie(Ydb::StatusIds::SCHEME_ERROR, NYql::TIssue(error));
         }
 
+        AFL_ENSURE(resp->ResultSet.size() <= 2);
+
         if (UserToken && !UserToken->GetSerializedToken().empty() && !CheckAlterAccess(*UserToken, resp)) {
             LOG_E("Access check failed");
             return ReplyErrorAndDie(Ydb::StatusIds::UNAUTHORIZED, NYql::TIssue("Unauthorized"));
@@ -941,6 +1091,7 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
     const TString ClientAddress;
     std::unique_ptr<TEvKqpExecuter::TEvTxResponse> ResponseEv;
     bool Temporary;
+    bool IsCreateTableAs;
     TString SessionId;
     ui64 TxId = 0;
     TActorId SchemePipeActorId_;
@@ -955,12 +1106,13 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
 IActor* CreateKqpSchemeExecuter(
     TKqpPhyTxHolder::TConstPtr phyTx, NKikimrKqp::EQueryType queryType, const TActorId& target,
     const TMaybe<TString>& requestType, const TString& database,
-    TIntrusiveConstPtr<NACLib::TUserToken> userToken, const TString& clientAddress, bool temporary, TString sessionId,
-    TIntrusivePtr<TUserRequestContext> ctx, const TActorId& kqpTempTablesAgentActor)
+    TIntrusiveConstPtr<NACLib::TUserToken> userToken, const TString& clientAddress,
+    bool temporary, bool isCreateTableAs,
+    TString sessionId, TIntrusivePtr<TUserRequestContext> ctx, const TActorId& kqpTempTablesAgentActor)
 {
     return new TKqpSchemeExecuter(
         phyTx, queryType, target, requestType, database, userToken, clientAddress,
-        temporary, sessionId, std::move(ctx), kqpTempTablesAgentActor);
+        temporary, isCreateTableAs, sessionId, std::move(ctx), kqpTempTablesAgentActor);
 }
 
 } // namespace NKikimr::NKqp
diff --git a/ydb/core/kqp/gateway/kqp_ic_gateway.cpp b/ydb/core/kqp/gateway/kqp_ic_gateway.cpp
@@ -615,7 +615,7 @@ class TKqpSchemeExecuterRequestHandler: public TActorBootstrapped<TKqpSchemeExec
     void Bootstrap() {
         auto ctx = MakeIntrusive<TUserRequestContext>();
         ctx->DatabaseId = DatabaseId;
-        IActor* actor = CreateKqpSchemeExecuter(PhyTx, QueryType, SelfId(), RequestType, Database, UserToken, ClientAddress, false /* temporary */, TString() /* sessionId */, ctx);
+        IActor* actor = CreateKqpSchemeExecuter(PhyTx, QueryType, SelfId(), RequestType, Database, UserToken, ClientAddress, false /* temporary */, false /* isCreateTableAs */, TString() /* sessionId */, ctx);
         Register(actor);
         Become(&TThis::WaitState);
     }
diff --git a/ydb/core/kqp/session_actor/kqp_session_actor.cpp b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
@@ -1632,7 +1632,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
         const bool temporary = GetTemporaryTableInfo(tx).has_value();
 
         auto executerActor = CreateKqpSchemeExecuter(tx, QueryState->GetType(), SelfId(), requestType, Settings.Database, userToken, clientAddress,
-            temporary, TempTablesState.SessionId, QueryState->UserRequestContext, KqpTempTablesAgentActor);
+            temporary, QueryState->IsCreateTableAs(), TempTablesState.SessionId, QueryState->UserRequestContext, KqpTempTablesAgentActor);
 
         ExecuterId = RegisterWithSameMailbox(executerActor);
 
diff --git a/ydb/core/kqp/ut/query/kqp_query_ut.cpp b/ydb/core/kqp/ut/query/kqp_query_ut.cpp
@@ -3103,6 +3103,39 @@ Y_UNIT_TEST_SUITE(KqpQuery) {
             CompareYson(output, R"([[[1u];[1u]];[[10u];[10u]];[[100u];[100u]]])");
         }
     }
+
+     Y_UNIT_TEST(CreateTableAs_MkDir) {
+        NKikimrConfig::TFeatureFlags featureFlags;
+        featureFlags.SetEnableMoveColumnTable(true);
+        auto settings = TKikimrSettings().SetFeatureFlags(featureFlags).SetWithSampleTables(false);
+        settings.AppConfig.MutableTableServiceConfig()->SetEnableOlapSink(true);
+        settings.AppConfig.MutableTableServiceConfig()->SetEnableCreateTableAs(true);
+        settings.AppConfig.MutableTableServiceConfig()->SetEnablePerStatementQueryExecution(true);
+        TKikimrRunner kikimr(settings);
+
+        const TString query = R"(
+            CREATE TABLE `/Root/test_dir/Destination` (
+                PRIMARY KEY (Col1)
+            )
+            WITH (STORE = ROW)
+            AS SELECT 1u As Col1, 1u As Col2;
+        )";
+
+        auto client = kikimr.GetQueryClient();
+        {
+            auto result = client.ExecuteQuery(query, NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync();
+            UNIT_ASSERT_C(result.GetStatus() == NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+
+        {
+            auto it = client.StreamExecuteQuery(R"(
+                SELECT * FROM `/Root/test_dir/Destination` ORDER BY Col1;
+            )", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(it.GetStatus(), EStatus::SUCCESS, it.GetIssues().ToString());
+            TString output = StreamResultToYson(it);
+            CompareYson(output, R"([[1u;1u]])");
+        }
+    }
 }
 
 } // namespace NKqp

Original file line number	Diff line number	Diff line change
`@@ -615,7 +615,7 @@ class TKqpSchemeExecuterRequestHandler: public TActorBootstrapped<TKqpSchemeExec`
`615`	`615`	`void Bootstrap() {`
`616`	`616`	`auto ctx = MakeIntrusive<TUserRequestContext>();`
`617`	`617`	`ctx->DatabaseId = DatabaseId;`
`618`		`- IActor* actor = CreateKqpSchemeExecuter(PhyTx, QueryType, SelfId(), RequestType, Database, UserToken, ClientAddress, false /* temporary /, TString() / sessionId */, ctx);`
	`618`	`+ IActor* actor = CreateKqpSchemeExecuter(PhyTx, QueryType, SelfId(), RequestType, Database, UserToken, ClientAddress, false /* temporary /, false / isCreateTableAs /, TString() / sessionId */, ctx);`
`619`	`619`	`Register(actor);`
`620`	`620`	`Become(&TThis::WaitState);`
`621`	`621`	`}`