Merge from 24-2-4-analytics-2

ydb/core/formats/arrow/size_calcer.cpp

@@ -241,11 +241,12 @@ ui64 GetArrayDataSize(const std::shared_ptr<arrow::Array>& column) {
 }
 
 NKikimr::NArrow::TSerializedBatch TSerializedBatch::Build(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context) {
-    std::optional<TFirstLastSpecialKeys> specialKeys;
+    std::optional<TString> specialKeys;
     if (context.GetFieldsForSpecialKeys().size()) {
-        specialKeys = TFirstLastSpecialKeys(batch, context.GetFieldsForSpecialKeys());
+        specialKeys = TFirstLastSpecialKeys(batch, context.GetFieldsForSpecialKeys()).SerializeToString();
     }
-    return TSerializedBatch(NArrow::SerializeSchema(*batch->schema()), NArrow::SerializeBatchNoCompression(batch), batch->num_rows(), NArrow::GetBatchDataSize(batch), specialKeys);
+    return TSerializedBatch(NArrow::SerializeSchema(*batch->schema()), NArrow::SerializeBatchNoCompression(batch), batch->num_rows(), 
+        NArrow::GetBatchDataSize(batch), specialKeys);
 }
 
 bool TSerializedBatch::BuildWithLimit(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context, std::optional<TSerializedBatch>& sbL, std::optional<TSerializedBatch>& sbR, TString* errorMessage) {

ydb/core/formats/arrow/size_calcer.h

@@ -72,13 +72,13 @@ class TSerializedBatch {
     YDB_READONLY_DEF(TString, Data);
     YDB_READONLY(ui32, RowsCount, 0);
     YDB_READONLY(ui32, RawBytes, 0);
-    std::optional<TFirstLastSpecialKeys> SpecialKeys;
+    std::optional<TString> SpecialKeys;
 public:
     size_t GetSize() const {
         return Data.size();
     }
 
-    const TFirstLastSpecialKeys& GetSpecialKeysSafe() const {
+    const TString& GetSpecialKeysSafe() const {
         AFL_VERIFY(SpecialKeys);
         return *SpecialKeys;
     }
@@ -93,7 +93,7 @@ class TSerializedBatch {
     static bool BuildWithLimit(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context, std::optional<TSerializedBatch>& sbL, std::optional<TSerializedBatch>& sbR, TString* errorMessage);
     static TSerializedBatch Build(std::shared_ptr<arrow::RecordBatch> batch, const TBatchSplitttingContext& context);
 
-    TSerializedBatch(TString&& schemaData, TString&& data, const ui32 rowsCount, const ui32 rawBytes, const std::optional<TFirstLastSpecialKeys>& specialKeys)
+    TSerializedBatch(TString&& schemaData, TString&& data, const ui32 rowsCount, const ui32 rawBytes, const std::optional<TString>& specialKeys)
         : SchemaData(schemaData)
         , Data(data)
         , RowsCount(rowsCount)

ydb/core/kqp/ut/olap/kqp_olap_ut.cpp

@@ -940,7 +940,6 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
             R"(`resource_id` = "10001")",
             R"(`resource_id` != "10001")",
             R"("XXX" == "YYY" OR `resource_id` != "10001")",
-            R"(`resource_id` != "10001" XOR "XXX" == "YYY")",
             R"(`level` = 1)",
             R"(`level` = Int8("1"))",
             R"(`level` = Int16("1"))",

ydb/core/tx/columnshard/background_controller.cpp

@@ -11,7 +11,7 @@ bool TBackgroundController::StartCompaction(const NOlap::TPlanCompactionInfo& in
 void TBackgroundController::CheckDeadlines() {
     for (auto&& i : ActiveCompactionInfo) {
         if (TMonotonic::Now() - i.second.GetStartTime() > NOlap::TCompactionLimits::CompactionTimeout) {
-            AFL_EMERG(NKikimrServices::TX_COLUMNSHARD)("event", "deadline_compaction");
+            AFL_CRIT(NKikimrServices::TX_COLUMNSHARD)("event", "deadline_compaction");
             Y_DEBUG_ABORT_UNLESS(false);
         }
     }
@@ -20,7 +20,7 @@ void TBackgroundController::CheckDeadlines() {
 void TBackgroundController::CheckDeadlinesIndexation() {
     for (auto&& i : ActiveIndexationTasks) {
         if (TMonotonic::Now() - i.second > NOlap::TCompactionLimits::CompactionTimeout) {
-            AFL_EMERG(NKikimrServices::TX_COLUMNSHARD)("event", "deadline_compaction")("task_id", i.first);
+            AFL_CRIT(NKikimrServices::TX_COLUMNSHARD)("event", "deadline_indexation")("task_id", i.first);
             Y_DEBUG_ABORT_UNLESS(false);
         }
     }

ydb/core/tx/columnshard/blobs_action/bs/blob_manager.cpp

@@ -233,16 +233,6 @@ std::shared_ptr<NBlobOperations::NBlobStorage::TGCTask> TBlobManager::BuildGCTas
     const ui32 channelIdx = BLOB_CHANNEL;
     NBlobOperations::NBlobStorage::TGCTask::TGCListsByGroup perGroupGCListsInFlight;
     // Clear all possibly not kept trash in channel's groups: create an event for each group
-    if (FirstGC) {
-        FirstGC = false;
-
-        // TODO: we need only actual channel history here
-        const auto& channelHistory = TabletInfo->ChannelInfo(channelIdx)->History;
-
-        for (auto it = channelHistory.begin(); it != channelHistory.end(); ++it) {
-            perGroupGCListsInFlight[it->GroupID];
-        }
-    }
 
     static const ui32 blobsGCCountLimit = 500000;
 
@@ -270,27 +260,22 @@ std::shared_ptr<NBlobOperations::NBlobStorage::TGCTask> TBlobManager::BuildGCTas
             BlobsManagerCounters.OnCollectDropExplicit(logoBlobId.BlobSize());
             gl.DontKeepList.insert(logoBlobId);
         }
-        if (extractedToRemoveFromDB.GetSize() >= blobsGCCountLimit) {
-            newCollectGenSteps.clear();
-        }
     }
 
 
     std::deque<TUnifiedBlobId> keepsToErase;
-    std::deque<TUnifiedBlobId> deleteIndex;
-    if (extractedToRemoveFromDB.GetSize() + keepsToErase.size() < blobsGCCountLimit) {
-        deleteIndex = BlobsToDelete.GroupByGenStep();
-    }
+    std::deque<TUnifiedBlobId> deleteIndex = BlobsToDelete.GroupByGenStep();
     for (auto&& newCollectGenStep : newCollectGenSteps) {
-        AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "PreparePerGroupGCRequests")("gen", std::get<0>(newCollectGenStep))("step", std::get<1>(newCollectGenStep));
+        AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "PreparePerGroupGCRequests")("gen_step", newCollectGenStep);
         BlobsManagerCounters.OnNewCollectStep(std::get<0>(newCollectGenStep), std::get<1>(newCollectGenStep));
 
         // Make per-group Keep/DontKeep lists
 
         {
             // Add all blobs to keep
             auto keepBlobIt = BlobsToKeep.begin();
-            for (; keepBlobIt != BlobsToKeep.end(); ++keepBlobIt) {
+
+            for (; keepBlobIt != BlobsToKeep.end();) {
                 TGenStep genStep{keepBlobIt->Generation(), keepBlobIt->Step()};
                 AFL_VERIFY(genStep > LastCollectedGenStep);
                 if (genStep > newCollectGenStep) {
@@ -300,16 +285,17 @@ std::shared_ptr<NBlobOperations::NBlobStorage::TGCTask> TBlobManager::BuildGCTas
                 perGroupGCListsInFlight[blobGroup].KeepList.insert(*keepBlobIt);
                 keepsToErase.emplace_back(TUnifiedBlobId(blobGroup, *keepBlobIt));
                 AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("to_keep_gc", *keepBlobIt);
-                if (extractedToRemoveFromDB.GetSize() + keepsToErase.size() > blobsGCCountLimit) {
-                    AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "a lot of blobs to gc")("to_remove", extractedToRemoveFromDB.GetSize())("keeps_to_erase", keepsToErase.size())("limit", blobsGCCountLimit);
+                keepBlobIt = BlobsToKeep.erase(keepBlobIt);
+                if (keepsToErase.size() > blobsGCCountLimit) {
+                    AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "a lot of blobs to gc")("to_remove", extractedToRemoveFromDB.GetSize())("keeps_to_erase", keepsToErase.size())("limit", blobsGCCountLimit)("has_border", !!CollectGenStepInFlight)("border", CollectGenStepInFlight.value_or(LastCollectedGenStep));
                     break;
                 }
             }
-            if (extractedToRemoveFromDB.GetSize() + keepsToErase.size() > blobsGCCountLimit) {
-                AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "a lot of blobs to gc")("to_remove", extractedToRemoveFromDB.GetSize())("keeps_to_erase", keepsToErase.size())("limit", blobsGCCountLimit);
+            AFL_VERIFY(!CollectGenStepInFlight || *CollectGenStepInFlight <= newCollectGenStep);
+            if (keepsToErase.size() > blobsGCCountLimit) {
+                AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "a lot of blobs to gc")("to_remove", extractedToRemoveFromDB.GetSize())("keeps_to_erase", keepsToErase.size())("limit", blobsGCCountLimit)("has_border", !!CollectGenStepInFlight)("border", CollectGenStepInFlight.value_or(LastCollectedGenStep));
                 break;
             }
-            BlobsToKeep.erase(BlobsToKeep.begin(), keepBlobIt);
             BlobsManagerCounters.OnBlobsKeep(BlobsToKeep);
 
             TTabletsByBlob extractedSelf;
@@ -320,7 +306,7 @@ std::shared_ptr<NBlobOperations::NBlobStorage::TGCTask> TBlobManager::BuildGCTas
                         break;
                     }
                     BlobsToDelete.ExtractBlobTo(deleteIndex.front(), extractedSelf);
-                    if (extractedToRemoveFromDB.GetSize() + extractedSelf.GetSize() + keepsToErase.size() > blobsGCCountLimit) {
+                    if (extractedToRemoveFromDB.GetSize() + extractedSelf.GetSize() > blobsGCCountLimit) {
                         AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "a lot of blobs to gc")("to_remove", extractedToRemoveFromDB.GetSize())("keeps_to_erase", keepsToErase.size())("limit", blobsGCCountLimit);
                         break;
                     }
@@ -358,20 +344,35 @@ std::shared_ptr<NBlobOperations::NBlobStorage::TGCTask> TBlobManager::BuildGCTas
             }
             BlobsManagerCounters.OnBlobsDelete(BlobsToDelete);
         }
-        CollectGenStepInFlight = newCollectGenStep;
-        if (extractedToRemoveFromDB.GetSize() + keepsToErase.size() > blobsGCCountLimit) {
+        if (std::get<0>(newCollectGenStep) == CurrentGen) {
+            CollectGenStepInFlight = newCollectGenStep;
+        }
+        if (keepsToErase.size() > blobsGCCountLimit) {
             AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "a lot of blobs to gc")("to_remove", extractedToRemoveFromDB.GetSize())("keeps_to_erase", keepsToErase.size())("limit", blobsGCCountLimit);
             break;
         }
     }
     if (CollectGenStepInFlight) {
+        if (FirstGC) {
+            FirstGC = false;
+
+            // TODO: we need only actual channel history here
+            const auto& channelHistory = TabletInfo->ChannelInfo(channelIdx)->History;
+
+            for (auto it = channelHistory.begin(); it != channelHistory.end(); ++it) {
+                perGroupGCListsInFlight[it->GroupID];
+            }
+        }
         PopGCBarriers(*CollectGenStepInFlight);
-    } else {
-        CollectGenStepInFlight = LastCollectedGenStep;
     }
+    if (BlobsToKeep.size() && CollectGenStepInFlight) {
+        TGenStep genStepFront{BlobsToKeep.begin()->Generation(), BlobsToKeep.begin()->Step()};
+        AFL_VERIFY(*CollectGenStepInFlight < genStepFront);
+    }
+    AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("notice", "collect_gen_step")("value", CollectGenStepInFlight)("current_gen", CurrentGen);
     auto removeCategories = sharedBlobsInfo->BuildRemoveCategories(std::move(extractedToRemoveFromDB));
 
-    auto result = std::make_shared<NBlobOperations::NBlobStorage::TGCTask>(storageId, std::move(perGroupGCListsInFlight), *CollectGenStepInFlight,
+    auto result = std::make_shared<NBlobOperations::NBlobStorage::TGCTask>(storageId, std::move(perGroupGCListsInFlight), CollectGenStepInFlight,
         std::move(keepsToErase), manager, std::move(removeCategories), counters, TabletInfo->TabletID, CurrentGen);
     if (result->IsEmpty()) {
         CollectGenStepInFlight = {};
@@ -383,6 +384,7 @@ std::shared_ptr<NBlobOperations::NBlobStorage::TGCTask> TBlobManager::BuildGCTas
 
 
 TBlobBatch TBlobManager::StartBlobBatch(ui32 channel) {
+    AFL_VERIFY(++CurrentStep < Max<ui32>() - 10);
     ++CountersUpdate.BatchesStarted;
     Y_ABORT_UNLESS(channel == BLOB_CHANNEL, "Support for mutiple blob channels is not implemented yet");
     ++CurrentStep;
@@ -443,13 +445,19 @@ void TBlobManager::DeleteBlobOnComplete(const TTabletId tabletId, const TUnified
     }
 }
 
-void TBlobManager::OnGCFinishedOnExecute(const TGenStep& genStep, IBlobManagerDb& db) {
-    db.SaveLastGcBarrier(genStep);
+void TBlobManager::OnGCFinishedOnExecute(const std::optional<TGenStep>& genStep, IBlobManagerDb& db) {
+    if (genStep) {
+        db.SaveLastGcBarrier(*genStep);
+    }
 }
 
-void TBlobManager::OnGCFinishedOnComplete(const TGenStep& genStep) {
-    LastCollectedGenStep = genStep;
-    CollectGenStepInFlight.reset();
+void TBlobManager::OnGCFinishedOnComplete(const std::optional<TGenStep>& genStep) {
+    if (genStep) {
+        LastCollectedGenStep = *genStep;
+        CollectGenStepInFlight.reset();
+    } else {
+        AFL_VERIFY(!CollectGenStepInFlight);
+    }
 }
 
 void TBlobManager::OnBlobFree(const TUnifiedBlobId& blobId) {

ydb/core/tx/columnshard/blobs_action/bs/blob_manager.h

@@ -195,8 +195,8 @@ class TBlobManager : public IBlobManager, public TCommonBlobsTracker {
         const std::shared_ptr<TBlobManager>& manager, const std::shared_ptr<NDataSharing::TStorageSharedBlobsManager>& sharedBlobsInfo,
         const std::shared_ptr<NBlobOperations::TRemoveGCCounters>& counters) noexcept;
 
-    void OnGCFinishedOnExecute(const TGenStep& genStep, IBlobManagerDb& db);
-    void OnGCFinishedOnComplete(const TGenStep& genStep);
+    void OnGCFinishedOnExecute(const std::optional<TGenStep>& genStep, IBlobManagerDb& db);
+    void OnGCFinishedOnComplete(const std::optional<TGenStep>& genStep);
 
     TBlobManagerCounters GetCountersUpdate() {
         TBlobManagerCounters res = CountersUpdate;

ydb/core/tx/columnshard/blobs_action/bs/gc.cpp

@@ -22,7 +22,7 @@ bool TGCTask::DoOnCompleteTxAfterCleaning(NColumnShard::TColumnShard& /*self*/,
     return true;
 }
 
-TGCTask::TGCTask(const TString& storageId, TGCListsByGroup&& listsByGroupId, const TGenStep& collectGenStepInFlight, std::deque<TUnifiedBlobId>&& keepsToErase,
+TGCTask::TGCTask(const TString& storageId, TGCListsByGroup&& listsByGroupId, const std::optional<TGenStep>& collectGenStepInFlight, std::deque<TUnifiedBlobId>&& keepsToErase,
     const std::shared_ptr<TBlobManager>& manager, TBlobsCategories&& blobsToRemove, const std::shared_ptr<TRemoveGCCounters>& counters,
     const ui64 tabletId, const ui64 currentGen)
     : TBase(storageId, std::move(blobsToRemove), counters)
@@ -53,8 +53,8 @@ std::unique_ptr<TEvBlobStorage::TEvCollectGarbage> TGCTask::BuildRequest(const u
     AFL_VERIFY(++it->second.RequestsCount < 10);
     auto result = std::make_unique<TEvBlobStorage::TEvCollectGarbage>(
         TabletId, CurrentGen, PerGenerationCounter.Val(),
-        channelIdx, true,
-        std::get<0>(CollectGenStepInFlight), std::get<1>(CollectGenStepInFlight),
+        channelIdx, !!CollectGenStepInFlight,
+        CollectGenStepInFlight ? std::get<0>(*CollectGenStepInFlight) : 0, CollectGenStepInFlight ? std::get<1>(*CollectGenStepInFlight) : 0,
         new TVector<TLogoBlobID>(it->second.KeepList.begin(), it->second.KeepList.end()),
         new TVector<TLogoBlobID>(it->second.DontKeepList.begin(), it->second.DontKeepList.end()),
         TInstant::Max(), true);

ydb/core/tx/columnshard/blobs_action/bs/gc.h

@@ -20,7 +20,7 @@ class TGCTask: public IBlobsGCAction {
     using TGCListsByGroup = THashMap<ui32, TGCLists>;
 private:
     TGCListsByGroup ListsByGroupId;
-    TGenStep CollectGenStepInFlight;
+    std::optional<TGenStep> CollectGenStepInFlight;
     const ui64 TabletId;
     const ui64 CurrentGen;
     std::deque<TUnifiedBlobId> KeepsToErase;
@@ -30,11 +30,11 @@ class TGCTask: public IBlobsGCAction {
     virtual void DoOnExecuteTxAfterCleaning(NColumnShard::TColumnShard& self, TBlobManagerDb& dbBlobs) override;
     virtual bool DoOnCompleteTxAfterCleaning(NColumnShard::TColumnShard& self, const std::shared_ptr<IBlobsGCAction>& taskAction) override;
     virtual bool DoIsEmpty() const override {
-        return false;
+        return !CollectGenStepInFlight && KeepsToErase.empty();
     }
 
 public:
-    TGCTask(const TString& storageId, TGCListsByGroup&& listsByGroupId, const TGenStep& collectGenStepInFlight, std::deque<TUnifiedBlobId>&& keepsToErase,
+    TGCTask(const TString& storageId, TGCListsByGroup&& listsByGroupId, const std::optional<TGenStep>& collectGenStepInFlight, std::deque<TUnifiedBlobId>&& keepsToErase,
         const std::shared_ptr<TBlobManager>& manager, TBlobsCategories&& blobsToRemove, const std::shared_ptr<TRemoveGCCounters>& counters, const ui64 tabletId, const ui64 currentGen);
 
     const TGCListsByGroup& GetListsByGroupId() const {

ydb/core/tx/columnshard/blobs_action/transaction/tx_write.cpp

@@ -7,7 +7,7 @@ bool TTxWrite::InsertOneBlob(TTransactionContext& txc, const NOlap::TWideSeriali
     meta.SetNumRows(batch->GetRowsCount());
     meta.SetRawBytes(batch->GetRawBytes());
     meta.SetDirtyWriteTimeSeconds(batch.GetStartInstant().Seconds());
-    meta.SetSpecialKeysRawData(batch->GetSpecialKeysSafe().SerializeToString());
+    meta.SetSpecialKeysRawData(batch->GetSpecialKeysSafe());
 
     const auto& blobRange = batch.GetRange();
     Y_ABORT_UNLESS(blobRange.GetBlobId().IsValid());

ydb/core/tx/columnshard/columnshard.cpp

@@ -212,7 +212,7 @@ void TColumnShard::UpdateInsertTableCounters() {
     SetCounter(COUNTER_COMMITTED_RECORDS, committed.Rows);
     SetCounter(COUNTER_COMMITTED_BYTES, committed.Bytes);
 
-    LOG_S_INFO("InsertTable. Prepared: " << prepared.Bytes << " in " << prepared.Rows
+    LOG_S_TRACE("InsertTable. Prepared: " << prepared.Bytes << " in " << prepared.Rows
         << " records, committed: " << committed.Bytes << " in " << committed.Rows
         << " records at tablet " << TabletID());
 }

ydb/core/tx/columnshard/columnshard__write.cpp

@@ -42,9 +42,8 @@ void TColumnShard::OverloadWriteFail(const EOverloadStatus overloadReason, const
             Y_ABORT("invalid function usage");
     }
 
-    LOG_S_INFO("Write (overload) " << writeData.GetSize() << " bytes into pathId " << writeData.GetWriteMeta().GetTableId()
-        << " overload reason: [" << overloadReason << "]"
-        << " at tablet " << TabletID());
+    AFL_TRACE(NKikimrServices::TX_COLUMNSHARD)("event", "write_overload")("size", writeData.GetSize())
+        ("path_id", writeData.GetWriteMeta().GetTableId())("reason", overloadReason);
 
     ctx.Send(writeData.GetWriteMeta().GetSource(), event.release(), 0, cookie);
 }

ydb/core/tx/columnshard/columnshard__write_index.cpp

@@ -49,14 +49,16 @@ void TColumnShard::Handle(TEvPrivate::TEvWriteIndex::TPtr& ev, const TActorConte
         } else {
             ACFL_DEBUG("event", "TEvWriteIndex")("count", ev->Get()->IndexChanges->GetWritePortionsCount());
             AFL_VERIFY(ev->Get()->IndexChanges->GetWritePortionsCount());
-
+            const bool needDiskLimiter = ev->Get()->IndexChanges->NeedDiskWriteLimiter();
             auto writeController = std::make_shared<NOlap::TCompactedWriteController>(ctx.SelfID, ev->Release());
             const TConclusion<bool> needDraftTransaction = writeController->GetBlobsAction().NeedDraftWritingTransaction();
             AFL_VERIFY(needDraftTransaction.IsSuccess())("error", needDraftTransaction.GetErrorMessage());
             if (*needDraftTransaction) {
                 Execute(new TTxWriteDraft(this, writeController));
-            } else {
+            } else if (needDiskLimiter) {
                 NLimiter::TCompDiskOperator::AskResource(std::make_shared<TDiskResourcesRequest>(writeController, TabletID()));
+            } else {
+                Register(CreateWriteActor(TabletID(), writeController, TInstant::Max()));
             }
         }
     } else {

ydb/core/tx/columnshard/engines/changes/abstract/abstract.h

@@ -234,6 +234,10 @@ class TColumnEngineChanges {
         virtual ~IMemoryPredictor() = default;
     };
 
+    virtual bool NeedDiskWriteLimiter() const {
+        return false;
+    }
+
     void OnFinish(NColumnShard::TColumnShard& self, TChangesFinishContext& context);
 
     ui64 CalcMemoryForUsage() const {

ydb/core/tx/columnshard/engines/changes/general_compaction.h

@@ -13,6 +13,11 @@ class TGeneralCompactColumnEngineChanges: public TCompactColumnEngineChanges {
     void BuildAppendedPortionsByChunks(TConstructionContext& context, std::vector<TPortionInfoWithBlobs>&& portions) noexcept;
 protected:
     virtual TConclusionStatus DoConstructBlobs(TConstructionContext& context) noexcept override;
+
+    virtual bool NeedDiskWriteLimiter() const override {
+        return true;
+    }
+
     virtual TPortionMeta::EProduced GetResultProducedClass() const override {
         return TPortionMeta::EProduced::SPLIT_COMPACTED;
     }