fix

Author: nikvas0

ydb/core/kqp/common/kqp_tx.cpp

@@ -310,40 +310,33 @@ bool HasOltpTableReadInTx(const NKqpProto::TKqpPhyQuery& physicalQuery) {
     return false;
 }
 
-bool HasOltpTableWriteInStage(const NKqpProto::TKqpPhyStage& stage, const google::protobuf::RepeatedPtrField< ::NKqpProto::TKqpPhyTable>& tables) {
-    for (const auto &tableOp : stage.GetTableOps()) {
-        if (tableOp.GetTypeCase() == NKqpProto::TKqpPhyTableOperation::kUpsertRows
-            || tableOp.GetTypeCase() == NKqpProto::TKqpPhyTableOperation::kDeleteRows) {
-            return true;
-        }
-    }
-
-    for (const auto& sink : stage.GetSinks()) {
-        if (sink.GetTypeCase() == NKqpProto::TKqpSink::kInternalSink && sink.GetInternalSink().GetSettings().Is<NKikimrKqp::TKqpTableSinkSettings>()) {
-            NKikimrKqp::TKqpTableSinkSettings settings;
-            YQL_ENSURE(sink.GetInternalSink().GetSettings().UnpackTo(&settings), "Failed to unpack settings");
-
-            const bool isOltpSink = std::any_of(
-                std::begin(tables),
-                std::end(tables),
-                [&](const NKqpProto::TKqpPhyTable& table) {
-                    return table.GetKind() == NKqpProto::EKqpPhyTableKind::TABLE_KIND_DS
-                        && google::protobuf::util::MessageDifferencer::Equals(table.GetId(), settings.GetTable());
-            });
-
-            if (isOltpSink) {
-                return true;
-            }
-        }
-    }
-    return false;
-}
-
 bool HasOltpTableWriteInTx(const NKqpProto::TKqpPhyQuery& physicalQuery) {
     for (const auto &tx : physicalQuery.GetTransactions()) {
         for (const auto &stage : tx.GetStages()) {
-            if (HasOltpTableWriteInStage(stage, tx.GetTables())) {
-                return true;
+            for (const auto &tableOp : stage.GetTableOps()) {
+                if (tableOp.GetTypeCase() == NKqpProto::TKqpPhyTableOperation::kUpsertRows
+                    || tableOp.GetTypeCase() == NKqpProto::TKqpPhyTableOperation::kDeleteRows) {
+                    return true;
+                }
+            }
+
+            for (const auto& sink : stage.GetSinks()) {
+                if (sink.GetTypeCase() == NKqpProto::TKqpSink::kInternalSink && sink.GetInternalSink().GetSettings().Is<NKikimrKqp::TKqpTableSinkSettings>()) {
+                    NKikimrKqp::TKqpTableSinkSettings settings;
+                    YQL_ENSURE(sink.GetInternalSink().GetSettings().UnpackTo(&settings), "Failed to unpack settings");
+
+                    const bool isOltpSink = std::any_of(
+                        std::begin(tx.GetTables()),
+                        std::end(tx.GetTables()),
+                        [&](const NKqpProto::TKqpPhyTable& table) {
+                            return table.GetKind() == NKqpProto::EKqpPhyTableKind::TABLE_KIND_DS
+                                && google::protobuf::util::MessageDifferencer::Equals(table.GetId(), settings.GetTable());
+                    });
+
+                    if (isOltpSink) {
+                        return true;
+                    }
+                }
             }
         }
     }

ydb/core/kqp/common/kqp_tx.h

@@ -454,14 +454,10 @@ std::pair<bool, std::vector<NYql::TIssue>> MergeLocks(const NKikimrMiniKQL::TTyp
 bool NeedSnapshot(const TKqpTransactionContext& txCtx, const NYql::TKikimrConfiguration& config, bool rollbackTx,
     bool commitTx, const NKqpProto::TKqpPhyQuery& physicalQuery);
 
+bool HasOlapTableReadInTx(const NKqpProto::TKqpPhyQuery& physicalQuery);
 bool HasOlapTableWriteInStage(
     const NKqpProto::TKqpPhyStage& stage,
     const google::protobuf::RepeatedPtrField< ::NKqpProto::TKqpPhyTable>& tables);
-bool HasOltpTableWriteInStage(
-    const NKqpProto::TKqpPhyStage& stage,
-    const google::protobuf::RepeatedPtrField< ::NKqpProto::TKqpPhyTable>& tables);
-
-bool HasOlapTableReadInTx(const NKqpProto::TKqpPhyQuery& physicalQuery);
 bool HasOlapTableWriteInTx(const NKqpProto::TKqpPhyQuery& physicalQuery);
 bool HasOltpTableReadInTx(const NKqpProto::TKqpPhyQuery& physicalQuery);
 bool HasOltpTableWriteInTx(const NKqpProto::TKqpPhyQuery& physicalQuery);