Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Remap path #15991

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Remap path #15991

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ydb/core/protos/tx_columnshard.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,4 +348,5 @@ message TInternalOperationData {
optional uint32 ModificationType = 2;
optional uint64 PathId = 3;
optional bool WritePortions = 4;
//optional uint64 LocalPathId = 5;
}
1 change: 1 addition & 0 deletions ydb/core/sys_view/common/schema.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,6 +290,7 @@ class TSystemViewResolver : public ISystemViewResolver {
RegisterColumnTableSystemView<Schema::PrimaryIndexPortionStats>(TablePrimaryIndexPortionStatsName);
RegisterColumnTableSystemView<Schema::PrimaryIndexGranuleStats>(TablePrimaryIndexGranuleStatsName);
RegisterColumnTableSystemView<Schema::PrimaryIndexOptimizerStats>(TablePrimaryIndexOptimizerStatsName);
RegisterColumnTableSystemView<Schema::TablePathIdMapping>(TablePathIdMappingName);

RegisterSystemView<Schema::TopPartitions>(TopPartitionsByCpu1MinuteName);
RegisterSystemView<Schema::TopPartitions>(TopPartitionsByCpu1HourName);
Expand Down
9 changes: 9 additions & 0 deletions ydb/core/sys_view/common/schema.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ constexpr TStringBuf TablePrimaryIndexStatsName = "primary_index_stats";
constexpr TStringBuf TablePrimaryIndexPortionStatsName = "primary_index_portion_stats";
constexpr TStringBuf TablePrimaryIndexGranuleStatsName = "primary_index_granule_stats";
constexpr TStringBuf TablePrimaryIndexOptimizerStatsName = "primary_index_optimizer_stats";
constexpr TStringBuf TablePathIdMappingName = "table_path_id_mapping";

constexpr TStringBuf TopPartitionsByCpu1MinuteName = "top_partitions_one_minute";
constexpr TStringBuf TopPartitionsByCpu1HourName = "top_partitions_one_hour";
Expand Down Expand Up @@ -797,6 +798,14 @@ struct Schema : NIceDb::Schema {
IndexSize,
FollowerId>;
};
struct TablePathIdMapping: Table<23> {
struct InternalPathId: Column<1, NScheme::NTypeIds::Uint64> {};
struct TabletId: Column<2, NScheme::NTypeIds::Uint64> {};
struct LocalPathId: Column<3, NScheme::NTypeIds::Uint64> {};

using TKey = TableKey<InternalPathId, TabletId>;
using TColumns = TableColumns<InternalPathId, TabletId, LocalPathId>;
};
};

