Prepare tests for Sinks (#16393)

Author: nikvas0

ydb/core/kqp/opt/kqp_query_plan.cpp

@@ -839,15 +839,17 @@ class TxPlanSerializer {
             op.Properties["Table"] = tableData.RelativePath ? *tableData.RelativePath : tablePath;
             op.Properties["Path"] = tablePath;
 
-            const auto& tupleType = stage.Ref().GetTypeAnn()->Cast<TTupleExprType>();
-            YQL_ENSURE(tupleType);
-            YQL_ENSURE(tupleType->GetSize() == 1);
-            const auto& listType = tupleType->GetItems()[0]->Cast<TListExprType>();
-            YQL_ENSURE(listType);
-            const auto& structType = listType->GetItemType()->Cast<TStructExprType>();
-            YQL_ENSURE(structType);
-            for (const auto& item : structType->GetItems()) {
-                writeInfo.Columns.push_back(TString(item->GetName()));
+            if (writeInfo.Type != EPlanTableWriteType::MultiErase) {
+                const auto& tupleType = stage.Ref().GetTypeAnn()->Cast<TTupleExprType>();
+                YQL_ENSURE(tupleType);
+                YQL_ENSURE(tupleType->GetSize() == 1);
+                const auto& listType = tupleType->GetItems()[0]->Cast<TListExprType>();
+                YQL_ENSURE(listType);
+                const auto& structType = listType->GetItemType()->Cast<TStructExprType>();
+                YQL_ENSURE(structType);
+                for (const auto& item : structType->GetItems()) {
+                    writeInfo.Columns.push_back(TString(item->GetName()));
+                }
             }
 
             SerializerCtx.Tables[tablePath].Writes.push_back(writeInfo);

ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp

@@ -10056,9 +10056,10 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
         );
     }
 
-    Y_UNIT_TEST(DoubleCreateResourcePoolClassifier) {
+    Y_UNIT_TEST_TWIN(DoubleCreateResourcePoolClassifier, UseSink) {
         NKikimrConfig::TAppConfig config;
         config.MutableFeatureFlags()->SetEnableResourcePools(true);
+        config.MutableTableServiceConfig()->SetEnableOltpSink(UseSink);
 
         TKikimrRunner kikimr(NKqp::TKikimrSettings()
             .SetAppConfig(config)
@@ -10085,7 +10086,11 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
                 );)";
             auto result = session.ExecuteSchemeQuery(query).GetValueSync();
             UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::GENERIC_ERROR);
-            UNIT_ASSERT_STRING_CONTAINS_C(result.GetIssues().ToString(), "Conflict with existing key", result.GetIssues().ToString());
+            if (UseSink) {
+                UNIT_ASSERT_STRING_CONTAINS_C(result.GetIssues().ToString(), "Duplicate keys have been found", result.GetIssues().ToString());
+            } else {
+                UNIT_ASSERT_STRING_CONTAINS_C(result.GetIssues().ToString(), "Conflict with existing key", result.GetIssues().ToString());
+            }
         }
     }
 

ydb/core/tx/datashard/check_write_unit.cpp

@@ -115,7 +115,7 @@ EExecutionStatus TCheckWriteUnit::Execute(TOperation::TPtr op,
         if (!Pipeline.AssignPlanInterval(op)) {
             TString err = TStringBuilder() << "Can't propose tx " << op->GetTxId() << " at blocked shard " << DataShard.TabletID();
 
-            writeOp->SetError(NKikimrDataEvents::TEvWriteResult::STATUS_INTERNAL_ERROR, err);
+            writeOp->SetError(NKikimrDataEvents::TEvWriteResult::STATUS_OVERLOADED, err);
             op->Abort(EExecutionUnitKind::FinishProposeWrite);
 
             LOG_NOTICE_S(ctx, NKikimrServices::TX_DATASHARD, err);

ydb/services/persqueue_v1/ut/topic_service_ut.cpp

@@ -412,7 +412,11 @@ Y_UNIT_TEST_F(MultiplePartitionsAndNoGapsInTheOffsets, TUpdateOffsetsInTransacti
     auto result = tx->Commit().ExtractValueSync();
     Cerr << ">>> CommitTx >>>" << Endl;
     UNIT_ASSERT_EQUAL(result.IsTransportError(), false);
-    UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), NYdb::EStatus::ABORTED);
+    if (server->ServerSettings.AppConfig->GetTableServiceConfig().GetEnableOltpSink()) {
+        UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), NYdb::EStatus::BAD_REQUEST);
+    } else {
+        UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), NYdb::EStatus::ABORTED);
+    }
 }
 
 }

ydb/services/ydb/ydb_table_ut.cpp

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 import logging
 
-from hamcrest import assert_that, raises, equal_to
+from hamcrest import assert_that, raises, equal_to, any_of
 
 from ydb.tests.library.harness.kikimr_runner import KiKiMR
 from ydb.tests.oss.ydb_sdk_import import ydb
@@ -189,9 +189,15 @@ def callee():
                 if first_query_kind in row_adding_operations and second_query_kind == 'insert':
                     assert_that(
                         callee,
-                        raises(
-                            ydb.PreconditionFailed,
-                            "Conflict with existing key."
+                        any_of(
+                            raises(
+                                ydb.PreconditionFailed,
+                                "Conflict with existing key."
+                            ),
+                            raises(
+                                ydb.PreconditionFailed,
+                                "Duplicate keys have been found."
+                            )
                         )
                     )
 

ydb/tests/functional/api/test_insert.py

@@ -651,6 +651,9 @@ def test_anti_dependency_cycles_g2_two_edges(self):
         def callee():
             t1.execute('{} update {} set value = 0 where id = 1;'.format(prefix, table_name))
 
+            # Sinks allow UPDATE statement to be delayed, so we do commit here to execute update.
+            t1.commit()
+
         assert_that(
             callee,
             raises(

ydb/tests/functional/api/test_isolation.py

@@ -4,6 +4,7 @@ FORK_TEST_FILES()
 SIZE(MEDIUM)
 
 ENV(YDB_DRIVER_BINARY="ydb/apps/ydbd/ydbd")
+ENV(YDB_HARD_MEMORY_LIMIT_BYTES="8000000000")
 
 TEST_SRCS(
     test_session_pool.py