compaction speedup (#10323)

Author: ivanmorozov333

.github/config/muted_ya.txt

@@ -24,6 +24,15 @@ ydb/core/kqp/ut/olap KqpOlapBlobsSharing.TableReshardingConsistency64
 ydb/core/kqp/ut/olap KqpOlapBlobsSharing.TableReshardingModuloN
 ydb/core/kqp/ut/olap KqpOlapBlobsSharing.UpsertWhileSplitTest
 ydb/core/kqp/ut/olap KqpOlapStatistics.StatsUsageWithTTL
+ydb/core/kqp/ut/olap KqpOlapSysView.StatsSysViewBytesDictStatActualization
+ydb/core/kqp/ut/olap KqpOlapAggregations.Aggregation_SumL_GroupL_OrderL
+ydb/core/kqp/ut/olap KqpOlapIndexes.IndexesActualization
+ydb/core/kqp/ut/olap KqpOlapIndexes.IndexesInBS
+ydb/core/kqp/ut/olap KqpOlapIndexes.IndexesInLocalMetadata
+ydb/core/tx/columnshard/ut_rw Normalizers.CleanEmptyPortionsNormalizer
+ydb/core/kqp/ut/pg KqpPg.CreateIndex
+ydb/core/kqp/ut/query KqpLimits.QueryReplySize
+ydb/core/kqp/ut/query KqpQuery.QueryTimeout
 ydb/core/kqp/ut/query KqpLimits.ComputeActorMemoryAllocationFailureQueryService
 ydb/core/kqp/ut/query KqpLimits.QueryExecTimeoutCancel
 ydb/core/kqp/ut/query KqpStats.SysViewClientLost

ydb/core/base/events.h

@@ -183,6 +183,7 @@ struct TKikimrEvents : TEvents {
         ES_GROUPED_ALLOCATIONS_MANAGER = 4260,
         ES_INCREMENTAL_RESTORE_SCAN = 4261,
         ES_FEATURE_FLAGS = 4262,
+        ES_PRIORITY_QUEUE = 4263,
     };
 };
 

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -185,6 +185,8 @@
 #include <ydb/core/tx/conveyor/service/service.h>
 #include <ydb/core/tx/conveyor/usage/config.h>
 #include <ydb/core/tx/conveyor/usage/service.h>
+#include <ydb/core/tx/priorities/usage/config.h>
+#include <ydb/core/tx/priorities/usage/service.h>
 #include <ydb/core/tx/limiter/service/service.h>
 #include <ydb/core/tx/limiter/usage/config.h>
 #include <ydb/core/tx/limiter/usage/service.h>
@@ -2199,6 +2201,28 @@ void TCompDiskLimiterInitializer::InitializeServices(NActors::TActorSystemSetup*
     }
 }
 
+TCompPrioritiesInitializer::TCompPrioritiesInitializer(const TKikimrRunConfig& runConfig)
+    : IKikimrServicesInitializer(runConfig) {
+}
+
+void TCompPrioritiesInitializer::InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) {
+    NPrioritiesQueue::TConfig serviceConfig;
+    if (Config.HasCompPrioritiesConfig()) {
+        Y_ABORT_UNLESS(serviceConfig.DeserializeFromProto(Config.GetCompPrioritiesConfig()));
+    }
+
+    if (serviceConfig.IsEnabled()) {
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> tabletGroup = GetServiceCounters(appData->Counters, "tablets");
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> conveyorGroup = tabletGroup->GetSubgroup("type", "TX_COMP_PRIORITIES");
+
+        auto service = NPrioritiesQueue::TCompServiceOperator::CreateService(serviceConfig, conveyorGroup);
+
+        setup->LocalServices.push_back(std::make_pair(
+            NPrioritiesQueue::TCompServiceOperator::MakeServiceId(NodeId),
+            TActorSetupCmd(service, TMailboxType::HTSwap, appData->UserPoolId)));
+    }
+}
+
 TCompConveyorInitializer::TCompConveyorInitializer(const TKikimrRunConfig& runConfig)
     : IKikimrServicesInitializer(runConfig) {
 }

ydb/core/driver_lib/run/kikimr_services_initializers.h

