Add index state to DataShard (#6210)

MBkkt · web-flow · commit bfbba8419146 · 2024-07-03T19:25:08.000+03:00
diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
@@ -1041,9 +1041,16 @@ message TBuildIndexOutcome {
         optional NKikimrProto.TPathID IndexPathId = 1;
     }
 
+    message TApply {
+        // Path id of the index whose creation was done
+        optional NKikimrProto.TPathID IndexPathId = 1;
+    }
+
+    reserved 1; // Apply type was changed from Empty to TApply
+
     oneof Result {
-        google.protobuf.Empty Apply = 1;
         TCancel Cancel = 2;
+        TApply Apply = 3;
     }
 }
 
diff --git a/ydb/core/tx/datashard/change_collector_async_index.cpp b/ydb/core/tx/datashard/change_collector_async_index.cpp
@@ -106,7 +106,7 @@ bool TAsyncIndexChangeCollector::Collect(const TTableId& tableId, ERowOp rop,
     const auto updatedTagToPos = MakeTagToPos(updates, [](const TUpdateOp& op) { return op.Tag; });
 
     for (const auto& [pathId, index] : userTable->Indexes) {
-        if (index.Type != TUserTable::TTableIndex::EIndexType::EIndexTypeGlobalAsync) {
+        if (index.Type != TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {
             continue;
         }
 
@@ -203,7 +203,7 @@ auto TAsyncIndexChangeCollector::CacheTags(const TTableId& tableId) const {
     TCachedTagsBuilder builder;
 
     for (const auto& [_, index] : userTable->Indexes) {
-        if (index.Type != TUserTable::TTableIndex::EIndexType::EIndexTypeGlobalAsync) {
+        if (index.Type != TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {
             continue;
         }
 
diff --git a/ydb/core/tx/datashard/change_sender.cpp b/ydb/core/tx/datashard/change_sender.cpp
@@ -276,7 +276,7 @@ class TChangeSender: public TActor<TChangeSender> {
             const auto fullTableId = TTableId(self->GetPathOwnerId(), tableId);
 
             for (const auto& [indexPathId, indexInfo] : tableInfo->Indexes) {
-                if (indexInfo.Type != TUserTable::TTableIndex::EIndexType::EIndexTypeGlobalAsync) {
+                if (indexInfo.Type != TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {
                     continue;
                 }
 
diff --git a/ydb/core/tx/datashard/datashard.cpp b/ydb/core/tx/datashard/datashard.cpp
@@ -1564,6 +1564,21 @@ TUserTable::TPtr TDataShard::AlterTableAddIndex(
     return tableInfo;
 }
 
+TUserTable::TPtr TDataShard::AlterTableSwitchIndexState(
+    const TActorContext& ctx, TTransactionContext& txc,
+    const TPathId& pathId, ui64 tableSchemaVersion,
+    const TPathId& streamPathId, NKikimrSchemeOp::EIndexState state)
+{
+    auto tableInfo = AlterTableSchemaVersion(ctx, txc, pathId, tableSchemaVersion, false);
+    tableInfo->SwitchIndexState(streamPathId, state);
+
+    // This isn't really necessary now, because no one rely on index state
+    NIceDb::TNiceDb db(txc.DB);
+    PersistUserTable(db, pathId.LocalPathId, *tableInfo);
+
+    return tableInfo;
+}
+
 TUserTable::TPtr TDataShard::AlterTableDropIndex(
     const TActorContext& ctx, TTransactionContext& txc,
     const TPathId& pathId, ui64 tableSchemaVersion,
diff --git a/ydb/core/tx/datashard/datashard_impl.h b/ydb/core/tx/datashard/datashard_impl.h
@@ -1857,6 +1857,11 @@ class TDataShard
         const TPathId& pathId, ui64 tableSchemaVersion,
         const NKikimrSchemeOp::TIndexDescription& indexDesc);
 
+    TUserTable::TPtr AlterTableSwitchIndexState(
+        const TActorContext& ctx, TTransactionContext& txc,
+        const TPathId& pathId, ui64 tableSchemaVersion,
+        const TPathId& streamPathId, NKikimrSchemeOp::EIndexState state);
+
     TUserTable::TPtr AlterTableDropIndex(
         const TActorContext& ctx, TTransactionContext& txc,
         const TPathId& pathId, ui64 tableSchemaVersion,
diff --git a/ydb/core/tx/datashard/datashard_user_table.cpp b/ydb/core/tx/datashard/datashard_user_table.cpp
@@ -72,7 +72,7 @@ void TUserTable::AddIndex(const NKikimrSchemeOp::TIndexDescription& indexDesc) {
     }
 
     Indexes.emplace(addIndexPathId, TTableIndex(indexDesc, Columns));
-    AsyncIndexCount += ui32(indexDesc.GetType() == TTableIndex::EIndexType::EIndexTypeGlobalAsync);
+    AsyncIndexCount += ui32(indexDesc.GetType() == TTableIndex::EType::EIndexTypeGlobalAsync);
 
     NKikimrSchemeOp::TTableDescription schema;
     GetSchema(schema);
@@ -81,13 +81,37 @@ void TUserTable::AddIndex(const NKikimrSchemeOp::TIndexDescription& indexDesc) {
     SetSchema(schema);
 }
 
+void TUserTable::SwitchIndexState(const TPathId& indexPathId, TTableIndex::EState state) {
+    auto it = Indexes.find(indexPathId);
+    if (it == Indexes.end()) {
+        return;
+    }
+
+    it->second.State = state;
+
+    // This isn't really necessary now, because no one rely on index state
+    NKikimrSchemeOp::TTableDescription schema;
+    GetSchema(schema);
+
+    auto& indexes = *schema.MutableTableIndexes();
+    for (auto it = indexes.begin(); it != indexes.end(); ++it) {
+        if (indexPathId == TPathId(it->GetPathOwnerId(), it->GetLocalPathId())) {
+            it->SetState(state);
+            SetSchema(schema);
+            return;
+        }
+    }
+
+    Y_ABORT("unreachable");
+}
+
 void TUserTable::DropIndex(const TPathId& indexPathId) {
     auto it = Indexes.find(indexPathId);
     if (it == Indexes.end()) {
         return;
     }
 
-    AsyncIndexCount -= ui32(it->second.Type == TTableIndex::EIndexType::EIndexTypeGlobalAsync);
+    AsyncIndexCount -= ui32(it->second.Type == TTableIndex::EType::EIndexTypeGlobalAsync);
     Indexes.erase(it);
 
     NKikimrSchemeOp::TTableDescription schema;
@@ -294,7 +318,7 @@ void TUserTable::ParseProto(const NKikimrSchemeOp::TTableDescription& descr)
     for (const auto& indexDesc : descr.GetTableIndexes()) {
         Y_ABORT_UNLESS(indexDesc.HasPathOwnerId() && indexDesc.HasLocalPathId());
         Indexes.emplace(TPathId(indexDesc.GetPathOwnerId(), indexDesc.GetLocalPathId()), TTableIndex(indexDesc, Columns));
-        AsyncIndexCount += ui32(indexDesc.GetType() == TTableIndex::EIndexType::EIndexTypeGlobalAsync);
+        AsyncIndexCount += ui32(indexDesc.GetType() == TTableIndex::EType::EIndexTypeGlobalAsync);
     }
 
     for (const auto& streamDesc : descr.GetCdcStreams()) {
diff --git a/ydb/core/tx/datashard/datashard_user_table.h b/ydb/core/tx/datashard/datashard_user_table.h
@@ -251,10 +251,12 @@ struct TUserTable : public TThrRefBase {
     };
 
     struct TTableIndex {
-        using EIndexType = NKikimrSchemeOp::EIndexType;
+        using EType = NKikimrSchemeOp::EIndexType;
+        using EState = NKikimrSchemeOp::EIndexState;
 
         TString Name;
-        EIndexType Type;
+        EType Type;
+        EState State;
         TVector<ui32> KeyColumnIds;
         TVector<ui32> DataColumnIds;
 
@@ -263,6 +265,7 @@ struct TUserTable : public TThrRefBase {
         TTableIndex(const NKikimrSchemeOp::TIndexDescription& indexDesc, const TMap<ui32, TUserColumn>& columns)
             : Name(indexDesc.GetName())
             , Type(indexDesc.GetType())
+            , State(indexDesc.GetState())
         {
             THashMap<TStringBuf, ui32> nameToId;
             for (const auto& [id, column] : columns) {
@@ -443,6 +446,7 @@ struct TUserTable : public TThrRefBase {
     bool ResetTableSchemaVersion();
 
     void AddIndex(const NKikimrSchemeOp::TIndexDescription& indexDesc);
+    void SwitchIndexState(const TPathId& indexPathId, TTableIndex::EState state);
     void DropIndex(const TPathId& indexPathId);
     bool HasAsyncIndexes() const;
 
diff --git a/ydb/core/tx/datashard/drop_table_unit.cpp b/ydb/core/tx/datashard/drop_table_unit.cpp
@@ -81,7 +81,7 @@ EExecutionStatus TDropTableUnit::Execute(TOperation::TPtr op,
     Y_ABORT_UNLESS(it != DataShard.GetUserTables().end());
     {
         for (const auto& [indexPathId, indexInfo] : it->second->Indexes) {
-            if (indexInfo.Type == TUserTable::TTableIndex::EIndexType::EIndexTypeGlobalAsync) {
+            if (indexInfo.Type == TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {
                 RemoveSenders.emplace_back(new TEvChangeExchange::TEvRemoveSender(indexPathId));
             }
         }
diff --git a/ydb/core/tx/datashard/finalize_build_index_unit.cpp b/ydb/core/tx/datashard/finalize_build_index_unit.cpp
@@ -37,14 +37,17 @@ class TFinalizeBuildIndexUnit : public TExecutionUnit {
         Y_ABORT_UNLESS(version);
 
         TUserTable::TPtr tableInfo;
-        if (params.HasOutcome() && params.GetOutcome().HasCancel()) {
+        if (params.HasOutcome() && params.GetOutcome().HasApply()) {
+            const auto indexPathId = PathIdFromPathId(params.GetOutcome().GetApply().GetIndexPathId());
+
+            tableInfo = DataShard.AlterTableSwitchIndexState(ctx, txc, pathId, version, indexPathId, NKikimrSchemeOp::EIndexStateReady);
+        } else if (params.HasOutcome() && params.GetOutcome().HasCancel()) {
+            const auto indexPathId = PathIdFromPathId(params.GetOutcome().GetCancel().GetIndexPathId());
+
             const auto& userTables = DataShard.GetUserTables();
             Y_ABORT_UNLESS(userTables.contains(pathId.LocalPathId));
             const auto& indexes = userTables.at(pathId.LocalPathId)->Indexes;
-
-            const auto indexPathId = PathIdFromPathId(params.GetOutcome().GetCancel().GetIndexPathId());
             auto it = indexes.find(indexPathId);
-
             if (it != indexes.end() && it->second.Type == NKikimrSchemeOp::EIndexType::EIndexTypeGlobalAsync) {
                 RemoveSender.Reset(new TEvChangeExchange::TEvRemoveSender(indexPathId));
             }
@@ -66,7 +69,7 @@ class TFinalizeBuildIndexUnit : public TExecutionUnit {
         Y_ABORT_UNLESS(step != 0);
 
         if (DataShard.GetBuildIndexManager().Contains(params.GetBuildIndexId())) {
-            auto  record = DataShard.GetBuildIndexManager().Get(params.GetBuildIndexId());
+            auto record = DataShard.GetBuildIndexManager().Get(params.GetBuildIndexId());
             DataShard.CancelScan(tableInfo->LocalTid, record.ScanId);
             DataShard.GetBuildIndexManager().Drop(params.GetBuildIndexId());
         }
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_apply_build_index.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_apply_build_index.cpp
@@ -30,7 +30,10 @@ TVector<ISubOperation::TPtr> ApplyBuildIndex(TOperationId nextId, const TTxTrans
         op->SetTableName(table.LeafName());
         op->SetSnapshotTxId(config.GetSnaphotTxId());  //TODO: fix spelling error in flat_scheme_op.proto first
         op->SetBuildIndexId(config.GetBuildIndexId());
-        op->MutableOutcome()->MutableApply();
+        if (!indexName.empty()) {
+            TPath index = table.Child(indexName);
+            PathIdFromPathId(index.Base()->PathId, op->MutableOutcome()->MutableApply()->MutableIndexPathId());
+        }
 
         result.push_back(CreateFinalizeBuildIndexMainTable(NextPartId(nextId, result), finalize));
     }

Original file line number	Diff line number	Diff line change
`@@ -1041,9 +1041,16 @@ message TBuildIndexOutcome {`
`1041`	`1041`	`optional NKikimrProto.TPathID IndexPathId = 1;`
`1042`	`1042`	`}`
`1043`	`1043`
	`1044`	`+ message TApply {`
	`1045`	`+ // Path id of the index whose creation was done`
	`1046`	`+ optional NKikimrProto.TPathID IndexPathId = 1;`
	`1047`	`+ }`
	`1048`	`+`
	`1049`	`+ reserved 1; // Apply type was changed from Empty to TApply`
	`1050`	`+`
`1044`	`1051`	`oneof Result {`
`1045`		`- google.protobuf.Empty Apply = 1;`
`1046`	`1052`	`TCancel Cancel = 2;`
	`1053`	`+ TApply Apply = 3;`
`1047`	`1054`	`}`
`1048`	`1055`	`}`
`1049`	`1056`
Original file line number	Diff line number	Diff line change
`@@ -106,7 +106,7 @@ bool TAsyncIndexChangeCollector::Collect(const TTableId& tableId, ERowOp rop,`
`106`	`106`	`const auto updatedTagToPos = MakeTagToPos(updates, [](const TUpdateOp& op) { return op.Tag; });`
`107`	`107`
`108`	`108`	`for (const auto& [pathId, index] : userTable->Indexes) {`
`109`		`- if (index.Type != TUserTable::TTableIndex::EIndexType::EIndexTypeGlobalAsync) {`
	`109`	`+ if (index.Type != TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {`
`110`	`110`	`continue;`
`111`	`111`	`}`
`112`	`112`
`@@ -203,7 +203,7 @@ auto TAsyncIndexChangeCollector::CacheTags(const TTableId& tableId) const {`
`203`	`203`	`TCachedTagsBuilder builder;`
`204`	`204`
`205`	`205`	`for (const auto& [_, index] : userTable->Indexes) {`
`206`		`- if (index.Type != TUserTable::TTableIndex::EIndexType::EIndexTypeGlobalAsync) {`
	`206`	`+ if (index.Type != TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {`
`207`	`207`	`continue;`
`208`	`208`	`}`
`209`	`209`
Original file line number	Diff line number	Diff line change
`@@ -276,7 +276,7 @@ class TChangeSender: public TActor<TChangeSender> {`
`276`	`276`	`const auto fullTableId = TTableId(self->GetPathOwnerId(), tableId);`
`277`	`277`
`278`	`278`	`for (const auto& [indexPathId, indexInfo] : tableInfo->Indexes) {`
`279`		`- if (indexInfo.Type != TUserTable::TTableIndex::EIndexType::EIndexTypeGlobalAsync) {`
	`279`	`+ if (indexInfo.Type != TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {`
`280`	`280`	`continue;`
`281`	`281`	`}`
`282`	`282`
Original file line number	Diff line number	Diff line change
`@@ -81,7 +81,7 @@ EExecutionStatus TDropTableUnit::Execute(TOperation::TPtr op,`
`81`	`81`	`Y_ABORT_UNLESS(it != DataShard.GetUserTables().end());`
`82`	`82`	`{`
`83`	`83`	`for (const auto& [indexPathId, indexInfo] : it->second->Indexes) {`
`84`		`- if (indexInfo.Type == TUserTable::TTableIndex::EIndexType::EIndexTypeGlobalAsync) {`
	`84`	`+ if (indexInfo.Type == TUserTable::TTableIndex::EType::EIndexTypeGlobalAsync) {`
`85`	`85`	`RemoveSenders.emplace_back(new TEvChangeExchange::TEvRemoveSender(indexPathId));`
`86`	`86`	`}`
`87`	`87`	`}`