KqpSinkTx.OlapInvalidateOnError has been fixed (#18950)

dorooleg · dorooleg · commit 30e4c7d8d624 · 2025-06-03T16:19:18.000Z
diff --git a/ydb/core/kqp/ut/service/kqp_qs_queries_ut.cpp b/ydb/core/kqp/ut/service/kqp_qs_queries_ut.cpp
@@ -4825,12 +4825,7 @@ Y_UNIT_TEST_SUITE(KqpQueryService) {
                 result = session.ExecuteQuery(fmt::format(R"(
                     INSERT INTO `/Root/DataShard` (Col1, Col2) VALUES ({}u, 0);
                 )", index), NYdb::NQuery::TTxControl::Tx(tx->GetId()).CommitTx()).ExtractValueSync();
-                if (GetIsOlap()) {
-                    // https://github.com/ydb-platform/ydb/issues/14383
-                    UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::BAD_REQUEST, result.GetIssues().ToString());
-                } else {
-                    UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());
-                }
+                UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());
             }
         }
     };
diff --git a/ydb/core/tx/columnshard/columnshard_private_events.h b/ydb/core/tx/columnshard/columnshard_private_events.h
@@ -225,7 +225,8 @@ struct TEvPrivate {
     public:
         enum EErrorClass {
             Internal,
-            Request
+            Request,
+            ConstraintViolation
         };
 
     private:
@@ -241,6 +242,8 @@ struct TEvPrivate {
                     return NKikimrDataEvents::TEvWriteResult::STATUS_INTERNAL_ERROR;
                 case EErrorClass::Request:
                     return NKikimrDataEvents::TEvWriteResult::STATUS_BAD_REQUEST;
+                case EErrorClass::ConstraintViolation:
+                    return NKikimrDataEvents::TEvWriteResult::STATUS_CONSTRAINT_VIOLATION;
             }
         }
 
