WIP

Author: Enjection

ydb/core/kqp/host/kqp_gateway_proxy.cpp

@@ -1368,7 +1368,8 @@ class TKqpGatewayProxy : public IKikimrGateway {
             op.SetPrefix(settings.Prefix);
 
             if (settings.Settings.IncrementalBackupEnabled) {
-                op.MutableIncrementalBackupConfig();
+                auto* config = op.MutableIncrementalBackupConfig();
+                config->SetOmitIndexes(settings.Settings.OmitIndexes);
             }
 
             auto errOpt = std::visit(

ydb/core/kqp/provider/yql_kikimr_exec.cpp

@@ -1021,6 +1021,13 @@ namespace {
                         "INCREMENTAL_BACKUP_ENABLED must be true or false"));
                     return false;
                 }
+            } else if (name == "omit_indexes") {
+                auto value = ToString(setting.Value().Cast<TCoDataCtor>().Literal().Cast<TCoAtom>().Value());
+                if (!TryFromString(value, dstSettings.OmitIndexes)) {
+                    ctx.AddError(TIssue(ctx.GetPosition(pos),
+                        "OMIT_INDEXES must be true or false"));
+                    return false;
+                }
             } else if (name == "storage") {
                 auto value = ToString(setting.Value().Cast<TCoDataCtor>().Literal().Cast<TCoAtom>().Value());
                 if (to_lower(value) != "cluster") {

ydb/core/kqp/provider/yql_kikimr_gateway.h

@@ -1052,7 +1052,8 @@ struct TAnalyzeSettings {
 };
 
 struct TBackupCollectionSettings {
-    bool IncrementalBackupEnabled;
+    bool IncrementalBackupEnabled = false;
+    bool OmitIndexes = false;
 };
 
 struct TCreateBackupCollectionSettings {

ydb/core/kqp/provider/yql_kikimr_type_ann.cpp

@@ -2448,6 +2448,7 @@ virtual TStatus HandleCreateTable(TKiCreateTable create, TExprContext& ctx) over
         const THashSet<TString> supportedSettings = {
             "incremental_backup_enabled",
             "storage",
+            "omit_indexes",
         };
 
         if (!CheckBackupCollectionSettings(node.BackupCollectionSettings(), supportedSettings, ctx)) {

ydb/core/protos/flat_scheme_op.proto

@@ -2312,7 +2312,7 @@ message TBackupCollectionDescription {
     }
 
     message TIncrementalBackupConfig {
-
+        optional bool OmitIndexes = 1 [default = false];
     }
 
     oneof Entries {

ydb/core/tx/datashard/change_collector_cdc_stream.cpp

@@ -123,8 +123,15 @@ bool TCdcStreamChangeCollector::NeedToReadKeys() const {
     }
 
     bool value = false;
-    for (const auto& [_, tableInfo] : Self->GetUserTables()) {
-        for (const auto& [_, streamInfo] : tableInfo->CdcStreams) {
+    const auto& userTables = Self->GetUserTables();
+    Cerr << "CDC_DEBUG: NeedToReadKeys checking " << userTables.size() << " user tables" << Endl;
+    for (const auto& [tableLocalPathId, tableInfo] : userTables) {
+        Cerr << "CDC_DEBUG: Table LocalPathId=" << tableLocalPathId 
+             << " has " << tableInfo->CdcStreams.size() << " CDC streams" << Endl;
+        for (const auto& [streamPathId, streamInfo] : tableInfo->CdcStreams) {
+            Cerr << "CDC_DEBUG:   Stream PathId=" << streamPathId 
+                 << " State=" << static_cast<ui32>(streamInfo.State)
+                 << " Mode=" << static_cast<ui32>(streamInfo.Mode) << Endl;
             if (streamInfo.State == NKikimrSchemeOp::ECdcStreamStateDisabled) {
                 continue;
             }
@@ -145,15 +152,29 @@ bool TCdcStreamChangeCollector::NeedToReadKeys() const {
     }
 
     CachedNeedToReadKeys = value;
+    Cerr << "CDC_DEBUG: NeedToReadKeys returning " << value << Endl;
     return *CachedNeedToReadKeys;
 }
 
 bool TCdcStreamChangeCollector::Collect(const TTableId& tableId, ERowOp rop,
         TArrayRef<const TRawTypeValue> key, TArrayRef<const TUpdateOp> updates)
 {
-    Y_ENSURE(Self->IsUserTable(tableId), "Unknown table: " << tableId);
+    Cerr << "CDC_DEBUG: Collect called for TableId OwnerId=" << tableId.PathId.OwnerId 
+         << " LocalPathId=" << tableId.PathId.LocalPathId 
+         << " RowOp=" << static_cast<ui32>(rop) << Endl;
+    
+    if (!Self->IsUserTable(tableId)) {
+        const auto& userTables = Self->GetUserTables();
+        Cerr << "CDC_DEBUG: IsUserTable returned FALSE! TableId not in UserTables map." << Endl;
+        Cerr << "CDC_DEBUG: UserTables contains " << userTables.size() << " tables:" << Endl;
+        for (const auto& [localPathId, _] : userTables) {
+            Cerr << "CDC_DEBUG:   LocalPathId=" << localPathId << Endl;
+        }
+        Y_ENSURE(false, "Unknown table: " << tableId);
+    }
 
     auto userTable = Self->GetUserTables().at(tableId.PathId.LocalPathId);
+    Cerr << "CDC_DEBUG: Found user table with " << userTable->CdcStreams.size() << " CDC streams" << Endl;
     const auto& keyTags = userTable->KeyColumnIds;
     const auto& keyTypes = userTable->KeyColumnTypes;
     const auto valueTags = MakeValueTags(userTable->Columns);
@@ -173,6 +194,9 @@ bool TCdcStreamChangeCollector::Collect(const TTableId& tableId, ERowOp rop,
     }
 
     for (const auto& [pathId, stream] : userTable->CdcStreams) {
+        Cerr << "CDC_DEBUG: Processing CDC stream PathId=" << pathId 
+             << " State=" << static_cast<ui32>(stream.State)
+             << " Mode=" << static_cast<ui32>(stream.Mode) << Endl;
         TMaybe<TRowState> initialState;
         TMaybe<TRowVersion> snapshotVersion;
 

ydb/core/tx/datashard/create_cdc_stream_unit.cpp

@@ -43,9 +43,17 @@ class TCreateCdcStreamUnit : public TExecutionUnit {
         const auto version = params.GetTableSchemaVersion();
         Y_ENSURE(version);
 
+        Cerr << "CDC_DEBUG: CreateCdcStreamUnit creating CDC stream on table PathId=" << pathId 
+             << " (OwnerId=" << pathId.OwnerId << " LocalPathId=" << pathId.LocalPathId << ")"
+             << " streamPathId=" << streamPathId
+             << " State=" << static_cast<ui32>(streamDesc.GetState())
+             << " Mode=" << static_cast<ui32>(streamDesc.GetMode()) << Endl;
+
         auto tableInfo = DataShard.AlterTableAddCdcStream(ctx, txc, pathId, version, streamDesc);
         TDataShardLocksDb locksDb(DataShard, txc);
         DataShard.AddUserTable(pathId, tableInfo, &locksDb);
+        
+        Cerr << "CDC_DEBUG: Added table to TableInfos with " << tableInfo->CdcStreams.size() << " CDC streams" << Endl;
 
         if (tableInfo->NeedSchemaSnapshots()) {
             DataShard.AddSchemaSnapshot(pathId, version, op->GetStep(), op->GetTxId(), txc, ctx);

ydb/core/tx/datashard/datashard_impl.h

@@ -1700,8 +1700,16 @@ class TDataShard
     }
 
     bool IsUserTable(const TTableId& tableId) const {
-        return (TableInfos.find(tableId.PathId.LocalPathId) != TableInfos.end())
-                && !TSysTables::IsSystemTable(tableId);
+        bool inTableInfos = TableInfos.find(tableId.PathId.LocalPathId) != TableInfos.end();
+        bool isSystemTable = TSysTables::IsSystemTable(tableId);
+        bool result = inTableInfos && !isSystemTable;
+        if (!result) {
+            Cerr << "CDC_DEBUG: IsUserTable returning FALSE for TableId OwnerId=" << tableId.PathId.OwnerId 
+                 << " LocalPathId=" << tableId.PathId.LocalPathId
+                 << " inTableInfos=" << inTableInfos
+                 << " isSystemTable=" << isSystemTable << Endl;
+        }
+        return result;
     }
 
     const THashMap<ui64, TUserTable::TCPtr> &GetUserTables() const { return TableInfos; }