Fix splitted commit message for EvWrite (#15062)

Author: nikvas0

ydb/core/kqp/node_service/kqp_node_service.cpp

@@ -457,7 +457,6 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
 
         ptr->InFlightMemoryLimitPerActorBytes = settings.GetInFlightMemoryLimitPerActorBytes();
         ptr->MemoryLimitPerMessageBytes = settings.GetMemoryLimitPerMessageBytes();
-        ptr->MaxBatchesPerMessage = settings.GetMaxBatchesPerMessage();
 
         ptr->StartRetryDelay = TDuration::MilliSeconds(settings.GetStartRetryDelayMs());
         ptr->MaxRetryDelay = TDuration::MilliSeconds(settings.GetMaxRetryDelayMs());

ydb/core/kqp/runtime/kqp_write_actor.cpp

@@ -239,12 +239,12 @@ class TKqpTableWriteActor : public TActorBootstrapped<TKqpTableWriteActor> {
         , Counters(counters)
         , TableWriteActorSpan(TWilsonKqp::TableWriteActor, NWilson::TTraceId(traceId), "TKqpTableWriteActor")
     {
+        AFL_ENSURE(MessageSettings.InFlightMemoryLimitPerActorBytes >= MessageSettings.MemoryLimitPerMessageBytes);
         LogPrefix = TStringBuilder() << "Table: `" << TablePath << "` (" << TableId << "), " << "SessionActorId: " << sessionActorId;
         ShardedWriteController = CreateShardedWriteController(
             TShardedWriteControllerSettings {
                 .MemoryLimitTotal = MessageSettings.InFlightMemoryLimitPerActorBytes,
                 .MemoryLimitPerMessage = MessageSettings.MemoryLimitPerMessageBytes,
-                .MaxBatchesPerMessage = MessageSettings.MaxBatchesPerMessage,
             },
             TypeEnv,
             Alloc);

ydb/core/kqp/runtime/kqp_write_actor_settings.h

@@ -10,7 +10,6 @@ namespace NKqp {
 struct TWriteActorSettings : TAtomicRefCount<TWriteActorSettings> {
     i64 InFlightMemoryLimitPerActorBytes = 64_MB;
     i64 MemoryLimitPerMessageBytes = 64_MB;
-    i64 MaxBatchesPerMessage = 1000;
     i64 MaxForwardedSize = 64_MB;
 
     TDuration StartRetryDelay = TDuration::Seconds(1);

ydb/core/kqp/runtime/kqp_write_table.cpp

@@ -927,20 +927,21 @@ class TShardsInfo {
             return IsClosed() && IsEmpty();
         }
 
-        void MakeNextBatches(i64 maxDataSize, ui64 maxCount) {
+        void MakeNextBatches(i64 maxDataSize, std::optional<ui64> maxCount) {
             YQL_ENSURE(BatchesInFlight == 0);
             YQL_ENSURE(!IsEmpty());
-            YQL_ENSURE(maxCount != 0);
             i64 dataSize = 0;
             // For columnshard batch can be slightly larger than the limit.
-            while (BatchesInFlight < maxCount
+            while ((!maxCount || BatchesInFlight < *maxCount)
                     && BatchesInFlight < Batches.size()
                     && (dataSize + GetBatch(BatchesInFlight).GetMemory() <= maxDataSize || BatchesInFlight == 0)) {
                 dataSize += GetBatch(BatchesInFlight).GetMemory();
                 ++BatchesInFlight;
             }
             YQL_ENSURE(BatchesInFlight != 0);
-            YQL_ENSURE(BatchesInFlight == Batches.size() || BatchesInFlight >= maxCount || dataSize + GetBatch(BatchesInFlight).GetMemory() > maxDataSize);
+            YQL_ENSURE(BatchesInFlight == Batches.size()
+                || (maxCount && BatchesInFlight >= *maxCount)
+                || dataSize + GetBatch(BatchesInFlight).GetMemory() > maxDataSize);
         }
 
         TBatchWithMetadata& GetBatch(size_t index) {
@@ -1451,9 +1452,11 @@ class TShardedWriteController : public IShardedWriteController {
     void BuildBatchesForShard(TShardsInfo::TShardInfo& shard) {
         if (shard.GetBatchesInFlight() == 0) {
             YQL_ENSURE(IsOlap != std::nullopt);
-            shard.MakeNextBatches(
-                Settings.MemoryLimitPerMessage,
-                (*IsOlap) ? 1 : Settings.MaxBatchesPerMessage);
+            if (*IsOlap) {
+                shard.MakeNextBatches(Settings.MemoryLimitPerMessage, 1);
+            } else {
+                shard.MakeNextBatches(Settings.MemoryLimitPerMessage, std::nullopt);
+            }
         }
     }
 

ydb/core/kqp/runtime/kqp_write_table.h

@@ -116,7 +116,6 @@ using IShardedWriteControllerPtr = TIntrusivePtr<IShardedWriteController>;
 struct TShardedWriteControllerSettings {
     i64 MemoryLimitTotal;
     i64 MemoryLimitPerMessage;
-    i64 MaxBatchesPerMessage;
     bool Inconsistent;
 };
 

ydb/core/protos/table_service_config.proto

@@ -330,7 +330,6 @@ message TTableServiceConfig {
     message TWriteActorSettings {
         optional uint64 InFlightMemoryLimitPerActorBytes = 1 [ default = 67108864 ];
         optional uint64 MemoryLimitPerMessageBytes = 2 [ default = 67108864 ];
-        optional uint64 MaxBatchesPerMessage = 3 [ default = 1000 ];
 
         optional uint64 StartRetryDelayMs = 4 [ default = 1000 ];
         optional uint64 MaxRetryDelayMs = 5 [ default = 10000 ];