stable-25-3-1: cherry-pick PR 25725, PR 26340 (#26501)

Author: ydbot

ydb/core/formats/arrow/program/abstract.h

@@ -48,12 +48,14 @@ class TFetchingCalculationPolicy: public IMemoryCalculationPolicy {
         return EStage::Fetching;
     }
     virtual ui64 GetReserveMemorySize(
-        const ui64 blobsSize, const ui64 rawSize, const std::optional<ui32> limit, const ui32 recordsCount) const override {
-        if (limit) {
-            return std::max<ui64>(blobsSize, rawSize * (1.0 * *limit) / recordsCount);
-        } else {
-            return std::max<ui64>(blobsSize, rawSize);
-        }
+        const ui64 blobsSize, const ui64 rawSize, const std::optional<ui32> /*limit*/, const ui32 /*recordsCount*/) const override {
+        return std::max<ui64>(blobsSize, rawSize);
+        // FIXME after futher memory usage investagiation
+        // if (limit) {
+        //     return std::max<ui64>(blobsSize, rawSize * (1.0 * *limit) / recordsCount);
+        // } else {
+        //     return std::max<ui64>(blobsSize, rawSize);
+        // }
     }
 };
 

ydb/core/tx/columnshard/engines/reader/abstract/read_metadata.h

@@ -24,6 +24,7 @@ class TReadMetadataBase {
 
 private:
     YDB_ACCESSOR_DEF(TString, ScanIdentifier);
+    YDB_ACCESSOR_DEF(bool, FakeSort);
     std::optional<ui64> FilteredCountLimit;
     std::optional<ui64> RequestedLimit;
     const ESorting Sorting = ESorting::ASC;   // Sorting inside returned batches

ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/scanner.cpp

@@ -21,14 +21,15 @@ TConclusionStatus TScanHead::Start() {
 
 TScanHead::TScanHead(std::unique_ptr<NCommon::ISourcesConstructor>&& sourcesConstructor, const std::shared_ptr<TSpecialReadContext>& context)
     : Context(context) {
+    auto readMetadataContext = context->GetReadMetadata();
     if (auto script = Context->GetSourcesAggregationScript()) {
         SourcesCollection =
-            std::make_shared<TNotSortedCollection>(Context, std::move(sourcesConstructor), Context->GetReadMetadata()->GetLimitRobustOptional());
+            std::make_shared<TNotSortedCollection>(Context, std::move(sourcesConstructor), readMetadataContext->GetLimitRobustOptional());
         SyncPoints.emplace_back(std::make_shared<TSyncPointResult>(SyncPoints.size(), context, SourcesCollection));
         SyncPoints.emplace_back(std::make_shared<TSyncPointResultsAggregationControl>(
             SourcesCollection, Context->GetSourcesAggregationScript(), Context->GetRestoreResultScript(), SyncPoints.size(), context));
-    } else if (Context->GetReadMetadata()->IsSorted()) {
-        if (Context->GetReadMetadata()->HasLimit()) {
+    } else if (readMetadataContext->IsSorted()) {
+        if (readMetadataContext->HasLimit() && !readMetadataContext->GetFakeSort()) {
             auto collection = std::make_shared<TScanWithLimitCollection>(Context, std::move(sourcesConstructor));
             SourcesCollection = collection;
             SyncPoints.emplace_back(std::make_shared<TSyncPointLimitControl>(
@@ -39,7 +40,7 @@ TScanHead::TScanHead(std::unique_ptr<NCommon::ISourcesConstructor>&& sourcesCons
         SyncPoints.emplace_back(std::make_shared<TSyncPointResult>(SyncPoints.size(), context, SourcesCollection));
     } else {
         SourcesCollection =
-            std::make_shared<TNotSortedCollection>(Context, std::move(sourcesConstructor), Context->GetReadMetadata()->GetLimitRobustOptional());
+            std::make_shared<TNotSortedCollection>(Context, std::move(sourcesConstructor), readMetadataContext->GetLimitRobustOptional());
         SyncPoints.emplace_back(std::make_shared<TSyncPointResult>(SyncPoints.size(), context, SourcesCollection));
     }
     for (ui32 i = 0; i + 1 < SyncPoints.size(); ++i) {

ydb/core/tx/columnshard/engines/reader/transaction/tx_scan.cpp

@@ -147,6 +147,9 @@ void TTxScan::Complete(const TActorContext& ctx) {
             }
             auto newRange = scannerConstructor->BuildReadMetadata(Self, read);
             if (newRange.IsSuccess()) {
+                if (!request.HasReverse() && deduplicationEnabled) {
+                    (*newRange)->SetFakeSort(true);
+                }
                 readMetadataRange = TValidator::CheckNotNull(newRange.DetachResult());
             } else {
                 return SendError("cannot build metadata", newRange.GetErrorMessage(), ctx);