diff --git a/core/trino-main/src/main/java/io/trino/operator/TopNOperator.java b/core/trino-main/src/main/java/io/trino/operator/TopNOperator.java
index aed9598f8222..0c6407e3c416 100644
--- a/core/trino-main/src/main/java/io/trino/operator/TopNOperator.java
+++ b/core/trino-main/src/main/java/io/trino/operator/TopNOperator.java
@@ -14,6 +14,7 @@
 package io.trino.operator;
 
 import com.google.common.collect.ImmutableList;
+import io.airlift.units.DataSize;
 import io.trino.memory.context.MemoryTrackingContext;
 import io.trino.operator.BasicWorkProcessorOperatorAdapter.BasicAdapterWorkProcessorOperatorFactory;
 import io.trino.operator.WorkProcessor.TransformationState;
@@ -24,6 +25,7 @@
 import io.trino.sql.planner.plan.PlanNodeId;
 
 import java.util.List;
+import java.util.Optional;
 
 import static com.google.common.base.Preconditions.checkState;
 import static io.trino.operator.BasicWorkProcessorOperatorAdapter.createAdapterOperatorFactory;
@@ -42,9 +44,10 @@ public static OperatorFactory createOperatorFactory(
             int n,
             List<Integer> sortChannels,
             List<SortOrder> sortOrders,
-            TypeOperators typeOperators)
+            TypeOperators typeOperators,
+            Optional<DataSize> maxPartialMemory)
     {
-        return createAdapterOperatorFactory(new Factory(operatorId, planNodeId, types, n, sortChannels, sortOrders, typeOperators));
+        return createAdapterOperatorFactory(new Factory(operatorId, planNodeId, types, n, sortChannels, sortOrders, typeOperators, maxPartialMemory));
     }
 
     private static class Factory
@@ -57,6 +60,7 @@ private static class Factory
         private final List<Integer> sortChannels;
         private final List<SortOrder> sortOrders;
         private final TypeOperators typeOperators;
+        private final Optional<DataSize> maxPartialMemory;
         private boolean closed;
 
         private Factory(
@@ -66,7 +70,8 @@ private Factory(
                 int n,
                 List<Integer> sortChannels,
                 List<SortOrder> sortOrders,
-                TypeOperators typeOperators)
+                TypeOperators typeOperators,
+                Optional<DataSize> maxPartialMemory)
         {
             this.operatorId = operatorId;
             this.planNodeId = requireNonNull(planNodeId, "planNodeId is null");
@@ -75,6 +80,7 @@ private Factory(
             this.sortChannels = ImmutableList.copyOf(requireNonNull(sortChannels, "sortChannels is null"));
             this.sortOrders = ImmutableList.copyOf(requireNonNull(sortOrders, "sortOrders is null"));
             this.typeOperators = typeOperators;
+            this.maxPartialMemory = requireNonNull(maxPartialMemory, "maxPartialMemory is null");
         }
 
         @Override
@@ -90,7 +96,8 @@ public WorkProcessorOperator create(
                     n,
                     sortChannels,
                     sortOrders,
-                    typeOperators);
+                    typeOperators,
+                    maxPartialMemory);
         }
 
         @Override
@@ -120,11 +127,10 @@ public void close()
         @Override
         public Factory duplicate()
         {
-            return new Factory(operatorId, planNodeId, sourceTypes, n, sortChannels, sortOrders, typeOperators);
+            return new Factory(operatorId, planNodeId, sourceTypes, n, sortChannels, sortOrders, typeOperators, maxPartialMemory);
         }
     }
 
