fix(query): aviod building TransformSortMergeLimit with a big number limit

src/query/pipeline/transforms/src/processors/transforms/transform_sort_merge.rs

@@ -43,7 +43,7 @@ use crate::processors::sort::Merger;
 pub struct TransformSortMerge<R: Rows> {
     schema: DataSchemaRef,
     enable_loser_tree: bool,
-
+    limit: Option<usize>,
     block_size: usize,
     buffer: Vec<Option<(DataBlock, Column)>>,
 
@@ -62,10 +62,12 @@ impl<R: Rows> TransformSortMerge<R> {
         _sort_desc: Arc<Vec<SortColumnDescription>>,
         block_size: usize,
         enable_loser_tree: bool,
+        limit: Option<usize>,
     ) -> Self {
         TransformSortMerge {
             schema,
             enable_loser_tree,
+            limit,
             block_size,
             buffer: vec![],
             aborting: Arc::new(AtomicBool::new(false)),
@@ -171,7 +173,8 @@ impl<R: Rows> TransformSortMerge<R> {
         let streams = self.buffer.drain(..).collect::<Vec<BlockStream>>();
         let mut result = Vec::with_capacity(size_hint);
 
-        let mut merger = Merger::<A, _>::create(self.schema.clone(), streams, batch_size, None);
+        let mut merger =
+            Merger::<A, _>::create(self.schema.clone(), streams, batch_size, self.limit);
 
         while let Some(block) = merger.next_block()? {
             if unlikely(self.aborting.load(Ordering::Relaxed)) {
@@ -218,7 +221,7 @@ pub fn sort_merge(
         0,
         0,
         sort_spilling_batch_bytes,
-        MergeSortCommonImpl::create(schema, sort_desc, block_size, enable_loser_tree),
+        MergeSortCommonImpl::create(schema, sort_desc, block_size, enable_loser_tree, None),
     )?;
     for block in data_blocks {
         processor.transform(block)?;

src/query/pipeline/transforms/src/processors/transforms/transform_sort_merge_base.rs

@@ -337,7 +337,7 @@ impl TransformSortMergeBuilder {
             !self.schema.has_field(ORDER_COL_NAME)
         });
 
-        if self.limit.is_some() {
+        if self.limit.map(|limit| limit < 10000).unwrap_or_default() {
             self.build_sort_limit()
         } else {
             self.build_sort()
@@ -400,6 +400,7 @@ impl TransformSortMergeBuilder {
                     self.sort_desc,
                     self.block_size,
                     self.enable_loser_tree,
+                    self.limit,
                 ),
             )?,
         ))

src/query/pipeline/transforms/src/processors/transforms/transform_sort_merge_limit.rs

@@ -127,7 +127,6 @@ impl<R: Rows> CursorOrder<R> for LocalCursorOrder {
 
 impl<R: Rows> TransformSortMergeLimit<R> {
     pub fn create(block_size: usize, limit: usize) -> Self {
-        debug_assert!(limit <= 10000, "Too large sort merge limit: {}", limit);
         TransformSortMergeLimit {
             heap: FixedHeap::new(limit),
             buffer: HashMap::with_capacity(limit),

src/query/service/tests/it/pipelines/transforms/sort/spill.rs

@@ -71,6 +71,7 @@ fn create_sort_spill_pipeline(
                 sort_desc.clone(),
                 block_size,
                 enable_loser_tree,
+                None,
             ),
         )
     })?;

tests/sqllogictests/suites/base/03_common/03_0004_select_order_by.test

@@ -113,6 +113,18 @@ INSERT INTO t3 VALUES(1, [1,2,3]), (2, [1,2,4]), (3, []), (4, [3,4,5]), (5, [4])
 statement ok
 DROP TABLE t3
 
+query I
+SELECT number FROM numbers(3) ORDER BY number LIMIT 214748364900;
+----
+0
+1
+2
+
+query I
+SELECT number FROM numbers(100000) ORDER BY number LIMIT 1 OFFSET 50000;
+----
+50000
+
 query I
 SELECT number FROM numbers(10000) ORDER BY number LIMIT 3
 ----