cs merge new changes from main

.github/config/muted_ya.txt

@@ -26,6 +26,10 @@ ydb/core/kqp/ut/olap KqpOlapSysView.StatsSysViewBytesDictActualization
 ydb/core/kqp/ut/olap KqpOlapSysView.StatsSysViewBytesDictStatActualization
 ydb/core/kqp/ut/olap KqpOlapJson.DuplicationCompactionVariants
 ydb/core/kqp/ut/olap KqpOlapWrite.TierDraftsGCWithRestart
+ydb/core/kqp/ut/batch_operations [*/*] chunk chunk
+ydb/core/kqp/ut/federated_query/large_results KqpScriptExecResults.ExecuteScriptWithLargeFile
+ydb/core/kqp/ut/federated_query/s3 sole chunk chunk
+ydb/core/kqp/ut/indexes KqpMultishardIndex.WriteIntoRenamingSyncIndex
 ydb/core/kqp/ut/olap [*/*] chunk chunk
 ydb/core/kqp/ut/join KqpIndexLookupJoin.LeftJoinRightNullFilter+StreamLookup
 ydb/core/kqp/ut/join KqpIndexLookupJoin.LeftJoinRightNullFilter-StreamLookup

ydb/core/base/events.h

@@ -185,6 +185,10 @@ struct TKikimrEvents : TEvents {
         ES_FEATURE_FLAGS = 4262,
         ES_PRIORITY_QUEUE = 4263,
         ES_SOLOMON_PROVIDER = 4264,
+        ES_CONVEYOR_COMPOSITE = 4265,
+        ES_GENERAL_CACHE_PUBLIC = 4266,
+        ES_GENERAL_CACHE_SOURCE = 4267,
+
     };
 };
 

ydb/core/base/localdb.cpp

@@ -262,4 +262,8 @@ const TString KqpResourceManagerTaskName = "kqp_query";
 const TString KqpResourceManagerQueue = "queue_kqp_resource_manager";
 const TString LegacyQueueIdTaskNamePrefix = "compaction_gen";
 
+const TString ColumnShardCompactionIndexationQueue = "queue_cs_indexation";
+const TString ColumnShardCompactionTtlQueue = "queue_cs_ttl";
+const TString ColumnShardCompactionGeneralQueue = "queue_cs_general";
+const TString ColumnShardCompactionNormalizerQueue = "queue_cs_normalizer";
 }}

ydb/core/base/localdb.h

@@ -152,4 +152,8 @@ extern const TString KqpResourceManagerTaskName;
 extern const TString KqpResourceManagerQueue;
 extern const TString LegacyQueueIdTaskNamePrefix;
 
+extern const TString ColumnShardCompactionIndexationQueue;
+extern const TString ColumnShardCompactionTtlQueue;
+extern const TString ColumnShardCompactionGeneralQueue;
+extern const TString ColumnShardCompactionNormalizerQueue;
 }}

ydb/core/driver_lib/run/kikimr_services_initializers.h