-    private final TopNProcessor topNProcessor;
     private final WorkProcessor<Page> pages;
 
     private TopNOperator(
@@ -134,21 +140,25 @@ private TopNOperator(
             int n,
             List<Integer> sortChannels,
             List<SortOrder> sortOrders,
-            TypeOperators typeOperators)
+            TypeOperators typeOperators,
+            Optional<DataSize> maxPartialMemory)
     {
-        this.topNProcessor = new TopNProcessor(
-                requireNonNull(memoryTrackingContext, "memoryTrackingContext is null").aggregateUserMemoryContext(),
-                types,
-                n,
-                sortChannels,
-                sortOrders,
-                typeOperators);
+        requireNonNull(memoryTrackingContext, "memoryTrackingContext is null");
 
         if (n == 0) {
             pages = WorkProcessor.of();
         }
         else {
-            pages = sourcePages.transform(new TopNPages());
+            TopNProcessor topNProcessor = new TopNProcessor(
+                    memoryTrackingContext.aggregateUserMemoryContext(),
+                    types,
+                    n,
+                    sortChannels,
+                    sortOrders,
+                    typeOperators);
+            long maxPartialMemoryWithDefaultValueIfAbsent = requireNonNull(maxPartialMemory, "maxPartialMemory is null")
+                    .map(DataSize::toBytes).orElse(Long.MAX_VALUE);
+            pages = sourcePages.transform(new TopNPages(topNProcessor, maxPartialMemoryWithDefaultValueIfAbsent));
         }
     }
 
@@ -158,18 +168,44 @@ public WorkProcessor<Page> getOutputPages()
         return pages;
     }
 
