Handle schema change in immediate kqp data transactions, KIKIMR-14518

ref:75592b3c8c6864c09a1cf84c0973aaafbb370f27

1 files changed, 13 insertions, 0 deletions

diff --git a/ydb/core/tx/datashard/execute_kqp_data_tx_unit.cpp b/ydb/core/tx/datashard/execute_kqp_data_tx_unit.cpp
index 78bfb34c32b..c49ee35beec 100644
--- a/ydb/core/tx/datashard/execute_kqp_data_tx_unit.cpp
+++ b/ydb/core/tx/datashard/execute_kqp_data_tx_unit.cpp
@@ -77,9 +77,22 @@ EExecutionStatus TExecuteKqpDataTxUnit::Execute(TOperation::TPtr op, TTransactio
         switch (Pipeline.RestoreDataTx(tx, txc, ctx)) {
             case ERestoreDataStatus::Ok:
                 break;
+
             case ERestoreDataStatus::Restart:
                 return EExecutionStatus::Restart;
+
             case ERestoreDataStatus::Error:
+                // For immediate transactions we want to translate this into a propose failure
+                if (op->IsImmediate()) {
+                    const auto& dataTx = tx->GetDataTx();
+                    Y_VERIFY(!dataTx->Ready());
+                    op->SetAbortedFlag();
+                    BuildResult(op, NKikimrTxDataShard::TEvProposeTransactionResult::ERROR);
+                    op->Result()->SetProcessError(dataTx->Code(), dataTx->GetErrors());
+                    return EExecutionStatus::Executed;
+                }
+
+                // For planned transactions errors are not expected
                 Y_FAIL("Failed to restore tx data: %s", tx->GetDataTx()->GetErrors().c_str());
         }
     }