@@ -410,6 +410,12 @@ class TGroupedMemoryLimiterInitializer: public IKikimrServicesInitializer {
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 
+class TCompPrioritiesInitializer: public IKikimrServicesInitializer {
+public:
+    TCompPrioritiesInitializer(const TKikimrRunConfig& runConfig);
+    void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
+};
+
 class TCompConveyorInitializer: public IKikimrServicesInitializer {
 public:
     TCompConveyorInitializer(const TKikimrRunConfig& runConfig);

ydb/core/driver_lib/run/run.cpp

@@ -1613,6 +1613,10 @@ TIntrusivePtr<TServiceInitializersList> TKikimrRunner::CreateServiceInitializers
         sil->AddServiceInitializer(new TScanConveyorInitializer(runConfig));
     }
 
+    if (serviceMask.EnableCompPriorities) {
+        sil->AddServiceInitializer(new TCompPrioritiesInitializer(runConfig));
+    }
+
     if (serviceMask.EnableCompConveyor) {
         sil->AddServiceInitializer(new TCompConveyorInitializer(runConfig));
     }

ydb/core/driver_lib/run/service_mask.h

@@ -80,6 +80,7 @@ union TBasicKikimrServicesMask {
         bool EnableCompDiskLimiter:1;
         bool EnableGroupedMemoryLimiter:1;
         bool EnableAwsService:1;
+        bool EnableCompPriorities : 1;
     };
 
     struct {

ydb/core/formats/arrow/accessor/plain/accessor.h

@@ -32,6 +32,10 @@ class TTrivialArray: public IChunkedArray {
     }
 
 public:
+    const std::shared_ptr<arrow::Array>& GetArray() const {
+        return Array;
+    }
+
     TTrivialArray(const std::shared_ptr<arrow::Array>& data)
         : TBase(data->length(), EType::Array, data->type())
         , Array(data) {

ydb/core/formats/arrow/accessor/plain/constructor.cpp

@@ -35,10 +35,15 @@ std::shared_ptr<arrow::Schema> TConstructor::DoGetExpectedSchema(const std::shar
 
 std::shared_ptr<arrow::RecordBatch> TConstructor::DoConstruct(
     const std::shared_ptr<IChunkedArray>& columnData, const TChunkConstructionData& externalInfo) const {
-    auto chunked = columnData->GetChunkedArray();
     auto schema = std::make_shared<arrow::Schema>(arrow::FieldVector({ std::make_shared<arrow::Field>("val", externalInfo.GetColumnType()) }));
-    auto table = arrow::Table::Make(schema, { chunked }, columnData->GetRecordsCount());
-    return NArrow::ToBatch(table, true);
+    if (columnData->GetType() == IChunkedArray::EType::Array) {
+        const auto* arr = static_cast<const TTrivialArray*>(columnData.get());
+        return arrow::RecordBatch::Make(schema, columnData->GetRecordsCount(), { arr->GetArray() });
+    } else {
+        auto chunked = columnData->GetChunkedArray();
+        auto table = arrow::Table::Make(schema, { chunked }, columnData->GetRecordsCount());
+        return NArrow::ToBatch(table, chunked->num_chunks() > 1);
+    }
 }
 
 }   // namespace NKikimr::NArrow::NAccessor::NPlain

ydb/core/formats/arrow/program.cpp

@@ -555,7 +555,7 @@ arrow::Status TDatumBatch::AddColumn(const std::string& name, arrow::Datum&& col
 
     auto field = arrow::field(name, column.type());
     if (!field || !field->type()->Equals(column.type())) {
-        return arrow::Status::Invalid("Cannot create field.");
+        return arrow::Status::Invalid("Cannot create field " + name + ". type:" + field->type()->ToString() + " vs " + column.type()->ToString());
     }
     if (!column.is_scalar() && column.length() != Rows) {
         return arrow::Status::Invalid("Wrong column length.");

ydb/core/formats/arrow/reader/position.cpp

@@ -133,6 +133,15 @@ TSortableScanData::TSortableScanData(
     BuildPosition(position);
 }
 
+TSortableScanData::TSortableScanData(
+    const ui64 position, const std::shared_ptr<arrow::RecordBatch>& batch) {
+    for (auto&& c : batch->columns()) {
+        Columns.emplace_back(std::make_shared<NAccessor::TTrivialArray>(c));
+    }
+    Fields = batch->schema()->fields();
+    BuildPosition(position);
+}
+
 TSortableScanData::TSortableScanData(const ui64 position, const std::shared_ptr<arrow::Table>& batch, const std::vector<std::string>& columns) {
     for (auto&& i : columns) {
         auto c = batch->GetColumnByName(i);