read_stream_threshold.patch

diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index b9d42b15a18..aac663edfdd 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -36,6 +36,7 @@
 #include "commands/progress.h"
 #include "executor/executor.h"
 #include "miscadmin.h"
+#include "optimizer/cost.h"
 #include "pgstat.h"
 #include "storage/bufmgr.h"
 #include "storage/bufpage.h"
@@ -46,6 +47,8 @@
 #include "utils/builtins.h"
 #include "utils/rel.h"
 
+int read_stream_threshold = DEFAULT_READ_STREAM_THRESHOLD;
+
 static void reform_and_rewrite_tuple(HeapTuple tuple,
 									 Relation OldHeap, Relation NewHeap,
 									 Datum *values, bool *isnull, RewriteState rwstate);
@@ -229,6 +232,7 @@ heap_batch_advance_pos(IndexScanDesc scan, struct BatchQueueItemPos *pos,
 					   ScanDirection direction)
 {
 	BatchIndexScan batch;
+	int proceed_items;
 
 	/* make sure we have batching initialized and consistent */
 	batch_assert_batches_valid(scan);
@@ -288,6 +292,24 @@ heap_batch_advance_pos(IndexScanDesc scan, struct BatchQueueItemPos *pos,
 	 */
 	batch = INDEX_SCAN_BATCH(scan, pos->batch);
 
+	proceed_items = ScanDirectionIsForward(direction)
+		? pos->item - batch->firstItem
+		: batch->lastItem - pos->item;
+
+	/* Delay initializing stream until proceeding */
+	if (proceed_items >= read_stream_threshold
+		&& !scan->xs_heapfetch->rs
+		&& !scan->batchqueue->disabled
+		&& !scan->xs_want_itup	/* XXX prefetching disabled for IoS, for now */
+		&& enable_indexscan_prefetch)
+	{
+		scan->xs_heapfetch->rs =
+			read_stream_begin_relation(READ_STREAM_DEFAULT, NULL,
+									   scan->heapRelation, MAIN_FORKNUM,
+									   scan->heapRelation->rd_tableam->index_getnext_stream,
+									   scan, 0);
+	}
+
 	if (ScanDirectionIsForward(direction))
 	{
 		if (++pos->item <= batch->lastItem)
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 8f6fa6843cb..0c2081e32ba 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -2322,6 +2322,15 @@
   max => 'DBL_MAX',
 },
 
+{ name => 'read_stream_threshold', type => 'int', context => 'PGC_USERSET', group => 'QUERY_TUNING_COST',
+  short_desc => 'Minimal number of heap tuples for creation read stream.',
+  flags => 'GUC_EXPLAIN',
+  variable => 'read_stream_threshold',
+  boot_val => 'DEFAULT_READ_STREAM_THRESHOLD',
+  min => '0',
+  max => 'INT_MAX',
+},
+
 { name => 'recovery_end_command', type => 'string', context => 'PGC_SIGHUP', group => 'WAL_ARCHIVE_RECOVERY',
   short_desc => 'Sets the shell command that will be executed once at the end of recovery.',
   variable => 'recoveryEndCommand',
diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h
index 00f4c3d0011..97150433c99 100644
--- a/src/include/optimizer/cost.h
+++ b/src/include/optimizer/cost.h
@@ -28,6 +28,7 @@
 #define DEFAULT_CPU_OPERATOR_COST  0.0025
 #define DEFAULT_PARALLEL_TUPLE_COST 0.1
 #define DEFAULT_PARALLEL_SETUP_COST  1000.0
+#define DEFAULT_READ_STREAM_THRESHOLD	10
 
 /* defaults for non-Cost parameters */
 #define DEFAULT_RECURSIVE_WORKTABLE_FACTOR  10.0
@@ -72,6 +73,7 @@ extern PGDLLIMPORT bool enable_partition_pruning;
 extern PGDLLIMPORT bool enable_presorted_aggregate;
 extern PGDLLIMPORT bool enable_async_append;
 extern PGDLLIMPORT int constraint_exclusion;
+extern PGDLLIMPORT int read_stream_threshold;
 
 extern double index_pages_fetched(double tuples_fetched, BlockNumber pages,
 								  double index_pages, PlannerInfo *root);