diff --git a/ydb/core/tx/columnshard/operations/batch_builder/merger.cpp b/ydb/core/tx/columnshard/operations/batch_builder/merger.cpp
@@ -4,18 +4,18 @@
 
 namespace NKikimr::NOlap {
 
-NKikimr::TConclusionStatus IMerger::Finish() {
+NKikimr::NOlap::IMerger::TYdbConclusionStatus IMerger::Finish() {
     while (!IncomingFinished) {
         auto result = OnIncomingOnly(IncomingPosition);
         if (result.IsFail()) {
             return result;
         }
         IncomingFinished = !IncomingPosition.NextPosition(1);
     }
-    return TConclusionStatus::Success();
+    return TYdbConclusionStatus::Success();
 }
 
-NKikimr::TConclusionStatus IMerger::AddExistsDataOrdered(const std::shared_ptr<arrow::Table>& data) {
+NKikimr::NOlap::IMerger::TYdbConclusionStatus IMerger::AddExistsDataOrdered(const std::shared_ptr<arrow::Table>& data) {
     AFL_VERIFY(data);
     NArrow::NMerger::TRWSortableBatchPosition existsPosition(data, 0, Schema->GetPKColumnNames(),
         Schema->GetIndexInfo().GetColumnSTLNames(false), false);
@@ -40,10 +40,10 @@ NKikimr::TConclusionStatus IMerger::AddExistsDataOrdered(const std::shared_ptr<a
         }
     }
     AFL_VERIFY(exsistFinished);
-    return TConclusionStatus::Success();
+    return TYdbConclusionStatus::Success();
 }
 
-NKikimr::TConclusionStatus TUpdateMerger::OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& incoming) {
+NKikimr::NOlap::IMerger::TYdbConclusionStatus TUpdateMerger::OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& incoming) {
     auto rGuard = Builder.StartRecord();
     AFL_VERIFY(Schema->GetIndexInfo().GetColumnIds(false).size() == exists.GetData().GetColumns().size())
         ("index", Schema->GetIndexInfo().GetColumnIds(false).size())("exists", exists.GetData().GetColumns().size());
@@ -57,7 +57,7 @@ NKikimr::TConclusionStatus TUpdateMerger::OnEqualKeys(const NArrow::NMerger::TSo
                 rGuard.Add(*exists.GetData().GetPositionAddress(columnIdx).GetArray(), idxChunk);
             }
         }
-    return TConclusionStatus::Success();
+    return TYdbConclusionStatus::Success();
 }
 
 TUpdateMerger::TUpdateMerger(const NArrow::TContainerWithIndexes<arrow::RecordBatch>& incoming,
diff --git a/ydb/core/tx/columnshard/operations/batch_builder/merger.h b/ydb/core/tx/columnshard/operations/batch_builder/merger.h
@@ -4,15 +4,18 @@
 #include <ydb/core/formats/arrow/reader/result_builder.h>
 #include <ydb/core/tx/columnshard/engines/scheme/versions/abstract_scheme.h>
 #include <ydb/library/conclusion/status.h>
+#include <ydb/public/api/protos/ydb_status_codes.pb.h>
 
 namespace NKikimr::NOlap {
 
 class IMerger {
+public:
+    using TYdbConclusionStatus = TConclusionSpecialStatus<Ydb::StatusIds::StatusCode, Ydb::StatusIds::SUCCESS, Ydb::StatusIds::BAD_REQUEST>;
 private:
     NArrow::NMerger::TRWSortableBatchPosition IncomingPosition;
 
-    virtual TConclusionStatus OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& incoming) = 0;
-    virtual TConclusionStatus OnIncomingOnly(const NArrow::NMerger::TSortableBatchPosition& incoming) = 0;
+    virtual TYdbConclusionStatus OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& incoming) = 0;
+    virtual TYdbConclusionStatus OnIncomingOnly(const NArrow::NMerger::TSortableBatchPosition& incoming) = 0;
 protected:
     std::shared_ptr<ISnapshotSchema> Schema;
     NArrow::TContainerWithIndexes<arrow::RecordBatch> IncomingData;
@@ -29,19 +32,19 @@ class IMerger {
 
     virtual NArrow::TContainerWithIndexes<arrow::RecordBatch> BuildResultBatch() = 0;
 
-    TConclusionStatus Finish();
+    TYdbConclusionStatus Finish();
 
-    TConclusionStatus AddExistsDataOrdered(const std::shared_ptr<arrow::Table>& data);
+    TYdbConclusionStatus AddExistsDataOrdered(const std::shared_ptr<arrow::Table>& data);
 };
 
 class TInsertMerger: public IMerger {
 private:
     using TBase = IMerger;
-    virtual TConclusionStatus OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& /*incoming*/) override {
-        return TConclusionStatus::Fail("Conflict with existing key. " + exists.GetSorting()->DebugJson(exists.GetPosition()).GetStringRobust());
+    virtual TYdbConclusionStatus OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& /*incoming*/) override {
+        return TYdbConclusionStatus::Fail(Ydb::StatusIds::PRECONDITION_FAILED, "Conflict with existing key. " + exists.GetSorting()->DebugJson(exists.GetPosition()).GetStringRobust());
     }
-    virtual TConclusionStatus OnIncomingOnly(const NArrow::NMerger::TSortableBatchPosition& /*incoming*/) override {
-        return TConclusionStatus::Success();
+    virtual TYdbConclusionStatus OnIncomingOnly(const NArrow::NMerger::TSortableBatchPosition& /*incoming*/) override {
+        return TYdbConclusionStatus::Success();
     }
 public:
     using TBase::TBase;
@@ -54,13 +57,13 @@ class TReplaceMerger: public IMerger {
 private:
     using TBase = IMerger;
     NArrow::TColumnFilter Filter = NArrow::TColumnFilter::BuildDenyFilter();
-    virtual TConclusionStatus OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& /*exists*/, const NArrow::NMerger::TSortableBatchPosition& /*incoming*/) override {
+    virtual TYdbConclusionStatus OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& /*exists*/, const NArrow::NMerger::TSortableBatchPosition& /*incoming*/) override {
         Filter.Add(true);
-        return TConclusionStatus::Success();
+        return TYdbConclusionStatus::Success();
     }
-    virtual TConclusionStatus OnIncomingOnly(const NArrow::NMerger::TSortableBatchPosition& /*incoming*/) override {
+    virtual TYdbConclusionStatus OnIncomingOnly(const NArrow::NMerger::TSortableBatchPosition& /*incoming*/) override {
         Filter.Add(false);
-        return TConclusionStatus::Success();
+        return TYdbConclusionStatus::Success();
     }
 public:
     using TBase::TBase;
@@ -80,13 +83,13 @@ class TUpdateMerger: public IMerger {
     std::vector<std::shared_ptr<arrow::BooleanArray>> HasIncomingDataFlags;
     const std::optional<NArrow::NMerger::TSortableBatchPosition> DefaultExists;
     const TString InsertDenyReason;
-    virtual TConclusionStatus OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& incoming) override;
-    virtual TConclusionStatus OnIncomingOnly(const NArrow::NMerger::TSortableBatchPosition& incoming) override {
+    virtual TYdbConclusionStatus OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& incoming) override;
+    virtual TYdbConclusionStatus OnIncomingOnly(const NArrow::NMerger::TSortableBatchPosition& incoming) override {
         if (!!InsertDenyReason) {
-            return TConclusionStatus::Fail("insertion is impossible: " + InsertDenyReason);
+            return TYdbConclusionStatus::Fail("insertion is impossible: " + InsertDenyReason);
         }
         if (!DefaultExists) {
-            return TConclusionStatus::Success();
+            return TYdbConclusionStatus::Success();
         } else {
             return OnEqualKeys(*DefaultExists, incoming);
         }
diff --git a/ydb/core/tx/columnshard/operations/batch_builder/restore.cpp b/ydb/core/tx/columnshard/operations/batch_builder/restore.cpp
@@ -25,9 +25,10 @@ TConclusionStatus TModificationRestoreTask::DoOnDataChunk(const std::shared_ptr<
     if (result.IsFail()) {
         AFL_WARN(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "merge_data_problems")("write_id", WriteData.GetWriteMeta().GetWriteId())(
             "tablet_id", GetTabletId())("message", result.GetErrorMessage());
-        SendErrorMessage(result.GetErrorMessage(), NColumnShard::TEvPrivate::TEvWriteBlobsResult::EErrorClass::Request);
+        SendErrorMessage(result.GetErrorMessage(), result.GetStatus() == Ydb::StatusIds::PRECONDITION_FAILED ? NColumnShard::TEvPrivate::TEvWriteBlobsResult::EErrorClass::ConstraintViolation :  NColumnShard::TEvPrivate::TEvWriteBlobsResult::EErrorClass::Request);
+        return TConclusionStatus::Fail(result.GetErrorMessage());
     }
-    return result;
+    return TConclusionStatus::Success();
 }
 
 void TModificationRestoreTask::DoOnError(const TString& errorMessage) {
@@ -37,12 +38,10 @@ void TModificationRestoreTask::DoOnError(const TString& errorMessage) {
 }
 
 NKikimr::TConclusionStatus TModificationRestoreTask::DoOnFinished() {
-    {
-        auto result = Merger->Finish();
-        if (result.IsFail()) {
-            OnError("cannot finish merger: " + result.GetErrorMessage());
-            return result;
-        }
+    auto result = Merger->Finish();
+    if (result.IsFail()) {
+        OnError("cannot finish merger: " + result.GetErrorMessage());
+        return TConclusionStatus::Fail(result.GetErrorMessage());
     }
 
     auto batchResult = Merger->BuildResultBatch();

Original file line number	Diff line number	Diff line change
`@@ -4825,12 +4825,7 @@ Y_UNIT_TEST_SUITE(KqpQueryService) {`
`4825`	`4825`	`result = session.ExecuteQuery(fmt::format(R"(`
`4826`	`4826`	INSERT INTO `/Root/DataShard` (Col1, Col2) VALUES ({}u, 0);
`4827`	`4827`	`)", index), NYdb::NQuery::TTxControl::Tx(tx->GetId()).CommitTx()).ExtractValueSync();`
`4828`		`- if (GetIsOlap()) {`
`4829`		`- // https://github.com/ydb-platform/ydb/issues/14383`
`4830`		`- UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::BAD_REQUEST, result.GetIssues().ToString());`
`4831`		`- } else {`
`4832`		`- UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());`
`4833`		`- }`
	`4828`	`+ UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());`
`4834`	`4829`	`}`
`4835`	`4830`	`}`
`4836`	`4831`	`};`
Original file line number	Diff line number	Diff line change
`@@ -4,18 +4,18 @@`
`4`	`4`
`5`	`5`	`namespace NKikimr::NOlap {`
`6`	`6`
`7`		`-NKikimr::TConclusionStatus IMerger::Finish() {`
	`7`	`+NKikimr::NOlap::IMerger::TYdbConclusionStatus IMerger::Finish() {`
`8`	`8`	`while (!IncomingFinished) {`
`9`	`9`	`auto result = OnIncomingOnly(IncomingPosition);`
`10`	`10`	`if (result.IsFail()) {`
`11`	`11`	`return result;`
`12`	`12`	`}`
`13`	`13`	`IncomingFinished = !IncomingPosition.NextPosition(1);`
`14`	`14`	`}`
`15`		`- return TConclusionStatus::Success();`
	`15`	`+ return TYdbConclusionStatus::Success();`
`16`	`16`	`}`
`17`	`17`
`18`		`-NKikimr::TConclusionStatus IMerger::AddExistsDataOrdered(const std::shared_ptr<arrow::Table>& data) {`
	`18`	`+NKikimr::NOlap::IMerger::TYdbConclusionStatus IMerger::AddExistsDataOrdered(const std::shared_ptr<arrow::Table>& data) {`
`19`	`19`	`AFL_VERIFY(data);`
`20`	`20`	`NArrow::NMerger::TRWSortableBatchPosition existsPosition(data, 0, Schema->GetPKColumnNames(),`
`21`	`21`	`Schema->GetIndexInfo().GetColumnSTLNames(false), false);`
`@@ -40,10 +40,10 @@ NKikimr::TConclusionStatus IMerger::AddExistsDataOrdered(const std::shared_ptr<a`
`40`	`40`	`}`
`41`	`41`	`}`
`42`	`42`	`AFL_VERIFY(exsistFinished);`
`43`		`- return TConclusionStatus::Success();`
	`43`	`+ return TYdbConclusionStatus::Success();`
`44`	`44`	`}`
`45`	`45`
`46`		`-NKikimr::TConclusionStatus TUpdateMerger::OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& incoming) {`
	`46`	`+NKikimr::NOlap::IMerger::TYdbConclusionStatus TUpdateMerger::OnEqualKeys(const NArrow::NMerger::TSortableBatchPosition& exists, const NArrow::NMerger::TSortableBatchPosition& incoming) {`
`47`	`47`	`auto rGuard = Builder.StartRecord();`
`48`	`48`	`AFL_VERIFY(Schema->GetIndexInfo().GetColumnIds(false).size() == exists.GetData().GetColumns().size())`
`49`	`49`	`("index", Schema->GetIndexInfo().GetColumnIds(false).size())("exists", exists.GetData().GetColumns().size());`
`@@ -57,7 +57,7 @@ NKikimr::TConclusionStatus TUpdateMerger::OnEqualKeys(const NArrow::NMerger::TSo`
`57`	`57`	`rGuard.Add(*exists.GetData().GetPositionAddress(columnIdx).GetArray(), idxChunk);`
`58`	`58`	`}`
`59`	`59`	`}`
`60`		`- return TConclusionStatus::Success();`
	`60`	`+ return TYdbConclusionStatus::Success();`
`61`	`61`	`}`
`62`	`62`
`63`	`63`	`TUpdateMerger::TUpdateMerger(const NArrow::TContainerWithIndexes<arrow::RecordBatch>& incoming,`