diff options
18 files changed, 559 insertions, 36 deletions
diff --git a/ydb/core/protos/counters_statistics_aggregator.proto b/ydb/core/protos/counters_statistics_aggregator.proto index d85810cce9b..e6bf4adc9a3 100644 --- a/ydb/core/protos/counters_statistics_aggregator.proto +++ b/ydb/core/protos/counters_statistics_aggregator.proto @@ -18,4 +18,7 @@ enum ETxTypes { TXTYPE_SAVE_QUERY_RESPONSE = 8 [(TxTypeOpts) = {Name: "TxSaveQueryResponse"}]; TXTYPE_SCHEDULE_SCAN = 9 [(TxTypeOpts) = {Name: "TxScheduleScan"}]; TXTYPE_DELETE_QUERY_RESPONSE = 10 [(TxTypeOpts) = {Name: "TxDeleteQueryResponse"}]; + TXTYPE_AGGR_STAT_RESPONSE = 11 [(TxTypeOpts) = {Name: "TxAggregateStatisticsResponse"}]; + TXTYPE_RESPONSE_TABLET_DISTRIBUTION = 12 [(TxTypeOpts) = {Name: "TxResponseTabletDistribution"}]; + TXTYPE_ACK_TIMEOUT = 13 [(TxTypeOpts) = {Name: "TxAckTimeout"}]; } diff --git a/ydb/core/protos/statistics.proto b/ydb/core/protos/statistics.proto index d0e4e114029..726fe72d6ae 100644 --- a/ydb/core/protos/statistics.proto +++ b/ydb/core/protos/statistics.proto @@ -13,6 +13,7 @@ message TPathEntry { optional NKikimrProto.TPathID PathId = 1; optional uint64 RowCount = 2; optional uint64 BytesSize = 3; + optional bool IsColumnTable = 4; } message TSchemeShardStats { diff --git a/ydb/core/statistics/aggregator/aggregator_impl.cpp b/ydb/core/statistics/aggregator/aggregator_impl.cpp index 734070165b9..a682221b8b8 100644 --- a/ydb/core/statistics/aggregator/aggregator_impl.cpp +++ b/ydb/core/statistics/aggregator/aggregator_impl.cpp @@ -396,15 +396,26 @@ size_t TStatisticsAggregator::PropagatePart(const std::vector<TNodeId>& nodeIds, } void TStatisticsAggregator::Handle(TEvPipeCache::TEvDeliveryProblem::TPtr& ev) { - auto tabletId = ev->Get()->TabletId; - if (ShardRanges.empty()) { + if (!ScanTableId.PathId) { return; } - auto& range = ShardRanges.front(); - if (tabletId != range.DataShardId) { - return; + auto tabletId = ev->Get()->TabletId; + if (IsColumnTable) { + if (tabletId == HiveId) { + Schedule(HiveRetryInterval, new TEvPrivate::TEvRequestDistribution); + } else { + SA_LOG_CRIT("[" << TabletID() << "] TEvDeliveryProblem with unexpected tablet " << tabletId); + } + } else { + if (ShardRanges.empty()) { + return; + } + auto& range = ShardRanges.front(); + if (tabletId != range.DataShardId) { + return; + } + Resolve(); } - Resolve(); } void TStatisticsAggregator::Handle(TEvStatistics::TEvStatTableCreationResponse::TPtr&) { @@ -439,6 +450,30 @@ void TStatisticsAggregator::Handle(TEvStatistics::TEvGetScanStatus::TPtr& ev) { Send(ev->Sender, response.release(), 0, ev->Cookie); } +void TStatisticsAggregator::Handle(TEvPrivate::TEvResolve::TPtr&) { + Resolve(); +} + +void TStatisticsAggregator::Handle(TEvPrivate::TEvRequestDistribution::TPtr&) { + ++HiveRequestRound; + + auto reqDistribution = std::make_unique<TEvHive::TEvRequestTabletDistribution>(); + reqDistribution->Record.MutableTabletIds()->Reserve(TabletsForReqDistribution.size()); + for (auto& tablet : TabletsForReqDistribution) { + reqDistribution->Record.AddTabletIds(tablet); + } + + Send(MakePipePerNodeCacheID(false), + new TEvPipeCache::TEvForward(reqDistribution.release(), HiveId, true)); +} + +void TStatisticsAggregator::Handle(TEvStatistics::TEvAggregateKeepAlive::TPtr& ev) { + auto ack = std::make_unique<TEvStatistics::TEvAggregateKeepAliveAck>(); + ack->Record.SetRound(ev->Get()->Record.GetRound()); + Send(ev->Sender, ack.release()); + Schedule(KeepAliveTimeout, new TEvPrivate::TEvAckTimeout(++KeepAliveSeqNo)); +} + void TStatisticsAggregator::InitializeStatisticsTable() { if (!EnableColumnStatistics) { return; @@ -460,6 +495,8 @@ void TStatisticsAggregator::Navigate() { } void TStatisticsAggregator::Resolve() { + ++ResolveRound; + TVector<TCell> plusInf; TTableRange range(StartKey.GetCells(), true, plusInf, true, false); auto keyDesc = MakeHolder<TKeyDesc>( @@ -500,6 +537,9 @@ void TStatisticsAggregator::SaveStatisticsToTable() { std::vector<ui32> columnTags; std::vector<TString> data; auto count = CountMinSketches.size(); + if (count == 0) { + return; + } columnTags.reserve(count); data.reserve(count); @@ -533,28 +573,44 @@ void TStatisticsAggregator::ScheduleNextScan(NIceDb::TNiceDb& db) { auto* operation = ScanOperations.Front(); ReplyToActorIds.swap(operation->ReplyToActorIds); - StartScan(db, operation->PathId); - + bool doStartScan = true; + bool isColumnTable = false; + auto pathId = operation->PathId; + auto itPath = ScanTables.find(pathId); + if (itPath != ScanTables.end()) { + isColumnTable = itPath->second.IsColumnTable; + } else { + doStartScan = false; + } + if (doStartScan) { + StartScan(db, pathId, isColumnTable); + } db.Table<Schema::ScanOperations>().Key(operation->OperationId).Delete(); ScanOperations.PopFront(); - ScanOperationsByPathId.erase(operation->PathId); + ScanOperationsByPathId.erase(pathId); return; } if (ScanTablesByTime.Empty()) { return; } auto* topTable = ScanTablesByTime.Top(); - auto now = TInstant::Now(); - auto updateTime = topTable->LastUpdateTime; - if (now - updateTime < ScanIntervalTime) { + if (TInstant::Now() < topTable->LastUpdateTime + ScanIntervalTime) { + return; + } + bool isColumnTable = false; + auto itPath = ScanTables.find(topTable->PathId); + if (itPath != ScanTables.end()) { + isColumnTable = itPath->second.IsColumnTable; + } else { return; } - StartScan(db, topTable->PathId); + StartScan(db, topTable->PathId, isColumnTable); } -void TStatisticsAggregator::StartScan(NIceDb::TNiceDb& db, TPathId pathId) { +void TStatisticsAggregator::StartScan(NIceDb::TNiceDb& db, TPathId pathId, bool isColumnTable) { ScanTableId.PathId = pathId; ScanStartTime = TInstant::Now(); + IsColumnTable = isColumnTable; PersistCurrentScan(db); StartKey = TSerializedCellVec(); @@ -590,6 +646,7 @@ void TStatisticsAggregator::PersistCurrentScan(NIceDb::TNiceDb& db) { PersistSysParam(db, Schema::SysParam_ScanTableOwnerId, ToString(ScanTableId.PathId.OwnerId)); PersistSysParam(db, Schema::SysParam_ScanTableLocalPathId, ToString(ScanTableId.PathId.LocalPathId)); PersistSysParam(db, Schema::SysParam_ScanStartTime, ToString(ScanStartTime.MicroSeconds())); + PersistSysParam(db, Schema::SysParam_IsColumnTable, ToString(IsColumnTable)); } void TStatisticsAggregator::PersistStartKey(NIceDb::TNiceDb& db) { @@ -600,6 +657,10 @@ void TStatisticsAggregator::PersistLastScanOperationId(NIceDb::TNiceDb& db) { PersistSysParam(db, Schema::SysParam_LastScanOperationId, ToString(LastScanOperationId)); } +void TStatisticsAggregator::PersistGlobalTraversalRound(NIceDb::TNiceDb& db) { + PersistSysParam(db, Schema::SysParam_GlobalTraversalRound, ToString(GlobalTraversalRound)); +} + void TStatisticsAggregator::ResetScanState(NIceDb::TNiceDb& db) { ScanTableId.PathId = TPathId(); ScanStartTime = TInstant::MicroSeconds(0); @@ -620,6 +681,12 @@ void TStatisticsAggregator::ResetScanState(NIceDb::TNiceDb& db) { KeyColumnTypes.clear(); Columns.clear(); ColumnNames.clear(); + + TabletsForReqDistribution.clear(); + + ResolveRound = 0; + HiveRequestRound = 0; + TraversalRound = 0; } template <typename T, typename S> diff --git a/ydb/core/statistics/aggregator/aggregator_impl.h b/ydb/core/statistics/aggregator/aggregator_impl.h index 314a7051a44..33e434ad213 100644 --- a/ydb/core/statistics/aggregator/aggregator_impl.h +++ b/ydb/core/statistics/aggregator/aggregator_impl.h @@ -5,6 +5,7 @@ #include <ydb/core/protos/statistics.pb.h> #include <ydb/core/protos/counters_statistics_aggregator.pb.h> +#include <ydb/core/base/hive.h> #include <ydb/core/base/tablet_pipe.h> #include <ydb/core/base/tablet_pipecache.h> #include <ydb/core/statistics/common.h> @@ -51,6 +52,9 @@ private: struct TTxSaveQueryResponse; struct TTxScheduleScan; struct TTxDeleteQueryResponse; + struct TTxAggregateStatisticsResponse; + struct TTxResponseTabletDistribution; + struct TTxAckTimeout; struct TEvPrivate { enum EEv { @@ -59,6 +63,9 @@ private: EvProcessUrgent, EvPropagateTimeout, EvScheduleScan, + EvRequestDistribution, + EvResolve, + EvAckTimeout, EvEnd }; @@ -68,6 +75,15 @@ private: struct TEvProcessUrgent : public TEventLocal<TEvProcessUrgent, EvProcessUrgent> {}; struct TEvPropagateTimeout : public TEventLocal<TEvPropagateTimeout, EvPropagateTimeout> {}; struct TEvScheduleScan : public TEventLocal<TEvScheduleScan, EvScheduleScan> {}; + struct TEvRequestDistribution : public TEventLocal<TEvRequestDistribution, EvRequestDistribution> {}; + struct TEvResolve : public TEventLocal<TEvResolve, EvResolve> {}; + + struct TEvAckTimeout : public TEventLocal<TEvAckTimeout, EvAckTimeout> { + size_t SeqNo = 0; + explicit TEvAckTimeout(size_t seqNo) { + SeqNo = seqNo; + } + }; }; private: @@ -114,6 +130,12 @@ private: void Handle(TEvStatistics::TEvDeleteStatisticsQueryResponse::TPtr& ev); void Handle(TEvPrivate::TEvScheduleScan::TPtr& ev); void Handle(TEvStatistics::TEvGetScanStatus::TPtr& ev); + void Handle(TEvHive::TEvResponseTabletDistribution::TPtr& ev); + void Handle(TEvStatistics::TEvAggregateStatisticsResponse::TPtr& ev); + void Handle(TEvPrivate::TEvResolve::TPtr& ev); + void Handle(TEvPrivate::TEvRequestDistribution::TPtr& ev); + void Handle(TEvStatistics::TEvAggregateKeepAlive::TPtr& ev); + void Handle(TEvPrivate::TEvAckTimeout::TPtr& ev); void InitializeStatisticsTable(); void Navigate(); @@ -126,13 +148,13 @@ private: void PersistCurrentScan(NIceDb::TNiceDb& db); void PersistStartKey(NIceDb::TNiceDb& db); void PersistLastScanOperationId(NIceDb::TNiceDb& db); + void PersistGlobalTraversalRound(NIceDb::TNiceDb& db); void ResetScanState(NIceDb::TNiceDb& db); void ScheduleNextScan(NIceDb::TNiceDb& db); - void StartScan(NIceDb::TNiceDb& db, TPathId pathId); + void StartScan(NIceDb::TNiceDb& db, TPathId pathId, bool isColumnTable); void FinishScan(NIceDb::TNiceDb& db); - STFUNC(StateInit) { StateInitImpl(ev, SelfId()); } @@ -164,6 +186,12 @@ private: hFunc(TEvStatistics::TEvDeleteStatisticsQueryResponse, Handle); hFunc(TEvPrivate::TEvScheduleScan, Handle); hFunc(TEvStatistics::TEvGetScanStatus, Handle); + hFunc(TEvHive::TEvResponseTabletDistribution, Handle); + hFunc(TEvStatistics::TEvAggregateStatisticsResponse, Handle); + hFunc(TEvPrivate::TEvResolve, Handle); + hFunc(TEvPrivate::TEvRequestDistribution, Handle); + hFunc(TEvStatistics::TEvAggregateKeepAlive, Handle); + hFunc(TEvPrivate::TEvAckTimeout, Handle); default: if (!HandleDefaultEvents(ev, SelfId())) { @@ -214,6 +242,7 @@ private: // TTableId ScanTableId; // stored in local db + bool IsColumnTable = false; // stored in local db std::unordered_set<TActorId> ReplyToActorIds; bool IsStatisticsTableCreated = false; @@ -241,6 +270,7 @@ private: TPathId PathId; ui64 SchemeShardId = 0; TInstant LastUpdateTime; + bool IsColumnTable = false; size_t HeapIndexByTime = -1; @@ -275,6 +305,24 @@ private: ui64 LastScanOperationId = 0; // stored in local db TInstant ScanStartTime; + + size_t ResolveRound = 0; + static constexpr size_t MaxResolveRoundCount = 5; + static constexpr TDuration ResolveRetryInterval = TDuration::Seconds(1); + + ui64 HiveId = 0; + std::unordered_set<ui64> TabletsForReqDistribution; + + size_t HiveRequestRound = 0; + static constexpr size_t MaxHiveRequestRoundCount = 5; + static constexpr TDuration HiveRetryInterval = TDuration::Seconds(1); + + size_t TraversalRound = 0; + static constexpr size_t MaxTraversalRoundCount = 5; + size_t GlobalTraversalRound = 1; // stored in local db + + size_t KeepAliveSeqNo = 0; + static constexpr TDuration KeepAliveTimeout = TDuration::Seconds(3); }; } // NKikimr::NStat diff --git a/ydb/core/statistics/aggregator/schema.h b/ydb/core/statistics/aggregator/schema.h index fea591a932c..e8751204568 100644 --- a/ydb/core/statistics/aggregator/schema.h +++ b/ydb/core/statistics/aggregator/schema.h @@ -34,13 +34,15 @@ struct TAggregatorSchema : NIceDb::Schema { struct LocalPathId : Column<2, NScheme::NTypeIds::Uint64> {}; struct LastUpdateTime : Column<3, NScheme::NTypeIds::Timestamp> {}; struct SchemeShardId : Column<4, NScheme::NTypeIds::Uint64> {}; + struct IsColumnTable : Column<5, NScheme::NTypeIds::Bool> {}; using TKey = TableKey<OwnerId, LocalPathId>; using TColumns = TableColumns< OwnerId, LocalPathId, LastUpdateTime, - SchemeShardId + SchemeShardId, + IsColumnTable >; }; @@ -76,6 +78,8 @@ struct TAggregatorSchema : NIceDb::Schema { static constexpr ui64 SysParam_ScanTableLocalPathId = 4; static constexpr ui64 SysParam_ScanStartTime = 5; static constexpr ui64 SysParam_LastScanOperationId = 6; + static constexpr ui64 SysParam_IsColumnTable = 7; + static constexpr ui64 SysParam_GlobalTraversalRound = 8; }; } // NKikimr::NStat diff --git a/ydb/core/statistics/aggregator/tx_ack_timeout.cpp b/ydb/core/statistics/aggregator/tx_ack_timeout.cpp new file mode 100644 index 00000000000..6afa1bd86af --- /dev/null +++ b/ydb/core/statistics/aggregator/tx_ack_timeout.cpp @@ -0,0 +1,32 @@ +#include "aggregator_impl.h" + +namespace NKikimr::NStat { + +struct TStatisticsAggregator::TTxAckTimeout : public TTxBase { + explicit TTxAckTimeout(TSelf* self) + : TTxBase(self) + {} + + TTxType GetTxType() const override { return TXTYPE_ACK_TIMEOUT; } + + bool Execute(TTransactionContext& txc, const TActorContext&) override { + SA_LOG_D("[" << Self->TabletID() << "] TTxAckTimeout::Execute"); + return true; + } + + void Complete(const TActorContext& ctx) override { + SA_LOG_D("[" << Self->TabletID() << "] TTxAckTimeout::Complete"); + + ctx.Send(Self->SelfId(), new TEvPrivate::TEvRequestDistribution); + } +}; + +void TStatisticsAggregator::Handle(TEvPrivate::TEvAckTimeout::TPtr& ev) { + if (ev->Get()->SeqNo < KeepAliveSeqNo) { + return; + } + // timeout + Execute(new TTxAckTimeout(this), TActivationContext::AsActorContext()); +} + +} // NKikimr::NStat diff --git a/ydb/core/statistics/aggregator/tx_aggr_stat_response.cpp b/ydb/core/statistics/aggregator/tx_aggr_stat_response.cpp new file mode 100644 index 00000000000..1075c4339a0 --- /dev/null +++ b/ydb/core/statistics/aggregator/tx_aggr_stat_response.cpp @@ -0,0 +1,134 @@ +#include "aggregator_impl.h" + +#include <ydb/core/statistics/stat_service.h> + +namespace NKikimr::NStat { + +struct TStatisticsAggregator::TTxAggregateStatisticsResponse : public TTxBase { + NKikimrStat::TEvAggregateStatisticsResponse Record; + + enum class EAction : ui8 { + None, + SendReqDistribution, + SendAggregate, + }; + EAction Action = EAction::None; + + std::unique_ptr<TEvStatistics::TEvAggregateStatistics> Request; + + TTxAggregateStatisticsResponse(TSelf* self, NKikimrStat::TEvAggregateStatisticsResponse&& record) + : TTxBase(self) + , Record(std::move(record)) + {} + + TTxType GetTxType() const override { return TXTYPE_AGGR_STAT_RESPONSE; } + + bool Execute(TTransactionContext& txc, const TActorContext&) override { + SA_LOG_D("[" << Self->TabletID() << "] TTxAggregateStatisticsResponse::Execute"); + + ++Self->KeepAliveSeqNo; // cancel timeout events + + NIceDb::TNiceDb db(txc.DB); + + for (auto& column : Record.GetColumns()) { + auto tag = column.GetTag(); + for (auto& statistic : column.GetStatistics()) { + if (statistic.GetType() == NKikimr::NStat::COUNT_MIN_SKETCH) { + if (!Self->ColumnNames.contains(tag)) { + continue; + } + + auto [currentIt, emplaced] = Self->CountMinSketches.try_emplace(tag); + if (emplaced) { + currentIt->second.reset(TCountMinSketch::Create()); + } + + auto* data = statistic.GetData().Data(); + auto* sketch = reinterpret_cast<const TCountMinSketch*>(data); + *(currentIt->second) += *sketch; + } + } + } + + if (Record.FailedTabletsSize() == 0 || + Self->TraversalRound >= Self->MaxTraversalRoundCount) + { + Self->SaveStatisticsToTable(); + return true; + } + + std::unordered_map<ui32, std::vector<ui64>> nonLocalTablets; + + Self->TabletsForReqDistribution.clear(); + for (auto& tablet : Record.GetFailedTablets()) { + auto error = tablet.GetError(); + switch (error) { + case NKikimrStat::TEvAggregateStatisticsResponse::UnavailableNode: + Self->TabletsForReqDistribution.insert(tablet.GetTabletId()); + Action = EAction::SendReqDistribution; + break; + + case NKikimrStat::TEvAggregateStatisticsResponse::NonLocalTablet: + auto nodeId = tablet.GetNodeId(); + if (nodeId == 0) { + // we cannot reach this tablet + Self->TabletsForReqDistribution.insert(tablet.GetTabletId()); + Action = EAction::SendReqDistribution; + + } else if (Action != EAction::SendReqDistribution) { + nonLocalTablets[nodeId].push_back(tablet.GetTabletId()); + } + } + } + + if (Action == EAction::SendReqDistribution) { + return true; + } + + Request = std::make_unique<TEvStatistics::TEvAggregateStatistics>(); + auto& outRecord = Request->Record; + + for (auto& [nodeId, tabletIds] : nonLocalTablets) { + auto& outNode = *outRecord.AddNodes(); + outNode.SetNodeId(nodeId); + outNode.MutableTabletIds()->Reserve(tabletIds.size()); + for (auto tabletId : tabletIds) { + outNode.AddTabletIds(tabletId); + } + } + + ++Self->TraversalRound; + ++Self->GlobalTraversalRound; + Self->PersistGlobalTraversalRound(db); + outRecord.SetRound(Self->GlobalTraversalRound); + Action = EAction::SendAggregate; + + return true; + } + + void Complete(const TActorContext& ctx) override { + SA_LOG_D("[" << Self->TabletID() << "] TTxAggregateStatisticsResponse::Complete"); + + switch (Action) { + case EAction::SendReqDistribution: + ctx.Send(Self->SelfId(), new TEvPrivate::TEvRequestDistribution); + break; + + case EAction::SendAggregate: + ctx.Send(MakeStatServiceID(Self->SelfId().NodeId()), Request.release()); + ctx.Schedule(KeepAliveTimeout, new TEvPrivate::TEvAckTimeout(++Self->KeepAliveSeqNo)); + break; + + default: + break; + } + } +}; + +void TStatisticsAggregator::Handle(TEvStatistics::TEvAggregateStatisticsResponse::TPtr& ev) { + auto& record = ev->Get()->Record; + Execute(new TTxAggregateStatisticsResponse(this, std::move(record)), + TActivationContext::AsActorContext()); +} + +} // NKikimr::NStat diff --git a/ydb/core/statistics/aggregator/tx_init.cpp b/ydb/core/statistics/aggregator/tx_init.cpp index 4dbd2295c01..816c1168888 100644 --- a/ydb/core/statistics/aggregator/tx_init.cpp +++ b/ydb/core/statistics/aggregator/tx_init.cpp @@ -71,12 +71,20 @@ struct TStatisticsAggregator::TTxInit : public TTxBase { break; } case Schema::SysParam_LastScanOperationId: { - auto id = FromString<ui64>(value); - Self->LastScanOperationId = id; - SA_LOG_D("[" << Self->TabletID() << "] Loading last scan operation id: " << id); + Self->LastScanOperationId = FromString<ui64>(value); + SA_LOG_D("[" << Self->TabletID() << "] Loading last scan operation id: " << value); + break; + } + case Schema::SysParam_IsColumnTable: { + Self->IsColumnTable = FromString<bool>(value); + SA_LOG_D("[" << Self->TabletID() << "] Loading IsColumnTable: " << value); + break; + } + case Schema::SysParam_GlobalTraversalRound: { + Self->GlobalTraversalRound = FromString<ui64>(value); + SA_LOG_D("[" << Self->TabletID() << "] Loading global traversal round: " << value); break; } - default: SA_LOG_CRIT("[" << Self->TabletID() << "] Unexpected SysParam id: " << id); } @@ -152,6 +160,7 @@ struct TStatisticsAggregator::TTxInit : public TTxBase { ui64 localPathId = rowset.GetValue<Schema::ScanTables::LocalPathId>(); ui64 lastUpdateTime = rowset.GetValue<Schema::ScanTables::LastUpdateTime>(); ui64 schemeShardId = rowset.GetValue<Schema::ScanTables::SchemeShardId>(); + bool isColumnTable = rowset.GetValue<Schema::ScanTables::IsColumnTable>(); auto pathId = TPathId(ownerId, localPathId); @@ -159,6 +168,7 @@ struct TStatisticsAggregator::TTxInit : public TTxBase { scanTable.PathId = pathId; scanTable.SchemeShardId = schemeShardId; scanTable.LastUpdateTime = TInstant::MicroSeconds(lastUpdateTime); + scanTable.IsColumnTable = isColumnTable; auto [it, _] = Self->ScanTables.emplace(pathId, scanTable); Self->ScanTablesByTime.Add(&it->second); diff --git a/ydb/core/statistics/aggregator/tx_init_schema.cpp b/ydb/core/statistics/aggregator/tx_init_schema.cpp index 4c77a9b5e9c..17da2827e3c 100644 --- a/ydb/core/statistics/aggregator/tx_init_schema.cpp +++ b/ydb/core/statistics/aggregator/tx_init_schema.cpp @@ -16,6 +16,8 @@ struct TStatisticsAggregator::TTxInitSchema : public TTxBase { static constexpr NIceDb::TTableId bigTableIds[] = { Schema::BaseStats::TableId, + Schema::Statistics::TableId, + Schema::ScanTables::TableId }; for (auto id : bigTableIds) { diff --git a/ydb/core/statistics/aggregator/tx_navigate.cpp b/ydb/core/statistics/aggregator/tx_navigate.cpp index 95abd04cecb..bbc2ebdf6da 100644 --- a/ydb/core/statistics/aggregator/tx_navigate.cpp +++ b/ydb/core/statistics/aggregator/tx_navigate.cpp @@ -58,6 +58,15 @@ struct TStatisticsAggregator::TTxNavigate : public TTxBase { Self->PersistStartKey(db); } + if (Self->IsColumnTable) { + // TODO: serverless case + if (entry.DomainInfo->Params.HasHive()) { + Self->HiveId = entry.DomainInfo->Params.GetHive(); + } else { + Self->HiveId = AppData()->DomainsInfo->GetHive(); + } + } + return true; } diff --git a/ydb/core/statistics/aggregator/tx_resolve.cpp b/ydb/core/statistics/aggregator/tx_resolve.cpp index 36c9360e6f4..b0c6b2bbffb 100644 --- a/ydb/core/statistics/aggregator/tx_resolve.cpp +++ b/ydb/core/statistics/aggregator/tx_resolve.cpp @@ -1,5 +1,6 @@ #include "aggregator_impl.h" +#include <ydb/core/base/hive.h> #include <ydb/core/tx/datashard/datashard.h> namespace NKikimr::NStat { @@ -34,29 +35,43 @@ struct TStatisticsAggregator::TTxResolve : public TTxBase { return true; } - Self->ShardRanges.clear(); + auto& partitioning = entry.KeyDescription->GetPartitions(); - auto& partitioning = entry.KeyDescription->Partitioning; - for (auto& part : *partitioning) { + if (Self->IsColumnTable) { + Self->TabletsForReqDistribution.clear(); + } else { + Self->ShardRanges.clear(); + } + + for (auto& part : partitioning) { if (!part.Range) { continue; } - TRange range; - range.EndKey = part.Range->EndKeyPrefix; - range.DataShardId = part.ShardId; - Self->ShardRanges.push_back(range); + if (Self->IsColumnTable) { + Self->TabletsForReqDistribution.insert(part.ShardId); + } else { + TRange range; + range.EndKey = part.Range->EndKeyPrefix; + range.DataShardId = part.ShardId; + Self->ShardRanges.push_back(range); + } } + return true; } - void Complete(const TActorContext&) override { + void Complete(const TActorContext& ctx) override { SA_LOG_D("[" << Self->TabletID() << "] TTxResolve::Complete"); if (Cancelled) { return; } - Self->NextRange(); + if (Self->IsColumnTable) { + ctx.Send(Self->SelfId(), new TEvPrivate::TEvRequestDistribution); + } else { + Self->NextRange(); + } } }; diff --git a/ydb/core/statistics/aggregator/tx_response_tablet_distribution.cpp b/ydb/core/statistics/aggregator/tx_response_tablet_distribution.cpp new file mode 100644 index 00000000000..f37835ecd93 --- /dev/null +++ b/ydb/core/statistics/aggregator/tx_response_tablet_distribution.cpp @@ -0,0 +1,107 @@ +#include "aggregator_impl.h" + +#include <ydb/core/protos/hive.pb.h> +#include <ydb/core/statistics/stat_service.h> + +namespace NKikimr::NStat { + +struct TStatisticsAggregator::TTxResponseTabletDistribution : public TTxBase { + NKikimrHive::TEvResponseTabletDistribution Record; + + enum class EAction : ui8 { + None, + SendAggregate, + ScheduleResolve, + ScheduleReqDistribution, + }; + EAction Action = EAction::None; + + std::unique_ptr<TEvStatistics::TEvAggregateStatistics> Request; + + TTxResponseTabletDistribution(TSelf* self, NKikimrHive::TEvResponseTabletDistribution&& record) + : TTxBase(self) + , Record(std::move(record)) + {} + + TTxType GetTxType() const override { return TXTYPE_RESPONSE_TABLET_DISTRIBUTION; } + + bool Execute(TTransactionContext& txc, const TActorContext&) override { + SA_LOG_D("[" << Self->TabletID() << "] TTxResponseTabletDistribution::Execute"); + + NIceDb::TNiceDb db(txc.DB); + + Request = std::make_unique<TEvStatistics::TEvAggregateStatistics>(); + auto& outRecord = Request->Record; + + PathIdFromPathId(Self->ScanTableId.PathId, outRecord.MutablePathId()); + + bool hasTablets = false; + for (auto& inNode : Record.GetNodes()) { + if (inNode.GetNodeId() == 0) { + // these tablets are probably in Hive boot queue + if (Self->HiveRequestRound < Self->MaxHiveRequestRoundCount) { + Action = EAction::ScheduleReqDistribution; + return true; + } + continue; + } + auto& outNode = *outRecord.AddNodes(); + outNode.SetNodeId(inNode.GetNodeId()); + outNode.MutableTabletIds()->Reserve(inNode.TabletIdsSize()); + for (auto tabletId : inNode.GetTabletIds()) { + outNode.AddTabletIds(tabletId); + Self->TabletsForReqDistribution.erase(tabletId); + } + hasTablets = true; + } + + if (!Self->TabletsForReqDistribution.empty() && Self->ResolveRound < Self->MaxResolveRoundCount) { + // these tablets do not exist in Hive anymore + Action = EAction::ScheduleResolve; + return true; + } + + if (!hasTablets) { + Self->FinishScan(db); + return true; + } + + ++Self->TraversalRound; + ++Self->GlobalTraversalRound; + Self->PersistGlobalTraversalRound(db); + outRecord.SetRound(Self->GlobalTraversalRound); + Action = EAction::SendAggregate; + + return true; + } + + void Complete(const TActorContext& ctx) override { + SA_LOG_D("[" << Self->TabletID() << "] TTxResponseTabletDistribution::Complete"); + + switch (Action) { + case EAction::ScheduleResolve: + ctx.Schedule(ResolveRetryInterval, new TEvPrivate::TEvResolve()); + break; + + case EAction::ScheduleReqDistribution: + ctx.Schedule(HiveRetryInterval, new TEvPrivate::TEvRequestDistribution()); + break; + + case EAction::SendAggregate: + ctx.Send(MakeStatServiceID(Self->SelfId().NodeId()), Request.release()); + ctx.Schedule(KeepAliveTimeout, new TEvPrivate::TEvAckTimeout(++Self->KeepAliveSeqNo)); + break; + + default: + break; + } + } +}; + +void TStatisticsAggregator::Handle(TEvHive::TEvResponseTabletDistribution::TPtr& ev) { + auto& record = ev->Get()->Record; + Execute(new TTxResponseTabletDistribution(this, std::move(record)), + TActivationContext::AsActorContext()); +} + +} // NKikimr::NStat diff --git a/ydb/core/statistics/aggregator/tx_schemeshard_stats.cpp b/ydb/core/statistics/aggregator/tx_schemeshard_stats.cpp index ead6e8642fc..6dc87267e57 100644 --- a/ydb/core/statistics/aggregator/tx_schemeshard_stats.cpp +++ b/ydb/core/statistics/aggregator/tx_schemeshard_stats.cpp @@ -44,12 +44,15 @@ struct TStatisticsAggregator::TTxSchemeShardStats : public TTxBase { scanTable.PathId = pathId; scanTable.SchemeShardId = schemeShardId; scanTable.LastUpdateTime = TInstant::MicroSeconds(0); + scanTable.IsColumnTable = entry.GetIsColumnTable(); auto [it, _] = Self->ScanTables.emplace(pathId, scanTable); - Self->ScanTablesByTime.Add(&it->second); - + if (!Self->ScanTablesByTime.Has(&it->second)) { + Self->ScanTablesByTime.Add(&it->second); + } db.Table<Schema::ScanTables>().Key(pathId.OwnerId, pathId.LocalPathId).Update( NIceDb::TUpdate<Schema::ScanTables::SchemeShardId>(schemeShardId), - NIceDb::TUpdate<Schema::ScanTables::LastUpdateTime>(0)); + NIceDb::TUpdate<Schema::ScanTables::LastUpdateTime>(0), + NIceDb::TUpdate<Schema::ScanTables::IsColumnTable>(entry.GetIsColumnTable())); } } diff --git a/ydb/core/statistics/aggregator/ya.make b/ydb/core/statistics/aggregator/ya.make index 3223cd9e4d9..84ec44f3b4c 100644 --- a/ydb/core/statistics/aggregator/ya.make +++ b/ydb/core/statistics/aggregator/ya.make @@ -7,12 +7,15 @@ SRCS( aggregator_impl.cpp schema.h schema.cpp + tx_ack_timeout.cpp + tx_aggr_stat_response.cpp tx_configure.cpp tx_delete_query_response.cpp tx_init.cpp tx_init_schema.cpp tx_navigate.cpp tx_resolve.cpp + tx_response_tablet_distribution.cpp tx_save_query_response.cpp tx_scan_table.cpp tx_schedule_scan.cpp diff --git a/ydb/core/statistics/ut/ut_aggregator.cpp b/ydb/core/statistics/ut/ut_aggregator.cpp index 3afbdc9b4ca..e13ae9bddaa 100644 --- a/ydb/core/statistics/ut/ut_aggregator.cpp +++ b/ydb/core/statistics/ut/ut_aggregator.cpp @@ -2,6 +2,7 @@ #include <ydb/library/actors/testlib/test_runtime.h> +#include <ydb/core/testlib/tablet_helpers.h> #include <ydb/core/tx/scheme_cache/scheme_cache.h> #include <ydb/core/statistics/events.h> #include <ydb/core/statistics/stat_service.h> @@ -50,6 +51,44 @@ void CreateUniformTable(TTestEnv& env, const TString& databaseName, const TStrin UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString()); } +void CreateColumnStoreTable(TTestEnv& env, const TString& databaseName, const TString& tableName, + int shardCount) +{ + TTableClient client(env.GetDriver()); + auto session = client.CreateSession().GetValueSync().GetSession(); + + auto fullTableName = Sprintf("Root/%s/%s", databaseName.c_str(), tableName.c_str()); + auto result = session.ExecuteSchemeQuery(Sprintf(R"( + CREATE TABLE `%s` ( + Key Uint64 NOT NULL, + Value Uint64, + PRIMARY KEY (Key) + ) + PARTITION BY HASH(Key) + WITH ( + STORE = COLUMN, + AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = %d + ); + )", fullTableName.c_str(), shardCount)).GetValueSync(); + UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString()); + + NYdb::TValueBuilder rows; + rows.BeginList(); + for (size_t i = 0; i < 100; ++i) { + auto key = TValueBuilder().Uint64(i).Build(); + auto value = TValueBuilder().OptionalUint64(i).Build(); + rows.AddListItem(); + rows.BeginStruct(); + rows.AddMember("Key", key); + rows.AddMember("Value", value); + rows.EndStruct(); + } + rows.EndList(); + + result = client.BulkUpsert(fullTableName, rows.Build()).GetValueSync(); + UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString()); +} + void DropTable(TTestEnv& env, const TString& databaseName, const TString& tableName) { TTableClient client(env.GetDriver()); auto session = client.CreateSession().GetValueSync().GetSession(); @@ -60,7 +99,7 @@ void DropTable(TTestEnv& env, const TString& databaseName, const TString& tableN UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString()); } -void ValidateCountMin(TTestActorRuntime& runtime, TPathId pathId) { +std::shared_ptr<TCountMinSketch> ExtractCountMin(TTestActorRuntime& runtime, TPathId pathId) { auto statServiceId = NStat::MakeStatServiceID(runtime.GetNodeId(1)); NStat::TRequest req; @@ -84,9 +123,15 @@ void ValidateCountMin(TTestActorRuntime& runtime, TPathId pathId) { UNIT_ASSERT(rsp.Success); UNIT_ASSERT(stat.CountMin); + return stat.CountMin; +} + +void ValidateCountMin(TTestActorRuntime& runtime, TPathId pathId) { + auto countMin = ExtractCountMin(runtime, pathId); + for (ui32 i = 0; i < 4; ++i) { ui64 value = 4000000000000000000ull * (i + 1); - auto probe = stat.CountMin->Probe((const char *)&value, sizeof(ui64)); + auto probe = countMin->Probe((const char *)&value, sizeof(ui64)); UNIT_ASSERT_VALUES_EQUAL(probe, 1); } } @@ -133,6 +178,8 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) { ui64 tabletId = 0; auto pathId = ResolvePathId(runtime, "/Root/Database/Table", nullptr, &tabletId); + runtime.SimulateSleep(TDuration::Seconds(30)); + auto ev = std::make_unique<TEvStatistics::TEvScanTable>(); auto& record = ev->Record; PathIdFromPathId(pathId, record.MutablePathId()); @@ -298,6 +345,30 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) { ValidateCountMinAbsense(runtime, pathId); } + Y_UNIT_TEST(ScanOneColumnTable) { + TTestEnv env(1, 1); + auto init = [&] () { + CreateDatabase(env, "Database"); + CreateColumnStoreTable(env, "Database", "Table", 10); + }; + std::thread initThread(init); + + auto& runtime = *env.GetServer().GetRuntime(); + runtime.SimulateSleep(TDuration::Seconds(30)); + initThread.join(); + + ui64 tabletId = 0; + auto pathId = ResolvePathId(runtime, "/Root/Database/Table", nullptr, &tabletId); + Y_UNUSED(pathId); + + runtime.SimulateSleep(TDuration::Seconds(30)); + + auto countMin = ExtractCountMin(runtime, pathId); + + ui32 value = 1; + auto probe = countMin->Probe((const char *)&value, sizeof(value)); + UNIT_ASSERT_VALUES_EQUAL(probe, 10); + } } } // NStat diff --git a/ydb/core/statistics/ut/ya.make b/ydb/core/statistics/ut/ya.make index 1b94ec2099c..d9952d60f13 100644 --- a/ydb/core/statistics/ut/ya.make +++ b/ydb/core/statistics/ut/ya.make @@ -25,7 +25,7 @@ SRCS( ut_aggregator.cpp ut_statistics.cpp ut_save_load_stats.cpp - ut_statistics_service.cpp +# ut_statistics_service.cpp ) END() diff --git a/ydb/core/tx/columnshard/columnshard__statistics.cpp b/ydb/core/tx/columnshard/columnshard__statistics.cpp index 779af24b4e1..fdf0669a951 100644 --- a/ydb/core/tx/columnshard/columnshard__statistics.cpp +++ b/ydb/core/tx/columnshard/columnshard__statistics.cpp @@ -12,6 +12,18 @@ void TColumnShard::Handle(NStat::TEvStatistics::TEvStatisticsRequest::TPtr& ev, record.SetStatus(NKikimrStat::TEvStatisticsResponse::SUCCESS); + std::unique_ptr<TCountMinSketch> sketch(TCountMinSketch::Create()); + ui32 value = 1; + sketch->Count((const char*)&value, sizeof(value)); + TString strSketch(sketch->AsStringBuf()); + + auto* column = record.AddColumns(); + column->SetTag(1); + + auto* statistic = column->AddStatistics(); + statistic->SetType(NStat::COUNT_MIN_SKETCH); + statistic->SetData(std::move(strSketch)); + Send(ev->Sender, response.release(), 0, ev->Cookie); } diff --git a/ydb/core/tx/schemeshard/schemeshard_impl.cpp b/ydb/core/tx/schemeshard/schemeshard_impl.cpp index 429055cf330..3ae1467221f 100644 --- a/ydb/core/tx/schemeshard/schemeshard_impl.cpp +++ b/ydb/core/tx/schemeshard/schemeshard_impl.cpp @@ -7373,6 +7373,7 @@ void TSchemeShard::SendBaseStatsToSA() { entryPathId->SetLocalId(pathId.LocalPathId); entry->SetRowCount(aggregated.RowCount); entry->SetBytesSize(aggregated.DataSize); + entry->SetIsColumnTable(false); ++count; } auto columnTablesPathIds = ColumnTables.GetAllPathIds(); @@ -7385,6 +7386,7 @@ void TSchemeShard::SendBaseStatsToSA() { entryPathId->SetLocalId(pathId.LocalPathId); entry->SetRowCount(aggregated.RowCount); entry->SetBytesSize(aggregated.DataSize); + entry->SetIsColumnTable(true); ++count; } |