dont use cpu for not abortable chunks (#7865)

Author: ivanmorozov333

ydb/core/tx/columnshard/columnshard_impl.cpp

@@ -271,7 +271,11 @@ bool TColumnShard::RemoveLongTxWrite(NIceDb::TNiceDb& db, const TWriteId writeId
             }
             LongTxWrites.erase(writeId);
             return true;
+        } else {
+            AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "cannot_remove_prepared_tx_insertion")("write_id", (ui64)writeId)("tx_id", txId);
         }
+    } else {
+        AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "cannot_remove_removed_tx_insertion")("write_id", (ui64)writeId)("tx_id", txId);
     }
     return false;
 }
@@ -284,9 +288,10 @@ void TColumnShard::TryAbortWrites(NIceDb::TNiceDb& db, NOlap::TDbWrapper& dbTabl
         }
     }
     if (failedAborts.size()) {
-        AFL_TRACE(NKikimrServices::TX_COLUMNSHARD)("event", "failed_aborts")("count", failedAborts.size())("writes_count", writesToAbort.size());
+        AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "failed_aborts")("count", failedAborts.size())("writes_count", writesToAbort.size());
     }
     for (auto& writeId : failedAborts) {
+        InsertTable->MarkAsNotAbortable(writeId);
         writesToAbort.erase(writeId);
     }
     if (!writesToAbort.empty()) {

ydb/core/tx/columnshard/engines/insert_table/data.h

@@ -33,8 +33,13 @@ struct TInsertedData {
 
 private:
     YDB_READONLY(ui64, SchemaVersion, 0);
+    YDB_READONLY_FLAG(NotAbortable, false);
 
 public:
+    void MarkAsNotAbortable() {
+        NotAbortableFlag = true;
+    }
+
     std::optional<TString> GetBlobData() const {
         if (BlobDataGuard) {
             return BlobDataGuard->GetData();
@@ -104,7 +109,7 @@ struct TInsertedData {
     /// One of them wins and becomes committed. Original DedupId would be lost then.
     /// After commit we use original Initiator:WriteId as DedupId of inserted blob inside {PlanStep, TxId}.
     /// pathId, initiator, {writeId}, {dedupId} -> pathId, planStep, txId, {dedupId}
-    void Commit(ui64 planStep, ui64 txId) {
+    void Commit(const ui64 planStep, const ui64 txId) {
         DedupId = ToString(PlanStep) + ":" + ToString((ui64)WriteTxId);
         PlanStep = planStep;
         WriteTxId = txId;

ydb/core/tx/columnshard/engines/insert_table/insert_table.h

@@ -82,6 +82,9 @@ class TInsertTable: public TInsertTableAccessor {
     TInsertionSummary::TCounters Commit(IDbWrapper& dbTable, ui64 planStep, ui64 txId,
                      const THashSet<TWriteId>& writeIds, std::function<bool(ui64)> pathExists);
     void Abort(IDbWrapper& dbTable, const THashSet<TWriteId>& writeIds);
+    void MarkAsNotAbortable(const TWriteId writeId) {
+        Summary.MarkAsNotAbortable(writeId);
+    }
     THashSet<TWriteId> OldWritesToAbort(const TInstant& now) const;
     THashSet<TWriteId> DropPath(IDbWrapper& dbTable, ui64 pathId);
 

ydb/core/tx/columnshard/engines/insert_table/rt_insertion.cpp

@@ -120,6 +120,9 @@ THashSet<NKikimr::NOlap::TWriteId> TInsertionSummary::GetExpiredInsertions(const
     TInstant newMin = TInstant::Max();
     for (auto& [writeId, data] : Inserted) {
         const TInstant dataInsertTs = data.GetMeta().GetDirtyWriteTime();
+        if (data.IsNotAbortable()) {
+            continue;
+        }
         if (dataInsertTs < timeBorder && toAbort.size() < limit) {
             toAbort.insert(writeId);
         } else {

ydb/core/tx/columnshard/engines/insert_table/rt_insertion.h

@@ -34,6 +34,14 @@ class TInsertionSummary {
     void OnEraseInserted(TPathInfo& pathInfo, const ui64 dataSize) noexcept;
     static TAtomicCounter CriticalInserted;
 public:
+    void MarkAsNotAbortable(const TWriteId writeId) {
+        auto it = Inserted.find(writeId);
+        if (it == Inserted.end()) {
+            return;
+        }
+        it->second.MarkAsNotAbortable();
+    }
+
     THashSet<TWriteId> GetInsertedByPathId(const ui64 pathId) const;
 
     THashSet<TWriteId> GetExpiredInsertions(const TInstant timeBorder, const ui64 limit) const;

ydb/core/tx/columnshard/engines/scheme/versions/abstract_scheme.cpp

@@ -117,7 +117,7 @@ TConclusion<std::shared_ptr<arrow::RecordBatch>> ISnapshotSchema::PrepareForModi
     switch (mType) {
         case NEvWrite::EModificationType::Replace:
         case NEvWrite::EModificationType::Upsert: {
-                AFL_VERIFY(batch->num_columns() <= dstSchema->num_fields());
+            AFL_VERIFY(batch->num_columns() <= dstSchema->num_fields());
             if (batch->num_columns() < dstSchema->num_fields()) {
                 for (auto&& f : dstSchema->fields()) {
                     if (GetIndexInfo().IsNullableVerified(f->name())) {