EvWrite: Immediate & Prepare (#10913)

Author: nikvas0

ydb/core/kqp/common/buffer/buffer.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <ydb/core/scheme/scheme_tabledefs.h>
+#include <ydb/core/kqp/common/kqp_tx_manager.h>
+
+namespace NKikimr {
+namespace NKqp {
+
+struct TKqpBufferWriterSettings {
+    TActorId SessionActorId;
+    IKqpTransactionManagerPtr TxManager;
+};
+
+NActors::IActor* CreateKqpBufferWriterActor(TKqpBufferWriterSettings&& settings);
+
+}
+}

ydb/core/kqp/common/buffer/events.cpp

@@ -0,0 +1,16 @@
+#include "events.h"
+
+namespace NKikimr {
+namespace NKqp {
+
+TEvKqpBuffer::TEvError::TEvError(
+    const TString& message,
+    NYql::NDqProto::StatusIds::StatusCode statusCode,
+    const NYql::TIssues& subIssues)
+    : Message(message)
+    , StatusCode(statusCode)
+    , SubIssues(subIssues) {
+}
+
+}
+}

ydb/core/kqp/common/buffer/events.h

@@ -0,0 +1,47 @@
+#pragma once
+
+#include <ydb/core/kqp/common/simple/kqp_event_ids.h>
+#include <ydb/library/yql/dq/actors/protos/dq_status_codes.pb.h>
+#include <ydb/library/yql/public/issue/yql_issue.h>
+
+
+namespace NKikimr {
+namespace NKqp {
+
+struct TEvKqpBuffer {
+
+struct TEvPrepare : public TEventLocal<TEvPrepare, TKqpBufferWriterEvents::EvPrepare> {
+    TActorId ExecuterActorId;
+};
+
+struct TEvCommit : public TEventLocal<TEvCommit, TKqpBufferWriterEvents::EvCommit> {
+    TActorId ExecuterActorId;
+    ui64 TxId;
+};
+
+struct TEvRollback : public TEventLocal<TEvRollback, TKqpBufferWriterEvents::EvRollback> {
+    TActorId ExecuterActorId;
+};
+
+struct TEvFlush : public TEventLocal<TEvFlush, TKqpBufferWriterEvents::EvFlush> {
+    TActorId ExecuterActorId;
+};
+
+struct TEvResult : public TEventLocal<TEvResult, TKqpBufferWriterEvents::EvResult> {
+};
+
+struct TEvError : public TEventLocal<TEvError, TKqpBufferWriterEvents::EvError> {
+    TString Message;
+    NYql::NDqProto::StatusIds::StatusCode StatusCode;
+    NYql::TIssues SubIssues;
+
+    TEvError(const TString& message, NYql::NDqProto::StatusIds::StatusCode statusCode, const NYql::TIssues& subIssues);
+};
+
+struct TEvTerminate : public TEventLocal<TEvTerminate, TKqpBufferWriterEvents::EvTerminate> {
+};
+
+};
+
+}
+}

ydb/core/kqp/common/buffer/ya.make

@@ -0,0 +1,14 @@
+LIBRARY()
+
+SRCS(
+    events.cpp
+)
+
+PEERDIR(
+    ydb/core/kqp/common/simple
+    ydb/library/yql/public/issue
+)
+
+YQL_LAST_ABI_VERSION()
+
+END()

ydb/core/kqp/common/kqp_tx.h

@@ -2,6 +2,7 @@
 
 #include <ydb/core/base/feature_flags.h>
 #include <ydb/core/kqp/common/kqp_yql.h>
+#include <ydb/core/kqp/common/kqp_tx_manager.h>
 #include <ydb/core/kqp/gateway/kqp_gateway.h>
 #include <ydb/core/kqp/provider/yql_kikimr_provider.h>
 
@@ -121,12 +122,6 @@ struct TDeferredEffects {
     friend class TKqpTransactionContext;
 };
 
-struct TTableInfo {
-    bool IsOlap = false;
-    THashSet<TStringBuf> Pathes;
-};
-
-
 class TShardIdToTableInfo {
 public:
     const TTableInfo& Get(ui64 shardId) const {
@@ -204,6 +199,8 @@ class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase  {
     void Finish() final {
         YQL_ENSURE(DeferredEffects.Empty());
         YQL_ENSURE(!Locks.HasLocks());
+        YQL_ENSURE(!TxManager);
+        YQL_ENSURE(!BufferActorId);
 
         FinishTime = TInstant::Now();
 
@@ -350,6 +347,9 @@ class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase  {
     bool NeedUncommittedChangesFlush = false;
     THashSet<NKikimr::TTableId> ModifiedTablesSinceLastFlush;
 
+    TActorId BufferActorId;
+    IKqpTransactionManagerPtr TxManager = nullptr;
+
     TShardIdToTableInfoPtr ShardIdToTableInfo = std::make_shared<TShardIdToTableInfo>();
 };