Merge f30a0dd into ecb9d76

Author: abyss7

ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp

@@ -124,6 +124,9 @@ class TKqpCaFactory : public IKqpNodeComputeActorFactory {
         memoryLimits.MinMemAllocSize = MinMemAllocSize.load();
         memoryLimits.MinMemFreeSize = MinMemFreeSize.load();
         memoryLimits.ArrayBufferMinFillPercentage = args.Task->GetArrayBufferMinFillPercentage();
+        if (args.Task->HasBufferPageAllocSize()) {
+            memoryLimits.BufferPageAllocSize = args.Task->GetBufferPageAllocSize();
+        }
 
         auto estimation = ResourceManager_->EstimateTaskResources(*args.Task, args.NumberOfTasks);
         NRm::TKqpResourcesRequest resourcesRequest;

ydb/core/kqp/executer_actor/kqp_executer_impl.h

@@ -169,6 +169,10 @@ class TKqpExecuterBase : public TActor<TDerived> {
             ArrayBufferMinFillPercentage = executerConfig.TableServiceConfig.GetArrayBufferMinFillPercentage();
         }
 
+        if (executerConfig.TableServiceConfig.HasBufferPageAllocSize()) {
+            BufferPageAllocSize = executerConfig.TableServiceConfig.GetBufferPageAllocSize();
+        }
+
         EnableReadsMerge = *MergeDatashardReadsControl() == 1;
         TasksGraph.GetMeta().Snapshot = IKqpGateway::TKqpSnapshot(Request.Snapshot.Step, Request.Snapshot.TxId);
         TasksGraph.GetMeta().Arena = MakeIntrusive<NActors::TProtoArenaHolder>();
@@ -1702,6 +1706,7 @@ class TKqpExecuterBase : public TActor<TDerived> {
             .CaFactory_ = Request.CaFactory_,
             .BlockTrackingMode = BlockTrackingMode,
             .ArrayBufferMinFillPercentage = ArrayBufferMinFillPercentage,
+            .BufferPageAllocSize = BufferPageAllocSize,
             .VerboseMemoryLimitException = VerboseMemoryLimitException,
         });
 
