Implemented split datashard status (wrong shard state) (#11955)

Author: dahbka-lis

ydb/core/kqp/runtime/kqp_write_actor.cpp

@@ -535,25 +535,38 @@ class TKqpTableWriteActor : public TActorBootstrapped<TKqpTableWriteActor> {
                 getIssues());
             return;
         }
-        case NKikimrDataEvents::TEvWriteResult::STATUS_INTERNAL_ERROR: {
-            CA_LOG_E("Got INTERNAL ERROR for table `"
+        case NKikimrDataEvents::TEvWriteResult::STATUS_WRONG_SHARD_STATE:
+            CA_LOG_E("Got WRONG SHARD STATE for table `"
                     << SchemeEntry->TableId.PathId.ToString() << "`."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
                     << " Sink=" << this->SelfId() << "."
                     << getIssues().ToOneLineString());
-            // TODO: Add new status for splits in datashard. This is tmp solution.
-            if (getIssues().ToOneLineString().Contains("in a pre/offline state assuming this is due to a finished split (wrong shard state)")) {
+            
+            if (InconsistentTx) {
                 ResetShardRetries(ev->Get()->Record.GetOrigin(), ev->Cookie);
                 RetryResolveTable();
             } else {
                 RuntimeError(
-                    TStringBuilder() << "Internal error for table `"
+                    TStringBuilder() << "Wrong shard state for table `"
                         << TablePath << "`. "
                         << getIssues().ToOneLineString(),
-                    NYql::NDqProto::StatusIds::INTERNAL_ERROR,
+                    NYql::NDqProto::StatusIds::PRECONDITION_FAILED,
                     getIssues());
             }
             return;
+        case NKikimrDataEvents::TEvWriteResult::STATUS_INTERNAL_ERROR: {
+            CA_LOG_E("Got INTERNAL ERROR for table `"
+                    << SchemeEntry->TableId.PathId.ToString() << "`."
+                    << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
+                    << " Sink=" << this->SelfId() << "."
+                    << getIssues().ToOneLineString());
+            RuntimeError(
+                TStringBuilder() << "Internal error for table `"
+                    << TablePath << "`. "
+                    << getIssues().ToOneLineString(),
+                NYql::NDqProto::StatusIds::INTERNAL_ERROR,
+                getIssues());
+            return;
         }
         case NKikimrDataEvents::TEvWriteResult::STATUS_DISK_SPACE_EXHAUSTED: {
             CA_LOG_E("Got DISK_SPACE_EXHAUSTED for table `"
@@ -1798,6 +1811,18 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
                 getIssues());
             return;
         }
+        case NKikimrDataEvents::TEvWriteResult::STATUS_WRONG_SHARD_STATE: {
+            CA_LOG_E("Got WRONG SHARD STATE for table."
+                    << " ShardID=" << ev->Get()->Record.GetOrigin() << ","
+                    << " Sink=" << this->SelfId() << "."
+                    << getIssues().ToOneLineString());
+            ReplyErrorAndDie(
+                TStringBuilder() << "Wrong shard state for table. "
+                    << getIssues().ToOneLineString(),
+                NYql::NDqProto::StatusIds::INTERNAL_ERROR,
+                getIssues());
+            return;
+        }
         case NKikimrDataEvents::TEvWriteResult::STATUS_INTERNAL_ERROR: {
             CA_LOG_E("Got INTERNAL ERROR for table."
                     << " ShardID=" << ev->Get()->Record.GetOrigin() << ","

ydb/core/protos/data_events.proto

@@ -130,6 +130,7 @@ message TEvWriteResult {
         STATUS_SCHEME_CHANGED = 8;
         STATUS_LOCKS_BROKEN = 9;
         STATUS_DISK_SPACE_EXHAUSTED = 10;
+        STATUS_WRONG_SHARD_STATE = 11;
     }
 
     // Status

ydb/core/tx/data_events/common/error_codes.cpp

@@ -26,6 +26,8 @@ TConclusion<NErrorCodes::TOperator::TYdbStatusInfo> TOperator::GetStatusInfo(
         case NKikimrDataEvents::TEvWriteResult::STATUS_LOCKS_BROKEN: {
             return TYdbStatusInfo(Ydb::StatusIds::ABORTED, NYql::TIssuesIds::KIKIMR_LOCKS_INVALIDATED, "Transaction locks invalidated.");
         }
+        case NKikimrDataEvents::TEvWriteResult::STATUS_WRONG_SHARD_STATE:
+            return TYdbStatusInfo(Ydb::StatusIds::PRECONDITION_FAILED, NYql::TIssuesIds::KIKIMR_PRECONDITION_FAILED, "Wrong shard state");
     }
 }
 

ydb/core/tx/datashard/datashard.cpp

@@ -3054,7 +3054,11 @@ bool TDataShard::CheckDataTxRejectAndReply(const NEvents::TDataEvents::TEvWrite:
                 status = NKikimrDataEvents::TEvWriteResult::STATUS_OVERLOADED;
                 break;
             case NKikimrTxDataShard::TEvProposeTransactionResult::ERROR:
-                status = NKikimrDataEvents::TEvWriteResult::STATUS_INTERNAL_ERROR;
+                if ((rejectReasons & ERejectReasons::WrongState) != ERejectReasons::None) {
+                    status = NKikimrDataEvents::TEvWriteResult::STATUS_WRONG_SHARD_STATE;
+                } else {
+                    status = NKikimrDataEvents::TEvWriteResult::STATUS_INTERNAL_ERROR;
+                }
                 break;
             default:
                 Y_FAIL_S("Unexpected rejectStatus " << rejectStatus);