-    private class TopNPages
+    private static class TopNPages
             implements WorkProcessor.Transformation<Page, Page>
     {
+        private final TopNProcessor topNProcessor;
+        private final long maxPartialMemory;
+
+        private boolean isPartialFlushing;
+
+        private TopNPages(TopNProcessor topNProcessor, long maxPartialMemory)
+        {
+            this.topNProcessor = topNProcessor;
+            this.maxPartialMemory = maxPartialMemory;
+        }
+
+        private boolean isBuilderFull()
+        {
+            return topNProcessor.getEstimatedSizeInBytes() >= maxPartialMemory;
+        }
+
+        private void addPage(Page page)
+        {
+            checkState(!isPartialFlushing, "TopN buffer is already full");
+            topNProcessor.addInput(page);
+            if (isBuilderFull()) {
+                isPartialFlushing = true;
+            }
+        }
+
         @Override
         public TransformationState<Page> process(Page inputPage)
         {
-            if (inputPage != null) {
-                topNProcessor.addInput(inputPage);
-                return TransformationState.needsMoreData();
+            if (!isPartialFlushing && inputPage != null) {
+                addPage(inputPage);
+                if (!isPartialFlushing) {
+                    return TransformationState.needsMoreData();
+                }
             }
 
-            // no more input, return results
             Page page = null;
             while (page == null && !topNProcessor.noMoreOutput()) {
                 page = topNProcessor.getOutput();
@@ -179,6 +215,14 @@ public TransformationState<Page> process(Page inputPage)
                 return TransformationState.ofResult(page, false);
             }
 
+            if (isPartialFlushing) {
+                checkState(inputPage != null, "inputPage that triggered partial flushing is null");
+                isPartialFlushing = false;
+                // resume receiving pages
+                return TransformationState.needsMoreData();
+            }
+
+            // all input pages are consumed
             return TransformationState.finished();
         }
     }
diff --git a/core/trino-main/src/main/java/io/trino/operator/TopNProcessor.java b/core/trino-main/src/main/java/io/trino/operator/TopNProcessor.java
index 92e10b236171..5c371d2c8c5f 100644
--- a/core/trino-main/src/main/java/io/trino/operator/TopNProcessor.java
+++ b/core/trino-main/src/main/java/io/trino/operator/TopNProcessor.java
@@ -24,6 +24,7 @@
 
 import java.util.Iterator;
 import java.util.List;
+import java.util.function.Supplier;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Verify.verify;
@@ -35,10 +36,13 @@
  */
 public class TopNProcessor
 {
-    private final LocalMemoryContext localUserMemoryContext;
+    private final LocalMemoryContext localMemoryContext;
+    @Nullable
+    private final Supplier<GroupedTopNBuilder> topNBuilderSupplier;
 
     @Nullable
     private GroupedTopNBuilder topNBuilder;
+    @Nullable
     private Iterator<Page> outputIterator;
 
     public TopNProcessor(
@@ -46,28 +50,33 @@ public TopNProcessor(
             List<Type> types,
             int n,
             List<Integer> sortChannels,
-            List<SortOrder> sortOrders, TypeOperators typeOperators)
+            List<SortOrder> sortOrders,
+            TypeOperators typeOperators)
     {
         requireNonNull(aggregatedMemoryContext, "aggregatedMemoryContext is null");
-        this.localUserMemoryContext = aggregatedMemoryContext.newLocalMemoryContext(TopNProcessor.class.getSimpleName());
+        this.localMemoryContext = aggregatedMemoryContext.newLocalMemoryContext(TopNProcessor.class.getSimpleName());
         checkArgument(n >= 0, "n must be positive");
 
         if (n == 0) {
-            outputIterator = emptyIterator();
+            topNBuilderSupplier = null;
         }
         else {
-            topNBuilder = new GroupedTopNRowNumberBuilder(
+            GroupByHash noChannelGroupByHash = new NoChannelGroupByHash();
+            PageWithPositionComparator comparator = new SimplePageWithPositionComparator(types, sortChannels, sortOrders, typeOperators);
+            topNBuilderSupplier = () -> new GroupedTopNRowNumberBuilder(
                     types,
-                    new SimplePageWithPositionComparator(types, sortChannels, sortOrders, typeOperators),
+                    comparator,
                     n,
                     false,
-                    new NoChannelGroupByHash());
+                    noChannelGroupByHash);
         }
     }
 
     public void addInput(Page page)
     {
-        requireNonNull(topNBuilder, "topNBuilder is null");
+        if (topNBuilder == null) {
+            topNBuilder = requireNonNull(topNBuilderSupplier.get(), "topNBuilderSupplier is null");
+        }
         boolean done = topNBuilder.processPage(requireNonNull(page, "page is null")).process();
         // there is no grouping so work will always be done
         verify(done);
@@ -78,7 +87,7 @@ public Page getOutput()
     {
         if (outputIterator == null) {
             // start flushing
-            outputIterator = topNBuilder.buildResult();
+            outputIterator = topNBuilder == null ? emptyIterator() : topNBuilder.buildResult();
         }
 
         Page output = null;
@@ -86,7 +95,8 @@ public Page getOutput()
             output = outputIterator.next();
         }
         else {
-            outputIterator = emptyIterator();
+            outputIterator = null;
+            topNBuilder = null;
         }
         updateMemoryReservation();
         return output;
@@ -94,12 +104,16 @@ public Page getOutput()
 
     public boolean noMoreOutput()
     {
-        return outputIterator != null && !outputIterator.hasNext();
+        return topNBuilder == null;
+    }
+
+    public long getEstimatedSizeInBytes()
+    {
+        return topNBuilder == null ? 0 : topNBuilder.getEstimatedSizeInBytes();
     }
 
     private void updateMemoryReservation()
     {
-        requireNonNull(topNBuilder, "topNBuilder is null");
-        localUserMemoryContext.setBytes(topNBuilder.getEstimatedSizeInBytes());
+        localMemoryContext.setBytes(getEstimatedSizeInBytes());
     }
 }
diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java b/core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java
index 9ffb9754f566..e435ce89a419 100644
--- a/core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java
+++ b/core/trino-main/src/main/java/io/trino/sql/planner/LocalExecutionPlanner.java
@@ -1633,11 +1633,18 @@ public PhysicalOperation visitTopN(TopNNode node, LocalExecutionPlanContext cont
                     (int) node.getCount(),
                     sortChannels,
                     sortOrders,
-                    plannerContext.getTypeOperators());
+                    plannerContext.getTypeOperators(),
+                    getMaxPartialTopNMemorySize(node.getStep()));
 
             return new PhysicalOperation(operator, source.getLayout(), context, source);
         }
 
