Refactoring: DataShard handling indexes (#6425)

Co-authored-by: Ilnaz Nizametdinov <[email protected]>

Author: MBkkt

ydb/core/tx/datashard/build_index.cpp

@@ -696,15 +696,15 @@ void TDataShard::HandleSafe(TEvDataShard::TEvBuildIndexCreateRequest::TPtr& ev,
     TUserTable::TCPtr userTable = GetUserTables().at(tableId.PathId.LocalPathId);
 
 
-    if (BuildIndexManager.Contains(buildIndexId)) {
-        TBuildIndexRecord recCard = BuildIndexManager.Get(buildIndexId);
-        if (recCard.SeqNo == seqNo) {
+    if (const auto* recCard = BuildIndexManager.Get(buildIndexId)) {
+        if (recCard->SeqNo == seqNo) {
             // do no start one more scan
             ctx.Send(ev->Sender, std::move(response));
             return;
         }
 
-        CancelScan(userTable->LocalTid, recCard.ScanId);
+        CancelScan(userTable->LocalTid, recCard->ScanId);
+        BuildIndexManager.Drop(buildIndexId);
     }
 
 

ydb/core/tx/datashard/build_index.h

@@ -33,26 +33,32 @@ struct TBuildIndexRecord {
 
 class TBuildIndexManager {
 public:
-    bool Contains(ui64 id) {
-        return Records.contains(id);
+    const TBuildIndexRecord* Get(ui64 id) const {
+        Y_ABORT_UNLESS(id != 0);
+        if (BuildIndexId == id) {
+            return &Record;
+        }
+        Y_ABORT_UNLESS(BuildIndexId == 0);
+        return nullptr;
     }
 
     void Set(ui64 id, TBuildIndexRecord record) {
-        Records.emplace(id, record);
-    }
-
-    TBuildIndexRecord Get(ui64 id) const {
-        return Records.at(id);
+        Y_ABORT_UNLESS(id != 0);
+        Y_ABORT_UNLESS(BuildIndexId == 0);
+        BuildIndexId = id;
+        Record = record;
     }
 
     void Drop(ui64 id) {
-        Records.erase(id);
+        Y_ABORT_UNLESS(Get(id) == &Record);
+        BuildIndexId = 0;
+        Record = {};
     }
 
 private:
-    using TBuildIndexIdToScanIdMap = TMap<ui64, TBuildIndexRecord>;
-
-    TBuildIndexIdToScanIdMap Records;
+    // Only single shard scan for build index possible now
+    ui64 BuildIndexId = 0;
+    TBuildIndexRecord Record;
 };
 
 }

ydb/core/tx/datashard/change_collector_async_index.cpp

@@ -105,7 +105,7 @@ bool TAsyncIndexChangeCollector::Collect(const TTableId& tableId, ERowOp rop,
     const auto tagToPos = MakeTagToPos(tagsToSelect, [](const auto tag) { return tag; });
     const auto updatedTagToPos = MakeTagToPos(updates, [](const TUpdateOp& op) { return op.Tag; });
 
-    userTable->ForEachAsyncIndex([&] (const auto& pathId, const auto& index) {
+    userTable->ForEachAsyncIndex([&](const auto& pathId, const TUserTable::TTableIndex& index) {
         if (generateDeletions) {
             bool needDeletion = rop == ERowOp::Erase || rop == ERowOp::Reset;
 
@@ -198,7 +198,7 @@ auto TAsyncIndexChangeCollector::CacheTags(const TTableId& tableId) const {
 
     TCachedTagsBuilder builder;
 
-    userTable->ForEachAsyncIndex([&] (const auto&, const auto& index) {
+    userTable->ForEachAsyncIndex([&](const auto&, const TUserTable::TTableIndex& index) {
         builder.AddIndexTags(index.KeyColumnIds);
         builder.AddDataTags(index.DataColumnIds);
     });

ydb/core/tx/datashard/change_record_cdc_serializer.cpp

@@ -370,6 +370,9 @@ class TDynamoDBStreamsJsonSerializer: public TJsonSerializer {
             } else if (name.StartsWith("__Hash_")) {
                 bool indexed = false;
                 for (const auto& [_, index] : schema->Indexes) {
+                    if (index.Type != TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {
+                        continue;
+                    }
                     Y_ABORT_UNLESS(index.KeyColumnIds.size() >= 1);
                     if (index.KeyColumnIds.at(0) == tag) {
                         indexed = true;

ydb/core/tx/datashard/change_sender.cpp

@@ -275,7 +275,7 @@ class TChangeSender: public TActor<TChangeSender> {
         for (const auto& [tableId, tableInfo] : self->GetUserTables()) {
             const auto fullTableId = TTableId(self->GetPathOwnerId(), tableId);
 
-            tableInfo->ForEachAsyncIndex([&] (const auto& indexPathId, const auto&) {
+            tableInfo->ForEachAsyncIndex([&](const auto& indexPathId, const auto&) {
                 AddChangeSender(indexPathId, fullTableId, ESenderType::AsyncIndex);
             });
 

ydb/core/tx/datashard/datashard.cpp

@@ -1825,9 +1825,7 @@ TUserTable::TPtr TDataShard::MoveUserIndex(TOperation::TPtr op, const NKikimrTxD
         node.key() = remapNewId;
         auto it = newTableInfo->Indexes.insert(std::move(node)).position;
 
-        Y_ABORT_UNLESS(move.GetReMapIndex().HasDstName());
-        indexDesc.SetName(dstIndexName);
-        it->second.Name = dstIndexName;
+        it->second.Rename(indexDesc, dstIndexName);
     }
 
     newTableInfo->SetSchema(schema);

ydb/core/tx/datashard/datashard_impl.h

@@ -1958,7 +1958,6 @@ class TDataShard
     }
 
     TBuildIndexManager& GetBuildIndexManager() { return BuildIndexManager; }
-    const TBuildIndexManager& GetBuildIndexManager() const { return BuildIndexManager; }
 
     // Returns true when datashard is working in mvcc mode
     bool IsMvccEnabled() const;

ydb/core/tx/datashard/datashard_user_table.h

@@ -254,7 +254,6 @@ struct TUserTable : public TThrRefBase {
         using EType = NKikimrSchemeOp::EIndexType;
         using EState = NKikimrSchemeOp::EIndexState;
 
-        TString Name;
         EType Type;
         EState State;
         TVector<ui32> KeyColumnIds;
@@ -263,10 +262,12 @@ struct TUserTable : public TThrRefBase {
         TTableIndex() = default;
 
         TTableIndex(const NKikimrSchemeOp::TIndexDescription& indexDesc, const TMap<ui32, TUserColumn>& columns)
-            : Name(indexDesc.GetName())
-            , Type(indexDesc.GetType())
+            : Type(indexDesc.GetType())
             , State(indexDesc.GetState())
         {
+            if (Type != EType::EIndexTypeGlobalAsync) {
+                return;
+            }
             THashMap<TStringBuf, ui32> nameToId;
             for (const auto& [id, column] : columns) {
                 Y_DEBUG_ABORT_UNLESS(!nameToId.contains(column.Name));
@@ -285,6 +286,10 @@ struct TUserTable : public TThrRefBase {
             fillColumnIds(indexDesc.GetKeyColumnNames(),  KeyColumnIds);
             fillColumnIds(indexDesc.GetDataColumnNames(), DataColumnIds);
         }
+
+        static void Rename(NKikimrSchemeOp::TIndexDescription& indexDesc, const TString& newName) {
+            indexDesc.SetName(newName);
+        }
     };
 
     struct TCdcStream {

ydb/core/tx/datashard/drop_table_unit.cpp

@@ -80,7 +80,7 @@ EExecutionStatus TDropTableUnit::Execute(TOperation::TPtr op,
     auto it = DataShard.GetUserTables().find(tableId);
     Y_ABORT_UNLESS(it != DataShard.GetUserTables().end());
     {
-        it->second->ForEachAsyncIndex([&] (const auto& indexPathId, const auto&) {
+        it->second->ForEachAsyncIndex([&](const auto& indexPathId, const auto&) {
             RemoveSenders.emplace_back(new TEvChangeExchange::TEvRemoveSender(indexPathId));
         });
         for (const auto& [streamPathId, _] : it->second->CdcStreams) {

ydb/core/tx/datashard/finalize_build_index_unit.cpp

@@ -46,7 +46,7 @@ class TFinalizeBuildIndexUnit : public TExecutionUnit {
 
             const auto& userTables = DataShard.GetUserTables();
             Y_ABORT_UNLESS(userTables.contains(pathId.LocalPathId));
-            userTables.at(pathId.LocalPathId)->ForAsyncIndex(indexPathId, [&] (const auto&) {
+            userTables.at(pathId.LocalPathId)->ForAsyncIndex(indexPathId, [&](const auto&) {
                 RemoveSender.Reset(new TEvChangeExchange::TEvRemoveSender(indexPathId));
             });
 
@@ -66,9 +66,8 @@ class TFinalizeBuildIndexUnit : public TExecutionUnit {
         ui64 txId = params.GetSnapshotTxId();
         Y_ABORT_UNLESS(step != 0);
 
-        if (DataShard.GetBuildIndexManager().Contains(params.GetBuildIndexId())) {
-            auto record = DataShard.GetBuildIndexManager().Get(params.GetBuildIndexId());
-            DataShard.CancelScan(tableInfo->LocalTid, record.ScanId);
+        if (const auto* record = DataShard.GetBuildIndexManager().Get(params.GetBuildIndexId())) {
+            DataShard.CancelScan(tableInfo->LocalTid, record->ScanId);
             DataShard.GetBuildIndexManager().Drop(params.GetBuildIndexId());
         }