fix filter usage for partial reading (#11835)

Author: ivanmorozov333

ydb/core/kqp/ut/olap/indexes_ut.cpp

@@ -22,6 +22,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
         csController->SetOverridePeriodicWakeupActivationPeriod(TDuration::Seconds(1));
         csController->SetOverrideLagForCompactionBeforeTierings(TDuration::Seconds(1));
         csController->SetOverrideReduceMemoryIntervalLimit(1LLU << 30);
+        csController->SetOverrideMemoryLimitForPortionReading(1e+10);
 
         TLocalHelper(kikimr).CreateTestOlapTable();
         auto tableClient = kikimr.GetTableClient();
@@ -322,6 +323,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
         void Execute() const {
             auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NOlap::TWaitCompactionController>();
             csController->SetOverrideReduceMemoryIntervalLimit(1LLU << 30);
+            csController->SetOverrideMemoryLimitForPortionReading(1e+10);
             TLocalHelper(*Kikimr).CreateTestOlapTable();
             auto tableClient = Kikimr->GetTableClient();
 

ydb/core/protos/config.proto

@@ -1698,6 +1698,7 @@ message TColumnShardConfig {
     optional uint32 SmallPortionDetectSizeLimit = 24 [default = 1048576];  // 1 << 20
     optional bool ColumnChunksV0Usage = 25 [default = true];
     optional bool ColumnChunksV1Usage = 26 [default = true];
+    optional uint64 MemoryLimitScanPortion = 27 [default = 100000000];
 }
 
 message TSchemeShardConfig {

ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetched_data.h

@@ -36,6 +36,7 @@ class TFetchedData {
             return;
         }
         AFL_VERIFY(!DataAdded);
+        UseFilter = value;
     }
 
     bool HasPortionAccessor() const {

ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetching.cpp

@@ -289,7 +289,8 @@ TConclusion<bool> TPortionAccessorFetchingStep::DoExecuteInplace(
 
 TConclusion<bool> TDetectInMem::DoExecuteInplace(const std::shared_ptr<IDataSource>& source, const TFetchingScriptCursor& /*step*/) const {
     if (Columns.GetColumnsCount()) {
-        source->SetSourceInMemory(source->GetColumnRawBytes(Columns.GetColumnIds()) < 1e+8);
+        source->SetSourceInMemory(
+            source->GetColumnRawBytes(Columns.GetColumnIds()) < NYDBTest::TControllers::GetColumnShardController()->GetMemoryLimitScanPortion());
     } else {
         source->SetSourceInMemory(true);
     }

ydb/core/tx/columnshard/hooks/abstract/abstract.h

@@ -130,6 +130,9 @@ class ICSController {
     virtual TDuration DoGetLagForCompactionBeforeTierings(const TDuration defaultValue) const {
         return defaultValue;
     }
+    virtual ui64 DoGetMemoryLimitScanPortion(const ui64 defaultValue) const {
+        return defaultValue;
+    }
 
 private:
     inline static const NKikimrConfig::TColumnShardConfig DefaultConfig = {};
@@ -146,6 +149,9 @@ class ICSController {
         const std::set<NOlap::TSnapshot>& /*snapshotsToSave*/, const std::set<NOlap::TSnapshot>& /*snapshotsToRemove*/) {
     }
 
+    ui64 GetMemoryLimitScanPortion() const {
+        return DoGetMemoryLimitScanPortion(GetConfig().GetMemoryLimitScanPortion());
+    }
     virtual bool CheckPortionForEvict(const NOlap::TPortionInfo& portion) const;
 
     TDuration GetPingCheckPeriod() const {

ydb/core/tx/columnshard/hooks/testing/controller.h

@@ -22,6 +22,7 @@ class TController: public TReadOnlyController {
     YDB_ACCESSOR_DEF(std::optional<TDuration>, OverrideCompactionActualizationLag);
     YDB_ACCESSOR_DEF(std::optional<TDuration>, OverrideTasksActualizationLag);
     YDB_ACCESSOR_DEF(std::optional<TDuration>, OverrideReadTimeoutClean);
+    YDB_ACCESSOR(std::optional<ui64>, OverrideMemoryLimitForPortionReading, 100);
     EOptimizerCompactionWeightControl CompactionControl = EOptimizerCompactionWeightControl::Force;
 
     YDB_ACCESSOR(std::optional<ui64>, OverrideReduceMemoryIntervalLimit, 1024);
@@ -130,6 +131,11 @@ class TController: public TReadOnlyController {
     THashSet<TString> SharingIds;
 protected:
     virtual ::NKikimr::NColumnShard::TBlobPutResult::TPtr OverrideBlobPutResultOnCompaction(const ::NKikimr::NColumnShard::TBlobPutResult::TPtr original, const NOlap::TWriteActionsCollection& actions) const override;
+
+    virtual ui64 DoGetMemoryLimitScanPortion(const ui64 defaultValue) const override {
+        return OverrideMemoryLimitForPortionReading.value_or(defaultValue);
+    }
+
     virtual TDuration DoGetLagForCompactionBeforeTierings(const TDuration def) const override {
         return OverrideLagForCompactionBeforeTierings.value_or(def);
     }