From 99f2f35030ebb3ec37d17afda2bbd758c49399b3 Mon Sep 17 00:00:00 2001 From: ivanmorozov333 Date: Thu, 21 Nov 2024 17:15:03 +0300 Subject: [PATCH] fix filter usage for partial reading (#11835) --- ydb/core/kqp/ut/olap/indexes_ut.cpp | 2 ++ ydb/core/protos/config.proto | 1 + .../engines/reader/plain_reader/iterator/fetched_data.h | 1 + .../engines/reader/plain_reader/iterator/fetching.cpp | 3 ++- ydb/core/tx/columnshard/hooks/abstract/abstract.h | 6 ++++++ ydb/core/tx/columnshard/hooks/testing/controller.h | 6 ++++++ 6 files changed, 18 insertions(+), 1 deletion(-) diff --git a/ydb/core/kqp/ut/olap/indexes_ut.cpp b/ydb/core/kqp/ut/olap/indexes_ut.cpp index caaf1b44c29d..e553efb34df6 100644 --- a/ydb/core/kqp/ut/olap/indexes_ut.cpp +++ b/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(); csController->SetOverrideReduceMemoryIntervalLimit(1LLU << 30); + csController->SetOverrideMemoryLimitForPortionReading(1e+10); TLocalHelper(*Kikimr).CreateTestOlapTable(); auto tableClient = Kikimr->GetTableClient(); diff --git a/ydb/core/protos/config.proto b/ydb/core/protos/config.proto index f3799a244b94..a18f14b8fbfd 100644 --- a/ydb/core/protos/config.proto +++ b/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 { diff --git a/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetched_data.h b/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetched_data.h index 1a6a78cf53c2..cfd2113bb15a 100644 --- a/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetched_data.h +++ b/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 { diff --git a/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetching.cpp b/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetching.cpp index fbeaba7f098b..a7461b6b76ae 100644 --- a/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetching.cpp +++ b/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/fetching.cpp @@ -289,7 +289,8 @@ TConclusion TPortionAccessorFetchingStep::DoExecuteInplace( TConclusion TDetectInMem::DoExecuteInplace(const std::shared_ptr& 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); } diff --git a/ydb/core/tx/columnshard/hooks/abstract/abstract.h b/ydb/core/tx/columnshard/hooks/abstract/abstract.h index 010e543e2aca..cc3856edface 100644 --- a/ydb/core/tx/columnshard/hooks/abstract/abstract.h +++ b/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& /*snapshotsToSave*/, const std::set& /*snapshotsToRemove*/) { } + ui64 GetMemoryLimitScanPortion() const { + return DoGetMemoryLimitScanPortion(GetConfig().GetMemoryLimitScanPortion()); + } virtual bool CheckPortionForEvict(const NOlap::TPortionInfo& portion) const; TDuration GetPingCheckPeriod() const { diff --git a/ydb/core/tx/columnshard/hooks/testing/controller.h b/ydb/core/tx/columnshard/hooks/testing/controller.h index a8e259877fd0..7b2028813412 100644 --- a/ydb/core/tx/columnshard/hooks/testing/controller.h +++ b/ydb/core/tx/columnshard/hooks/testing/controller.h @@ -22,6 +22,7 @@ class TController: public TReadOnlyController { YDB_ACCESSOR_DEF(std::optional, OverrideCompactionActualizationLag); YDB_ACCESSOR_DEF(std::optional, OverrideTasksActualizationLag); YDB_ACCESSOR_DEF(std::optional, OverrideReadTimeoutClean); + YDB_ACCESSOR(std::optional, OverrideMemoryLimitForPortionReading, 100); EOptimizerCompactionWeightControl CompactionControl = EOptimizerCompactionWeightControl::Force; YDB_ACCESSOR(std::optional, OverrideReduceMemoryIntervalLimit, 1024); @@ -130,6 +131,11 @@ class TController: public TReadOnlyController { THashSet 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); }