@@ -2313,6 +2318,7 @@ class TKqpExecuterBase : public TActor<TDerived> {
     const NKikimrConfig::TTableServiceConfig::EBlockTrackingMode BlockTrackingMode;
     const bool VerboseMemoryLimitException;
     TMaybe<ui8> ArrayBufferMinFillPercentage;
+    TMaybe<ui8> BufferPageAllocSize;
 
     ui64 StatCollectInflightBytes = 0;
     ui64 StatFinishInflightBytes = 0;

ydb/core/kqp/executer_actor/kqp_planner.cpp

@@ -105,6 +105,7 @@ TKqpPlanner::TKqpPlanner(TKqpPlanner::TArgs&& args)
     , CaFactory_(args.CaFactory_)
     , BlockTrackingMode(args.BlockTrackingMode)
     , ArrayBufferMinFillPercentage(args.ArrayBufferMinFillPercentage)
+    , BufferPageAllocSize(args.BufferPageAllocSize)
     , VerboseMemoryLimitException(args.VerboseMemoryLimitException)
 {
     Y_UNUSED(MkqlMemoryLimit);
@@ -488,6 +489,10 @@ TString TKqpPlanner::ExecuteDataComputeTask(ui64 taskId, ui32 computeTasksSize)
         taskDesc->SetArrayBufferMinFillPercentage(*ArrayBufferMinFillPercentage);
     }
 
+    if (BufferPageAllocSize) {
+        taskDesc->SetBufferPageAllocSize(*BufferPageAllocSize);
+    }
+
     auto startResult = CaFactory_->CreateKqpComputeActor({
         .ExecuterId = ExecuterId,
         .TxId = TxId,

ydb/core/kqp/executer_actor/kqp_planner.h

@@ -65,6 +65,7 @@ class TKqpPlanner {
         const std::shared_ptr<NKikimr::NKqp::NComputeActor::IKqpNodeComputeActorFactory>& CaFactory_;
         const NKikimrConfig::TTableServiceConfig::EBlockTrackingMode BlockTrackingMode;
         const TMaybe<ui8> ArrayBufferMinFillPercentage;
+        const TMaybe<ui8> BufferPageAllocSize;
         const bool VerboseMemoryLimitException;
     };
 
@@ -140,6 +141,7 @@ class TKqpPlanner {
     TVector<TProgressStat> LastStats;
     const NKikimrConfig::TTableServiceConfig::EBlockTrackingMode BlockTrackingMode;
     const TMaybe<ui8> ArrayBufferMinFillPercentage;
+    const TMaybe<ui8> BufferPageAllocSize;
     const bool VerboseMemoryLimitException;
 
 public:

ydb/core/protos/table_service_config.proto

@@ -394,4 +394,6 @@ message TTableServiceConfig {
     optional bool EnableIndexStreamWrite = 88 [default = false];
 
     optional bool EnableParallelPointReadConsolidation = 89 [default = true];
+
+    optional uint64 BufferPageAllocSize = 90  [ default = 4096 ];
 };

ydb/library/yql/dq/actors/compute/dq_compute_actor.h

@@ -380,6 +380,7 @@ struct TComputeMemoryLimits {
     ui64 OutputChunkMaxSize = GetDqExecutionSettings().FlowControl.MaxOutputChunkSize;
     ui64 ChunkSizeLimit = 48_MB;
     TMaybe<ui8> ArrayBufferMinFillPercentage; // Used by DqOutputHashPartitionConsumer and DqOutputChannel
+    TMaybe<size_t> BufferPageAllocSize;
 
     IMemoryQuotaManager::TPtr MemoryQuotaManager;
 };

ydb/library/yql/dq/actors/compute/dq_sync_compute_actor_base.h

@@ -216,6 +216,7 @@ class TDqSyncComputeActorBase: public TDqComputeActorBase<TDerived, TComputeActo
         limits.OutputChunkMaxSize = this->MemoryLimits.OutputChunkMaxSize;
         limits.ChunkSizeLimit = this->MemoryLimits.ChunkSizeLimit;
         limits.ArrayBufferMinFillPercentage = this->MemoryLimits.ArrayBufferMinFillPercentage;
+        limits.BufferPageAllocSize = this->MemoryLimits.BufferPageAllocSize;
 
         if (!limits.OutputChunkMaxSize) {
             limits.OutputChunkMaxSize = GetDqExecutionSettings().FlowControl.MaxOutputChunkSize;

ydb/library/yql/dq/proto/dq_tasks.proto

@@ -227,4 +227,5 @@ message TDqTask {
     optional bool EnableSpilling = 19;
     optional uint32 ArrayBufferMinFillPercentage = 20;
     optional bool DisableMetering = 22;
+    optional uint64 BufferPageAllocSize = 23;
 }

ydb/library/yql/dq/runtime/dq_output_channel.cpp

@@ -22,8 +22,6 @@ namespace {
 
 using namespace NKikimr;
 
-using NKikimr::NMiniKQL::TPagedBuffer;
-
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 template<bool FastPack>
@@ -36,7 +34,7 @@ class TDqOutputChannel : public IDqOutputChannel {
         const NMiniKQL::THolderFactory& holderFactory, const TDqOutputChannelSettings& settings, const TLogFunc& logFunc,
         NDqProto::EDataTransportVersion transportVersion)
         : OutputType(outputType)
-        , Packer(OutputType)
+        , Packer(OutputType, settings.BufferPageAllocSize)
         , Width(OutputType->IsMulti() ? static_cast<NMiniKQL::TMultiType*>(OutputType)->GetElementsCount() : 1u)
         , Storage(settings.ChannelStorage)
         , HolderFactory(holderFactory)

ydb/library/yql/dq/runtime/dq_output_channel.h

@@ -37,6 +37,7 @@ struct TDqOutputChannelSettings {
     IDqChannelStorage::TPtr ChannelStorage;
     TCollectStatsLevel Level = TCollectStatsLevel::None;
     TMaybe<ui8> ArrayBufferMinFillPercentage;
+    TMaybe<size_t> BufferPageAllocSize;
     TMutable MutableSettings;
 };
 

ydb/library/yql/dq/runtime/dq_tasks_runner.cpp

@@ -683,6 +683,7 @@ class TDqTaskRunner : public IDqTaskRunner {
                     settings.MaxChunkBytes = memoryLimits.OutputChunkMaxSize;
                     settings.ChunkSizeLimit = memoryLimits.ChunkSizeLimit;
                     settings.ArrayBufferMinFillPercentage = memoryLimits.ArrayBufferMinFillPercentage;
+                    settings.BufferPageAllocSize = memoryLimits.BufferPageAllocSize;
                     settings.TransportVersion = outputChannelDesc.GetTransportVersion();
                     settings.Level = StatsModeToCollectStatsLevel(Settings.StatsMode);
 

ydb/library/yql/dq/runtime/dq_tasks_runner.h

@@ -220,6 +220,7 @@ struct TDqTaskRunnerMemoryLimits {
     ui32 OutputChunkMaxSize = 0;
     ui32 ChunkSizeLimit = 48_MB;
     TMaybe<ui8> ArrayBufferMinFillPercentage;
+    TMaybe<size_t> BufferPageAllocSize;
 };
 
 NUdf::TUnboxedValue DqBuildInputValue(const NDqProto::TTaskInput& inputDesc, const NKikimr::NMiniKQL::TType* type,