stable-25-2: cherry-pick PR 25095, PR 25725, PR 26340 (#26484)

ydbot · web-flow · commit 305a43884057 · 2025-10-08T12:34:32.000+02:00
diff --git a/ydb/core/formats/arrow/program/abstract.h b/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);
+        // }
     }
 };
 
diff --git a/ydb/core/kqp/ut/olap/aggregations_ut.cpp b/ydb/core/kqp/ut/olap/aggregations_ut.cpp
@@ -488,8 +488,7 @@ Y_UNIT_TEST_SUITE(KqpOlapAggregations) {
                 LIMIT 2
             )")
             .AddExpectedPlanOptions("KqpOlapFilter")
-            .MutableLimitChecker().SetExpectedLimit(2)
-            .SetExpectedResultCount(400); // FIXME (delete this line)
+            .MutableLimitChecker().SetExpectedLimit(2);
         TestAggregations({ testCase });
     }
 
@@ -504,8 +503,7 @@ Y_UNIT_TEST_SUITE(KqpOlapAggregations) {
                 LIMIT 2
             )")
             .AddExpectedPlanOptions("KqpOlapFilter")
-            .MutableLimitChecker().SetExpectedLimit(2)
-            .SetExpectedResultCount(400); // FIXME (delete this line)
+            .MutableLimitChecker().SetExpectedLimit(2);
         TestAggregations({ testCase });
     }
 
@@ -520,8 +518,7 @@ Y_UNIT_TEST_SUITE(KqpOlapAggregations) {
             )")
             .AddExpectedPlanOptions("KqpOlapFilter")
             .AddExpectedPlanOptions("KqpOlapExtractMembers")
-            .MutableLimitChecker().SetExpectedLimit(2)
-            .SetExpectedResultCount(400); // FIXME (delete this line)
+            .MutableLimitChecker().SetExpectedLimit(2);
         TestAggregations({ testCase });
     }
 
@@ -537,8 +534,7 @@ Y_UNIT_TEST_SUITE(KqpOlapAggregations) {
             )")
             .AddExpectedPlanOptions("KqpOlapFilter")
             .AddExpectedPlanOptions("KqpOlapExtractMembers")
-            .MutableLimitChecker().SetExpectedLimit(2)
-            .SetExpectedResultCount(400); // FIXME (delete this line)
+            .MutableLimitChecker().SetExpectedLimit(2);
         TestAggregations({ testCase });
     }
 
