Register pathes for insert table (#9881)

ydb-platform · Oct 1, 2024 · b5341bd · b5341bd
1 parent ce9f20d
commit b5341bd
Show file tree

Hide file tree

Showing 8 changed files with 42 additions and 23 deletions.
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;