+        private Optional<DataSize> getMaxPartialTopNMemorySize(TopNNode.Step step)
+        {
+            DataSize maxPartialTopNMemorySize = SystemSessionProperties.getMaxPartialTopNMemory(session);
+            return step == TopNNode.Step.PARTIAL && maxPartialTopNMemorySize.compareTo(DataSize.ofBytes(0)) > 0 ? Optional.of(maxPartialTopNMemorySize) : Optional.empty();
+        }
+
         @Override
         public PhysicalOperation visitSort(SortNode node, LocalExecutionPlanContext context)
         {
diff --git a/core/trino-main/src/test/java/io/trino/operator/BenchmarkTopNOperator.java b/core/trino-main/src/test/java/io/trino/operator/BenchmarkTopNOperator.java
index 2303a05e41f3..c677d104c61b 100644
--- a/core/trino-main/src/test/java/io/trino/operator/BenchmarkTopNOperator.java
+++ b/core/trino-main/src/test/java/io/trino/operator/BenchmarkTopNOperator.java
@@ -38,6 +38,7 @@
 
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -97,7 +98,8 @@ public void setup()
                     Integer.valueOf(topN),
                     ImmutableList.of(0, 2),
                     ImmutableList.of(DESC_NULLS_LAST, ASC_NULLS_FIRST),
-                    new TypeOperators());
+                    new TypeOperators(),
+                    Optional.of(DataSize.of(16, DataSize.Unit.MEGABYTE)));
         }
 
         @TearDown
diff --git a/core/trino-main/src/test/java/io/trino/operator/TestTopNOperator.java b/core/trino-main/src/test/java/io/trino/operator/TestTopNOperator.java
index d6629f1fc34a..e59d4c4c4bac 100644
--- a/core/trino-main/src/test/java/io/trino/operator/TestTopNOperator.java
+++ b/core/trino-main/src/test/java/io/trino/operator/TestTopNOperator.java
@@ -24,9 +24,11 @@
 import io.trino.testing.MaterializedResult;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 
@@ -45,6 +47,7 @@
 import static java.util.concurrent.Executors.newScheduledThreadPool;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
 
@@ -73,8 +76,14 @@ public void tearDown()
         scheduledExecutor.shutdownNow();
     }
 
-    @Test
-    public void testSingleFieldKey()
+    @DataProvider
+    public Object[][] partial()
+    {
+        return new Object[][] {{true}, {false}};
+    }
+
+    @Test(dataProvider = "partial")
+    public void testSingleFieldKey(boolean partial)
     {
         List<Page> input = rowPagesBuilder(BIGINT, DOUBLE)
                 .row(1L, 0.1)
@@ -93,18 +102,32 @@ public void testSingleFieldKey()
                 ImmutableList.of(BIGINT, DOUBLE),
                 2,
                 ImmutableList.of(0),
-                ImmutableList.of(DESC_NULLS_LAST));
+                ImmutableList.of(DESC_NULLS_LAST),
+                partial ? Optional.of(DataSize.ofBytes(1)) : Optional.empty());
 
-        MaterializedResult expected = resultBuilder(driverContext.getSession(), BIGINT, DOUBLE)
-                .row(6L, 0.6)
-                .row(5L, 0.5)
-                .build();
+        MaterializedResult expected;
+        if (partial) {
+            expected = resultBuilder(driverContext.getSession(), BIGINT, DOUBLE)
+                    .row(2L, 0.2)
+                    .row(1L, 0.1)
+                    .row(4L, 0.4)
+                    .row(-1L, -0.1)
+                    .row(6L, 0.6)
+                    .row(5L, 0.5)
+                    .build();
+        }
+        else {
+            expected = resultBuilder(driverContext.getSession(), BIGINT, DOUBLE)
+                    .row(6L, 0.6)
+                    .row(5L, 0.5)
+                    .build();
+        }
 
         assertOperatorEquals(operatorFactory, driverContext, input, expected);
     }
 
-    @Test
-    public void testMultiFieldKey()
+    @Test(dataProvider = "partial")
+    public void testMultiFieldKey(boolean partial)
     {
         List<Page> input = rowPagesBuilder(VARCHAR, BIGINT)
                 .row("a", 1L)
@@ -122,19 +145,34 @@ public void testMultiFieldKey()
                 ImmutableList.of(VARCHAR, BIGINT),
                 3,
                 ImmutableList.of(0, 1),
-                ImmutableList.of(DESC_NULLS_LAST, DESC_NULLS_LAST));
+                ImmutableList.of(DESC_NULLS_LAST, DESC_NULLS_LAST),
+                partial ? Optional.of(DataSize.ofBytes(1)) : Optional.empty());
 
