Add delete rows functionality for fulltext indexed tables

ydb/core/kqp/opt/physical/effects/kqp_opt_phy_delete_index.cpp

@@ -74,34 +74,47 @@ TExprBase BuildDeleteIndexStagesImpl(const TKikimrTableDescription& table,
 
         auto deleteIndexKeys = project(indexTableColumns);
 
-        if (indexDesc->Type == TIndexDescription::EType::GlobalSyncVectorKMeansTree) {
-            if (indexDesc->KeyColumns.size() > 1) {
-                const auto& prefixTable = kqpCtx.Tables->ExistingTable(kqpCtx.Cluster, TStringBuilder() << del.Table().Path().Value()
-                    << "/" << indexDesc->Name << "/" << NKikimr::NTableIndex::NKMeans::PrefixTable);
-                deleteIndexKeys = BuildVectorIndexPrefixRows(table, prefixTable, false, indexDesc, deleteIndexKeys, indexTableColumns, del.Pos(), ctx);
+        switch (indexDesc->Type) {
+            case TIndexDescription::EType::GlobalSync:
+            case TIndexDescription::EType::GlobalAsync:
+            case TIndexDescription::EType::GlobalSyncUnique: {
+                // deleteIndexKeys are already correct 
+                break;
+            }
+            case TIndexDescription::EType::GlobalSyncVectorKMeansTree: {
+                if (indexDesc->KeyColumns.size() > 1) {
+                    const auto& prefixTable = kqpCtx.Tables->ExistingTable(kqpCtx.Cluster, TStringBuilder() << del.Table().Path().Value()
+                        << "/" << indexDesc->Name << "/" << NKikimr::NTableIndex::NKMeans::PrefixTable);
+                    deleteIndexKeys = BuildVectorIndexPrefixRows(table, prefixTable, false, indexDesc, deleteIndexKeys, indexTableColumns, del.Pos(), ctx);
+                }
+                deleteIndexKeys = BuildVectorIndexPostingRows(table, del.Table(), indexDesc->Name,
+                    indexTableColumns, deleteIndexKeys, false, del.Pos(), ctx);
+                break;
+            }
+            case TIndexDescription::EType::GlobalFulltext: {
+                // For fulltext indexes, we need to tokenize the text from the rows being deleted
+                // and then delete the corresponding token rows from the index table
+                auto deleteKeysPrecompute = Build<TDqPhyPrecompute>(ctx, del.Pos())
+                    .Connection<TDqCnUnionAll>()
+                        .Output()
+                            .Stage(ReadTableToStage(deleteIndexKeys, ctx))
+                            .Index().Build("0")
+                            .Build()
+                        .Build()
+                    .Done();
+                deleteIndexKeys = BuildFulltextIndexRows(table, indexDesc, deleteKeysPrecompute, indexTableColumnsSet, indexTableColumns, /*includeDataColumns=*/false,
+                    del.Pos(), ctx);
+                break;
             }
-
-            auto resolveUnion = BuildVectorIndexPostingRows(table, del.Table(), indexDesc->Name,
-                indexTableColumns, deleteIndexKeys, false, del.Pos(), ctx);
-
-            auto indexDelete = Build<TKqlDeleteRows>(ctx, del.Pos())
-                .Table(tableNode)
-                .Input(resolveUnion)
-                .ReturningColumns<TCoAtomList>().Build()
-                .IsBatch(ctx.NewAtom(del.Pos(), "false"))
-                .Done();
-
-            effects.emplace_back(indexDelete);
-        } else {
-            auto indexDelete = Build<TKqlDeleteRows>(ctx, del.Pos())
-                .Table(tableNode)
-                .Input(deleteIndexKeys)
-                .ReturningColumns<TCoAtomList>().Build()
-                .IsBatch(ctx.NewAtom(del.Pos(), "false"))
-                .Done();
-
-            effects.emplace_back(std::move(indexDelete));
         }
+
+        auto indexDelete = Build<TKqlDeleteRows>(ctx, del.Pos())
+            .Table(tableNode)
+            .Input(deleteIndexKeys)
+            .ReturningColumns<TCoAtomList>().Build()
+            .IsBatch(ctx.NewAtom(del.Pos(), "false"))
+            .Done();
+        effects.emplace_back(std::move(indexDelete));
     }
 
     return Build<TExprList>(ctx, del.Pos())

ydb/core/kqp/opt/physical/effects/kqp_opt_phy_effects_impl.h

@@ -98,6 +98,8 @@ NYql::NNodes::TKqpCnStreamLookup BuildStreamLookupOverPrecompute(const NYql::TKi
     NYql::NNodes::TExprBase input,
     const NYql::NNodes::TKqpTable& kqpTableNode, const NYql::TPositionHandle& pos, NYql::TExprContext& ctx, const TVector<TString>& extraColumnsToRead = {});
 
+NYql::NNodes::TDqStageBase ReadTableToStage(const NYql::NNodes::TExprBase& expr, NYql::TExprContext& ctx);
+
 NYql::NNodes::TExprBase BuildVectorIndexPostingRows(const NYql::TKikimrTableDescription& table,
     const NYql::NNodes::TKqpTable& tableNode,
     const TString& indexName,
@@ -118,9 +120,7 @@ std::pair<NYql::NNodes::TExprBase, NYql::NNodes::TExprBase> BuildVectorIndexPref
     TVector<TStringBuf>& indexTableColumns, NYql::TPositionHandle pos, NYql::TExprContext& ctx);
 
 NYql::NNodes::TExprBase BuildFulltextIndexRows(const NYql::TKikimrTableDescription& table, const NYql::TIndexDescription* indexDesc,
-    const NYql::NNodes::TExprBase& inputRows, const THashSet<TStringBuf>& inputColumns, const TVector<TStringBuf>& indexTableColumns,
+    const NYql::NNodes::TExprBase& inputRows, const THashSet<TStringBuf>& inputColumns, TVector<TStringBuf>& indexTableColumns, bool includeDataColumns,
     NYql::TPositionHandle pos, NYql::TExprContext& ctx);
 
-TVector<TStringBuf> BuildFulltextIndexColumns(const NYql::TKikimrTableDescription& table, const NYql::TIndexDescription* indexDesc);
-
 } // NKikimr::NKqp::NOpt

ydb/core/kqp/opt/physical/effects/kqp_opt_phy_fulltext_index.cpp

@@ -8,7 +8,7 @@ using namespace NYql::NDq;
 using namespace NYql::NNodes;
 
 TExprBase BuildFulltextIndexRows(const TKikimrTableDescription& table, const TIndexDescription* indexDesc,
-    const NNodes::TExprBase& inputRows, const THashSet<TStringBuf>& inputColumns, const TVector<TStringBuf>& indexTableColumns,
+    const NNodes::TExprBase& inputRows, const THashSet<TStringBuf>& inputColumns, TVector<TStringBuf>& indexTableColumns, bool includeDataColumns,
     TPositionHandle pos, NYql::TExprContext& ctx)
 {
     // Extract fulltext index settings
@@ -31,17 +31,10 @@ TExprBase BuildFulltextIndexRows(const TKikimrTableDescription& table, const TIn
     // Build output row structure for each token
     TVector<TExprBase> tokenRowTuples;
 
-    // Add token column (first column in fulltext index)
-    auto tokenTuple = Build<TCoNameValueTuple>(ctx, pos)
-        .Name().Build(NTableIndex::NFulltext::TokenColumn)
-        .Value(tokenArg)
-        .Done();
-    tokenRowTuples.emplace_back(tokenTuple);
-
-    // Add all other columns (primary key + data columns)
-    for (const auto& column : indexTableColumns) {
+    indexTableColumns.clear();
+    auto addIndexColumn = [&](const TStringBuf& column) {
+        indexTableColumns.emplace_back(column);
         auto columnAtom = ctx.NewAtom(pos, column);
-
         if (inputColumns.contains(column)) {
             auto tuple = Build<TCoNameValueTuple>(ctx, pos)
                 .Name(columnAtom)
@@ -50,20 +43,37 @@ TExprBase BuildFulltextIndexRows(const TKikimrTableDescription& table, const TIn
                     .Name(columnAtom)
                     .Build()
                 .Done();
-
             tokenRowTuples.emplace_back(tuple);
         } else {
             auto columnType = table.GetColumnType(TString(column));
-
             auto tuple = Build<TCoNameValueTuple>(ctx, pos)
                 .Name(columnAtom)
                 .Value<TCoNothing>()
                     .OptionalType(NCommon::BuildTypeExpr(pos, *columnType, ctx))
                     .Build()
                 .Done();
-
             tokenRowTuples.emplace_back(tuple);
         }
+    };
+
+    // Add token column first
+    indexTableColumns.emplace_back(NTableIndex::NFulltext::TokenColumn);
+    auto tokenTuple = Build<TCoNameValueTuple>(ctx, pos)
+        .Name().Build(NTableIndex::NFulltext::TokenColumn)
+        .Value(tokenArg)
+        .Done();
+    tokenRowTuples.emplace_back(tokenTuple);
+
+    // Add primary key columns
+    for (const auto& column : table.Metadata->KeyColumnNames) {
+        addIndexColumn(column);
+    }
+
+    // Add data columns (covered columns)
+    if (includeDataColumns) {
+        for (const auto& column : indexDesc->DataColumns) {
+            addIndexColumn(column);
+        }
     }
 
     // Create lambda that builds output row for each token
@@ -126,29 +136,4 @@ TExprBase BuildFulltextIndexRows(const TKikimrTableDescription& table, const TIn
         .Done();
 }
 
-TVector<TStringBuf> BuildFulltextIndexColumns(const TKikimrTableDescription& table, const TIndexDescription* indexDesc) {
-    TVector<TStringBuf> indexTableColumns;
-    THashSet<TStringBuf> indexTableColumnSet;
-
-    // Add token column first (replaces the text column)
-    indexTableColumns.emplace_back(NTableIndex::NFulltext::TokenColumn);
-    indexTableColumnSet.insert(NTableIndex::NFulltext::TokenColumn);
-
-    // Add primary key columns
-    for (const auto& column : table.Metadata->KeyColumnNames) {
-        if (indexTableColumnSet.insert(column).second) {
-            indexTableColumns.emplace_back(column);
-        }
-    }
-
-    // Add data columns (covered columns)
-    for (const auto& column : indexDesc->DataColumns) {
-        if (indexTableColumnSet.insert(column).second) {
-            indexTableColumns.emplace_back(column);
-        }
-    }
-
-    return indexTableColumns;
-}
-
 }

ydb/core/kqp/opt/physical/effects/kqp_opt_phy_insert_index.cpp

@@ -227,11 +227,9 @@ TExprBase KqpBuildInsertIndexStages(TExprBase node, TExprContext& ctx, const TKq
                     break;
                 }
                 case TIndexDescription::EType::GlobalFulltext: {
-                    // For fulltext indexes, we need to tokenize the text and create index rows
-                    upsertIndexRows = BuildFulltextIndexRows(table, indexDesc, insertRowsPrecompute, inputColumnsSet, indexTableColumns,
+                    // For fulltext indexes, we need to tokenize the text and create index rows and refill index columns
+                    upsertIndexRows = BuildFulltextIndexRows(table, indexDesc, insertRowsPrecompute, inputColumnsSet, indexTableColumns, /*includeDataColumns=*/true,
                         insert.Pos(), ctx);
-                    // Update columns to reflect transformation: text column -> __ydb_token
-                    indexTableColumns = BuildFulltextIndexColumns(table, indexDesc);
                     break;
                 }
             }

ydb/core/kqp/opt/physical/effects/kqp_opt_phy_update.cpp

@@ -240,4 +240,40 @@ TExprBase KqpBuildUpdateStages(TExprBase node, TExprContext& ctx, const TKqpOpti
     }
 }
 
+TDqStageBase ReadTableToStage(const TExprBase& expr, TExprContext& ctx) {
+    if (expr.Maybe<TDqStageBase>()) {
+        return expr.Cast<TDqStageBase>();
+    }
+    if (expr.Maybe<TDqCnUnionAll>()) {
+        return expr.Cast<TDqCnUnionAll>().Output().Stage();
+    }
+    auto pos = expr.Pos();
+    TVector<TExprNode::TPtr> inputs;
+    TVector<TExprNode::TPtr> args;
+    TNodeOnNodeOwnedMap replaces;
+    int i = 1;
+    VisitExpr(expr.Ptr(), [&](const TExprNode::TPtr& node) {
+        TExprBase expr(node);
+        if (auto cast = expr.Maybe<TDqCnUnionAll>()) {
+            auto newArg = ctx.NewArgument(pos, TStringBuilder() << "rows" << i);
+            inputs.emplace_back(node);
+            args.emplace_back(newArg);
+            replaces.emplace(expr.Raw(), newArg);
+            return false;
+        }
+        return true;
+    });
+    return Build<TDqStage>(ctx, pos)
+        .Inputs()
+            .Add(inputs)
+            .Build()
+        .Program()
+            .Args(args)
+            .Body(ctx.ReplaceNodes(expr.Ptr(), replaces))
+            .Build()
+        .Settings()
+            .Build()
+        .Done();
+}
+
 } // namespace NKikimr::NKqp::NOpt

ydb/core/kqp/opt/physical/effects/kqp_opt_phy_vector_index.cpp

@@ -6,42 +6,6 @@ using namespace NYql;
 using namespace NYql::NDq;
 using namespace NYql::NNodes;
 
-TDqStageBase ReadTableToStage(const TExprBase& expr, TExprContext& ctx) {
-    if (expr.Maybe<TDqStageBase>()) {
-        return expr.Cast<TDqStageBase>();
-    }
-    if (expr.Maybe<TDqCnUnionAll>()) {
-        return expr.Cast<TDqCnUnionAll>().Output().Stage();
-    }
-    auto pos = expr.Pos();
-    TVector<TExprNode::TPtr> inputs;
-    TVector<TExprNode::TPtr> args;
-    TNodeOnNodeOwnedMap replaces;
-    int i = 1;
-    VisitExpr(expr.Ptr(), [&](const TExprNode::TPtr& node) {
-        TExprBase expr(node);
-        if (auto cast = expr.Maybe<TDqCnUnionAll>()) {
-            auto newArg = ctx.NewArgument(pos, TStringBuilder() << "rows" << i);
-            inputs.emplace_back(node);
-            args.emplace_back(newArg);
-            replaces.emplace(expr.Raw(), newArg);
-            return false;
-        }
-        return true;
-    });
-    return Build<TDqStage>(ctx, pos)
-        .Inputs()
-            .Add(inputs)
-            .Build()
-        .Program()
-            .Args(args)
-            .Body(ctx.ReplaceNodes(expr.Ptr(), replaces))
-            .Build()
-        .Settings()
-            .Build()
-        .Done();
-}
-
 TExprBase BuildVectorIndexPostingRows(const TKikimrTableDescription& table,
     const TKqpTable& tableNode,
     const TString& indexName,