Evwrite better errors (#14320)

10 files changed, 117 insertions, 69 deletions

diff --git a/.github/config/muted_ya.txt b/.github/config/muted_ya.txt
index e9649525c9..1feb93362f 100644
--- a/.github/config/muted_ya.txt
+++ b/.github/config/muted_ya.txt
@@ -62,6 +62,7 @@ ydb/core/kqp/ut/service KqpQueryService.ExecuteQueryWithResourcePoolClassifier
 ydb/core/kqp/ut/service [*/*] chunk chunk
 ydb/core/kqp/ut/service [*/*]+chunk+chunk
 ydb/core/kqp/ut/spilling KqpScanSpilling.SelfJoinQueryService
+ydb/core/kqp/ut/tx KqpSinkTx.OlapInvalidateOnError
 ydb/core/kqp/ut/tx KqpSnapshotIsolation.*Olap*
 ydb/core/kqp/ut/tx KqpSnapshotIsolation.*Oltp*
 ydb/core/kqp/ut/yql KqpScripting.StreamExecuteYqlScriptScanOperationTmeoutBruteForce
diff --git a/ydb/core/kqp/runtime/kqp_write_actor.cpp b/ydb/core/kqp/runtime/kqp_write_actor.cpp
index 7188e71957..a6e5fd0ea3 100644
--- a/ydb/core/kqp/runtime/kqp_write_actor.cpp
+++ b/ydb/core/kqp/runtime/kqp_write_actor.cpp
@@ -154,6 +154,7 @@ struct IKqpTableWriterCallbacks {
     virtual void OnMessageAcknowledged(ui64 dataSize) = 0;
 
     virtual void OnError(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::EYqlIssueCode id, const TString& message, const NYql::TIssues& subIssues) = 0;
+    virtual void OnError(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::TIssues&& issues) = 0;
 };
 
 struct TKqpTableWriterStatistics {
@@ -547,8 +548,7 @@ public:
                 NYql::NDqProto::StatusIds::UNSPECIFIED,
                 NYql::TIssuesIds::DEFAULT_ERROR,
                 TStringBuilder() << "Unspecified error for table `"
-                    << TablePath << "`. "
-                    << getIssues().ToOneLineString(),
+                    << TablePath << "`.",
                 getIssues());
             return;
         }
@@ -571,8 +571,7 @@ public:
                 NYql::NDqProto::StatusIds::ABORTED,
                 NYql::TIssuesIds::KIKIMR_OPERATION_ABORTED,
                 TStringBuilder() << "Aborted for table `"
-                    << TablePath << "`. "
-                    << getIssues().ToOneLineString(),
+                    << TablePath << "`.",
                 getIssues());
             return;
         }
@@ -592,8 +591,7 @@ public:
                     NYql::NDqProto::StatusIds::PRECONDITION_FAILED,
                     NYql::TIssuesIds::KIKIMR_TEMPORARILY_UNAVAILABLE,
                     TStringBuilder() << "Wrong shard state for table `"
-                        << TablePath << "`. "
-                        << getIssues().ToOneLineString(),
+                        << TablePath << "`.",
                     getIssues());
             }
             return;
@@ -608,8 +606,7 @@ public:
                 NYql::NDqProto::StatusIds::INTERNAL_ERROR,
                 NYql::TIssuesIds::KIKIMR_INTERNAL_ERROR,
                 TStringBuilder() << "Internal error for table `"
-                    << TablePath << "`. "
-                    << getIssues().ToOneLineString(),
+                    << TablePath << "`.",
                 getIssues());
             return;
         }
@@ -624,8 +621,7 @@ public:
                 NYql::NDqProto::StatusIds::UNAVAILABLE,
                 NYql::TIssuesIds::KIKIMR_DISK_SPACE_EXHAUSTED,
                 TStringBuilder() << "Disk space exhausted for table `"
-                    << TablePath << "`. "
-                    << getIssues().ToOneLineString(),
+                    << TablePath << "`.",
                 getIssues());
             return;
         }
@@ -643,8 +639,7 @@ public:
                     NYql::NDqProto::StatusIds::OVERLOADED,
                     NYql::TIssuesIds::KIKIMR_OVERLOADED,
                     TStringBuilder() << "Tablet " << ev->Get()->Record.GetOrigin() << " is overloaded. Table `"
