GG-399: Enable incremental sort for greengage

src/backend/cdb/cdbgroupingpaths.c

@@ -2302,6 +2302,12 @@ fetch_single_dqa_info(PlannerInfo *root,
  * is already sorted, we prefer to gather it to a single node to make
  * use of the pre-existing order, instead of redistributing and resorting
  * it.
+ * 
+ * The presorted_keys parameter is the number of keys by which the input path 
+ * is known to be sorted. 
+ * If this parameter is greater than 0 (NO_INCREMENTAL_SORT), the input is 
+ * considered to be partially sorted and incremental sort is applied.
+ * Otherwise, a full sort is used. 
  */
 Path *
 cdb_prepare_path_for_sorted_agg(PlannerInfo *root,
@@ -2311,6 +2317,7 @@ cdb_prepare_path_for_sorted_agg(PlannerInfo *root,
 								Path *subpath,
 								PathTarget *target,
 								List *group_pathkeys,
+								int presorted_keys,
 								double limit_tuples,
 								/* extra arguments */
 								List *groupClause,
@@ -2344,11 +2351,23 @@ cdb_prepare_path_for_sorted_agg(PlannerInfo *root,
 	{
 		if (!is_sorted)
 		{
-			subpath = (Path *) create_sort_path(root,
-												rel,
-												subpath,
-												group_pathkeys,
-												-1.0);
+			if (presorted_keys == NO_INCREMENTAL_SORT)
+			{
+				subpath = (Path *) create_sort_path(root,
+													rel,
+													subpath,
+													group_pathkeys,
+													-1.0);
+			} 
+			else
+			{
+				subpath = (Path *) create_incremental_sort_path(root,
+																rel,
+																subpath,
+																group_pathkeys,
+																presorted_keys,
+																-1.0);
+			}
 		}
 		return subpath;
 	}
@@ -2380,14 +2399,42 @@ cdb_prepare_path_for_sorted_agg(PlannerInfo *root,
 		 * to merge the inputs.
 		 */
 		if (CdbPathLocus_IsPartitioned(locus))
-			subpath = cdbpath_create_motion_path(root, subpath, NIL,
-												 false, locus);
+		{
+			Path *motion_path = cdbpath_create_motion_path(root, subpath, subpath->pathkeys,
+												 presorted_keys != NO_INCREMENTAL_SORT, locus);
 
-		subpath = (Path *) create_sort_path(root,
-											rel,
-											subpath,
-											group_pathkeys,
-											-1.0);
+
+			/*
+			 * We can not return NULL here, so make a possibly redundant path with 
+			 * no incremental sort
+			 */
+			if (motion_path == NULL) 
+			{
+				presorted_keys = NO_INCREMENTAL_SORT;
+				motion_path = cdbpath_create_motion_path(root, subpath, NIL,
+													false, locus);
+			}
+			Assert(motion_path != NULL);
+			subpath = motion_path;
+		}
+
+		if (presorted_keys == NO_INCREMENTAL_SORT)
+		{
+			subpath = (Path *) create_sort_path(root,
+												rel,
+												subpath,
+												group_pathkeys,
+												-1.0);
+		}
+		else
+		{
+			subpath = (Path *) create_incremental_sort_path(root,
+															rel,
+															subpath,
+															group_pathkeys,
+															presorted_keys,
+															-1.0);
+		}
 
 		if (!CdbPathLocus_IsPartitioned(locus))
 			subpath = cdbpath_create_motion_path(root, subpath,

src/backend/cdb/cdbllize.c

@@ -1485,6 +1485,7 @@ motion_sanity_walker(Node *node, sanity_result_t *result)
 		case T_SetOp:
 		case T_Limit:
 		case T_Sort:
+		case T_IncrementalSort:
 		case T_Material:
 		case T_ForeignScan:
 			if (plan_tree_walker(node, motion_sanity_walker, result, true))

src/backend/cdb/cdbplan.c

@@ -621,6 +621,20 @@ plan_tree_mutator(Node *node,
 			}
 			break;
 
+		case T_IncrementalSort:
+			{
+				IncrementalSort	   *sort = (IncrementalSort *) node;
+				IncrementalSort	   *newsort;
+
+				FLATCOPY(newsort, sort, IncrementalSort);
+				PLANMUTATE(newsort, sort);
+				COPYARRAY(&newsort->sort, &sort->sort, numCols, sortColIdx);
+				COPYARRAY(&newsort->sort, &sort->sort, numCols, sortOperators);
+				COPYARRAY(&newsort->sort, &sort->sort, numCols, nullsFirst);
+				return (Node *) newsort;
+			}
+			break;
+
 		case T_Agg:
 			{
 				Agg		   *agg = (Agg *) node;

src/backend/cdb/cdbtargeteddispatch.c

@@ -521,6 +521,7 @@ DirectDispatchUpdateContentIdsFromPlan(PlannerInfo *root, Plan *plan)
 			break;
 		case T_Material:
 		case T_Sort:
+		case T_IncrementalSort:
 		case T_Agg:
 		case T_TupleSplit:
 		case T_Unique:

src/backend/executor/execAmi.c

@@ -832,6 +832,10 @@ ExecSquelchNode(PlanState *node)
 			ExecSquelchSort((SortState *) node);
 			break;
 
+		case T_IncrementalSortState:
+			ExecSquelchIncrementalSort((IncrementalSortState *) node);
+			break;
+
 		case T_AggState:
 			ExecSquelchAgg((AggState*) node);
 			break;

src/backend/executor/nodeIncrementalSort.c

@@ -115,6 +115,7 @@
 		} \
 	} while (0)
 
+static void ExecEagerFreeIncrementalSort(IncrementalSortState *node);
 
 /* ----------------------------------------------------------------
  * instrumentSortedGroup
@@ -977,6 +978,12 @@ ExecIncrementalSort(PlanState *pstate)
 	slot = node->ss.ps.ps_ResultTupleSlot;
 	(void) tuplesort_gettupleslot(read_sortstate, ScanDirectionIsForward(dir),
 								  false, slot, NULL);
+
+	if (TupIsNull(slot) && !node->delayEagerFree)
+	{
+		ExecEagerFreeIncrementalSort(node);
+	}
+
 	return slot;
 }
 
@@ -1046,6 +1053,11 @@ ExecInitIncrementalSort(IncrementalSort *node, EState *estate, int eflags)
 	 * ExecQual or ExecProject.
 	 */
 
+	/*
+	 * If eflag contains EXEC_FLAG_REWIND then this node is not eager free safe.
+	 */
+	incrsortstate->delayEagerFree = ((eflags & (EXEC_FLAG_REWIND)) != 0);
+
 	/*
 	 * Initialize child nodes.
 	 *
@@ -1056,6 +1068,15 @@ ExecInitIncrementalSort(IncrementalSort *node, EState *estate, int eflags)
 	 */
 	outerPlanState(incrsortstate) = ExecInitNode(outerPlan(node), estate, eflags);
 
+
+	/*
+	 * If the child node is a Motion, then this node is not eager free safe.
+	 */
+	if (IsA(outerPlan((Plan *)node), Motion))
+	{
+		incrsortstate->delayEagerFree = true;
+	}
+
 	/*
 	 * Initialize scan slot and type.
 	 */
@@ -1093,27 +1114,7 @@ ExecEndIncrementalSort(IncrementalSortState *node)
 {
 	SO_printf("ExecEndIncrementalSort: shutting down sort node\n");
 
-	/* clean out the scan tuple */
-	ExecClearTuple(node->ss.ss_ScanTupleSlot);
-	/* must drop pointer to sort result tuple */
-	ExecClearTuple(node->ss.ps.ps_ResultTupleSlot);
-	/* must drop standalone tuple slots from outer node */
-	ExecDropSingleTupleTableSlot(node->group_pivot);
-	ExecDropSingleTupleTableSlot(node->transfer_tuple);
-
-	/*
-	 * Release tuplesort resources.
-	 */
-	if (node->fullsort_state != NULL)
-	{
-		tuplesort_end(node->fullsort_state);
-		node->fullsort_state = NULL;
-	}
-	if (node->prefixsort_state != NULL)
-	{
-		tuplesort_end(node->prefixsort_state);
-		node->prefixsort_state = NULL;
-	}
+	ExecEagerFreeIncrementalSort(node);
 
 	/*
 	 * Shut down the subplan.
@@ -1185,6 +1186,52 @@ ExecReScanIncrementalSort(IncrementalSortState *node)
 		ExecReScan(outerPlan);
 }
 
+static void
+ExecEagerFreeIncrementalSort(IncrementalSortState *node)
+{
+	/* clean out the scan tuple */
+	ExecClearTuple(node->ss.ss_ScanTupleSlot);
+	/* must drop pointer to sort result tuple */
+	ExecClearTuple(node->ss.ps.ps_ResultTupleSlot);
+	/* must drop standalone tuple slots from outer node */
+	if (node->group_pivot != NULL)
+	{
+		ExecDropSingleTupleTableSlot(node->group_pivot);
+		node->group_pivot = NULL;
+	}
+
+	if (node->transfer_tuple != NULL)
+	{
+		ExecDropSingleTupleTableSlot(node->transfer_tuple);
+		node->transfer_tuple = NULL;
+	}
+
+	/*
+	 * Release tuplesort resources.
+	 */
+	if (node->fullsort_state != NULL)
+	{
+		tuplesort_end(node->fullsort_state);
+		node->fullsort_state = NULL;
+	}
+	if (node->prefixsort_state != NULL)
+	{
+		tuplesort_end(node->prefixsort_state);
+		node->prefixsort_state = NULL;
+	}
+}
+
+void
+ExecSquelchIncrementalSort(IncrementalSortState *node)
+{
+	if (!node->delayEagerFree)
+	{
+		ExecEagerFreeIncrementalSort(node);
+		ExecSquelchNode(outerPlanState(node));
+	}
+}
+
+
 /* ----------------------------------------------------------------
  *						Parallel Query Support
  * ----------------------------------------------------------------

src/backend/nodes/outfast.c

@@ -1082,6 +1082,9 @@ _outNode(StringInfo str, void *obj)
 			case T_Sort:
 				_outSort(str, obj);
 				break;
+			case T_IncrementalSort:
+				_outIncrementalSort(str, obj);
+				break;
 			case T_Unique:
 				_outUnique(str, obj);
 				break;

src/backend/nodes/outfuncs.c

@@ -2472,6 +2472,17 @@ _outSortPath(StringInfo str, const SortPath *node)
 	WRITE_NODE_FIELD(subpath);
 }
 
+static void
+_outIncrementalSortPath(StringInfo str, const IncrementalSortPath *node)
+{
+	WRITE_NODE_TYPE("INCREMENTALSORTPATH");
+
+	_outPathInfo(str, (const Path *) node);
+
+	WRITE_NODE_FIELD(spath.subpath);
+	WRITE_INT_FIELD(nPresortedCols);
+}
+
 static void
 _outGroupPath(StringInfo str, const GroupPath *node)
 {
@@ -5970,6 +5981,9 @@ outNode(StringInfo str, const void *obj)
 			case T_SortPath:
 				_outSortPath(str, obj);
 				break;
+			case T_IncrementalSortPath:
+				_outIncrementalSortPath(str, obj);
+				break;
 			case T_GroupPath:
 				_outGroupPath(str, obj);
 				break;

src/backend/nodes/readfast.c

@@ -1854,6 +1854,9 @@ readNodeBinary(void)
 			case T_Sort:
 				return_value = _readSort();
 				break;
+			case T_IncrementalSort:
+				return_value = _readIncrementalSort();
+				break;
 			case T_Unique:
 				return_value = _readUnique();
 				break;

src/backend/optimizer/path/costsize.c

@@ -132,8 +132,7 @@ bool		enable_indexonlyscan = true;
 bool		enable_bitmapscan = true;
 bool		enable_tidscan = true;
 bool		enable_sort = true;
-/* GPDB_13_MERGE_FIXME: enable incremental sort */
-bool		enable_incremental_sort = false;
+bool		enable_incremental_sort = true;
 bool		enable_hashagg = true;
 bool		enable_groupagg = true;
 bool		enable_nestloop = false;