Flush sink on INSERT (#15003)

Author: nikvas0

ydb/core/kqp/common/kqp_tx.cpp

@@ -335,7 +335,7 @@ bool HasOltpTableWriteInTx(const NKqpProto::TKqpPhyQuery& physicalQuery) {
     return false;
 }
 
-bool HasUncommittedChangesRead(THashSet<NKikimr::TTableId>& modifiedTables, const NKqpProto::TKqpPhyQuery& physicalQuery) {
+bool HasUncommittedChangesRead(THashSet<NKikimr::TTableId>& modifiedTables, const NKqpProto::TKqpPhyQuery& physicalQuery, const bool commit) {
     auto getTable = [](const NKqpProto::TKqpPhyTableId& table) {
         return NKikimr::TTableId(table.GetOwnerId(), table.GetTableId());
     };
@@ -402,6 +402,10 @@ bool HasUncommittedChangesRead(THashSet<NKikimr::TTableId>& modifiedTables, cons
                     NKikimrKqp::TKqpTableSinkSettings settings;
                     YQL_ENSURE(sink.GetInternalSink().GetSettings().UnpackTo(&settings), "Failed to unpack settings");
                     modifiedTables.insert(getTable(settings.GetTable()));
+                    if (settings.GetType() == NKikimrKqp::TKqpTableSinkSettings::MODE_INSERT && !commit) {
+                        // INSERT with sink should be executed immediately, because it returns an error in case of duplicate rows.
+                        return true;
+                    }
                 } else {
                     return true;
                 }

ydb/core/kqp/common/kqp_tx.h

@@ -160,7 +160,7 @@ class TShardIdToTableInfo {
 };
 using TShardIdToTableInfoPtr = std::shared_ptr<TShardIdToTableInfo>;
 
-bool HasUncommittedChangesRead(THashSet<NKikimr::TTableId>& modifiedTables, const NKqpProto::TKqpPhyQuery& physicalQuery);
+bool HasUncommittedChangesRead(THashSet<NKikimr::TTableId>& modifiedTables, const NKqpProto::TKqpPhyQuery& physicalQuery, const bool commit);
 
 class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase  {
 public:
@@ -316,8 +316,8 @@ class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase  {
         return true;
     }
 
-    void ApplyPhysicalQuery(const NKqpProto::TKqpPhyQuery& phyQuery) {
-        NeedUncommittedChangesFlush = HasUncommittedChangesRead(ModifiedTablesSinceLastFlush, phyQuery);
+    void ApplyPhysicalQuery(const NKqpProto::TKqpPhyQuery& phyQuery, const bool commit) {
+        NeedUncommittedChangesFlush = HasUncommittedChangesRead(ModifiedTablesSinceLastFlush, phyQuery, commit);
         if (NeedUncommittedChangesFlush) {
             ModifiedTablesSinceLastFlush.clear();   
         }

ydb/core/kqp/common/kqp_tx_manager.cpp

@@ -489,7 +489,8 @@ class TKqpTransactionManager : public IKqpTransactionManager {
 
     void MakeLocksIssue(const TShardInfo& shardInfo) {
         TStringBuilder message;
-        message << "Transaction locks invalidated. Tables: ";
+        message << "Transaction locks invalidated. ";
+        message << (shardInfo.Pathes.size() == 1 ? "Table: " : "Tables: ");
         bool first = true;
         // TODO: add error by pathid
         for (const auto& path : shardInfo.Pathes) {

ydb/core/kqp/runtime/kqp_write_actor.cpp

@@ -2460,7 +2460,10 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::ABORTED,
                 NYql::TIssuesIds::KIKIMR_LOCKS_INVALIDATED,
-                TStringBuilder() << "Transaction locks invalidated. Tables: " << getPathes() << ".",
+                TStringBuilder()
+                    << "Transaction locks invalidated. "
+                    << (TxManager->GetShardTableInfo(ev->Get()->Record.GetOrigin()).Pathes.size() == 1 ? "Table: " : "Tables: ")
+                    << getPathes() << ".",
                 getIssues());
             return;
         }

ydb/core/kqp/session_actor/kqp_session_actor.cpp

@@ -944,7 +944,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
         }
 
         QueryState->TxCtx->SetTempTables(QueryState->TempTablesState);
-        QueryState->TxCtx->ApplyPhysicalQuery(phyQuery);
+        QueryState->TxCtx->ApplyPhysicalQuery(phyQuery, QueryState->Commit);
         auto [success, issues] = QueryState->TxCtx->ApplyTableOperations(phyQuery.GetTableOps(), phyQuery.GetTableInfos(),
             EKikimrQueryType::Dml);
         if (!success) {

ydb/core/kqp/ut/tx/kqp_locks_ut.cpp

@@ -241,8 +241,9 @@ Y_UNIT_TEST_SUITE(KqpLocks) {
             }), commitResult.GetIssues().ToString());
     }
 
-    Y_UNIT_TEST(MixedTxFail) {
+    Y_UNIT_TEST_TWIN(MixedTxFail, useSink) {
         NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableTableServiceConfig()->SetEnableOltpSink(useSink);
         appConfig.MutableTableServiceConfig()->SetEnableOlapSink(true);
         appConfig.MutableTableServiceConfig()->SetEnableHtapTx(true);
         auto settings = TKikimrSettings().SetAppConfig(appConfig).SetWithSampleTables(false);

ydb/core/kqp/ut/tx/kqp_sink_tx_ut.cpp

@@ -177,7 +177,6 @@ Y_UNIT_TEST_SUITE(KqpSinkTx) {
 
             auto result = session.ExecuteQuery(Q_(R"(
                 INSERT INTO `/Root/KV` (Key, Value) VALUES (1u, "New");
-                SELECT COUNT(*) FROM `/Root/KV`;
             )"), TTxControl::Tx(tx.GetId())).ExtractValueSync();
             result.GetIssues().PrintTo(Cerr);
             UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());