-                        << TablePath << "`. "
-                        << getIssues().ToOneLineString(),
+                        << TablePath << "`.",
                     getIssues());
             }
             return;
@@ -660,8 +655,7 @@ public:
                 NYql::NDqProto::StatusIds::CANCELLED,
                 NYql::TIssuesIds::KIKIMR_OPERATION_CANCELLED,
                 TStringBuilder() << "Cancelled request to table `"
-                    << TablePath << "`."
-                    << getIssues().ToOneLineString(),
+                    << TablePath << "`.",
                 getIssues());
             return;
         }
@@ -675,9 +669,8 @@ public:
             RuntimeError(
                 NYql::NDqProto::StatusIds::BAD_REQUEST,
                 NYql::TIssuesIds::KIKIMR_BAD_REQUEST,
-                TStringBuilder() << "Bad request. Table `"
-                    << TablePath << "`. "
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Bad request. Table: `"
+                    << TablePath << "`.",
                 getIssues());
             return;
         }
@@ -695,9 +688,8 @@ public:
                 RuntimeError(
                     NYql::NDqProto::StatusIds::SCHEME_ERROR,
                     NYql::TIssuesIds::KIKIMR_SCHEME_MISMATCH,
-                    TStringBuilder() << "Scheme changed. Table `"
-                        << TablePath << "`. "
-                        << getIssues().ToOneLineString(),
+                    TStringBuilder() << "Scheme changed. Table: `"
+                        << TablePath << "`.",
                     getIssues());
             }
             return;
@@ -715,9 +707,22 @@ public:
             RuntimeError(
                 NYql::NDqProto::StatusIds::ABORTED,
                 NYql::TIssuesIds::KIKIMR_LOCKS_INVALIDATED,
-                TStringBuilder() << "Transaction locks invalidated. Table `"
-                    << TablePath << "`. "
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Transaction locks invalidated. Table: `"
+                    << TablePath << "`.",
+                getIssues());
+            return;
+        }
+        case NKikimrDataEvents::TEvWriteResult::STATUS_CONSTRAINT_VIOLATION: {
+            CA_LOG_E("Got CONSTRAINT VIOLATION for table."
+                    << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
+                    << " Sink=" << this->SelfId() << "."
+                    << getIssues().ToOneLineString());
+            TxManager->SetError(ev->Get()->Record.GetOrigin());
+            RuntimeError(
+                NYql::NDqProto::StatusIds::PRECONDITION_FAILED,
+                NYql::TIssuesIds::KIKIMR_CONSTRAINT_VIOLATION,
+                TStringBuilder() << "Constraint violated. Table: `"
+                    << TablePath << "`.",
                 getIssues());
             return;
         }
@@ -1019,16 +1024,18 @@ public:
             TableWriteActorSpan.EndError(message);
         }
 
-        NYql::TIssue issue(message);
-        SetIssueCode(id, issue);
-        for (const auto& i : subIssues) {
-            issue.AddSubIssue(MakeIntrusive<NYql::TIssue>(i));
-        }
+        Callbacks->OnError(statusCode, id, message, subIssues);
+    }
 
-        NYql::TIssues issues;
-        issues.AddIssue(std::move(issue));
+    void RuntimeError(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::TIssues&& issues) {
+        if (TableWriteActorStateSpan) {
+            TableWriteActorStateSpan.EndError(issues.ToOneLineString());
+        }
+        if (TableWriteActorSpan) {
+            TableWriteActorSpan.EndError(issues.ToOneLineString());
+        }
 
-        Callbacks->OnError(statusCode, id, message, issues);
+        Callbacks->OnError(statusCode, std::move(issues));
     }
 
     void PassAway() override {;
@@ -1333,6 +1340,12 @@ private:
         Callbacks->OnAsyncOutputError(OutputIndex, std::move(issues), statusCode);
     }
 
