stable-25-2-1: added verification for schema version (ydb-platform#28567)

Author: ydbot

ydb/core/tx/columnshard/columnshard__write.cpp

@@ -543,6 +543,35 @@ void TColumnShard::Handle(NEvents::TDataEvents::TEvWrite::TPtr& ev, const TActor
 
     const auto& mvccSnapshot = record.HasMvccSnapshot() ? NOlap::TSnapshot{record.GetMvccSnapshot().GetStep(), record.GetMvccSnapshot().GetTxId()} : NOlap::TSnapshot::Zero();
 
+    if (mvccSnapshot != NOlap::TSnapshot::Zero()) {
+        auto snapshotSchema = TablesManager.GetPrimaryIndex()->GetVersionedIndex().GetSchemaVerified(mvccSnapshot);
+        if (snapshotSchema->GetVersion() != schema->GetVersion()) {
+            const TString errorMessage = TStringBuilder() << "schema version mismatch with snapshot: "
+                << "tx_id=" << record.GetTxId() 
+                << ", snapshot=" << mvccSnapshot
+                << ", snapshot_schema_version=" << snapshotSchema->GetVersion()
+                << ", request_schema_version=" << schema->GetVersion()
+                << ", table_id=" << schemeShardLocalPathId
+                << ", lock_id=" << lockId;
+            
+            AFL_ERROR(NKikimrServices::TX_COLUMNSHARD_WRITE)("event", "schema_version_mismatch")
+                ("tx_id", record.GetTxId())
+                ("snapshot", TStringBuilder() << mvccSnapshot)
+                ("snapshot_schema_version", snapshotSchema->GetVersion())
+                ("request_schema_version", schema->GetVersion())
+                ("table_id", schemeShardLocalPathId)
+                ("lock_id", lockId)
+                ("path_id", pathId)
+                ("source", source.ToString())
+                ("cookie", cookie);
+            
+            LWPROBE(EvWrite, TabletID(), source.ToString(), cookie, record.GetTxId(), writeTimeout.value_or(TDuration::Max()), 0, "", false,
+                operation.GetIsBulk(), ToString(NKikimrDataEvents::TEvWriteResult::STATUS_BAD_REQUEST), errorMessage);
+            sendError(errorMessage, NKikimrDataEvents::TEvWriteResult::STATUS_BAD_REQUEST);
+            return;
+        }
+    }
+
     LWPROBE(EvWrite, TabletID(), source.ToString(), cookie, record.GetTxId(), writeTimeout.value_or(TDuration::Max()), arrowData->GetSize(), "", true, operation.GetIsBulk(), "", "");
 
     WriteTasksQueue->Enqueue(TWriteTask(arrowData, schema, source, ev->Recipient, granuleShardingVersionId, pathId, cookie, mvccSnapshot, lockId, *mType, behaviour, writeTimeout, record.GetTxId(),