-        MaterializedResult expected = MaterializedResult.resultBuilder(driverContext.getSession(), VARCHAR, BIGINT)
-                .row("f", 3L)
-                .row("e", 6L)
-                .row("d", 7L)
-                .build();
+        MaterializedResult expected;
+        if (partial) {
+            expected = MaterializedResult.resultBuilder(driverContext.getSession(), VARCHAR, BIGINT)
+                    .row("b", 2L)
+                    .row("a", 1L)
+                    .row("f", 3L)
+                    .row("a", 4L)
+                    .row("e", 6L)
+                    .row("d", 7L)
+                    .row("d", 5L)
+                    .build();
+        }
+        else {
+            expected = MaterializedResult.resultBuilder(driverContext.getSession(), VARCHAR, BIGINT)
+                    .row("f", 3L)
+                    .row("e", 6L)
+                    .row("d", 7L)
+                    .build();
+        }
 
         assertOperatorEquals(operatorFactory, driverContext, input, expected);
     }
 
-    @Test
-    public void testReverseOrder()
+    @Test(dataProvider = "partial")
+    public void testReverseOrder(boolean partial)
     {
         List<Page> input = rowPagesBuilder(BIGINT, DOUBLE)
                 .row(1L, 0.1)
@@ -153,12 +191,26 @@ public void testReverseOrder()
                 ImmutableList.of(BIGINT, DOUBLE),
                 2,
                 ImmutableList.of(0),
-                ImmutableList.of(ASC_NULLS_LAST));
+                ImmutableList.of(ASC_NULLS_LAST),
+                partial ? Optional.of(DataSize.ofBytes(1)) : Optional.empty());
 
-        MaterializedResult expected = resultBuilder(driverContext.getSession(), BIGINT, DOUBLE)
-                .row(-1L, -0.1)
-                .row(1L, 0.1)
-                .build();
+        MaterializedResult expected;
+        if (partial) {
+            expected = resultBuilder(driverContext.getSession(), BIGINT, DOUBLE)
+                    .row(1L, 0.1)
+                    .row(2L, 0.2)
+                    .row(-1L, -0.1)
+                    .row(4L, 0.4)
+                    .row(4L, 0.41)
+                    .row(5L, 0.5)
+                    .build();
+        }
+        else {
+            expected = resultBuilder(driverContext.getSession(), BIGINT, DOUBLE)
+                    .row(-1L, -0.1)
+                    .row(1L, 0.1)
+                    .build();
+        }
 
         assertOperatorEquals(operatorFactory, driverContext, input, expected);
     }
@@ -171,7 +223,8 @@ public void testLimitZero()
                 ImmutableList.of(BIGINT),
                 0,
                 ImmutableList.of(0),
-                ImmutableList.of(DESC_NULLS_LAST));
+                ImmutableList.of(DESC_NULLS_LAST),
+                Optional.of(DataSize.ofBytes(1)));
 
         try (Operator operator = factory.createOperator(driverContext)) {
             assertNull(operator.getOutput());
@@ -181,8 +234,8 @@ public void testLimitZero()
         }
     }
 
-    @Test
-    public void testExceedMemoryLimit()
+    @Test(dataProvider = "partial")
+    public void testExceedMemoryLimit(boolean partial)
             throws Exception
     {
         List<Page> input = rowPagesBuilder(BIGINT)
@@ -197,7 +250,8 @@ public void testExceedMemoryLimit()
                 ImmutableList.of(BIGINT),
                 100,
                 ImmutableList.of(0),
-                ImmutableList.of(ASC_NULLS_LAST));
+                ImmutableList.of(ASC_NULLS_LAST),
+                partial ? Optional.of(DataSize.ofBytes(1)) : Optional.empty());
         Operator operator = operatorFactory.createOperator(smallDiverContext);
         operator.addInput(input.get(0));
         assertThatThrownBy(() -> operator.getOutput())