+    void RuntimeError(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::TIssues&& issues) {
+        DirectWriteActorSpan.EndError(issues.ToOneLineString());
+
+        Callbacks->OnAsyncOutputError(OutputIndex, std::move(issues), statusCode);
+    }
+
     void PassAway() override {
         if (WriteTableActor) {
             WriteTableActor->Terminate();
@@ -1370,6 +1383,10 @@ private:
         RuntimeError(statusCode, id, message, subIssues);
     }
 
+    void OnError(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::TIssues&& issues) override {
+        RuntimeError(statusCode, std::move(issues));
+    }
+
     void FillExtraStats(NYql::NDqProto::TDqTaskStats* stats, bool last, const NYql::NDq::TDqMeteringStats*) override {
         if (last && WriteTableActor) {
             WriteTableActor->FillStats(stats);
@@ -2180,10 +2197,22 @@ public:
             }()
             << ", Cookie=" << ev->Cookie);
 
+        auto getPathes = [&]() -> TString {
+            const auto tableInfo = TxManager->GetShardTableInfo(ev->Get()->Record.GetOrigin());
+            TStringBuilder builder;
+            for (const auto& path : tableInfo.Pathes) {
+                if (!builder.empty()) {
+                    builder << ", ";
+                }
+                builder << "`" << path << "`";
+            }
+            return builder;
+        };
+
         // TODO: get rid of copy-paste
         switch (ev->Get()->GetStatus()) {
         case NKikimrDataEvents::TEvWriteResult::STATUS_UNSPECIFIED: {
-            CA_LOG_E("Got UNSPECIFIED for table."
+            CA_LOG_E("Got UNSPECIFIED for tables."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                     << " Sink=" << this->SelfId() << "."
                     << getIssues().ToOneLineString());
@@ -2191,7 +2220,7 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::UNSPECIFIED,
                 NYql::TIssuesIds::DEFAULT_ERROR,
-                TStringBuilder() << "Unspecified error for table. "
+                TStringBuilder() << "Unspecified error for tables `" << getPathes() << "`. "
                     << getIssues().ToOneLineString(),
                 getIssues());
             return;
@@ -2205,7 +2234,7 @@ public:
             return;
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_ABORTED: {
-            CA_LOG_E("Got ABORTED for table."
+            CA_LOG_E("Got ABORTED for tables."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                     << " Sink=" << this->SelfId() << "."
                     << getIssues().ToOneLineString());
@@ -2213,13 +2242,12 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::ABORTED,
                 NYql::TIssuesIds::KIKIMR_OPERATION_ABORTED,
-                TStringBuilder() << "Aborted for table. "
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Aborted for tables " << getPathes() << ". ",
                 getIssues());
             return;
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_WRONG_SHARD_STATE: {
-            CA_LOG_E("Got WRONG SHARD STATE for table."
+            CA_LOG_E("Got WRONG SHARD STATE for tables."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                     << " Sink=" << this->SelfId() << "."
                     << getIssues().ToOneLineString());
@@ -2227,13 +2255,12 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::PRECONDITION_FAILED,
                 NYql::TIssuesIds::KIKIMR_TEMPORARILY_UNAVAILABLE,
-                TStringBuilder() << "Wrong shard state for table. "
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Wrong shard state for tables " << getPathes() << ".",
                 getIssues());
             return;
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_INTERNAL_ERROR: {
-            CA_LOG_E("Got INTERNAL ERROR for table."
+            CA_LOG_E("Got INTERNAL ERROR for tables."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                     << " Sink=" << this->SelfId() << "."
                     << getIssues().ToOneLineString());
@@ -2241,13 +2268,12 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::INTERNAL_ERROR,
                 NYql::TIssuesIds::KIKIMR_INTERNAL_ERROR,
-                TStringBuilder() << "Internal error for table. "
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Internal error for tables " << getPathes() << ".",
                 getIssues());
             return;
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_DISK_SPACE_EXHAUSTED: {
-            CA_LOG_E("Got DISK_SPACE_EXHAUSTED for table."
+            CA_LOG_E("Got DISK_SPACE_EXHAUSTED for tables."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                     << " Sink=" << this->SelfId() << "."
                     << getIssues().ToOneLineString());
@@ -2255,13 +2281,12 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::UNAVAILABLE,
                 NYql::TIssuesIds::KIKIMR_DISK_SPACE_EXHAUSTED,
-                TStringBuilder() << "Disk space exhausted for table. "
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Disk space exhausted for tables " << getPathes() << ".",
                 getIssues());
                 return;
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_OVERLOADED: {
-            CA_LOG_W("Got OVERLOADED for table ."
+            CA_LOG_W("Got OVERLOADED for tables."
                 << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                 << " Sink=" << this->SelfId() << "."
                 << " Ignored this error."
@@ -2270,13 +2295,12 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::OVERLOADED,
                 NYql::TIssuesIds::KIKIMR_OVERLOADED,
-                TStringBuilder() << "Tablet " << ev->Get()->Record.GetOrigin() << " is overloaded."
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Tablet " << ev->Get()->Record.GetOrigin() << "(" << getPathes() << ")" << " is overloaded.",
                 getIssues());
             return;
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_CANCELLED: {
-            CA_LOG_E("Got CANCELLED for table."
+            CA_LOG_E("Got CANCELLED for tables."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                     << " Sink=" << this->SelfId() << "."
                     << getIssues().ToOneLineString());
@@ -2284,13 +2308,12 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::CANCELLED,
                 NYql::TIssuesIds::KIKIMR_OPERATION_CANCELLED,
-                TStringBuilder() << "Cancelled request to table."
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Cancelled request to tables " << getPathes() << ".",
                 getIssues());
             return;
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_BAD_REQUEST: {
-            CA_LOG_E("Got BAD REQUEST for table."
+            CA_LOG_E("Got BAD REQUEST for tables."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                     << " Sink=" << this->SelfId() << "."
                     << getIssues().ToOneLineString());
@@ -2298,8 +2321,7 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::BAD_REQUEST,
                 NYql::TIssuesIds::KIKIMR_BAD_REQUEST,
-                TStringBuilder() << "Bad request. "
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Bad request. Tables: " << getPathes() << ".",
                 getIssues());
             return;
         }
@@ -2312,8 +2334,7 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::SCHEME_ERROR,
                 NYql::TIssuesIds::KIKIMR_SCHEME_MISMATCH,
-                TStringBuilder() << "Scheme changed. "
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Scheme changed. Tables: " << getPathes() << ".",
                 getIssues());
             return;
         }
@@ -2328,8 +2349,20 @@ public:
             ReplyErrorAndDie(
                 NYql::NDqProto::StatusIds::ABORTED,
                 NYql::TIssuesIds::KIKIMR_LOCKS_INVALIDATED,
-                TStringBuilder() << "Transaction locks invalidated."
-                    << getIssues().ToOneLineString(),
+                TStringBuilder() << "Transaction locks invalidated. Tables: " << getPathes() << ".",
+                getIssues());
+            return;
+        }
+        case NKikimrDataEvents::TEvWriteResult::STATUS_CONSTRAINT_VIOLATION: {
+            CA_LOG_E("Got CONSTRAINT VIOLATION for table."
+                    << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
+                    << " Sink=" << this->SelfId() << "."
+                    << getIssues().ToOneLineString());
+            TxManager->SetError(ev->Get()->Record.GetOrigin());
+            ReplyErrorAndDie(
+                NYql::NDqProto::StatusIds::PRECONDITION_FAILED,
+                NYql::TIssuesIds::KIKIMR_CONSTRAINT_VIOLATION,
+                TStringBuilder() << "Constraint violated. Tables: " << getPathes() << ".",
                 getIssues());
             return;
         }
@@ -2493,6 +2526,10 @@ public:
         ReplyErrorAndDie(statusCode, id, message, subIssues);
     }
 
+    void OnError(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::TIssues&& issues) override {
+        ReplyErrorAndDie(statusCode, std::move(issues));
+    }
+
     void ReplyErrorAndDie(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::EYqlIssueCode id, const TString& message, const NYql::TIssues& subIssues = {}) {
         BufferWriteActorState.EndError(message);
         BufferWriteActor.EndError(message);
@@ -2506,6 +2543,17 @@ public:
         NYql::TIssues issues;
         issues.AddIssue(std::move(issue));
 
+        ReplyErrorAndDieImpl(statusCode, std::move(issues));
+    }
+
+    void ReplyErrorAndDie(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::TIssues&& issues) {
+        BufferWriteActorState.EndError(issues.ToOneLineString());
+        BufferWriteActor.EndError(issues.ToOneLineString());
+
+        ReplyErrorAndDieImpl(statusCode, std::move(issues));
+    }
+
+    void ReplyErrorAndDieImpl(NYql::NDqProto::StatusIds::StatusCode statusCode, NYql::TIssues&& issues) {
         CA_LOG_E("statusCode=" << NYql::NDqProto::StatusIds_StatusCode_Name(statusCode) << ". Issue=" << issues.ToString() << ". sessionActorId=" << SessionActorId << ". isRollback=" << (State == EState::ROLLINGBACK));
 
         Y_ABORT_UNLESS(!HasError);
diff --git a/ydb/core/kqp/session_actor/kqp_session_actor.cpp b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
index eea2c802d3..7af8ae5886 100644
--- a/ydb/core/kqp/session_actor/kqp_session_actor.cpp
+++ b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
@@ -1579,7 +1579,7 @@ public:
             switch (status) {
                 case Ydb::StatusIds::ABORTED: {
                     if (QueryState->TxCtx->TxManager && QueryState->TxCtx->TxManager->BrokenLocks()) {
-                        issues.AddIssue(*QueryState->TxCtx->TxManager->GetLockIssue());
+                        YQL_ENSURE(!issues.Empty());
                     } else if (ev->BrokenLockPathId) {
                         YQL_ENSURE(!QueryState->TxCtx->TxManager);
                         issues.AddIssue(GetLocksInvalidatedIssue(*QueryState->TxCtx, *ev->BrokenLockPathId));
diff --git a/ydb/core/kqp/ut/cost/kqp_cost_ut.cpp b/ydb/core/kqp/ut/cost/kqp_cost_ut.cpp
index 7cc8678ccd..994d9fab51 100644
--- a/ydb/core/kqp/ut/cost/kqp_cost_ut.cpp
+++ b/ydb/core/kqp/ut/cost/kqp_cost_ut.cpp
@@ -841,11 +841,7 @@ Y_UNIT_TEST_SUITE(KqpCost) {
             auto txControl = NYdb::NQuery::TTxControl::BeginTx().CommitTx();
 
             auto result = session.ExecuteQuery(query, txControl, GetQuerySettings()).ExtractValueSync();
-            if (isSink) { // TODO: fix status?
-                UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::BAD_REQUEST);
-            } else {
-                UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::PRECONDITION_FAILED);
-            }
+            UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::PRECONDITION_FAILED);
 
             auto stats = NYdb::TProtoAccessor::GetProto(*result.GetStats());
 
diff --git a/ydb/core/kqp/ut/tx/kqp_locks_ut.cpp b/ydb/core/kqp/ut/tx/kqp_locks_ut.cpp
index 768e1d5a66..735e4f8535 100644
--- a/ydb/core/kqp/ut/tx/kqp_locks_ut.cpp
+++ b/ydb/core/kqp/ut/tx/kqp_locks_ut.cpp
@@ -37,10 +37,10 @@ Y_UNIT_TEST_SUITE(KqpLocks) {
         )"), TTxControl::Tx(*tx1).CommitTx()).ExtractValueSync();
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::ABORTED, result.GetIssues().ToString());
         result.GetIssues().PrintTo(Cerr);
-        UNIT_ASSERT(HasIssue(result.GetIssues(), NYql::TIssuesIds::KIKIMR_LOCKS_INVALIDATED,
+        UNIT_ASSERT_C(HasIssue(result.GetIssues(), NYql::TIssuesIds::KIKIMR_LOCKS_INVALIDATED,
             [] (const auto& issue) {
                 return issue.GetMessage().contains("/Root/Test");
-            }));
+            }), result.GetIssues().ToString());
 
         result = session2.ExecuteDataQuery(Q_(R"(
             SELECT * FROM `/Root/Test` WHERE Name == "Paul" ORDER BY Group, Name;
diff --git a/ydb/core/kqp/ut/tx/kqp_sink_tx_ut.cpp b/ydb/core/kqp/ut/tx/kqp_sink_tx_ut.cpp
index a780ea2f92..085bbf6f12 100644
--- a/ydb/core/kqp/ut/tx/kqp_sink_tx_ut.cpp
+++ b/ydb/core/kqp/ut/tx/kqp_sink_tx_ut.cpp
@@ -180,7 +180,7 @@ Y_UNIT_TEST_SUITE(KqpSinkTx) {
                 SELECT COUNT(*) FROM `/Root/KV`;
             )"), TTxControl::Tx(tx.GetId())).ExtractValueSync();
             result.GetIssues().PrintTo(Cerr);
-            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::BAD_REQUEST, result.GetIssues().ToString());
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());
 
             result = session.ExecuteQuery(Q_(R"(
                 UPSERT INTO `/Root/KV` (Key, Value) VALUES (1u, "New");
diff --git a/ydb/core/protos/data_events.proto b/ydb/core/protos/data_events.proto
index e856858f49..2dfe758f06 100644
--- a/ydb/core/protos/data_events.proto
+++ b/ydb/core/protos/data_events.proto
@@ -138,6 +138,7 @@ message TEvWriteResult {
         STATUS_LOCKS_BROKEN = 9;
         STATUS_DISK_SPACE_EXHAUSTED = 10;
         STATUS_WRONG_SHARD_STATE = 11;
+        STATUS_CONSTRAINT_VIOLATION = 12;
     }
 
     // Status
diff --git a/ydb/core/tx/data_events/common/error_codes.cpp b/ydb/core/tx/data_events/common/error_codes.cpp
index 4a64146ba6..2c86d4b521 100644
--- a/ydb/core/tx/data_events/common/error_codes.cpp
+++ b/ydb/core/tx/data_events/common/error_codes.cpp
@@ -28,6 +28,8 @@ TConclusion<NErrorCodes::TOperator::TYdbStatusInfo> TOperator::GetStatusInfo(
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_WRONG_SHARD_STATE:
             return TYdbStatusInfo(Ydb::StatusIds::PRECONDITION_FAILED, NYql::TIssuesIds::KIKIMR_PRECONDITION_FAILED, "Wrong shard state");
+        case NKikimrDataEvents::TEvWriteResult::STATUS_CONSTRAINT_VIOLATION:
+            return TYdbStatusInfo(Ydb::StatusIds::PRECONDITION_FAILED, NYql::TIssuesIds::KIKIMR_PRECONDITION_FAILED, "Constraint violated");
     }
 }
 
diff --git a/ydb/core/tx/datashard/datashard_ut_write.cpp b/ydb/core/tx/datashard/datashard_ut_write.cpp
index 906ce5d909..b3065f3322 100644
--- a/ydb/core/tx/datashard/datashard_ut_write.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_write.cpp
@@ -392,7 +392,7 @@ Y_UNIT_TEST_SUITE(DataShardWrite) {
         Cout << "========= Send immediate insert with duplicate, keys -2, 0, 2 =========\n";
         {
             auto request = MakeWriteRequest(++txId, NKikimrDataEvents::TEvWrite::MODE_IMMEDIATE, NKikimrDataEvents::TEvWrite::TOperation::OPERATION_INSERT, tableId, opts.Columns_, rowCount, -2);
-            const auto writeResult = Write(runtime, sender, shard, std::move(request), NKikimrDataEvents::TEvWriteResult::STATUS_BAD_REQUEST);
+            const auto writeResult = Write(runtime, sender, shard, std::move(request), NKikimrDataEvents::TEvWriteResult::STATUS_CONSTRAINT_VIOLATION);
         }
 
         Cout << "========= Read table =========\n";
diff --git a/ydb/core/tx/datashard/execute_write_unit.cpp b/ydb/core/tx/datashard/execute_write_unit.cpp
index eccd4b1a77..afe7ee0154 100644
--- a/ydb/core/tx/datashard/execute_write_unit.cpp
+++ b/ydb/core/tx/datashard/execute_write_unit.cpp
@@ -96,7 +96,7 @@ public:
             return EExecutionStatus::Continue;
         
         LOG_TRACE_S(ctx, NKikimrServices::TX_DATASHARD, "Operation " << writeOp << " at " << DataShard.TabletID() << " aborting because an duplicate key");
-        writeOp.SetError(NKikimrDataEvents::TEvWriteResult::STATUS_BAD_REQUEST, "Operation is aborting because an duplicate key");
+        writeOp.SetError(NKikimrDataEvents::TEvWriteResult::STATUS_CONSTRAINT_VIOLATION, "Operation is aborting because an duplicate key");
         ResetChanges(userDb, writeOp, txc);
         return EExecutionStatus::Executed;
     }