docs: AbstractPlanNode applyOptimization(OrderByPlanNode orderbyNode) {
          // Find all child RECEIVE nodes. We are not interested in the MERGERECEIVE nodes there
          // because they could only come from subqueries.
          List<AbstractPlanNode> receives = orderbyNode.findAllNodesOfType(PlanNodeType.RECEIVE);
          if (receives.isEmpty()) {
              return orderbyNode;
          }
          assert(receives.size() == 1);
  
          ReceivePlanNode receive = (ReceivePlanNode)receives.get(0);
          // Make sure that this receive node belongs to the same coordinator fragment that
          // the ORDER BY node does. Alternatively, it could belong to a distributed subquery.
          // Walk up the tree starting at the receive node until we hit either a scan node
          // (distributed subquery) or the original order by node (distributed order by)
          // Collect all nodes that are currently in between ORDER BY and RECEIVE nodes
          // If the optimization is possible, they will be converted to inline nodes of
          // the MERGE RECEIVE node. The expected node types are:
          //      LIMIT, AGGREGATE/PARTIALAGGREGATE/HASHAGGREGATE
          // The HASHAGGREGATE must be convertible to AGGREGATE or PARTIALAGGREGATE for optimization
          // to be applicable.
  
          // LIMIT can be already inline with ORDER BY node
          AbstractPlanNode limitNode = orderbyNode.getInlinePlanNode(PlanNodeType.LIMIT);
          AbstractPlanNode aggregateNode = null;
          AbstractPlanNode inlineCandidate = receive.getParent(0);
          while (orderbyNode != inlineCandidate) {
              if (inlineCandidate instanceof AbstractScanPlanNode) {
                  // it's a subquery
                  return orderbyNode;
              }
              PlanNodeType nodeType = inlineCandidate.getPlanNodeType();
              if (nodeType == PlanNodeType.LIMIT && limitNode == null) {
                  limitNode = inlineCandidate;
              } else if ((nodeType == PlanNodeType.AGGREGATE || nodeType == PlanNodeType.PARTIALAGGREGATE) &&
                      aggregateNode == null) {
                  aggregateNode = inlineCandidate;
              } else if (nodeType == PlanNodeType.HASHAGGREGATE && aggregateNode == null) {
                  aggregateNode = convertToSerialAggregation(inlineCandidate, orderbyNode);
                  if (PlanNodeType.HASHAGGREGATE == aggregateNode.getPlanNodeType()) {
                      return orderbyNode;
                  }
              } else {
                  // Don't know how to handle this node or there is already a node of this type
                  return orderbyNode;
              }
              // move up one node
              assert(inlineCandidate.getParentCount() == 1);
              inlineCandidate = inlineCandidate.getParent(0);
          }
  
          assert(receive.getChildCount() == 1);
          AbstractPlanNode partitionRoot = receive.getChild(0);
          if (!partitionRoot.isOutputOrdered(orderbyNode.getSortExpressions(), orderbyNode.getSortDirections())) {
              // Partition results are not ordered. The MERGE RECEIVE optimization is still possible if
              //  - the coordinator's plan below the ORDER BY node is trivial. At the moment only RECEIVE node is allowed.
              //    A future enhancement may allow PROJECTION/RECEIVE
              //  - Coordinator's ORDER BY node does not have inline LIMIT/OFFSET node. If a plan has an ORDER BY and
              // LIMIT/OFFSET nodes, PlanAssembeler tries to push both of them to a partition fragment
              // as LIMIT pushed down optimization during the regular planning phase (PlanAssembler.handleSelectLimitOperator).
              // If it succeeds, the partition's output would be ordered and we wouldn't get there.
              // If we get there and ODER BY node does have inline LIMIT it means that the PlanAssembler
              // fails to apply the original optimization and we also can not do it here.
              // For example, an OFFSET without a LIMIT can not be pushed down.
              if (isOptimizationPossible(orderbyNode)) {
                  // Build fragment's ORDER BY plan node
                  OrderByPlanNode fragmentOrderbyNode = new OrderByPlanNode();
                  fragmentOrderbyNode.addSortExpressions(orderbyNode.getSortExpressions(), orderbyNode.getSortDirections());
                  for(Map.Entry<PlanNodeType, AbstractPlanNode> inlineEntry : orderbyNode.getInlinePlanNodes().entrySet()) {
                      fragmentOrderbyNode.addInlinePlanNode(inlineEntry.getValue());
                  }
  
                  // Insert the fragment's ORDER BY plan node below the partition's root (Send node)
                  assert(partitionRoot.getChildCount() == 1);
                  AbstractPlanNode partitionRootInput = partitionRoot.getChild(0);
                  partitionRootInput.clearParents();
                  partitionRoot.clearChildren();
                  fragmentOrderbyNode.addAndLinkChild(partitionRootInput);
                  partitionRoot.addAndLinkChild(fragmentOrderbyNode);
              } else {
                  return orderbyNode;
              }
          }
  
          // At this point we confirmed that the optimization is applicable.
          // Short circuit the current ORDER BY parent (if such exists) and
          // the new MERGERECIEVE node.. All in-between nodes will be inlined
          assert (orderbyNode.getParentCount() <= 1);
          AbstractPlanNode rootNode = (orderbyNode.getParentCount() == 1) ? orderbyNode.getParent(0) : null;
          MergeReceivePlanNode mergeReceive = new MergeReceivePlanNode();
          assert(receive.getChildCount() == 1);
          mergeReceive.addAndLinkChild(receive.getChild(0));
          receive.removeFromGraph();
          if (rootNode == null) {
              rootNode = mergeReceive;
          } else {
              rootNode.clearChildren();
              rootNode.addAndLinkChild(mergeReceive);
          }
  
          // Add inline ORDER BY node and remove inline LIMIT node if any
          mergeReceive.addInlinePlanNode(orderbyNode);
          if (limitNode != null) {
              orderbyNode.removeInlinePlanNode(PlanNodeType.LIMIT);
          }
  
          // Add inline aggregate
          if (aggregateNode != null) {
              if (limitNode != null) {
                  // Inline LIMIT with aggregate
                  aggregateNode.addInlinePlanNode(limitNode);
              }
              mergeReceive.addInlinePlanNode(aggregateNode);
          }
          // Add LIMIT if it is exist and wasn't inline with aggregate node
          if (limitNode != null && aggregateNode == null) {
              mergeReceive.addInlinePlanNode(limitNode);
          }
  
          // return the new root
          return rootNode;
      }