@@ -404,33 +404,33 @@ class TCompDiskLimiterInitializer: public IKikimrServicesInitializer {
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 
-class TGroupedMemoryLimiterInitializer: public IKikimrServicesInitializer {
+class TScanGroupedMemoryLimiterInitializer: public IKikimrServicesInitializer {
 public:
-    TGroupedMemoryLimiterInitializer(const TKikimrRunConfig& runConfig);
+    TScanGroupedMemoryLimiterInitializer(const TKikimrRunConfig& runConfig);
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 
-class TCompPrioritiesInitializer: public IKikimrServicesInitializer {
+class TCompGroupedMemoryLimiterInitializer: public IKikimrServicesInitializer {
 public:
-    TCompPrioritiesInitializer(const TKikimrRunConfig& runConfig);
+    TCompGroupedMemoryLimiterInitializer(const TKikimrRunConfig& runConfig);
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 
-class TCompConveyorInitializer: public IKikimrServicesInitializer {
+class TCompPrioritiesInitializer: public IKikimrServicesInitializer {
 public:
-    TCompConveyorInitializer(const TKikimrRunConfig& runConfig);
+    TCompPrioritiesInitializer(const TKikimrRunConfig& runConfig);
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 
-class TScanConveyorInitializer: public IKikimrServicesInitializer {
+class TCompositeConveyorInitializer : public IKikimrServicesInitializer {
 public:
-    TScanConveyorInitializer(const TKikimrRunConfig& runConfig);
-    void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
+    TCompositeConveyorInitializer(const TKikimrRunConfig& runConfig);
+	void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 
-class TInsertConveyorInitializer: public IKikimrServicesInitializer {
+class TGeneralCachePortionsMetadataInitializer: public IKikimrServicesInitializer {
 public:
-    TInsertConveyorInitializer(const TKikimrRunConfig& runConfig);
+    TGeneralCachePortionsMetadataInitializer(const TKikimrRunConfig& runConfig);
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 

ydb/core/driver_lib/run/run.cpp

@@ -1700,23 +1700,20 @@ TIntrusivePtr<TServiceInitializersList> TKikimrRunner::CreateServiceInitializers
     }
 
     if (serviceMask.EnableGroupedMemoryLimiter) {
-        sil->AddServiceInitializer(new TGroupedMemoryLimiterInitializer(runConfig));
-    }
-
-    if (serviceMask.EnableScanConveyor) {
-        sil->AddServiceInitializer(new TScanConveyorInitializer(runConfig));
+        sil->AddServiceInitializer(new TScanGroupedMemoryLimiterInitializer(runConfig));
+        sil->AddServiceInitializer(new TCompGroupedMemoryLimiterInitializer(runConfig));
     }
 
     if (serviceMask.EnableCompPriorities) {
         sil->AddServiceInitializer(new TCompPrioritiesInitializer(runConfig));
     }
 
-    if (serviceMask.EnableCompConveyor) {
-        sil->AddServiceInitializer(new TCompConveyorInitializer(runConfig));
+    if (serviceMask.EnableCompConveyor || serviceMask.EnableInsertConveyor || serviceMask.EnableScanConveyor) {
+        sil->AddServiceInitializer(new TCompositeConveyorInitializer(runConfig));
     }
 
-    if (serviceMask.EnableInsertConveyor) {
-        sil->AddServiceInitializer(new TInsertConveyorInitializer(runConfig));
+    if (serviceMask.EnableGeneralCachePortionsMetadata) {
+        sil->AddServiceInitializer(new TGeneralCachePortionsMetadataInitializer(runConfig));
     }
 
     if (serviceMask.EnableCms) {

ydb/core/driver_lib/run/service_mask.h

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

ydb/core/driver_lib/run/ya.make

@@ -106,10 +106,11 @@ PEERDIR(
     ydb/core/tx
     ydb/core/tx/columnshard
     ydb/core/tx/conveyor/service
+    ydb/core/tx/general_cache
+    ydb/core/tx/columnshard/data_accessor/cache_policy
     ydb/core/tx/coordinator
     ydb/core/tx/datashard
     ydb/core/tx/limiter/grouped_memory/usage
-    ydb/core/tx/limiter/service
     ydb/core/tx/long_tx_service
     ydb/core/tx/long_tx_service/public
     ydb/core/tx/mediator
@@ -142,6 +143,7 @@ PEERDIR(
     ydb/library/security
     ydb/library/signal_backtrace
     ydb/library/yql/providers/pq/cm_client
+    ydb/library/slide_limiter/service
     ydb/library/yql/providers/s3/actors
     ydb/public/lib/base
     ydb/public/lib/deprecated/client

ydb/core/formats/arrow/arrow_filter.cpp

@@ -297,6 +297,9 @@ const std::vector<bool>& TColumnFilter::BuildSimpleFilter() const {
 class TMergePolicyAnd {
 private:
 public:
+    static bool HasValue(const bool /*a*/, const bool /*b*/) {
+        return true;
+    }
     static bool Calc(const bool a, const bool b) {
         return a && b;
     }
@@ -307,11 +310,17 @@ class TMergePolicyAnd {
             return TColumnFilter::BuildDenyFilter();
         }
     }
+    static TColumnFilter MergeWithSimple(const bool simpleValue, const TColumnFilter& filter) {
+        return MergeWithSimple(filter, simpleValue);
+    }
 };
 
 class TMergePolicyOr {
 private:
 public:
+    static bool HasValue(const bool /*a*/, const bool /*b*/) {
+        return true;
+    }
     static bool Calc(const bool a, const bool b) {
         return a || b;
     }
@@ -322,6 +331,43 @@ class TMergePolicyOr {
             return filter;
         }
     }
+    static TColumnFilter MergeWithSimple(const bool simpleValue, const TColumnFilter& filter) {
+        return MergeWithSimple(filter, simpleValue);
+    }
+};
+
+class TMergePolicyApplyFilter {
+private:
+public:
+    static bool HasValue(const bool /*a*/, const bool b) {
+        return b;
+    }
+    static bool Calc(const bool a, const bool /*b*/) {
+        return a;
+    }
+    static TColumnFilter MergeWithSimple(const TColumnFilter& filter, const bool simpleValue) {
+        if (simpleValue) {
+            return filter;
+        } else {
+            return TColumnFilter::BuildDenyFilter();
+        }
+    }
+    static TColumnFilter MergeWithSimple(const bool simpleValue, const TColumnFilter& filter) {
+        const auto count = filter.GetRecordsCount();
+        if (simpleValue) {
+            TColumnFilter result = TColumnFilter::BuildAllowFilter();
+            if (count) {
+                result.Add(true, *count);
+            }
+            return result;
+        } else {
+            TColumnFilter result = TColumnFilter::BuildDenyFilter();
+            if (count) {
+                result.Add(false, *count);
+            }
+            return result;
+        }
+    }
 };
 
 class TColumnFilter::TMergerImpl {
@@ -340,7 +386,7 @@ class TColumnFilter::TMergerImpl {
         if (Filter1.empty() && Filter2.empty()) {
             return TColumnFilter(TMergePolicy::Calc(Filter1.DefaultFilterValue, Filter2.DefaultFilterValue));
         } else if (Filter1.empty()) {
-            return TMergePolicy::MergeWithSimple(Filter2, Filter1.DefaultFilterValue);
+            return TMergePolicy::MergeWithSimple(Filter1.DefaultFilterValue, Filter2);
         } else if (Filter2.empty()) {
             return TMergePolicy::MergeWithSimple(Filter1, Filter2.DefaultFilterValue);
         } else {
@@ -359,7 +405,7 @@ class TColumnFilter::TMergerImpl {
 
             while (it1 != Filter1.Filter.cend() && it2 != Filter2.Filter.cend()) {
                 const ui32 delta = TColumnFilter::CrossSize(pos2, pos2 + *it2, pos1, pos1 + *it1);
-                if (delta) {
+                if (delta && TMergePolicy::HasValue(curValue1, curValue2)) {
                     if (!count || curCurrent != TMergePolicy::Calc(curValue1, curValue2)) {
                         resultFilter.emplace_back(delta);
                         curCurrent = TMergePolicy::Calc(curValue1, curValue2);
@@ -405,10 +451,15 @@ TColumnFilter TColumnFilter::Or(const TColumnFilter& extFilter) const {
     return TMergerImpl(*this, extFilter).Merge<TMergePolicyOr>();
 }
 
+TColumnFilter TColumnFilter::ApplyFilterFrom(const TColumnFilter& extFilter) const {
+    ResetCaches();
+    return TMergerImpl(*this, extFilter).Merge<TMergePolicyApplyFilter>();
+}
+
 TColumnFilter TColumnFilter::CombineSequentialAnd(const TColumnFilter& extFilter) const {
     ResetCaches();
     if (Filter.empty()) {
-        return TMergePolicyAnd::MergeWithSimple(extFilter, DefaultFilterValue);
+        return TMergePolicyAnd::MergeWithSimple(DefaultFilterValue, extFilter);
     } else if (extFilter.Filter.empty()) {
         return TMergePolicyAnd::MergeWithSimple(*this, extFilter.DefaultFilterValue);
     } else {
@@ -467,14 +518,19 @@ TColumnFilter TColumnFilter::CombineSequentialAnd(const TColumnFilter& extFilter
     }
 }
 
-TColumnFilter::TIterator TColumnFilter::GetIterator(const bool reverse, const ui32 expectedSize) const {
+TColumnFilter::TIterator TColumnFilter::GetBegin(const bool reverse, const ui32 expectedSize) const {
+    return GetIterator(reverse, expectedSize, 0);
+}
+
+TColumnFilter::TIterator TColumnFilter::GetIterator(const bool reverse, const ui32 expectedSize, const ui64 startOffset) const {
+    AFL_VERIFY(expectedSize >= startOffset);
     if (IsTotalAllowFilter()) {
-        return TIterator(reverse, expectedSize, true);
+        return TIterator(reverse, expectedSize, true, startOffset);
     } else if (IsTotalDenyFilter()) {
-        return TIterator(reverse, expectedSize, false);
+        return TIterator(reverse, expectedSize, false, startOffset);
     } else {
         AFL_VERIFY(expectedSize == GetRecordsCountVerified())("expected", expectedSize)("count", GetRecordsCountVerified())("reverse", reverse);
-        return TIterator(reverse, Filter, GetStartValue(reverse));
+        return TIterator(reverse, Filter, GetStartValue(reverse), startOffset);
     }
 }
 

ydb/core/formats/arrow/arrow_filter.h

@@ -150,7 +150,7 @@ class TColumnFilter {
     public:
         TString DebugString() const;
 
-        TIterator(const bool reverse, const std::vector<ui32>& filter, const bool startValue)
+        TIterator(const bool reverse, const std::vector<ui32>& filter, const bool startValue, const ui64 startOffset)
             : FilterPointer(&filter)
             , CurrentValue(startValue)
             , FinishPosition(reverse ? -1 : FilterPointer->size())
@@ -163,9 +163,12 @@ class TColumnFilter {
                 }
                 CurrentRemainVolume = (*FilterPointer)[Position];
             }
+            if (startOffset) {
+                Next(startOffset);
+            }
         }
 
-        TIterator(const bool reverse, const ui32 size, const bool startValue)
+        TIterator(const bool reverse, const ui32 size, const bool startValue, const ui64 startOffset)
             : CurrentValue(startValue)
             , FinishPosition(reverse ? -1 : 1)
             , DeltaPosition(reverse ? -1 : 1) {
@@ -177,6 +180,9 @@ class TColumnFilter {
                 }
                 CurrentRemainVolume = size;
             }
+            if (startOffset) {
+                Next(startOffset);
+            }
         }
 
         bool GetCurrentAcceptance() const {
@@ -194,7 +200,8 @@ class TColumnFilter {
 
     TString DebugString() const;
 
-    TIterator GetIterator(const bool reverse, const ui32 expectedSize) const;
+    TIterator GetBegin(const bool reverse, const ui32 expectedSize) const;
+    TIterator GetIterator(const bool reverse, const ui32 expectedSize, const ui64 startOffset) const;
 
     bool CheckSlice(const ui32 offset, const ui32 count) const;
 
@@ -263,8 +270,9 @@ class TColumnFilter {
         return TColumnFilter(false);
     }
 
-    TColumnFilter And(const TColumnFilter& extFilter) const Y_WARN_UNUSED_RESULT;
-    TColumnFilter Or(const TColumnFilter& extFilter) const Y_WARN_UNUSED_RESULT;
+    [[nodiscard]] TColumnFilter And(const TColumnFilter& extFilter) const;
+    [[nodiscard]] TColumnFilter Or(const TColumnFilter& extFilter) const;
+    [[nodiscard]] TColumnFilter ApplyFilterFrom(const TColumnFilter& filter) const;
 
     class TApplyContext {
     private: