diff --git a/dbcon/mysql/ha_exists_sub.cpp b/dbcon/mysql/ha_exists_sub.cpp
index 9db9294c9..4c6ebd424 100644
--- a/dbcon/mysql/ha_exists_sub.cpp
+++ b/dbcon/mysql/ha_exists_sub.cpp
@@ -112,7 +112,7 @@ execplan::ParseTree* ExistsSub::transform()
         return NULL;
     }
 
-    if (getSelectPlan(gwi, *(fSub->get_select_lex()), csep, false, true) != 0)
+    if (getSelectPlan(gwi, *(fSub->get_select_lex()), csep, false) != 0)
     {
         fGwip.fatalParseError = true;
 
diff --git a/dbcon/mysql/ha_from_sub.cpp b/dbcon/mysql/ha_from_sub.cpp
index 12b2b24de..d34bc8c2e 100644
--- a/dbcon/mysql/ha_from_sub.cpp
+++ b/dbcon/mysql/ha_from_sub.cpp
@@ -323,8 +323,7 @@ FromSubQuery::FromSubQuery(gp_walk_info& gwip) : SubQuery(gwip)
 {}
 
 FromSubQuery::FromSubQuery(gp_walk_info& gwip, 
-    SELECT_LEX* sub, 
-    bool isPushdownHandler) :
+    SELECT_LEX* sub) :
         SubQuery(gwip),
         fFromSub(sub)
 {}
@@ -348,7 +347,7 @@ SCSEP FromSubQuery::transform()
     csep->derivedTbAlias(fAlias); // always lower case
     csep->derivedTbView(fGwip.viewName.alias);
 
