Configurable MaxCdcInitialScanShardsInFlight

Author: CyberROFL

ydb/core/protos/config.proto

@@ -1531,6 +1531,9 @@ message TSchemeShardConfig {
     optional uint32 StatsMaxExecuteMs = 3 [default = 10];
 
     repeated TInFlightCounterConfig InFlightCounterConfig = 4;
+
+    // number of shards per table
+    optional uint32 MaxCdcInitialScanShardsInFlight = 5 [default = 10];
 }
 
 message TCompactionConfig {

ydb/core/tx/schemeshard/schemeshard_cdc_stream_scan.cpp

@@ -172,7 +172,7 @@ struct TSchemeShard::TCdcStreamScan::TTxProgress: public TTransactionBase<TSchem
         }
 
         while (!streamInfo->PendingShards.empty()) {
-            if (streamInfo->InProgressShards.size() >= streamInfo->MaxInProgressShards) {
+            if (streamInfo->InProgressShards.size() >= Self->MaxCdcInitialScanShardsInFlight) {
                 break;
             }
 

ydb/core/tx/schemeshard/schemeshard_impl.cpp

@@ -4450,6 +4450,7 @@ void TSchemeShard::OnActivateExecutor(const TActorContext &ctx) {
     ConfigureCompactionQueues(appData->CompactionConfig, ctx);
     ConfigureStatsBatching(appData->SchemeShardConfig, ctx);
     ConfigureStatsOperations(appData->SchemeShardConfig, ctx);
+    MaxCdcInitialScanShardsInFlight = appData->SchemeShardConfig.GetMaxCdcInitialScanShardsInFlight();
 
     ConfigureBackgroundCleaningQueue(appData->BackgroundCleaningConfig, ctx);
 
@@ -6951,6 +6952,7 @@ void TSchemeShard::ApplyConsoleConfigs(const NKikimrConfig::TAppConfig& appConfi
         const auto& schemeShardConfig = appConfig.GetSchemeShardConfig();
         ConfigureStatsBatching(schemeShardConfig, ctx);
         ConfigureStatsOperations(schemeShardConfig, ctx);
+        MaxCdcInitialScanShardsInFlight = schemeShardConfig.GetMaxCdcInitialScanShardsInFlight();
     }
 
     if (appConfig.HasTableProfilesConfig()) {

ydb/core/tx/schemeshard/schemeshard_impl.h

@@ -338,7 +338,9 @@ class TSchemeShard
     TActorId SysPartitionStatsCollector;
 
     TActorId TabletMigrator;
+
     TActorId CdcStreamScanFinalizer;
+    ui32 MaxCdcInitialScanShardsInFlight = 10;
 
     TDuration StatsMaxExecuteTime;
     TDuration StatsBatchTimeout;

ydb/core/tx/schemeshard/schemeshard_info_types.h

@@ -2395,8 +2395,6 @@ struct TCdcStreamInfo : public TSimpleRefCount<TCdcStreamInfo> {
         {}
     };
 
-    static constexpr ui32 MaxInProgressShards = 10;
-
     TCdcStreamInfo(ui64 version, EMode mode, EFormat format, bool vt, const TDuration& rt, const TString& awsRegion, EState state)
         : AlterVersion(version)
         , Mode(mode)