@@ -205,11 +259,61 @@ public void testExceedMemoryLimit()
                 .hasMessageStartingWith("Query exceeded per-node memory limit of ");
     }
 
+    @Test(dataProvider = "partial")
+    public void testPartialFlushing(boolean partial)
+    {
+        List<Page> input = rowPagesBuilder(BIGINT, DOUBLE)
+                .row(1L, 0.1)
+                .row(2L, 0.2)
+                .pageBreak()
+                .row(-1L, -0.1)
+                .row(4L, 0.4)
+                .pageBreak()
+                .row(5L, 0.5)
+                .row(4L, 0.41)
+                .row(6L, 0.6)
+                .pageBreak()
+                .build();
+
+        OperatorFactory operatorFactory = topNOperatorFactory(
+                ImmutableList.of(BIGINT, DOUBLE),
+                2,
+                ImmutableList.of(0),
+                ImmutableList.of(DESC_NULLS_LAST),
+                partial ? Optional.of(DataSize.ofBytes(1)) : Optional.empty());
+
+        WorkProcessorOperatorAdapter topNOperator = (WorkProcessorOperatorAdapter) operatorFactory.createOperator(driverContext);
+        for (Page inputPage : input) {
+            topNOperator.addInput(inputPage);
+            if (partial) {
+                assertNotNull(topNOperator.getOutput()); // get partial flush result
+                assertFalse(topNOperator.isFinished()); // not finished. just partial flushing.
+                assertNull(topNOperator.getOutput()); // clear flushing
+                assertFalse(topNOperator.isFinished()); // not finished. just yield,
+            }
+            else {
+                assertNull(topNOperator.getOutput());
+            }
+        }
+
+        topNOperator.finish();
+        assertFalse(topNOperator.isFinished()); // not finished. finishing.
+        if (partial) {
+            assertNull(topNOperator.getOutput()); // finished
+            assertTrue(topNOperator.isFinished());
+        }
+        else {
+            assertNotNull(topNOperator.getOutput()); // start flushing final result
+            assertFalse(topNOperator.isFinished());
+        }
+    }
+
     private OperatorFactory topNOperatorFactory(
             List<? extends Type> types,
             int n,
             List<Integer> sortChannels,
-            List<SortOrder> sortOrders)
+            List<SortOrder> sortOrders,
+            Optional<DataSize> maxPartialMemory)
     {
         return TopNOperator.createOperatorFactory(
                 0,
@@ -218,6 +322,7 @@ private OperatorFactory topNOperatorFactory(
                 n,
                 sortChannels,
                 sortOrders,
-                typeOperators);
+                typeOperators,
+                maxPartialMemory);
     }
 }
diff --git a/testing/trino-benchmark/src/main/java/io/trino/benchmark/Top100Benchmark.java b/testing/trino-benchmark/src/main/java/io/trino/benchmark/Top100Benchmark.java
index 0f38b93c016b..11fec3c191f0 100644
--- a/testing/trino-benchmark/src/main/java/io/trino/benchmark/Top100Benchmark.java
+++ b/testing/trino-benchmark/src/main/java/io/trino/benchmark/Top100Benchmark.java
@@ -14,6 +14,7 @@
 package io.trino.benchmark;
 
 import com.google.common.collect.ImmutableList;
+import io.airlift.units.DataSize;
 import io.trino.operator.OperatorFactory;
 import io.trino.operator.TopNOperator;
 import io.trino.spi.type.Type;
@@ -21,6 +22,7 @@
 import io.trino.testing.LocalQueryRunner;
 
 import java.util.List;
+import java.util.Optional;
 
 import static io.trino.benchmark.BenchmarkQueryRunner.createLocalQueryRunner;
 import static io.trino.spi.connector.SortOrder.ASC_NULLS_LAST;
@@ -45,7 +47,8 @@ protected List<? extends OperatorFactory> createOperatorFactories()
                 100,
                 ImmutableList.of(0),
                 ImmutableList.of(ASC_NULLS_LAST),
-                localQueryRunner.getTypeOperators());
+                localQueryRunner.getTypeOperators(),
+                Optional.of(DataSize.of(16, DataSize.Unit.MEGABYTE)));
         return ImmutableList.of(tableScanOperator, topNOperator);
     }
 