-    if (getSelectPlan(gwi, *fFromSub, csep, false, true) != 0)
+    if (getSelectPlan(gwi, *fFromSub, csep, false) != 0)
     {
         fGwip.fatalParseError = true;
 
diff --git a/dbcon/mysql/ha_in_sub.cpp b/dbcon/mysql/ha_in_sub.cpp
index 4d1e828ce..5ae5e18f2 100644
--- a/dbcon/mysql/ha_in_sub.cpp
+++ b/dbcon/mysql/ha_in_sub.cpp
@@ -148,7 +148,7 @@ execplan::ParseTree* InSub::transform()
     gwi.tbList.insert(gwi.tbList.begin(), fGwip.tbList.begin(), fGwip.tbList.end());
     gwi.derivedTbList.insert(gwi.derivedTbList.begin(), fGwip.derivedTbList.begin(), fGwip.derivedTbList.end());
 
-    if (getSelectPlan(gwi, *(fSub->get_select_lex()), csep, false, true) != 0)
+    if (getSelectPlan(gwi, *(fSub->get_select_lex()), csep, false) != 0)
     {
         fGwip.fatalParseError = true;
 
diff --git a/dbcon/mysql/ha_mcs_execplan.cpp b/dbcon/mysql/ha_mcs_execplan.cpp
index 1d11be0eb..da10645f0 100755
--- a/dbcon/mysql/ha_mcs_execplan.cpp
+++ b/dbcon/mysql/ha_mcs_execplan.cpp
@@ -3117,7 +3117,10 @@ CalpontSystemCatalog::ColType colType_MysqlToIDB (const Item* item)
     return ct;
 }
 
-ReturnedColumn* buildReturnedColumn(Item* item, gp_walk_info& gwi, bool& nonSupport, bool pushdownHand, bool isRefItem)
+ReturnedColumn* buildReturnedColumn(
+    Item* item, gp_walk_info& gwi,
+    bool& nonSupport,
+    bool isRefItem)
 {
     ReturnedColumn* rc = NULL;
 
@@ -3275,9 +3278,9 @@ ReturnedColumn* buildReturnedColumn(Item* item, gp_walk_info& gwi, bool& nonSupp
             }
 
             if (func_name == "+" || func_name == "-" || func_name == "*" || func_name == "/" )
-                return buildArithmeticColumn(ifp, gwi, nonSupport, pushdownHand);
+                return buildArithmeticColumn(ifp, gwi, nonSupport);
             else
-                return buildFunctionColumn(ifp, gwi, nonSupport, pushdownHand);
+                return buildFunctionColumn(ifp, gwi, nonSupport);
         }
 
         case Item::SUM_FUNC_ITEM:
@@ -3410,8 +3413,7 @@ ReturnedColumn* buildReturnedColumn(Item* item, gp_walk_info& gwi, bool& nonSupp
 ArithmeticColumn* buildArithmeticColumn(
     Item_func* item,
     gp_walk_info& gwi,
-    bool& nonSupport,
-    bool pushdownHand)
+    bool& nonSupport)
 {
     if (get_fe_conn_info_ptr() == NULL)
         set_fe_conn_info_ptr((void*)new cal_connection_info());
@@ -3435,7 +3437,7 @@ ArithmeticColumn* buildArithmeticColumn(
     {
         if (gwi.clauseType == SELECT || /*gwi.clauseType == HAVING || */gwi.clauseType == GROUP_BY || gwi.clauseType == FROM) // select list
         {
-            lhs = new ParseTree(buildReturnedColumn(sfitempp[0], gwi, nonSupport, pushdownHand));
+            lhs = new ParseTree(buildReturnedColumn(sfitempp[0], gwi, nonSupport));
 
             if (!lhs->data() && (sfitempp[0]->type() == Item::FUNC_ITEM))
             {
@@ -3443,7 +3445,7 @@ ArithmeticColumn* buildArithmeticColumn(
                 Item_func* ifp = (Item_func*)sfitempp[0];
                 lhs = buildParseTree(ifp, gwi, nonSupport);
             }
-            else if(pushdownHand && !lhs->data() && (sfitempp[0]->type() == Item::REF_ITEM))
+            else if(!lhs->data() && (sfitempp[0]->type() == Item::REF_ITEM))
             {
                 // There must be an aggregation column in extended SELECT
                 // list so find the corresponding column.
@@ -3454,7 +3456,7 @@ ArithmeticColumn* buildArithmeticColumn(
                 if(rc)
                     lhs = new ParseTree(rc);
             }
-            rhs = new ParseTree(buildReturnedColumn(sfitempp[1], gwi, nonSupport, pushdownHand));
+            rhs = new ParseTree(buildReturnedColumn(sfitempp[1], gwi, nonSupport));
 
             if (!rhs->data() && (sfitempp[1]->type() == Item::FUNC_ITEM))
             {
@@ -3462,7 +3464,7 @@ ArithmeticColumn* buildArithmeticColumn(
                 Item_func* ifp = (Item_func*)sfitempp[1];
                 rhs = buildParseTree(ifp, gwi, nonSupport);
             }
-            else if(pushdownHand && !rhs->data() && (sfitempp[1]->type() == Item::REF_ITEM))
+            else if(!rhs->data() && (sfitempp[1]->type() == Item::REF_ITEM))
             {
                 // There must be an aggregation column in extended SELECT
                 // list so find the corresponding column.
@@ -3638,7 +3640,6 @@ ReturnedColumn* buildFunctionColumn(
     Item_func* ifp,
     gp_walk_info& gwi,
     bool& nonSupport,
-    bool pushdownHand,
     bool selectBetweenIn)
 {
     if (get_fe_conn_info_ptr() == NULL)
@@ -3680,7 +3681,7 @@ ReturnedColumn* buildFunctionColumn(
     // Arithmetic exp
     if (funcName == "+" || funcName == "-" || funcName == "*" || funcName == "/" )
     {
-        ArithmeticColumn* ac = buildArithmeticColumn(ifp, gwi, nonSupport, pushdownHand);
+        ArithmeticColumn* ac = buildArithmeticColumn(ifp, gwi, nonSupport);
         return ac;
     }
 
@@ -3834,11 +3835,10 @@ ReturnedColumn* buildFunctionColumn(
                     return NULL;
                 }
 
-                ReturnedColumn* rc = buildReturnedColumn(ifp->arguments()[i], gwi, nonSupport, pushdownHand);
+                ReturnedColumn* rc = buildReturnedColumn(ifp->arguments()[i], gwi, nonSupport);
 
                 // MCOL-1510 It must be a temp table field, so find the corresponding column.
-                if (!rc && pushdownHand 
-                    && ifp->arguments()[i]->type() == Item::REF_ITEM)
+                if (!rc && ifp->arguments()[i]->type() == Item::REF_ITEM)
                 {
                     gwi.fatalParseError = false;
                     rc = buildAggFrmTempField(ifp->arguments()[i], gwi);
@@ -5652,7 +5652,7 @@ void gp_walk(const Item* item, void* arg)
 
             if (col->type() != Item::COND_ITEM)
             {
-                rc = buildReturnedColumn(col, *gwip, gwip->fatalParseError, false, true);
+                rc = buildReturnedColumn(col, *gwip, gwip->fatalParseError, false);
 
                 if ( col->type() == Item::FIELD_ITEM )
                     gwip->fatalParseError = false;
@@ -6191,7 +6191,7 @@ int processFrom(bool &isUnion,
             if (table_ptr->derived)
             {
                 SELECT_LEX* select_cursor = table_ptr->derived->first_select();
-                FromSubQuery fromSub(gwi, select_cursor, true);
+                FromSubQuery fromSub(gwi, select_cursor);
                 string alias(table_ptr->alias.str);
                 fromSub.alias(lower(alias));
 
@@ -6303,7 +6303,7 @@ int processFrom(bool &isUnion,
             union_gwi.thd = gwi.thd;
             uint32_t err = 0;
 
-            if ((err = getSelectPlan(union_gwi, *sl, plan, unionSel, true)) != 0)
+            if ((err = getSelectPlan(union_gwi, *sl, plan, unionSel)) != 0)
                 return err;
 
             unionVec.push_back(SCEP(plan));
@@ -6543,25 +6543,176 @@ int processWhere(SELECT_LEX &select_lex,
     return 0;
 }
 
+/*@brief  Process LIMIT part of a query or sub-query      */
+/***********************************************************
+ * DESCRIPTION:
+ *  Processes LIMIT and OFFSET parts
+ * RETURNS
+ *  error id as an int
+ ***********************************************************/
+int processLimitAndOffset(
+    SELECT_LEX& select_lex,
+    gp_walk_info& gwi,
+    SCSEP& csep,
+    bool unionSel,
+    bool isUnion,
+    bool isSelectHandlerTop
+)
+{
+    // LIMIT processing part
+    uint64_t limitNum = std::numeric_limits<uint64_t>::max();
+
+    // non-MAIN union branch
+    if (unionSel || gwi.subSelectType != CalpontSelectExecutionPlan::MAIN_SELECT)
+    {
+        /* Consider the following query:
+           "select a from t1 where exists (select b from t2 where a=b);"
+           CS first builds a hash table for t2, then pushes down the hash to
+           PrimProc for a distributed hash join execution, with t1 being the
+           large-side table. However, the server applies an optimization in
+           Item_exists_subselect::fix_length_and_dec in sql/item_subselect.cc
+           (see server commit ae476868a5394041a00e75a29c7d45917e8dfae8)
+           where it sets explicit_limit to true, which causes csep->limitNum set to 1.
+           This causes the hash table for t2 to only contain a single record for the
+           hash join, giving less number of rows in the output result set than expected.
+           We therefore do not allow limit set to 1 here for such queries.
+        */
+        if (gwi.subSelectType != CalpontSelectExecutionPlan::IN_SUBS
+             && gwi.subSelectType != CalpontSelectExecutionPlan::EXISTS_SUBS
+             && select_lex.master_unit()->global_parameters()->explicit_limit)
+        {
+            if (select_lex.master_unit()->global_parameters()->offset_limit)
+            {
+                Item_int* offset = (Item_int*)select_lex.master_unit()->global_parameters()->offset_limit;
+                csep->limitStart(offset->val_int());
+            }
+
+            if (select_lex.master_unit()->global_parameters()->select_limit)
+            {
+                Item_int* select = (Item_int*)select_lex.master_unit()->global_parameters()->select_limit;
+                csep->limitNum(select->val_int());
+                // MCOL-894 Activate parallel ORDER BY
+                csep->orderByThreads(get_orderby_threads(gwi.thd));
+            }
+        }
+    }
+    // union with explicit select at the top level
+    else if (isUnion && select_lex.explicit_limit)
+    {
+        if (select_lex.braces)
+        {
+            if (select_lex.offset_limit)
+                csep->limitStart(((Item_int*)select_lex.offset_limit)->val_int());
+
+            if (select_lex.select_limit)
+                csep->limitNum(((Item_int*)select_lex.select_limit)->val_int());
+        }
+    }
+    // other types of queries that have explicit LIMIT
+    else if (select_lex.explicit_limit)
+    {
+        uint32_t limitOffset = 0;
+
+        if (select_lex.join)
+        {
+            JOIN* join = select_lex.join;
+#if MYSQL_VERSION_ID >= 50172
+
+            // @bug5729. After upgrade, join->unit sometimes is uninitialized pointer
+            // (not null though) and will cause seg fault. Prefer checking
+            // select_lex->offset_limit if not null.
+            if (join->select_lex &&
+                    join->select_lex->offset_limit &&
+                    join->select_lex->offset_limit->is_fixed() &&
+                    join->select_lex->select_limit &&
+                    join->select_lex->select_limit->is_fixed())
+            {
+                limitOffset = join->select_lex->offset_limit->val_int();
+                limitNum = join->select_lex->select_limit->val_int();
+            }
+            else if (join->unit)
+            {
+                limitOffset = join->unit->offset_limit_cnt;
+                limitNum = join->unit->select_limit_cnt - limitOffset;
+            }
+
+#else
+            limitOffset = (join->unit)->offset_limit_cnt;
+            limitNum = (join->unit)->select_limit_cnt - (join->unit)->offset_limit_cnt;
+#endif
+        }
+        else
+        {
+            if (select_lex.master_unit()->global_parameters()->offset_limit)
+            {
+                Item_int* offset = (Item_int*)select_lex.master_unit()->global_parameters()->offset_limit;
+                limitOffset = offset->val_int();
+            }
+
+            if (select_lex.master_unit()->global_parameters()->select_limit)
+            {
+                Item_int* select = (Item_int*)select_lex.master_unit()->global_parameters()->select_limit;
+                limitNum = select->val_int();
+            }
+        }
+
+        csep->limitStart(limitOffset);
+        csep->limitNum(limitNum);
+    }
+    // If an explicit limit is not specified, use the system variable value
+    else
+    {
+        csep->limitNum(gwi.thd->variables.select_limit);
+    }
+
+    // We don't currently support limit with correlated subquery
+    if (gwi.subQuery && !gwi.correlatedTbNameVec.empty() && csep->hasOrderBy())
+    {
+        gwi.fatalParseError = true;
+        gwi.parseErrorText = IDBErrorInfo::instance()->errorMsg(ERR_NON_SUPPORT_LIMIT_SUB);
+        setError(gwi.thd, ER_INTERNAL_ERROR, gwi.parseErrorText, gwi);
+        return ER_CHECK_NOT_IMPLEMENTED;
+    }
+
+    // MDB applies OFFSET on its own for SH processing.
+    // See related MDEV-16327 for details.
+    if (isSelectHandlerTop && csep->limitStart())
+    {
+        if (std::numeric_limits<uint64_t>::max()-csep->limitStart()
+            < csep->limitNum())
+        {
+            csep->limitNum(std::numeric_limits<uint64_t>::max());
+            csep->limitStart(0);
+        }
+        else
+        {
+            csep->limitNum(csep->limitNum()+csep->limitStart());
+            csep->limitStart(0);
+        }
+    }
+
+    return 0;
+}
+
 /*@brief  Translates SELECT_LEX into CSEP                  */
 /***********************************************************
  * DESCRIPTION:
  *  This function takes SELECT_LEX and tries to produce
  *  a corresponding CSEP out of it. It is made of parts that
  *  process parts of the query, e.g. FROM, WHERE, SELECT,
- *  HAVING, GROUP BY, ORDER BY. FROM and WHERE are processed
- *  by processFrom(), processWhere(). CS calls getSelectPlan()
+ *  HAVING, GROUP BY, ORDER BY. FROM, WHERE, LIMIT are processed
+ *  by corresponding methods. CS calls getSelectPlan()
  *  recursively to process subqueries.
  * ARGS
  *  isUnion if true CS processes UNION unit now
- *  isPushdownHand legacy to be removed
+ *  isSelectHandlerTop removes offset at the top of SH query.
  * RETURNS
  *  error id as an int
  ***********************************************************/
 int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
     SCSEP& csep,
     bool isUnion,
-    bool isPushdownHand)
+    bool isSelectHandlerTop)
 {
 #ifdef DEBUG_WALK_COND
     cerr << "getSelectPlan()" << endl;
@@ -6769,7 +6920,7 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
                 ReturnedColumn* rc;
                 if (funcName == "in" || funcName == " IN " || funcName == "between")
                 {
-                    rc = buildFunctionColumn(ifp, gwi, hasNonSupportItem, false, true);
+                    rc = buildFunctionColumn(ifp, gwi, hasNonSupportItem, false);
                 }
                 else
                 {
@@ -6801,15 +6952,11 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
                     parse_item(ifp, funcFieldVec, hasNonSupportItem, parseInfo, &gwi);
                     uint32_t after_size = funcFieldVec.size();
 
-                    // group by func and func in subquery can not be post processed
                     // pushdown handler projection functions
                     // @bug3881. set_user_var can not be treated as constant function
                     // @bug5716. Try to avoid post process function for union query.
-                    if ((gwi.subQuery || select_lex.group_list.elements != 0 ||
-                            !csep->unionVec().empty() || isUnion || isPushdownHand ) &&
-                            !hasNonSupportItem && (after_size - before_size) == 0 &&
-                            !(parseInfo & AGG_BIT) && !(parseInfo & SUB_BIT)
-                       ) 
+                    if (!hasNonSupportItem && (after_size - before_size) == 0 &&
+                            !(parseInfo & AGG_BIT) && !(parseInfo & SUB_BIT))
                     {
                         String val, *str = ifp->val_str(&val);
                         string valStr;
@@ -7490,6 +7637,7 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
         }
     }
 
+    // ORDER BY processing
     {
         SQL_I_List<ORDER> order_list = select_lex.order_list;
         ORDER* ordercol = reinterpret_cast<ORDER*>(order_list.first);
@@ -7503,9 +7651,9 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
             // MCOL-2166 Looking for this sorting item in GROUP_BY items list.
             // Shouldn't look into this if query doesn't have GROUP BY or 
             // aggregations
-            if(isPushdownHand
-                && select_lex.agg_func_used() && select_lex.group_list.first
-              && !sortItemIsInGrouping(*ordercol->item, select_lex.group_list.first))
+            if(select_lex.agg_func_used() && select_lex.group_list.first
+              && !sortItemIsInGrouping(*ordercol->item,
+                                        select_lex.group_list.first))
             {
                 std::ostringstream ostream;
                 std::ostringstream& osr = ostream;
@@ -7522,10 +7670,6 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
         // re-visit the first of ordercol list
         ordercol = reinterpret_cast<ORDER*>(order_list.first);
 
-        // for subquery or pushdown query, order+limit by will be supported in CS
-        // union order by and limit are supported
-        if (gwi.hasWindowFunc || isPushdownHand || ( isUnion && ordercol )
-        || gwi.subSelectType != CalpontSelectExecutionPlan::MAIN_SELECT )
         {
             for (; ordercol; ordercol = ordercol->next)
             {
@@ -7601,158 +7745,6 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
                 gwi.orderByCols.push_back(SRCP(rc));
             }
         }
-        // DRRTUY The whole block is marked for removal in 1.4.1
-#if 0
-        else if (!isUnion)
-        {
-            vector <Item_field*> fieldVec;
-
-            // the following order by is just for redo phase
-            if (!unionSel)
-            {
-                for (; ordercol; ordercol = ordercol->next)
-                {
-                    Item* ord_item = *(ordercol->item);
-
-                    // @bug5993. Could be nested ref.
-                    while (ord_item->type() == Item::REF_ITEM)
-                        ord_item = (*((Item_ref*)ord_item)->ref);
-
-                    //ReturnedColumn* rc = 0;
-                    // check if this order by column is on the select list
-                    //Item_func* ifp = (Item_func*)(*(ordercol->item));
-                    //rc = buildFunctionColumn(ifp, gwi, gwi.fatalParseError);
-
-                    if (ord_item->type() == Item::FUNC_ITEM)
-                    {
-                        //FunctionColumn* fc = dynamic_cast<FunctionColumn*>(rc);
-                    }
-                    else if (ord_item->type() == Item::SUBSELECT_ITEM)
-                    {
-                        string emsg = IDBErrorInfo::instance()->errorMsg(ERR_NON_SUPPORT_ORDER_BY);
-                        setError(gwi.thd, ER_CHECK_NOT_IMPLEMENTED, emsg, gwi);
-                        return ER_CHECK_NOT_IMPLEMENTED;
-                    }
-                    else if (ord_item->type() == Item::SUM_FUNC_ITEM)
-                    {
-                        ReturnedColumn* ac = 0;
-
-                        Item_sum* ifp = (Item_sum*)(*(ordercol->item));
-                        // @bug3477. add aggregate column to the select list of the create phase.
-                        ac = buildAggregateColumn(ifp, gwi);
-
-                        if (!ac)
-                        {
-                            setError(gwi.thd, ER_CHECK_NOT_IMPLEMENTED,
-                                     IDBErrorInfo::instance()->errorMsg(ERR_NON_SUPPORT_ORDER_BY), gwi);
-                            return ER_CHECK_NOT_IMPLEMENTED;
-                        }
-
-                        // check if this order by column is on the select list
-                        for (uint32_t i = 0; i < gwi.returnedCols.size(); i++)
-                        {
-                            AggregateColumn* ret = dynamic_cast<AggregateColumn*>(gwi.returnedCols[i].get());
-
-                            if (!ret)
-                                continue;
-
-                        }
-
-                        if (ac || !gwi.groupByCols.empty())
-                        {
-                            SRCP srcp(ac);
-                            gwi.returnedCols.push_back(srcp);
-
-                            continue;
-                        }
-                    }
-                    else if (ord_item->name.length && ord_item->type() == Item::FIELD_ITEM)
-                    {
-                        Item_field* field = reinterpret_cast<Item_field*>(ord_item);
-                        ReturnedColumn* rc = buildSimpleColumn(field, gwi);
-
-                        for (uint32_t i = 0; i < gwi.returnedCols.size(); i++)
-                        {
-                            SimpleColumn* sc = dynamic_cast<SimpleColumn*>(gwi.returnedCols[i].get());
-
-                            if (sc && ((Item_field*)ord_item)->cached_table &&
-                                    (strcasecmp(getViewName(((Item_field*)ord_item)->cached_table).c_str(), sc->viewName().c_str()) != 0))
-                            {
-                                continue;
-                            }
-
-                            if (sc && sc->sameColumn(rc))
-                            {
-                                break;
-                            }
-                        }
-                    }
-
-                    {
-                        // @bug 2719. Error out order by not on the distinct select list.
-                        if (select_lex.options & SELECT_DISTINCT)
-                        {
-                            gwi.parseErrorText = IDBErrorInfo::instance()->errorMsg(ERR_ORDERBY_NOT_IN_DISTINCT);
-                            setError(gwi.thd, ER_CHECK_NOT_IMPLEMENTED, gwi.parseErrorText, gwi);
-                            return ER_CHECK_NOT_IMPLEMENTED;
-                        }
-
-                        bool hasNonSupportItem = false;
-                        uint16_t parseInfo = 0;
-                        parse_item(ord_item, fieldVec, hasNonSupportItem, parseInfo, &gwi);
-
-                        if (hasNonSupportItem)
-                        {
-                            string emsg = IDBErrorInfo::instance()->errorMsg(ERR_NON_SUPPORT_ORDER_BY);
-                            setError(gwi.thd, ER_CHECK_NOT_IMPLEMENTED, emsg, gwi);
-                            return ER_CHECK_NOT_IMPLEMENTED;
-                        }
-
-                    }
-                }
-            }
-
-            // populate string to be added to the select list for order by
-            for (uint32_t i = 0; i < fieldVec.size(); i++)
-            {
-                SimpleColumn* sc = buildSimpleColumn(fieldVec[i], gwi);
-
-                if (!sc)
-                {
-                    string emsg = IDBErrorInfo::instance()->errorMsg(ERR_NON_SUPPORT_ORDER_BY);
-                    setError(gwi.thd, ER_CHECK_NOT_IMPLEMENTED, emsg, gwi);
-                    return ER_CHECK_NOT_IMPLEMENTED;
-                }
-
-                String str;
-                fieldVec[i]->print(&str, QT_ORDINARY);
-                sc->alias(string(str.c_ptr()));
-                SRCP srcp(sc);
-                uint32_t j = 0;
-
-                for (; j < gwi.returnedCols.size(); j++)
-                {
-                    if (sc->sameColumn(gwi.returnedCols[j].get()))
-                    {
-                        SimpleColumn* field = dynamic_cast<SimpleColumn*>(gwi.returnedCols[j].get());
-
-                        if (field && field->alias() == sc->alias())
-                            break;
-                    }
-                }
-
-                if (j == gwi.returnedCols.size())
-                {
-                    gwi.returnedCols.push_back(srcp);
-                    gwi.columnMap.insert(CalpontSelectExecutionPlan::ColumnMap::value_type(string(fieldVec[i]->field_name.str), srcp));
-                    TABLE_LIST* tmp = (fieldVec[i]->cached_table ? fieldVec[i]->cached_table : 0);
-                    gwi.tableMap[make_aliastable(sc->schemaName(), sc->tableName(), sc->tableAlias(), sc->isColumnStore())] =
-                        make_pair(1, tmp);
-                }
-            }
-        }
-#endif
-
         // make sure columnmap, returnedcols and count(*) arg_list are not empty
         TableMap::iterator tb_iter = gwi.tableMap.begin();
 
@@ -7878,129 +7870,16 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
             {
                 csep->hasOrderBy(true);
                 // To activate LimitedOrderBy
-                if(isPushdownHand)
-                {
-                    csep->orderByThreads(get_orderby_threads(gwi.thd));
-                    csep->specHandlerProcessed(true);
-                }
+                csep->orderByThreads(get_orderby_threads(gwi.thd));
+                csep->specHandlerProcessed(true);
             }
         }
 
-        // LIMIT processing part
-        uint64_t limitNum = std::numeric_limits<uint64_t>::max();
-
-        // non-MAIN union branch
-        if (unionSel || gwi.subSelectType != CalpontSelectExecutionPlan::MAIN_SELECT)
+        if ((rc = processLimitAndOffset(select_lex, gwi, csep, unionSel, isUnion, isSelectHandlerTop)))
         {
-            /* Consider the following query:
-               "select a from t1 where exists (select b from t2 where a=b);"
-               CS first builds a hash table for t2, then pushes down the hash to
-               PrimProc for a distributed hash join execution, with t1 being the
-               large-side table. However, the server applies an optimization in
-               Item_exists_subselect::fix_length_and_dec in sql/item_subselect.cc
-               (see server commit ae476868a5394041a00e75a29c7d45917e8dfae8)
-               where it sets explicit_limit to true, which causes csep->limitNum set to 1.
-               This causes the hash table for t2 to only contain a single record for the
-               hash join, giving less number of rows in the output result set than expected.
-               We therefore do not allow limit set to 1 here for such queries.
-            */
-            if (gwi.subSelectType != CalpontSelectExecutionPlan::IN_SUBS
-                 && gwi.subSelectType != CalpontSelectExecutionPlan::EXISTS_SUBS
-                 && select_lex.master_unit()->global_parameters()->explicit_limit)
-            {
-                if (select_lex.master_unit()->global_parameters()->offset_limit)
-                {
-                    Item_int* offset = (Item_int*)select_lex.master_unit()->global_parameters()->offset_limit;
-                    csep->limitStart(offset->val_int());
-                }
-
-                if (select_lex.master_unit()->global_parameters()->select_limit)
-                {
-                    Item_int* select = (Item_int*)select_lex.master_unit()->global_parameters()->select_limit;
-                    csep->limitNum(select->val_int());
-                    // MCOL-894 Activate parallel ORDER BY
-                    csep->orderByThreads(get_orderby_threads(gwi.thd));
-                }
-            }
+            return rc;
         }
-        // union with explicit select at the top level
-        else if (isUnion && select_lex.explicit_limit)
-        {
-            if (select_lex.braces)
-            {
-                if (select_lex.offset_limit)
-                    csep->limitStart(((Item_int*)select_lex.offset_limit)->val_int());
-
-                if (select_lex.select_limit)
-                    csep->limitNum(((Item_int*)select_lex.select_limit)->val_int());
-            }
-        }
-        // other types of queries that have explicit LIMIT
-        else if (select_lex.explicit_limit)
-        {
-            uint32_t limitOffset = 0;
-
-            if (select_lex.join)
-            {
-                JOIN* join = select_lex.join;
-#if MYSQL_VERSION_ID >= 50172
-
-                // @bug5729. After upgrade, join->unit sometimes is uninitialized pointer
-                // (not null though) and will cause seg fault. Prefer checking
-                // select_lex->offset_limit if not null.
-                if (join->select_lex &&
-                        join->select_lex->offset_limit &&
-                        join->select_lex->offset_limit->is_fixed() &&
-                        join->select_lex->select_limit &&
-                        join->select_lex->select_limit->is_fixed())
-                {
-                    limitOffset = join->select_lex->offset_limit->val_int();
-                    limitNum = join->select_lex->select_limit->val_int();
-                }
-                else if (join->unit)
-                {
-                    limitOffset = join->unit->offset_limit_cnt;
-                    limitNum = join->unit->select_limit_cnt - limitOffset;
-                }
-
-#else
-                limitOffset = (join->unit)->offset_limit_cnt;
-                limitNum = (join->unit)->select_limit_cnt - (join->unit)->offset_limit_cnt;
-#endif
-            }
-            else
-            {
-                if (select_lex.master_unit()->global_parameters()->offset_limit)
-                {
-                    Item_int* offset = (Item_int*)select_lex.master_unit()->global_parameters()->offset_limit;
-                    limitOffset = offset->val_int();
-                }
-
-                if (select_lex.master_unit()->global_parameters()->select_limit)
-                {
-                    Item_int* select = (Item_int*)select_lex.master_unit()->global_parameters()->select_limit;
-                    limitNum = select->val_int();
-                }
-            }
-
-            csep->limitStart(limitOffset);
-            csep->limitNum(limitNum);
-        }
-        // If an explicit limit is not specified, use the system variable value
-        else
-        {
-            csep->limitNum(gwi.thd->variables.select_limit);
-        }
-
-        // We don't currently support limit with correlated subquery
-        if (gwi.subQuery && !gwi.correlatedTbNameVec.empty() && csep->hasOrderBy())
-        {
-            gwi.fatalParseError = true;
-            gwi.parseErrorText = IDBErrorInfo::instance()->errorMsg(ERR_NON_SUPPORT_LIMIT_SUB);
-            setError(gwi.thd, ER_INTERNAL_ERROR, gwi.parseErrorText, gwi);
-            return ER_CHECK_NOT_IMPLEMENTED;
-        }
-    } // LIMIT processing finishes here
+    } // ORDER BY end
 
     if (select_lex.options & SELECT_DISTINCT)
         csep->distinct(true);
@@ -8214,7 +8093,7 @@ int cp_get_group_plan(THD* thd, SCSEP& csep, cal_impl_if::cal_group_info& gi)
 int cs_get_derived_plan(derived_handler* handler, THD* thd, SCSEP& csep, gp_walk_info& gwi)
 {
     SELECT_LEX select_lex = *handler->select;
-    int status = getSelectPlan(gwi, select_lex, csep, false, true);
+    int status = getSelectPlan(gwi, select_lex, csep, false);
 
     if (status > 0)
         return ER_INTERNAL_ERROR;
@@ -8412,7 +8291,7 @@ int getGroupPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, cal_gro
 
                 SELECT_LEX* select_cursor = table_ptr->derived->first_select();
                 // Use Pushdown handler for subquery processing
-                FromSubQuery fromSub(gwi, select_cursor, true);
+                FromSubQuery fromSub(gwi, select_cursor);
                 string alias(table_ptr->alias.str);
                 fromSub.alias(lower(alias));
 
diff --git a/dbcon/mysql/ha_mcs_impl.cpp b/dbcon/mysql/ha_mcs_impl.cpp
index 5e35c383c..e93275be6 100644
--- a/dbcon/mysql/ha_mcs_impl.cpp
+++ b/dbcon/mysql/ha_mcs_impl.cpp
@@ -1780,7 +1780,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi)
 
         gwi.clauseType = WHERE;
 
-        if (getSelectPlan(gwi, select_lex, updateCP, false, true) != 0) //@Bug 3030 Modify the error message for unsupported functions
+        if (getSelectPlan(gwi, select_lex, updateCP, false) != 0) //@Bug 3030 Modify the error message for unsupported functions
         {
             if (gwi.cs_vtable_is_update_with_derive)
             {
diff --git a/dbcon/mysql/ha_mcs_impl_if.h b/dbcon/mysql/ha_mcs_impl_if.h
index 3c1d2ada6..9975605b7 100644
--- a/dbcon/mysql/ha_mcs_impl_if.h
+++ b/dbcon/mysql/ha_mcs_impl_if.h
@@ -342,7 +342,7 @@ int cp_get_table_plan(THD* thd, execplan::SCSEP& csep, cal_impl_if::cal_table_in
 int cp_get_group_plan(THD* thd, execplan::SCSEP& csep, cal_impl_if::cal_group_info& gi);
 int cs_get_derived_plan(derived_handler* handler, THD* thd, execplan::SCSEP& csep, gp_walk_info& gwi);
 int cs_get_select_plan(select_handler* handler, THD* thd, execplan::SCSEP& csep, gp_walk_info& gwi);
-int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, execplan::SCSEP& csep, bool isUnion = false, bool isPushdownHand = false);
+int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, execplan::SCSEP& csep, bool isUnion = false, bool isSelectHandlerTop = false);
 int getGroupPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, execplan::SCSEP& csep, cal_group_info& gi, bool isUnion = false);
 void setError(THD* thd, uint32_t errcode, const std::string errmsg, gp_walk_info* gwi);
 void setError(THD* thd, uint32_t errcode, const std::string errmsg);
@@ -352,9 +352,9 @@ const std::string bestTableName(const Item_field* ifp);
 bool isMCSTable(TABLE* table_ptr);
 
 // execution plan util functions prototypes
-execplan::ReturnedColumn* buildReturnedColumn(Item* item, gp_walk_info& gwi, bool& nonSupport, bool pushdownHand = false, bool isRefItem = false);
-execplan::ReturnedColumn* buildFunctionColumn(Item_func* item, gp_walk_info& gwi, bool& nonSupport, bool pushdownHand = false, bool selectBetweenIn = false);
-execplan::ArithmeticColumn* buildArithmeticColumn(Item_func* item, gp_walk_info& gwi, bool& nonSupport, bool pushdownHand = false);
+execplan::ReturnedColumn* buildReturnedColumn(Item* item, gp_walk_info& gwi, bool& nonSupport, bool isRefItem = false);
+execplan::ReturnedColumn* buildFunctionColumn(Item_func* item, gp_walk_info& gwi, bool& nonSupport, bool selectBetweenIn = false);
+execplan::ArithmeticColumn* buildArithmeticColumn(Item_func* item, gp_walk_info& gwi, bool& nonSupport);
 execplan::ConstantColumn* buildDecimalColumn(Item* item, gp_walk_info& gwi);
 execplan::SimpleColumn* buildSimpleColumn(Item_field* item, gp_walk_info& gwi);
 execplan::FunctionColumn* buildCaseFunction(Item_func* item, gp_walk_info& gwi, bool& nonSupport);
diff --git a/dbcon/mysql/ha_mcs_pushdown.cpp b/dbcon/mysql/ha_mcs_pushdown.cpp
index 3b0764a66..64ad79eae 100644
--- a/dbcon/mysql/ha_mcs_pushdown.cpp
+++ b/dbcon/mysql/ha_mcs_pushdown.cpp
@@ -787,21 +787,9 @@ create_columnstore_select_handler(THD* thd, SELECT_LEX* select_lex)
         return handler;
     }
 
-    // Remove this in 1.4.3
-    // Save the original group_list as it can be mutated by the
-    // optimizer which calls the remove_const() function
-    Group_list_ptrs *group_list_ptrs = NULL;
-    if (save_group_list(thd, select_lex, &group_list_ptrs))
-    {
-        return handler;
-    }
-
-    // Select_handler use the short-cut that effectively disables
-    // INSERT..SELECT, LDI, SELECT..INTO OUTFILE
+    // Select_handler couldn't properly process UPSERT..SELECT
     if ((thd->lex)->sql_command == SQLCOM_INSERT_SELECT
-        || (thd->lex)->sql_command == SQLCOM_CREATE_TABLE
-        || (thd->lex)->exchange)
-        
+            && thd->lex->duplicates == DUP_UPDATE)
     {
         return handler;
     }
diff --git a/dbcon/mysql/ha_scalar_sub.cpp b/dbcon/mysql/ha_scalar_sub.cpp
index f41482ba2..7d6d2dd11 100644
--- a/dbcon/mysql/ha_scalar_sub.cpp
+++ b/dbcon/mysql/ha_scalar_sub.cpp
@@ -253,7 +253,7 @@ execplan::ParseTree* ScalarSub::buildParseTree(PredicateOperator* op)
     gwi.tbList.insert(gwi.tbList.begin(), fGwip.tbList.begin(), fGwip.tbList.end());
     gwi.derivedTbList.insert(gwi.derivedTbList.begin(), fGwip.derivedTbList.begin(), fGwip.derivedTbList.end());
 
-    if (getSelectPlan(gwi, *(fSub->get_select_lex()), csep, false, true) != 0)
+    if (getSelectPlan(gwi, *(fSub->get_select_lex()), csep, false) != 0)
     {
         //@todo more in error handling
         if (!gwi.fatalParseError)
diff --git a/dbcon/mysql/ha_select_sub.cpp b/dbcon/mysql/ha_select_sub.cpp
index 708d2cca8..6d5dfd611 100644
--- a/dbcon/mysql/ha_select_sub.cpp
+++ b/dbcon/mysql/ha_select_sub.cpp
@@ -77,7 +77,7 @@ SCSEP SelectSubQuery::transform()
     gwi.tbList.insert(gwi.tbList.begin(), fGwip.tbList.begin(), fGwip.tbList.end());
     gwi.derivedTbList.insert(gwi.derivedTbList.begin(), fGwip.derivedTbList.begin(), fGwip.derivedTbList.end());
 
-    if (getSelectPlan(gwi, *(fSelSub->get_select_lex()), csep, false, true) != 0)
+    if (getSelectPlan(gwi, *(fSelSub->get_select_lex()), csep, false) != 0)
     {
         if (!gwi.fatalParseError)
         {
diff --git a/dbcon/mysql/ha_subquery.h b/dbcon/mysql/ha_subquery.h
index d14c95131..506cdf0eb 100644
--- a/dbcon/mysql/ha_subquery.h
+++ b/dbcon/mysql/ha_subquery.h
@@ -181,7 +181,7 @@ class FromSubQuery : public SubQuery
 {
 public:
     FromSubQuery(gp_walk_info&);
-    FromSubQuery(gp_walk_info&, SELECT_LEX* fromSub, bool isPushdownHand=false);
+    FromSubQuery(gp_walk_info&, SELECT_LEX* fromSub);
     ~FromSubQuery();
     const SELECT_LEX* fromSub() const
     {
diff --git a/utils/loggingcpp/ErrorMessage.txt b/utils/loggingcpp/ErrorMessage.txt
index 3b066fc0f..08233bbf6 100755
--- a/utils/loggingcpp/ErrorMessage.txt
+++ b/utils/loggingcpp/ErrorMessage.txt
@@ -17,7 +17,7 @@
 # Non support errors 1000 ~ 2000. 
 # The query will go through the optimizer again with some optimization turned off
 1000	ERR_MISS_JOIN	%1% not joined.
-1001	ERR_NON_SUPPORTED_FUNCTION	Function '%1%' can only be used in the outermost select or order by clause and cannot be used in conjunction with an aggregate function.
+1001	ERR_NON_SUPPORTED_FUNCTION	Function '%1%' isn't supported.
 1002	ERR_INCOMPATIBLE_JOIN	%1% incompatible column type specified for join condition.
 1003	ERR_CIRCULAR_JOIN	Circular joins are not supported.
 1004	ERR_MIX_JOIN	Mixed %1% JOIN is not supported.