@@ -554,7 +554,7 @@ class TChangesTask: public NConveyor::ITask {
554
554
virtual bool DoExecute () override {
555
555
NActors::TLogContextGuard g (NActors::TLogContextBuilder::Build (NKikimrServices::TX_COLUMNSHARD)(" tablet_id" , TabletId)(" parent_id" , ParentActorId));
556
556
{
557
- NOlap::TConstructionContext context (TxEvent->IndexInfo , Counters);
557
+ NOlap::TConstructionContext context (* TxEvent->IndexInfo , Counters);
558
558
Y_ABORT_UNLESS (TxEvent->IndexChanges ->ConstructBlobs (context).Ok ());
559
559
if (!TxEvent->IndexChanges ->GetWritePortionsCount ()) {
560
560
TxEvent->SetPutStatus (NKikimrProto::OK);
@@ -601,7 +601,8 @@ class TChangesReadTask: public NOlap::NBlobOperations::NRead::ITask {
601
601
}
602
602
virtual bool DoOnError (const TString& storageId, const NOlap::TBlobRange& range, const NOlap::IBlobsReadingAction::TErrorStatus& status) override {
603
603
AFL_DEBUG (NKikimrServices::TX_COLUMNSHARD)(" event" , " DoOnError" )(" storage_id" , storageId)(" blob_id" , range)(" status" , status.GetErrorMessage ())(" status_code" , status.GetStatus ());
604
- AFL_VERIFY (false )(" blob_id" , range)(" status" , status.GetStatus ());
604
+ AFL_VERIFY (false )(" blob_id" , range)(" status" , status.GetStatus ())(" error" , status.GetErrorMessage ())(" type" , TxEvent->IndexChanges ->TypeString ())(" task_id" , TxEvent->IndexChanges ->GetTaskIdentifier ())
605
+ (" debug" , TxEvent->IndexChanges ->DebugString ());
605
606
TxEvent->SetPutStatus (NKikimrProto::ERROR);
606
607
TActorContext::AsActorContext ().Send (ParentActorId, std::move (TxEvent));
607
608
return false ;
@@ -631,9 +632,9 @@ void TColumnShard::StartIndexTask(std::vector<const NOlap::TInsertedData*>&& dat
631
632
auto indexChanges = TablesManager.MutablePrimaryIndex ().StartInsert (std::move (data));
632
633
Y_ABORT_UNLESS (indexChanges);
633
634
634
- auto actualIndexInfo = TablesManager.GetPrimaryIndex ()->GetVersionedIndex ();
635
+ auto actualIndexInfo = std::make_shared<NOlap::TVersionedIndex>( TablesManager.GetPrimaryIndex ()->GetVersionedIndex () );
635
636
indexChanges->Start (*this );
636
- auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(std::move ( actualIndexInfo) , indexChanges, Settings.CacheDataAfterIndexing );
637
+ auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(actualIndexInfo, indexChanges, Settings.CacheDataAfterIndexing );
637
638
638
639
const TString externalTaskId = indexChanges->GetTaskIdentifier ();
639
640
AFL_DEBUG (NKikimrServices::TX_COLUMNSHARD)(" event" , " indexation" )(" bytes" , bytesToIndex)(" blobs_count" , dataToIndex.size ())(" max_limit" , (i64)Limits.MaxInsertBytes )
@@ -712,8 +713,8 @@ void TColumnShard::SetupCompaction() {
712
713
713
714
indexChanges->Start (*this );
714
715
715
- auto actualIndexInfo = TablesManager.GetPrimaryIndex ()->GetVersionedIndex ();
716
- auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(std::move ( actualIndexInfo) , indexChanges, Settings.CacheDataAfterCompaction );
716
+ auto actualIndexInfo = std::make_shared<NOlap::TVersionedIndex>( TablesManager.GetPrimaryIndex ()->GetVersionedIndex () );
717
+ auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(actualIndexInfo, indexChanges, Settings.CacheDataAfterCompaction );
717
718
const TString externalTaskId = indexChanges->GetTaskIdentifier ();
718
719
AFL_DEBUG (NKikimrServices::TX_COLUMNSHARD)(" event" , " compaction" )(" external_task_id" , externalTaskId);
719
720
@@ -731,37 +732,34 @@ bool TColumnShard::SetupTtl(const THashMap<ui64, NOlap::TTiering>& pathTtls) {
731
732
return false ;
732
733
}
733
734
CSCounters.OnSetupTtl ();
734
- if (BackgroundController.IsTtlActive ()) {
735
- ACFL_DEBUG (" background" , " ttl" )(" skip_reason" , " in_progress" );
736
- return false ;
737
- }
738
735
THashMap<ui64, NOlap::TTiering> eviction = pathTtls;
739
736
for (auto && i : eviction) {
740
737
ACFL_DEBUG (" background" , " ttl" )(" path" , i.first )(" info" , i.second .GetDebugString ());
741
738
}
742
739
743
- auto actualIndexInfo = TablesManager.GetPrimaryIndex ()->GetVersionedIndex ();
740
+ auto actualIndexInfo = std::make_shared<NOlap::TVersionedIndex>( TablesManager.GetPrimaryIndex ()->GetVersionedIndex () );
744
741
const ui64 memoryUsageLimit = HasAppData () ? AppDataVerified ().ColumnShardConfig .GetTieringsMemoryLimit () : ((ui64)512 * 1024 * 1024 );
745
- std::shared_ptr<NOlap::TTTLColumnEngineChanges> indexChanges = TablesManager.MutablePrimaryIndex ().StartTtl (eviction, DataLocksManager, memoryUsageLimit);
742
+ std::vector<std:: shared_ptr<NOlap::TTTLColumnEngineChanges> > indexChanges = TablesManager.MutablePrimaryIndex ().StartTtl (eviction, DataLocksManager, memoryUsageLimit);
746
743
747
- if (! indexChanges) {
744
+ if (indexChanges. empty () ) {
748
745
ACFL_DEBUG (" background" , " ttl" )(" skip_reason" , " no_changes" );
749
746
return false ;
750
747
}
751
- const TString externalTaskId = indexChanges->GetTaskIdentifier ();
752
- const bool needWrites = indexChanges->NeedConstruction ();
753
- ACFL_DEBUG (" background" , " ttl" )(" need_writes" , needWrites);
754
-
755
- indexChanges->Start (*this );
756
- auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(std::move (actualIndexInfo), indexChanges, false );
757
- NYDBTest::TControllers::GetColumnShardController ()->OnWriteIndexStart (TabletID (), indexChanges->TypeString ());
758
- if (needWrites) {
759
- NOlap::NResourceBroker::NSubscribe::ITask::StartResourceSubscription (
760
- ResourceSubscribeActor, std::make_shared<NOlap::NBlobOperations::NRead::ITask::TReadSubscriber>(
761
- std::make_shared<TChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters), 0, indexChanges->CalcMemoryForUsage(), externalTaskId, TTLTaskSubscription));
762
- } else {
763
- ev->SetPutStatus (NKikimrProto::OK);
764
- ActorContext ().Send (SelfId (), std::move (ev));
748
+ for (auto && i : indexChanges) {
749
+ const TString externalTaskId = i->GetTaskIdentifier ();
750
+ const bool needWrites = i->NeedConstruction ();
751
+ ACFL_DEBUG (" background" , " ttl" )(" need_writes" , needWrites);
752
+ i->Start (*this );
753
+ auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(actualIndexInfo, i, false );
754
+ NYDBTest::TControllers::GetColumnShardController ()->OnWriteIndexStart (TabletID (), i->TypeString ());
755
+ if (needWrites) {
756
+ NOlap::NResourceBroker::NSubscribe::ITask::StartResourceSubscription (
757
+ ResourceSubscribeActor, std::make_shared<NOlap::NBlobOperations::NRead::ITask::TReadSubscriber>(
758
+ std::make_shared<TChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters), 0, i->CalcMemoryForUsage(), externalTaskId, TTLTaskSubscription));
759
+ } else {
760
+ ev->SetPutStatus (NKikimrProto::OK);
761
+ ActorContext ().Send (SelfId (), std::move (ev));
762
+ }
765
763
}
766
764
return true ;
767
765
}
@@ -782,8 +780,8 @@ void TColumnShard::SetupCleanup() {
782
780
}
783
781
784
782
ACFL_DEBUG (" background" , " cleanup" )(" changes_info" , changes->DebugString ());
785
- auto actualIndexInfo = TablesManager.GetPrimaryIndex ()->GetVersionedIndex ();
786
- auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(std::move ( actualIndexInfo) , changes, false );
783
+ auto actualIndexInfo = std::make_shared<NOlap::TVersionedIndex>( TablesManager.GetPrimaryIndex ()->GetVersionedIndex () );
784
+ auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(actualIndexInfo, changes, false );
787
785
ev->SetPutStatus (NKikimrProto::OK); // No new blobs to write
788
786
789
787
changes->Start (*this );
0 commit comments