IGNITE-28437 Sql. Reduce number of rows fetched during ordered index scan (#7919)

Author: korlov42

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNode.java

@@ -17,6 +17,9 @@
 
 package org.apache.ignite.internal.sql.engine.exec.rel;
 
+import static java.lang.Integer.max;
+import static java.lang.Integer.min;
+
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.Iterator;
@@ -38,7 +41,6 @@
 import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptor;
 import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
 import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
-import org.apache.ignite.internal.sql.engine.util.Commons;
 import org.apache.ignite.internal.util.SubscriptionUtils;
 import org.apache.ignite.internal.util.TransformingIterator;
 import org.jetbrains.annotations.Nullable;
@@ -131,8 +133,18 @@ private Publisher<RowT> indexPublisher(
                 partWithConsistencyToken -> partitionPublisher(partWithConsistencyToken, cond)
         );
 
+        int bufferSize = context().bufferSize();
+
+        // Let's prefetch equal share of a buffer from each partition.
+        int fetchSize = max(context().bufferSize() / partsWithConsistencyTokens.size(), 1);
+
+        // Adds some buffer to improve chances to fulfill entire request without need go to storage once again.
+        // This renders over-prefetching over all local partitions in total, but at least it's capped now at
+        // 2x bufferSize within up to 512 local partitions.
+        fetchSize = min(fetchSize * 2, bufferSize);
+
         if (comp != null) {
-            return SubscriptionUtils.orderedMerge(comp, Commons.SORTED_IDX_PART_PREFETCH_SIZE, it);
+            return SubscriptionUtils.orderedMerge(comp, fetchSize, it);
         } else {
             return SubscriptionUtils.concat(it);
         }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/Commons.java

@@ -138,13 +138,6 @@ public final class Commons {
             SqlKind.OTHER_DDL
     );
 
-    /**
-     * The number of elements to be prefetched from each partition when scanning the sorted index.
-     * The higher the value, the fewer calls to the upstream will be, but at the same time, the bigger
-     * internal buffer will be.
-     */
-    public static final int SORTED_IDX_PART_PREFETCH_SIZE = 100;
-
     @SuppressWarnings("rawtypes")
     public static final List<RelTraitDef> DISTRIBUTED_TRAITS_SET = List.of(
             ConventionTraitDef.INSTANCE,