Register pathes for insert table (#9881)

ivanmorozov333 · web-flow · commit b5341bdbae62 · 2024-10-01T10:20:42.000+03:00
diff --git a/ydb/core/tx/columnshard/columnshard__init.cpp b/ydb/core/tx/columnshard/columnshard__init.cpp
@@ -101,18 +101,6 @@ bool TTxInit::ReadEverything(TTransactionContext& txc, const TActorContext& ctx)
     NIceDb::TNiceDb db(txc.DB);
     TBlobGroupSelector dsGroupSelector(Self->Info());
     NOlap::TDbWrapper dbTable(txc.DB, &dsGroupSelector);
-    {
-        ACFL_DEBUG("step", "TInsertTable::Load_Start");
-        TMemoryProfileGuard g("TTxInit/InsertTable");
-        auto localInsertTable = std::make_unique<NOlap::TInsertTable>();
-        if (!localInsertTable->Load(db, dbTable, TAppData::TimeProvider->Now())) {
-            ACFL_ERROR("step", "TInsertTable::Load_Fails");
-            return false;
-        }
-        ACFL_DEBUG("step", "TInsertTable::Load_Finish");
-        Self->InsertTable.swap(localInsertTable);
-    }
-
     {
         ACFL_DEBUG("step", "TTablesManager::Load_Start");
         TTablesManager tManagerLocal(Self->StoragesManager, Self->TabletID());
@@ -138,6 +126,21 @@ bool TTxInit::ReadEverything(TTransactionContext& txc, const TActorContext& ctx)
         ACFL_DEBUG("step", "TTablesManager::Load_Finish");
     }
 
+    {
+        ACFL_DEBUG("step", "TInsertTable::Load_Start");
+        TMemoryProfileGuard g("TTxInit/InsertTable");
+        auto localInsertTable = std::make_unique<NOlap::TInsertTable>();
+        for (auto&& i : Self->TablesManager.GetTables()) {
+            localInsertTable->RegisterPathInfo(i.first);
+        }
+        if (!localInsertTable->Load(db, dbTable, TAppData::TimeProvider->Now())) {
+            ACFL_ERROR("step", "TInsertTable::Load_Fails");
+            return false;
+        }
+        ACFL_DEBUG("step", "TInsertTable::Load_Finish");
+        Self->InsertTable.swap(localInsertTable);
+    }
+
     {
         ACFL_DEBUG("step", "TTxController::Load_Start");
         TMemoryProfileGuard g("TTxInit/TTxController");
diff --git a/ydb/core/tx/columnshard/columnshard_impl.cpp b/ydb/core/tx/columnshard/columnshard_impl.cpp
@@ -418,6 +418,7 @@ void TColumnShard::RunEnsureTable(const NKikimrTxColumnShard::TCreateTable& tabl
     tableVerProto.SetSchemaPresetVersionAdj(tableProto.GetSchemaPresetVersionAdj());
 
     TablesManager.AddTableVersion(pathId, version, tableVerProto, db, Tiers);
+    InsertTable->RegisterPathInfo(pathId);
 
     Counters.GetTabletCounters()->SetCounter(COUNTER_TABLES, TablesManager.GetTables().size());
     Counters.GetTabletCounters()->SetCounter(COUNTER_TABLE_PRESETS, TablesManager.GetSchemaPresets().size());
diff --git a/ydb/core/tx/columnshard/engines/insert_table/insert_table.cpp b/ydb/core/tx/columnshard/engines/insert_table/insert_table.cpp
@@ -65,7 +65,7 @@ TInsertionSummary::TCounters TInsertTable::CommitEphemeral(IDbWrapper& dbTable,
 
     AddBlobLink(data.GetBlobRange().BlobId);
     const ui64 pathId = data.GetPathId();
-    auto& pathInfo = Summary.GetPathInfo(pathId);
+    auto& pathInfo = Summary.GetPathInfoVerified(pathId);
     AFL_TRACE(NKikimrServices::TX_COLUMNSHARD)("event", "commit_insertion")("path_id", pathId)("blob_range", data.GetBlobRange().ToString());
     dbTable.Commit(data);
     pathInfo.AddCommitted(std::move(data));
diff --git a/ydb/core/tx/columnshard/engines/insert_table/insert_table.h b/ydb/core/tx/columnshard/engines/insert_table/insert_table.h
@@ -25,6 +25,10 @@ class TInsertTableAccessor {
     bool RemoveBlobLinkOnComplete(const TUnifiedBlobId& blobId);
 
 public:
+    TPathInfo& RegisterPathInfo(const ui64 pathId) {
+        return Summary.RegisterPathInfo(pathId);
+    }
+
     void ErasePath(const ui64 pathId) {
         Summary.ErasePath(pathId);
     }
@@ -64,7 +68,7 @@ class TInsertTableAccessor {
             AddBlobLink(data.GetBlobRange().BlobId);
         }
         const ui64 pathId = data.GetPathId();
-        return Summary.GetPathInfo(pathId).AddCommitted(std::move(data), load);
+        return Summary.GetPathInfoVerified(pathId).AddCommitted(std::move(data), load);
     }
     bool HasPathIdData(const ui64 pathId) const {
         return Summary.HasPathIdData(pathId);
diff --git a/ydb/core/tx/columnshard/engines/insert_table/rt_insertion.cpp b/ydb/core/tx/columnshard/engines/insert_table/rt_insertion.cpp
@@ -39,23 +39,23 @@ void TInsertionSummary::AddPriority(const TPathInfo& pathInfo) noexcept {
     }
 }
 
-NKikimr::NOlap::TPathInfo& TInsertionSummary::GetPathInfo(const ui64 pathId) {
+TPathInfo& TInsertionSummary::RegisterPathInfo(const ui64 pathId) {
     auto it = PathInfo.find(pathId);
     if (it == PathInfo.end()) {
         it = PathInfo.emplace(pathId, TPathInfo(*this, pathId)).first;
     }
     return it->second;
 }
 
-NKikimr::NOlap::TPathInfo* TInsertionSummary::GetPathInfoOptional(const ui64 pathId) {
+TPathInfo* TInsertionSummary::GetPathInfoOptional(const ui64 pathId) {
     auto it = PathInfo.find(pathId);
     if (it == PathInfo.end()) {
         return nullptr;
     }
     return &it->second;
 }
 
-const NKikimr::NOlap::TPathInfo* TInsertionSummary::GetPathInfoOptional(const ui64 pathId) const {
+const TPathInfo* TInsertionSummary::GetPathInfoOptional(const ui64 pathId) const {
     auto it = PathInfo.find(pathId);
     if (it == PathInfo.end()) {
         return nullptr;
@@ -134,7 +134,7 @@ bool TInsertionSummary::HasCommitted(const TCommittedData& data) {
     return pathInfo->HasCommitted(data);
 }
 
-const NKikimr::NOlap::TInsertedData* TInsertionSummary::AddAborted(TInsertedData&& data, const bool load /*= false*/) {
+const TInsertedData* TInsertionSummary::AddAborted(TInsertedData&& data, const bool load /*= false*/) {
     const TInsertWriteId writeId = data.GetInsertWriteId();
     Counters.Aborted.Add(data.BlobSize(), load);
     AFL_VERIFY_DEBUG(!Inserted.contains(writeId));
@@ -143,7 +143,7 @@ const NKikimr::NOlap::TInsertedData* TInsertionSummary::AddAborted(TInsertedData
     return &insertInfo.first->second;
 }
 
-std::optional<NKikimr::NOlap::TInsertedData> TInsertionSummary::ExtractInserted(const TInsertWriteId id) {
+std::optional<TInsertedData> TInsertionSummary::ExtractInserted(const TInsertWriteId id) {
     auto result = Inserted.ExtractOptional(id);
     if (result) {
         auto pathInfo = GetPathInfoOptional(result->GetPathId());
@@ -154,10 +154,10 @@ std::optional<NKikimr::NOlap::TInsertedData> TInsertionSummary::ExtractInserted(
     return result;
 }
 
-const NKikimr::NOlap::TInsertedData* TInsertionSummary::AddInserted(TInsertedData&& data, const bool load /*= false*/) {
+const TInsertedData* TInsertionSummary::AddInserted(TInsertedData&& data, const bool load /*= false*/) {
     auto* insertInfo = Inserted.AddVerified(std::move(data));
     AFL_VERIFY_DEBUG(!Aborted.contains(insertInfo->GetInsertWriteId()));
-    OnNewInserted(GetPathInfo(insertInfo->GetPathId()), insertInfo->BlobSize(), load);
+    OnNewInserted(GetPathInfoVerified(insertInfo->GetPathId()), insertInfo->BlobSize(), load);
     return insertInfo;
 }
 
diff --git a/ydb/core/tx/columnshard/engines/insert_table/rt_insertion.h b/ydb/core/tx/columnshard/engines/insert_table/rt_insertion.h
@@ -199,9 +199,19 @@ class TInsertionSummary {
     const NColumnShard::TInsertTableCounters& GetCounters() const {
         return Counters;
     }
-    NKikimr::NOlap::TPathInfo& GetPathInfo(const ui64 pathId);
+    NKikimr::NOlap::TPathInfo& RegisterPathInfo(const ui64 pathId);
     TPathInfo* GetPathInfoOptional(const ui64 pathId);
     const TPathInfo* GetPathInfoOptional(const ui64 pathId) const;
+    TPathInfo& GetPathInfoVerified(const ui64 pathId) {
+        auto* result = GetPathInfoOptional(pathId);
+        AFL_VERIFY(result);
+        return *result;
+    }
+    const TPathInfo& GetPathInfoVerified(const ui64 pathId) const {
+        auto* result = GetPathInfoOptional(pathId);
+        AFL_VERIFY(result);
+        return *result;
+    }
 
     const THashMap<ui64, TPathInfo>& GetPathInfo() const {
         return PathInfo;
diff --git a/ydb/core/tx/columnshard/engines/ut/ut_insert_table.cpp b/ydb/core/tx/columnshard/engines/ut/ut_insert_table.cpp
@@ -84,6 +84,7 @@ Y_UNIT_TEST_SUITE(TColumnEngineTestInsertTable) {
         
         // insert, not commited
         auto userData1 = std::make_shared<TUserData>(tableId, TBlobRange(blobId1), TLocalHelper::GetMetaProto(), indexSnapshot, std::nullopt);
+        insertTable.RegisterPathInfo(tableId);
         bool ok = insertTable.Insert(dbTable, TInsertedData(writeId, userData1));
         UNIT_ASSERT(ok);
 
diff --git a/ydb/core/tx/columnshard/engines/writer/indexed_blob_constructor.h b/ydb/core/tx/columnshard/engines/writer/indexed_blob_constructor.h
@@ -96,7 +96,7 @@ class TWriteAggregation {
     YDB_READONLY(ui64, Size, 0);
     YDB_READONLY(ui64, Rows, 0);
     YDB_ACCESSOR_DEF(std::vector<TWideSerializedBatch>, SplittedBlobs);
-    YDB_READONLY_DEF(TVector<TInsertWriteId>, InsertWriteIds);
+    YDB_READONLY_DEF(std::vector<TInsertWriteId>, InsertWriteIds);
     YDB_READONLY_DEF(std::shared_ptr<NOlap::IBlobsWritingAction>, BlobsAction);
     YDB_READONLY_DEF(NArrow::TSchemaSubset, SchemaSubset);
     std::shared_ptr<arrow::RecordBatch> RecordBatch;

Original file line number	Diff line number	Diff line change
`@@ -25,6 +25,10 @@ class TInsertTableAccessor {`
`25`	`25`	`bool RemoveBlobLinkOnComplete(const TUnifiedBlobId& blobId);`
`26`	`26`
`27`	`27`	`public:`
	`28`	`+ TPathInfo& RegisterPathInfo(const ui64 pathId) {`
	`29`	`+ return Summary.RegisterPathInfo(pathId);`
	`30`	`+ }`
	`31`	`+`
`28`	`32`	`void ErasePath(const ui64 pathId) {`
`29`	`33`	`Summary.ErasePath(pathId);`
`30`	`34`	`}`
`@@ -64,7 +68,7 @@ class TInsertTableAccessor {`
`64`	`68`	`AddBlobLink(data.GetBlobRange().BlobId);`
`65`	`69`	`}`
`66`	`70`	`const ui64 pathId = data.GetPathId();`
`67`		`- return Summary.GetPathInfo(pathId).AddCommitted(std::move(data), load);`
	`71`	`+ return Summary.GetPathInfoVerified(pathId).AddCommitted(std::move(data), load);`
`68`	`72`	`}`
`69`	`73`	`bool HasPathIdData(const ui64 pathId) const {`
`70`	`74`	`return Summary.HasPathIdData(pathId);`
Original file line number	Diff line number	Diff line change
`@@ -39,23 +39,23 @@ void TInsertionSummary::AddPriority(const TPathInfo& pathInfo) noexcept {`
`39`	`39`	`}`
`40`	`40`	`}`
`41`	`41`
`42`		`-NKikimr::NOlap::TPathInfo& TInsertionSummary::GetPathInfo(const ui64 pathId) {`
	`42`	`+TPathInfo& TInsertionSummary::RegisterPathInfo(const ui64 pathId) {`
`43`	`43`	`auto it = PathInfo.find(pathId);`
`44`	`44`	`if (it == PathInfo.end()) {`
`45`	`45`	`it = PathInfo.emplace(pathId, TPathInfo(*this, pathId)).first;`
`46`	`46`	`}`
`47`	`47`	`return it->second;`
`48`	`48`	`}`
`49`	`49`
`50`		`-NKikimr::NOlap::TPathInfo* TInsertionSummary::GetPathInfoOptional(const ui64 pathId) {`
	`50`	`+TPathInfo* TInsertionSummary::GetPathInfoOptional(const ui64 pathId) {`
`51`	`51`	`auto it = PathInfo.find(pathId);`
`52`	`52`	`if (it == PathInfo.end()) {`
`53`	`53`	`return nullptr;`
`54`	`54`	`}`
`55`	`55`	`return &it->second;`
`56`	`56`	`}`
`57`	`57`
`58`		`-const NKikimr::NOlap::TPathInfo* TInsertionSummary::GetPathInfoOptional(const ui64 pathId) const {`
	`58`	`+const TPathInfo* TInsertionSummary::GetPathInfoOptional(const ui64 pathId) const {`
`59`	`59`	`auto it = PathInfo.find(pathId);`
`60`	`60`	`if (it == PathInfo.end()) {`
`61`	`61`	`return nullptr;`
`@@ -134,7 +134,7 @@ bool TInsertionSummary::HasCommitted(const TCommittedData& data) {`
`134`	`134`	`return pathInfo->HasCommitted(data);`
`135`	`135`	`}`
`136`	`136`
`137`		`-const NKikimr::NOlap::TInsertedData* TInsertionSummary::AddAborted(TInsertedData&& data, const bool load /= false/) {`
	`137`	`+const TInsertedData* TInsertionSummary::AddAborted(TInsertedData&& data, const bool load /= false/) {`
`138`	`138`	`const TInsertWriteId writeId = data.GetInsertWriteId();`
`139`	`139`	`Counters.Aborted.Add(data.BlobSize(), load);`
`140`	`140`	`AFL_VERIFY_DEBUG(!Inserted.contains(writeId));`
`@@ -143,7 +143,7 @@ const NKikimr::NOlap::TInsertedData* TInsertionSummary::AddAborted(TInsertedData`
`143`	`143`	`return &insertInfo.first->second;`
`144`	`144`	`}`
`145`	`145`
`146`		`-std::optional<NKikimr::NOlap::TInsertedData> TInsertionSummary::ExtractInserted(const TInsertWriteId id) {`
	`146`	`+std::optional<TInsertedData> TInsertionSummary::ExtractInserted(const TInsertWriteId id) {`
`147`	`147`	`auto result = Inserted.ExtractOptional(id);`
`148`	`148`	`if (result) {`
`149`	`149`	`auto pathInfo = GetPathInfoOptional(result->GetPathId());`
`@@ -154,10 +154,10 @@ std::optional<NKikimr::NOlap::TInsertedData> TInsertionSummary::ExtractInserted(`
`154`	`154`	`return result;`
`155`	`155`	`}`
`156`	`156`
`157`		`-const NKikimr::NOlap::TInsertedData* TInsertionSummary::AddInserted(TInsertedData&& data, const bool load /= false/) {`
	`157`	`+const TInsertedData* TInsertionSummary::AddInserted(TInsertedData&& data, const bool load /= false/) {`
`158`	`158`	`auto* insertInfo = Inserted.AddVerified(std::move(data));`
`159`	`159`	`AFL_VERIFY_DEBUG(!Aborted.contains(insertInfo->GetInsertWriteId()));`
`160`		`- OnNewInserted(GetPathInfo(insertInfo->GetPathId()), insertInfo->BlobSize(), load);`
	`160`	`+ OnNewInserted(GetPathInfoVerified(insertInfo->GetPathId()), insertInfo->BlobSize(), load);`
`161`	`161`	`return insertInfo;`
`162`	`162`	`}`
`163`	`163`