bool MaybeSystemViewPath(const TVector<TString>& path);
Expand Down
10 changes: 5 additions & 5 deletions ydb/core/tx/columnshard/blobs_action/transaction/tx_blobs_written.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ bool TTxBlobsWritingFinished::DoExecute(TTransactionContext& txc, const TActorCo
if (!operationIds.emplace(operation->GetWriteId()).second) {
continue;
}
AFL_VERIFY(Self->TablesManager.IsReadyForFinishWrite(writeMeta.GetTableId(), minReadSnapshot));
AFL_VERIFY(Self->TablesManager.IsReadyForFinishWrite(writeMeta.GetPathId(), minReadSnapshot));
Y_ABORT_UNLESS(operation->GetStatus() == EOperationStatus::Started);
if (operation->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
operation->OnWriteFinish(txc, {}, true);
Expand All @@ -75,7 +75,7 @@ bool TTxBlobsWritingFinished::DoExecute(TTransactionContext& txc, const TActorCo
lock.SetDataShard(Self->TabletID());
lock.SetGeneration(info.GetGeneration());
lock.SetCounter(info.GetInternalGenerationCounter());
lock.SetPathId(writeMeta.GetTableId().GetRawValue());
lock.SetPathId(writeMeta.GetPathId().GetInternalPathId().GetRawValue());
auto ev = NEvents::TDataEvents::TEvWriteResult::BuildCompleted(Self->TabletID(), operation->GetLockId(), lock);
Results.emplace_back(std::move(ev), writeMeta.GetSource(), operation->GetCookie());
}
Expand Down Expand Up @@ -103,11 +103,11 @@ void TTxBlobsWritingFinished::DoComplete(const TActorContext& ctx) {
continue;
}
auto op = Self->GetOperationsManager().GetOperationVerified((TOperationWriteId)writeMeta.GetWriteId());
pathIds.emplace(op->GetPathId());
pathIds.emplace(op->GetPathId().GetInternalPathId());
if (op->GetBehaviour() == EOperationBehaviour::WriteWithLock || op->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
if (op->GetBehaviour() != EOperationBehaviour::NoTxWrite || Self->GetOperationsManager().HasReadLocks(writeMeta.GetTableId())) {
if (op->GetBehaviour() != EOperationBehaviour::NoTxWrite || Self->GetOperationsManager().HasReadLocks(writeMeta.GetPathId().GetInternalPathId())) {
auto evWrite = std::make_shared<NOlap::NTxInteractions::TEvWriteWriter>(
writeMeta.GetTableId(), writeResult.GetPKBatchVerified(), Self->GetIndexOptional()->GetVersionedIndex().GetPrimaryKey());
writeMeta.GetPathId().GetInternalPathId(), writeResult.GetPKBatchVerified(), Self->GetIndexOptional()->GetVersionedIndex().GetPrimaryKey());
Self->GetOperationsManager().AddEventForLock(*Self, op->GetLockId(), evWrite);
}
}
Expand Down
10 changes: 5 additions & 5 deletions ydb/core/tx/columnshard/blobs_action/transaction/tx_write.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ bool TTxWrite::DoExecute(TTransactionContext& txc, const TActorContext&) {
const auto minReadSnapshot = Self->GetMinReadSnapshot();
for (auto&& aggr : buffer.GetAggregations()) {
const auto& writeMeta = aggr->GetWriteMeta();
Y_ABORT_UNLESS(Self->TablesManager.IsReadyForFinishWrite(writeMeta.GetTableId(), minReadSnapshot));
Y_ABORT_UNLESS(Self->TablesManager.IsReadyForFinishWrite(writeMeta.GetPathId(), minReadSnapshot));
txc.DB.NoMoreReadsForTx();
TWriteOperation::TPtr operation;
if (writeMeta.HasLongTxId()) {
Expand Down Expand Up @@ -105,14 +105,14 @@ bool TTxWrite::DoExecute(TTransactionContext& txc, const TActorContext&) {
lock.SetDataShard(Self->TabletID());
lock.SetGeneration(info.GetGeneration());
lock.SetCounter(info.GetInternalGenerationCounter());
lock.SetPathId(writeMeta.GetTableId().GetRawValue());
lock.SetPathId(writeMeta.GetPathId().GetInternalPathId().GetRawValue());
auto ev = NEvents::TDataEvents::TEvWriteResult::BuildCompleted(Self->TabletID(), operation->GetLockId(), lock);
Results.emplace_back(std::move(ev), writeMeta.GetSource(), operation->GetCookie());
}
} else {
Y_ABORT_UNLESS(aggr->GetInsertWriteIds().size() == 1);
auto ev = std::make_unique<TEvColumnShard::TEvWriteResult>(
Self->TabletID(), writeMeta, (ui64)aggr->GetInsertWriteIds().front(), NKikimrTxColumnShard::EResultStatus::SUCCESS);
Self->TabletID(), writeMeta.GetPathId().GetLocalPathId(), writeMeta.GetDedupId(), (ui64)aggr->GetInsertWriteIds().front(), NKikimrTxColumnShard::EResultStatus::SUCCESS);
Results.emplace_back(std::move(ev), writeMeta.GetSource(), 0);
}
}
Expand Down Expand Up @@ -141,8 +141,8 @@ void TTxWrite::DoComplete(const TActorContext& ctx) {
if (!writeMeta.HasLongTxId()) {
auto op = Self->GetOperationsManager().GetOperationVerified((TOperationWriteId)writeMeta.GetWriteId());
if (op->GetBehaviour() == EOperationBehaviour::WriteWithLock || op->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
if (op->GetBehaviour() != EOperationBehaviour::NoTxWrite || Self->GetOperationsManager().HasReadLocks(writeMeta.GetTableId())) {
auto evWrite = std::make_shared<NOlap::NTxInteractions::TEvWriteWriter>(writeMeta.GetTableId(),
if (op->GetBehaviour() != EOperationBehaviour::NoTxWrite || Self->GetOperationsManager().HasReadLocks(writeMeta.GetPathId().GetInternalPathId())) {
auto evWrite = std::make_shared<NOlap::NTxInteractions::TEvWriteWriter>(writeMeta.GetPathId().GetInternalPathId(),
buffer.GetAggregations()[i]->GetRecordBatch(), Self->GetIndexOptional()->GetVersionedIndex().GetPrimaryKey());
Self->GetOperationsManager().AddEventForLock(*Self, op->GetLockId(), evWrite);
}
Expand Down
42 changes: 21 additions & 21 deletions ydb/core/tx/columnshard/columnshard.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,29 +406,29 @@ void TColumnShard::FillOlapStats(const TActorContext& ctx, std::unique_ptr<TEvDa
}

void TColumnShard::FillColumnTableStats(const TActorContext& ctx, std::unique_ptr<TEvDataShard::TEvPeriodicTableStats>& ev) {
auto tables = TablesManager.GetTables();
TTableStatsBuilder tableStatsBuilder(Counters, Executor());

LOG_S_DEBUG("There are stats for " << tables.size() << " tables");
for (const auto& [pathId, _] : tables) {
auto* periodicTableStats = ev->Record.AddTables();
periodicTableStats->SetDatashardId(TabletID());
periodicTableStats->SetTableLocalId(pathId.GetRawValue());

periodicTableStats->SetShardState(2); // NKikimrTxDataShard.EDatashardState.Ready
periodicTableStats->SetGeneration(Executor()->Generation());
periodicTableStats->SetRound(StatsReportRound++);
periodicTableStats->SetNodeId(ctx.SelfID.NodeId());
periodicTableStats->SetStartTime(StartTime().MilliSeconds());

if (auto* resourceMetrics = Executor()->GetResourceMetrics()) {
resourceMetrics->Fill(*periodicTableStats->MutableTabletMetrics());
LOG_S_DEBUG("There are stats for " << TablesManager.GetTableCount() << " tables");
TablesManager.ForEachPathId(
[&](const TInternalPathId pathId) {
auto* periodicTableStats = ev->Record.AddTables();
periodicTableStats->SetDatashardId(TabletID());
periodicTableStats->SetTableLocalId(pathId.GetRawValue());

periodicTableStats->SetShardState(2); // NKikimrTxDataShard.EDatashardState.Ready
periodicTableStats->SetGeneration(Executor()->Generation());
periodicTableStats->SetRound(StatsReportRound++);
periodicTableStats->SetNodeId(ctx.SelfID.NodeId());
periodicTableStats->SetStartTime(StartTime().MilliSeconds());

if (auto* resourceMetrics = Executor()->GetResourceMetrics()) {
resourceMetrics->Fill(*periodicTableStats->MutableTabletMetrics());
}

tableStatsBuilder.FillTableStats(pathId, *(periodicTableStats->MutableTableStats()));

LOG_S_TRACE("Add stats for table, tableLocalID=" << pathId);
}

tableStatsBuilder.FillTableStats(pathId, *(periodicTableStats->MutableTableStats()));

LOG_S_TRACE("Add stats for table, tableLocalID=" << pathId);
}
);
}

void TColumnShard::SendPeriodicStats() {
Expand Down
11 changes: 3 additions & 8 deletions ydb/core/tx/columnshard/columnshard.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,17 +270,12 @@ namespace TEvColumnShard {
struct TEvWriteResult : public TEventPB<TEvWriteResult, NKikimrTxColumnShard::TEvWriteResult, TEvColumnShard::EvWriteResult> {
TEvWriteResult() = default;

TEvWriteResult(ui64 origin, const NEvWrite::TWriteMeta& writeMeta, ui32 status)
: TEvWriteResult(origin, writeMeta, writeMeta.GetWriteId(), status)
{
}

TEvWriteResult(ui64 origin, const NEvWrite::TWriteMeta& writeMeta, const i64 writeId, ui32 status) {
TEvWriteResult(ui64 origin, const NColumnShard::TLocalPathId& localPathId, TString dedupId, const i64 writeId, ui32 status) {
Record.SetOrigin(origin);
Record.SetTxInitiator(0);
Record.SetWriteId(writeId);
Record.SetTableId(writeMeta.GetTableId().GetRawValue());
Record.SetDedupId(writeMeta.GetDedupId());
Record.SetTableId(localPathId.GetRawValue());
Record.SetDedupId(dedupId);
Record.SetStatus(status);
}

Expand Down
24 changes: 22 additions & 2 deletions ydb/core/tx/columnshard/columnshard__scan.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
#include "columnshard_impl.h"
#include "engines/reader/transaction/tx_scan.h"
#include "engines/reader/transaction/tx_internal_scan.h"
#include "ydb/core/kqp/compute_actor/kqp_compute_events.h"

#include <ydb/core/protos/kqp.pb.h>
#include <ydb/core/base/appdata_fwd.h>
Expand Down Expand Up @@ -30,10 +31,29 @@ void TColumnShard::Handle(TEvDataShard::TEvKqpScan::TPtr& ev, const TActorContex
return;
}

Counters.GetColumnTablesCounters()->GetPathIdCounter(TInternalPathId::FromRawValue(record.GetLocalPathId()))->OnReadEvent();
const auto localPathId = TLocalPathId::FromRawValue(record.GetLocalPathId());
const auto internalPathId = TablesManager.ResolveInternalPathId(localPathId);
if (!internalPathId) {
//TODO FIXME
const auto& request = ev->Get()->Record;
const TString table = request.GetTablePath();
const ui32 scanGen = request.GetGeneration();
const auto scanComputeActor = ev->Sender;

auto ev = MakeHolder<NKqp::TEvKqpCompute::TEvScanError>(scanGen, TabletID());
ev->Record.SetStatus(Ydb::StatusIds::BAD_REQUEST);
auto issue = NYql::YqlIssue({}, NYql::TIssuesIds::KIKIMR_BAD_REQUEST,
TStringBuilder() << "table not found");
NYql::IssueToMessage(issue, ev->Record.MutableIssues()->Add());

ctx.Send(scanComputeActor, ev.Release());
return;
}

Counters.GetColumnTablesCounters()->GetPathIdCounter(*internalPathId)->OnReadEvent();
ScanTxInFlight.insert({txId, TAppData::TimeProvider->Now()});
Counters.GetTabletCounters()->SetCounter(COUNTER_SCAN_IN_FLY, ScanTxInFlight.size());
Execute(new NOlap::NReader::TTxScan(this, ev), ctx);
Execute(new NOlap::NReader::TTxScan(this, ev, *internalPathId), ctx);
}

void TColumnShard::Handle(TEvColumnShard::TEvInternalScan::TPtr& ev, const TActorContext& ctx) {
Expand Down
10 changes: 9 additions & 1 deletion ydb/core/tx/columnshard/columnshard__statistics.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,10 +285,18 @@ void TColumnShard::Handle(NStat::TEvStatistics::TEvStatisticsRequest::TPtr& ev,
Send(ev->Sender, response.release(), 0, ev->Cookie);
return;
}
auto internalPathId = TablesManager.ResolveInternalPathId(TLocalPathId::FromRawValue(record.GetTable().GetPathId().GetLocalId()));
if (!internalPathId) {
AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("error", "Statistic request for unknown GetLocalPathId()");
respRecord.SetStatus(NKikimrStat::TEvStatisticsResponse::STATUS_ERROR);

Send(ev->Sender, response.release(), 0, ev->Cookie);
return;
}

AFL_VERIFY(HasIndex());
auto index = GetIndexAs<NOlap::TColumnEngineForLogs>();
auto spg = index.GetGranuleOptional(TInternalPathId::FromRawValue(record.GetTable().GetPathId().GetLocalId()));
auto spg = index.GetGranuleOptional(*internalPathId);
AFL_VERIFY(spg);

std::set<ui32> columnTagsRequested;
Expand Down
Loading
Loading