diff --git a/testing/trino-tests/src/test/java/io/trino/tests/TestPartialFlushingOrderByWithLimitQueries.java b/testing/trino-tests/src/test/java/io/trino/tests/TestPartialFlushingOrderByWithLimitQueries.java
new file mode 100644
index 000000000000..a9e8ec202a2c
--- /dev/null
+++ b/testing/trino-tests/src/test/java/io/trino/tests/TestPartialFlushingOrderByWithLimitQueries.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.tests;
+
+import io.trino.Session;
+import io.trino.SystemSessionProperties;
+import io.trino.plugin.tpch.TpchPlugin;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.DistributedQueryRunner;
+import io.trino.testing.QueryRunner;
+import org.testng.annotations.Test;
+
+import static io.trino.plugin.tpch.TpchMetadata.TINY_SCHEMA_NAME;
+import static io.trino.testing.TestingSession.testSessionBuilder;
+
+public class TestPartialFlushingOrderByWithLimitQueries
+        extends AbstractTestQueryFramework
+{
+    @Override
+    protected QueryRunner createQueryRunner()
+            throws Exception
+    {
+        Session defaultSession = testSessionBuilder()
+                .setCatalog("tpch")
+                .setSchema(TINY_SCHEMA_NAME)
+                .setSystemProperty(SystemSessionProperties.TASK_CONCURRENCY, "2")
+                .setSystemProperty(SystemSessionProperties.MAX_PARTIAL_TOP_N_MEMORY, "1B")
+                .build();
+
+        DistributedQueryRunner queryRunner = DistributedQueryRunner.builder(defaultSession)
+                .setNodeCount(2)
+                .build();
+
+        try {
+            queryRunner.installPlugin(new TpchPlugin());
+            queryRunner.createCatalog("tpch", "tpch");
+            return queryRunner;
+        }
+        catch (Exception e) {
+            queryRunner.close();
+            throw e;
+        }
+    }
+
+    @Test
+    public void testOrderByLimit()
+    {
+        assertQueryOrdered("SELECT custkey, orderstatus FROM orders ORDER BY orderkey DESC LIMIT 10");
+        assertQueryOrdered("SELECT custkey, orderstatus FROM orders ORDER BY orderkey + 1 DESC LIMIT 10");
+        assertQuery("SELECT custkey, totalprice FROM orders ORDER BY orderkey LIMIT 0");
+    }
+
+    @Test
+    public void testOrderByLimitAll()
+    {
+        assertQuery("SELECT custkey, totalprice FROM orders ORDER BY orderkey LIMIT ALL", "SELECT custkey, totalprice FROM orders ORDER BY orderkey");
+    }
+
+    @Test
+    public void testOrderByWithSimilarExpressions()
+    {
+        assertQuery(
+                "WITH t AS (SELECT 1 x, 2 y) SELECT x, y FROM t ORDER BY x, y",
+                "SELECT 1, 2");
+        assertQuery(
+                "WITH t AS (SELECT 1 x, 2 y) SELECT x, y FROM t ORDER BY x, y LIMIT 1",
+                "SELECT 1, 2");
+        assertQuery(
+                "WITH t AS (SELECT 1 x, 1 y) SELECT x, y FROM t ORDER BY x, y LIMIT 1",
+                "SELECT 1, 1");
+        assertQuery(
+                "WITH t AS (SELECT orderkey x, orderkey y FROM orders) SELECT x, y FROM t ORDER BY x, y LIMIT 1",
+                "SELECT 1, 1");
+        assertQuery(
+                "WITH t AS (SELECT orderkey x, orderkey y FROM orders) SELECT x, y FROM t ORDER BY x, y DESC LIMIT 1",
+                "SELECT 1, 1");
+        assertQuery(
+                "WITH t AS (SELECT orderkey x, totalprice y, orderkey z FROM orders) SELECT x, y, z FROM t ORDER BY x, y, z LIMIT 1",
+                "SELECT 1, 172799.49, 1");
+    }
+}