diff --git a/ydb/core/tx/columnshard/engines/reader/abstract/read_metadata.h b/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
diff --git a/ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/scanner.cpp b/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) {
diff --git a/ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/sync_points/limit.cpp b/ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/sync_points/limit.cpp
@@ -17,11 +17,8 @@ bool TSyncPointLimitControl::DrainToLimit() {
     if (Collection->GetNextSource()) {
         nextInHeap = TSourceIterator(Collection->GetNextSource());
     }
-    if (Iterators.empty() || (nextInHeap && Iterators.front() < *nextInHeap)) {
-        return false;
-    }
 
-    while (Iterators.size()) {
+    while (Iterators.size() && (!nextInHeap || !(Iterators.front() < *nextInHeap))) {
         if (!Iterators.front().IsFilled()) {
             return false;
         }
diff --git a/ydb/core/tx/columnshard/engines/reader/transaction/tx_scan.cpp b/ydb/core/tx/columnshard/engines/reader/transaction/tx_scan.cpp
@@ -46,9 +46,7 @@ void TTxScan::Complete(const TActorContext& ctx) {
             return TReadMetadataBase::ESorting::NONE;
         }
     }();
-    /* FIXME: #22992 */
-    const auto useLimit = request.HasItemsLimit() && (sorting == ERequestSorting::NONE || !deduplicationEnabled);
-    TScannerConstructorContext context(snapshot, useLimit ? request.GetItemsLimit() : 0, sorting);
+    TScannerConstructorContext context(snapshot, request.HasItemsLimit() ? request.GetItemsLimit() : 0, sorting);
     const auto scanId = request.GetScanId();
     const ui64 txId = request.GetTxId();
     const ui32 scanGen = request.GetGeneration();
@@ -150,6 +148,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);
diff --git a/ydb/tests/olap/order_by_with_limit.py b/ydb/tests/olap/order_by_with_limit.py
@@ -2,6 +2,7 @@
 import os
 import yatest.common
 import ydb
+import random
 
 from ydb.tests.library.harness.kikimr_config import KikimrConfigGenerator
 from ydb.tests.library.harness.kikimr_runner import KiKiMR
@@ -12,9 +13,11 @@
 
 class TestOrderBy(object):
     test_name = "order_by"
+    n = 200
 
     @classmethod
     def setup_class(cls):
+        random.seed(0xBEDA)
         ydb_path = yatest.common.build_path(os.environ.get("YDB_DRIVER_BINARY"))
         logger.info(yatest.common.execute([ydb_path, "-V"], wait=True).stdout.decode("utf-8"))
         config = KikimrConfigGenerator(
@@ -33,20 +36,21 @@ def setup_class(cls):
     def write_data(self, table: str):
         column_types = ydb.BulkUpsertColumns()
         column_types.add_column("id", ydb.PrimitiveType.Uint64)
+        column_types.add_column("value", ydb.PrimitiveType.Uint64)
 
-        for i in range(100):
-            data = [
-                {
-                    "id": i,
-                }
-            ]
-
-            data.extend({"id": i * 100000 + j} for j in range(100))
+        # [2,  4,  6,  8, 10, ...]
+        # [3,  6,  9, 12, 15, ...]
+        # [4,  8, 12, 16, 20, ...]
+        # [5, 10, 15, 20, 25, ...]
+        # ...
+        data = [[{"id": j, "value": j} for j in range(1, self.n) if j % i == 0] for i in range(2, self.n)]
+        random.shuffle(data)
 
+        for row in data:
             self.ydb_client.bulk_upsert(
                 table,
                 column_types,
-                data,
+                row,
             )
 
     def test(self):
@@ -57,6 +61,7 @@ def test(self):
             f"""
             CREATE TABLE `{table_path}` (
                 id Uint64 NOT NULL,
+                value Uint64 NOT NULL,
                 PRIMARY KEY(id),
             )
             WITH (
@@ -68,12 +73,30 @@ def test(self):
 
         self.write_data(table_path)
 
-        result_sets = self.ydb_client.query(
-            f"""
-            select id from `{table_path}` order by id limit 10
-            """
-        )
+        for i in range(100):
+            limit = random.randint(1, self.n)
+            offset = random.randint(1, self.n)
+            is_desc = random.randint(0, 1)
+            is_le = random.randint(0, 1)
+            order = ["asc", "desc"]
+            condition = [">", "<"]
+            data = list(range(2, self.n))
+            if is_le:
+                answer = [i for i in data if i < offset]
+            else:
+                answer = [i for i in data if i > offset]
+            if is_desc:
+                answer = answer[::-1]
+            answer = answer[:limit]
+
+            result_sets = self.ydb_client.query(
+                f"""
+                select id from `{table_path}`
+                where value {condition[is_le]} {offset}
+                order by id {order[is_desc]} limit {limit}
+                """
+            )
 
-        keys = [row['id'] for result_set in result_sets for row in result_set.rows]
+            keys = [row['id'] for result_set in result_sets for row in result_set.rows]
 
-        assert keys == [i for i in range(10)], keys
+            assert keys == answer, keys

Original file line number	Diff line number	Diff line change
`@@ -17,11 +17,8 @@ bool TSyncPointLimitControl::DrainToLimit() {`
`17`	`17`	`if (Collection->GetNextSource()) {`
`18`	`18`	`nextInHeap = TSourceIterator(Collection->GetNextSource());`
`19`	`19`	`}`
`20`		`- if (Iterators.empty() \|\| (nextInHeap && Iterators.front() < *nextInHeap)) {`
`21`		`- return false;`
`22`		`- }`
`23`	`20`
`24`		`- while (Iterators.size()) {`
	`21`	`+ while (Iterators.size() && (!nextInHeap \|\| !(Iterators.front() < *nextInHeap))) {`
`25`	`22`	`if (!Iterators.front().IsFilled()) {`
`26`	`23`	`return false;`
`27`	`24`	`}`