Add limit and sort (#86)

* add limit and sort

* delete extra space

* add sort validate

* add test

* rebase upstream

* update

* add null test

* rebase upstream

* add union iterator eraseRange

* address dutor's comment

* address comment

* update

* address comments

* rebase upstream

Co-authored-by: dutor <440396+dutor@users.noreply.github.com>

src/context/ExecutionContext.h

@@ -175,3 +175,4 @@ private:
 }  // namespace graph
 }  // namespace nebula
 #endif  // CONTEXT_EXECUTIONCONTEXT_H_
+

src/context/Iterator.h

@@ -14,6 +14,7 @@
 #include "common/datatypes/Value.h"
 #include "common/datatypes/List.h"
 #include "common/datatypes/DataSet.h"
+#include "parser/TraverseSentences.h"
 
 namespace nebula {
 namespace graph {
@@ -24,7 +25,6 @@ public:
         kDefault,
         kGetNeighbors,
         kSequential,
-        kUnion,
     };
 
     explicit Iterator(std::shared_ptr<Value> value, Kind kind)
@@ -42,10 +42,14 @@ public:
 
     virtual void next() = 0;
 
+    // erase current iter
     virtual void erase() = 0;
 
     virtual const Row* row() const = 0;
 
+    // erase range, no include last position, if last > size(), erase to the end position
+    virtual void eraseRange(size_t first, size_t last) = 0;
+
     // Reset iterator position to `pos' from begin. Must be sure that the `pos' position
     // is lower than `size()' before resetting
     void reset(size_t pos = 0) {
@@ -53,6 +57,8 @@ public:
         doReset(pos);
     }
 
+    virtual void clear() = 0;
+
     void operator++() {
         next();
     }
@@ -71,6 +77,10 @@ public:
 
     virtual size_t size() const = 0;
 
+    bool isDefaultIter() const {
+        return kind_ == Kind::kDefault;
+    }
+
     bool isGetNeighborsIter() const {
         return kind_ == Kind::kGetNeighbors;
     }
@@ -132,6 +142,14 @@ public:
         counter_--;
     }
 
+    void eraseRange(size_t, size_t) override {
+        return;
+    }
+
+    void clear() override {
+        reset();
+    }
+
     size_t size() const override {
         return 1;
     }
@@ -172,12 +190,30 @@ public:
         }
     }
 
+    void clear() override {
+        valid_ = false;
+        dsIndices_.clear();
+        logicalRows_.clear();
+    }
+
     void erase() override {
         if (valid()) {
             iter_ = logicalRows_.erase(iter_);
         }
     }
 
+    void eraseRange(size_t first, size_t last) override {
+        if (first >= last || first >= size()) {
+            return;
+        }
+        if (last > size()) {
+            logicalRows_.erase(logicalRows_.begin() + first, logicalRows_.end());
+        } else {
+            logicalRows_.erase(logicalRows_.begin() + first, logicalRows_.begin() + last);
+        }
+        reset();
+    }
+
     size_t size() const override {
         return logicalRows_.size();
     }
@@ -233,12 +269,6 @@ private:
         iter_ = logicalRows_.begin() + pos;
     }
 
-    void clear() {
-        valid_ = false;
-        dsIndices_.clear();
-        logicalRows_.clear();
-    }
-
     inline size_t currentSeg() const {
         return iter_->dsIdx;
     }
@@ -306,10 +336,27 @@ public:
         }
         iter_ = rows_.begin();
         for (size_t i = 0; i < ds.colNames.size(); ++i) {
-            colIndex_.emplace(ds.colNames[i], i);
+            colIndexes_.emplace(ds.colNames[i], i);
         }
     }
 
+    SequentialIter(std::unique_ptr<Iterator> left, std::unique_ptr<Iterator> right)
+        : Iterator(nullptr, Kind::kSequential) {
+        DCHECK(left->isSequentialIter());
+        DCHECK(right->isSequentialIter());
+        auto lIter = static_cast<SequentialIter*>(left.get());
+        auto rIter = static_cast<SequentialIter*>(right.get());
+        rows_.insert(rows_.end(),
+                     std::make_move_iterator(lIter->begin()),
+                     std::make_move_iterator(lIter->end()));
+
+        rows_.insert(rows_.end(),
+                     std::make_move_iterator(rIter->begin()),
+                     std::make_move_iterator(rIter->end()));
+        iter_ = rows_.begin();
+        colIndexes_ = lIter->getColIndexes();
+    }
+
     std::unique_ptr<Iterator> copy() const override {
         auto copy = std::make_unique<SequentialIter>(*this);
         copy->reset();
@@ -330,6 +377,35 @@ public:
         iter_ = rows_.erase(iter_);
     }
 
+    void eraseRange(size_t first, size_t last) override {
+        if (first >= last || first >= size()) {
+            return;
+        }
+        if (last > size()) {
+            rows_.erase(rows_.begin() + first, rows_.end());
+        } else {
+            rows_.erase(rows_.begin() + first, rows_.begin() + last);
+        }
+        reset();
+    }
+
+    void clear() override {
+        rows_.clear();
+        reset();
+    }
+
+    std::vector<const Row*>::iterator begin() {
+        return rows_.begin();
+    }
+
+    std::vector<const Row*>::iterator end() {
+        return rows_.end();
+    }
+
+    const std::unordered_map<std::string, int64_t>& getColIndexes() const {
+        return colIndexes_;
+    }
+
     size_t size() const override {
         return rows_.size();
     }
@@ -339,8 +415,8 @@ public:
             return Value::kNullValue;
         }
         auto row = *iter_;
-        auto index = colIndex_.find(col);
-        if (index == colIndex_.end()) {
+        auto index = colIndexes_.find(col);
+        if (index == colIndexes_.end()) {
             return Value::kNullValue;
         } else {
             DCHECK_LT(index->second, row->values.size());
@@ -349,6 +425,9 @@ public:
     }
 
     const Row* row() const override {
+        if (!valid()) {
+            return nullptr;
+        }
         return *iter_;
     }
 
@@ -364,84 +443,13 @@ private:
         iter_ = rows_.begin() + pos;
     }
 
+private:
     std::vector<const Row*>                      rows_;
     std::vector<const Row*>::iterator            iter_;
-    std::unordered_map<std::string, int64_t>     colIndex_;
-};
-
-class UnionIterator final : public Iterator {
-public:
-    UnionIterator(std::unique_ptr<Iterator> left, std::unique_ptr<Iterator> right)
-        : Iterator(left->valuePtr(), Kind::kUnion),
-          left_(std::move(left)),
-          right_(std::move(right)) {}
-
-    std::unique_ptr<Iterator> copy() const override {
-        auto iter = std::make_unique<UnionIterator>(left_->copy(), right_->copy());
-        iter->reset();
-        return iter;
-    }
-
-    bool valid() const override {
-        return left_->valid() || right_->valid();
-    }
-
-    void next() override {
-        if (left_->valid()) {
-            left_->next();
-        } else {
-            if (right_->valid()) {
-                right_->next();
-            }
-        }
-    }
-
-    size_t size() const override {
-        return left_->size() + right_->size();
-    }
-
-    void erase() override {
-        if (left_->valid()) {
-            left_->erase();
-        } else {
-            if (right_->valid()) {
-                right_->erase();
-            }
-        }
-    }
-
-    const Value& getColumn(const std::string& col) const override {
-        if (left_->valid()) {
-            return left_->getColumn(col);
-        }
-        if (right_->valid()) {
-            return right_->getColumn(col);
-        }
-        return Value::kEmpty;
-    }
-
-    const Row* row() const override {
-        if (left_->valid()) return left_->row();
-        if (right_->valid()) return right_->row();
-        return nullptr;
-    }
-
-private:
-    void doReset(size_t poc) override {
-        if (poc < left_->size()) {
-            left_->reset(poc);
-            right_->reset();
-        } else {
-            right_->reset(poc - left_->size());
-        }
-    }
-
-    std::unique_ptr<Iterator> left_;
-    std::unique_ptr<Iterator> right_;
+    std::unordered_map<std::string, int64_t>     colIndexes_;
 };
 }  // namespace graph
 }  // namespace nebula
-
 namespace std {
 template <>
 struct equal_to<const nebula::Row*> {

src/context/test/IteratorTest.cpp

@@ -546,56 +546,45 @@ TEST(IteratorTest, TestHead) {
     }
 }
 
-TEST(IteratorTest, TestUnionIterator) {
-    std::vector<std::string> colNames = {"col1", "col2"};
-    DataSet lds;
-    lds.colNames = colNames;
-    lds.rows = {
-        Row({Value(1), Value("row1")}),
-        Row({Value(2), Value("row2")}),
-    };
-    auto lIter = std::make_unique<SequentialIter>(std::make_shared<Value>(lds));
-
-    DataSet rds;
-    rds.colNames = colNames;
-    rds.rows = {
-        Row({Value(3), Value("row3")}),
-    };
-    auto rIter = std::make_unique<SequentialIter>(std::make_shared<Value>(rds));
-
-    // next and valid
+TEST(IteratorTest, EraseRange) {
+    // Sequential iterator
     {
-        auto uIter = std::make_unique<UnionIterator>(lIter->copy(), rIter->copy());
-        for (; lIter->valid(); lIter->next()) {
-            EXPECT_TRUE(uIter->valid());
-            for (auto &col : colNames) {
-                EXPECT_EQ(lIter->getColumn(col), uIter->getColumn(col));
-            }
-            uIter->next();
+        DataSet ds({"col1", "col2"});
+        for (auto i = 0; i < 10; ++i) {
+            ds.rows.emplace_back(Row({i, folly::to<std::string>(i)}));
         }
-
-        for (; rIter->valid(); rIter->next()) {
-            EXPECT_TRUE(uIter->valid());
-            for (auto &col : colNames) {
-                EXPECT_EQ(rIter->getColumn(col), uIter->getColumn(col));
+        // erase out of range pos
+        {
+            auto val = std::make_shared<Value>(ds);
+            SequentialIter iter(val);
+            iter.eraseRange(5, 11);
+            ASSERT_EQ(iter.size(), 5);
+            auto i = 0;
+            for (; iter.valid(); iter.next()) {
+                ASSERT_EQ(iter.getColumn("col1"), i);
+                ASSERT_EQ(iter.getColumn("col2"), folly::to<std::string>(i));
+                ++i;
             }
-            uIter->next();
         }
-    }
-
-    // erase
-    {
-        auto uIter = std::make_unique<UnionIterator>(lIter->copy(), rIter->copy());
-        for (; lIter->valid(); lIter->next()) {
-            uIter->erase();
+        // erase in range
+        {
+            auto val = std::make_shared<Value>(ds);
+            SequentialIter iter(val);
+            iter.eraseRange(0, 10);
+            ASSERT_EQ(iter.size(), 0);
         }
-
-        for (; rIter->valid(); rIter->next()) {
-            EXPECT_TRUE(uIter->valid());
-            for (auto &col : colNames) {
-                EXPECT_EQ(rIter->getColumn(col), uIter->getColumn(col));
+        // erase part
+        {
+            auto val = std::make_shared<Value>(ds);
+            SequentialIter iter(val);
+            iter.eraseRange(0, 5);
+            EXPECT_EQ(iter.size(), 5);
+            auto i = 5;
+            for (; iter.valid(); iter.next()) {
+                ASSERT_EQ(iter.getColumn("col1"), i);
+                ASSERT_EQ(iter.getColumn("col2"), folly::to<std::string>(i));
+                ++i;
             }
-            uIter->next();
         }
     }
 }

src/exec/query/DedupExecutor.cpp

@@ -19,7 +19,7 @@ folly::Future<Status> DedupExecutor::execute() {
         return Status::Error("Internal Error: iterator is nullptr");
     }
 
-    if (iter->kind() == Iterator::Kind::kGetNeighbors) {
+    if (iter->isGetNeighborsIter()) {
         LOG(INFO) << "Invalid iterator kind: " << static_cast<uint16_t>(iter->kind());
         return Status::Error("Invalid iterator kind, %d", static_cast<uint16_t>(iter->kind()));
     }

src/exec/query/LimitExecutor.cpp

@@ -5,16 +5,32 @@
  */
 
 #include "exec/query/LimitExecutor.h"
-
-#include "planner/PlanNode.h"
+#include "context/ExpressionContextImpl.h"
+#include "planner/Query.h"
 
 namespace nebula {
 namespace graph {
 
 folly::Future<Status> LimitExecutor::execute() {
     dumpLog();
-    // TODO(yee): Get all neighbors by storage client
-    return start();
+    auto* limit = asNode<Limit>(node());
+    auto iter = ectx_->getResult(limit->inputVar()).iter();
+    auto result = ExecResult::buildDefault(iter->valuePtr());
+    ExpressionContextImpl ctx(ectx_, iter.get());
+    auto offset = limit->offset();
+    auto count = limit->count();
+    auto size = iter->size();
+    if (size <= static_cast<size_t>(offset)) {
+        iter->clear();
+    } else if (size > static_cast<size_t>(offset + count)) {
+        iter->eraseRange(0, offset);
+        iter->eraseRange(count, size - offset);
+    } else if (size > static_cast<size_t>(offset) &&
+               size <= static_cast<size_t>(offset + count)) {
+        iter->eraseRange(0, offset);
+    }
+    result.setIter(std::move(iter));
+    return finish(std::move(result));
 }
 
 }   // namespace graph

src/exec/query/LimitExecutor.h

@@ -17,7 +17,6 @@ public:
     LimitExecutor(const PlanNode *node, QueryContext *qctx)
         : Executor("LimitExecutor", node, qctx) {}
 
-private:
     folly::Future<Status> execute() override;
 };
 

src/exec/query/SortExecutor.cpp

@@ -5,16 +5,60 @@
  */
 
 #include "exec/query/SortExecutor.h"
-
-#include "planner/PlanNode.h"
+#include "context/ExpressionContextImpl.h"
+#include "planner/Query.h"
 
 namespace nebula {
 namespace graph {
 
 folly::Future<Status> SortExecutor::execute() {
     dumpLog();
-    // TODO(yee):
-    return start();
+    auto* sort = asNode<Sort>(node());
+    auto iter = ectx_->getResult(sort->inputVar()).iter();
+    if (UNLIKELY(iter == nullptr)) {
+        return Status::Error("Internal error: nullptr iterator in sort executor");
+    }
+    if (UNLIKELY(iter->isGetNeighborsIter())) {
+        std::string errMsg = "Internal error: Sort executor does not supported GetNeighborsIter";
+        LOG(ERROR) << errMsg;
+        return Status::Error(errMsg);
+    }
+    if (iter->isSequentialIter()) {
+        auto seqIter = static_cast<SequentialIter*>(iter.get());
+        auto &factors = sort->factors();
+        auto &colIndexes = seqIter->getColIndexes();
+        std::vector<std::pair<size_t, OrderFactor::OrderType>> indexes;
+        for (auto &factor : factors) {
+            auto indexFind = colIndexes.find(factor.first);
+            if (indexFind == colIndexes.end()) {
+                LOG(ERROR) << "Column name `" << factor.first << "' is not exist.";
+                return Status::Error("Column name `%s' is not exist.", factor.first.c_str());
+            }
+            indexes.emplace_back(std::make_pair(indexFind->second, factor.second));
+        }
+        auto comparator = [&indexes] (const Row* lhs, const Row *rhs) {
+            const auto &lhsColumns = lhs->values;
+            const auto &rhsColumns = rhs->values;
+            for (auto &item : indexes) {
+                auto index = item.first;
+                auto orderType = item.second;
+                if (lhsColumns[index] == rhsColumns[index]) {
+                    continue;
+                }
+
+                if (orderType == OrderFactor::OrderType::ASCEND) {
+                    return lhsColumns[index] < rhsColumns[index];
+                } else if (orderType == OrderFactor::OrderType::DESCEND) {
+                    return lhsColumns[index] > rhsColumns[index];
+                }
+            }
+            return false;
+        };
+        std::sort(seqIter->begin(), seqIter->end(), comparator);
+    }
+    auto result = ExecResult::buildDefault(iter->valuePtr());
+    result.setIter(std::move(iter));
+    return finish(std::move(result));
 }
 
 }   // namespace graph

src/exec/query/UnionExecutor.cpp

@@ -17,7 +17,7 @@ folly::Future<Status> UnionExecutor::execute() {
     auto left = getLeftInputDataIter();
     auto right = getRightInputDataIter();
     auto result = ExecResult::buildDefault(left->valuePtr());
-    auto iter = std::make_unique<UnionIterator>(std::move(left), std::move(right));
+    auto iter = std::make_unique<SequentialIter>(std::move(left), std::move(right));
     result.setIter(std::move(iter));
     return finish(std::move(result));
 }

src/exec/query/test/CMakeLists.txt

@@ -62,6 +62,8 @@ nebula_add_test(
         SetExecutorTest.cpp
         FilterTest.cpp
         DedupTest.cpp
+        LimitTest.cpp
+        SortTest.cpp
     OBJECTS
         ${EXEC_QUERY_TEST_OBJS}
     LIBRARIES

src/exec/query/test/DedupTest.cpp

@@ -43,7 +43,8 @@ public:
         auto* project =                                                        \
             Project::make(plan, nullptr, yieldSentence->yieldColumns());       \
         project->setInputVar(dedupNode->varName());                            \
-        project->setColNames(std::vector<std::string>{"name"});                \
+        auto colNames = expected.colNames;                                     \
+        project->setColNames(std::move(colNames));                             \
                                                                                \
         auto proExe = std::make_unique<ProjectExecutor>(project, qctx_.get()); \
         EXPECT_TRUE(proExe->execute().get().ok());                             \
@@ -54,14 +55,18 @@ public:
     } while (false)
 
 TEST_F(DedupTest, TestSequential) {
-    DataSet expected({"name"});
-    expected.emplace_back(Row({Value("School1")}));
-    expected.emplace_back(Row({Value("School1")}));
-    expected.emplace_back(Row({Value("School2")}));
+    DataSet expected({"vid", "name", "age", "dst", "start", "end"});
+    expected.emplace_back(Row({"Ann", "Ann", 18, "School1", 2010, 2014}));
+    expected.emplace_back(Row({"Joy", "Joy", Value::kNullValue, "School2", 2009, 2012}));
+    expected.emplace_back(Row({"Tom", "Tom", 20, "School2", 2008, 2012}));
+    expected.emplace_back(Row({"Kate", "Kate", 19, "School2", 2009, 2013}));
+    expected.emplace_back(Row({"Lily", "Lily", 20, "School2", 2009, 2012}));
 
+    auto sentence = "YIELD DISTINCT $-.vid as vid, $-.v_name as name, $-.v_age as age, "
+                    "$-.v_dst as dst, $-.e_start_year as start, $-.e_end_year as end";
     DEDUP_RESUTL_CHECK("input_sequential",
                        "dedup_sequential",
-                       "YIELD DISTINCT $-.v_dst as name",
+                       sentence,
                        expected);
 }
 

src/exec/query/test/LimitTest.cpp

+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include <gtest/gtest.h>
+
+#include "context/QueryContext.h"
+#include "planner/Query.h"
+#include "exec/query/LimitExecutor.h"
+#include "exec/query/ProjectExecutor.h"
+#include "exec/query/test/QueryTestBase.h"
+
+namespace nebula {
+namespace graph {
+class LimitTest : public QueryTestBase {
+};
+
+#define LIMIT_RESUTL_CHECK(outputName, offset, count, expected)                       \
+    do {                                                                              \
+        auto* plan = qctx_->plan();                                                   \
+        auto* limitNode = Limit::make(plan, nullptr, offset, count);                  \
+        limitNode->setInputVar("input_neighbor");                                     \
+        limitNode->setOutputVar(outputName);                                          \
+        auto limitExec = std::make_unique<LimitExecutor>(limitNode, qctx_.get());     \
+        EXPECT_TRUE(limitExec->execute().get().ok());                                 \
+        auto& limitResult = qctx_->ectx()->getResult(limitNode->varName());           \
+        EXPECT_EQ(limitResult.state().state(), StateDesc::State::kSuccess);           \
+        auto yieldSentence = getYieldSentence(                                        \
+                "YIELD study._dst AS name, study.start_year AS start");               \
+        auto* project = Project::make(plan, nullptr, yieldSentence->yieldColumns());  \
+        project->setInputVar(limitNode->varName());                                   \
+        project->setColNames(std::vector<std::string>{"name", "start"});              \
+        auto proExe = std::make_unique<ProjectExecutor>(project, qctx_.get());        \
+        EXPECT_TRUE(proExe->execute().get().ok());                                    \
+        auto& proResult = qctx_->ectx()->getResult(project->varName());               \
+        EXPECT_EQ(proResult.value().getDataSet(), expected);                          \
+        EXPECT_EQ(proResult.state().state(), StateDesc::State::kSuccess);             \
+    } while (false)
+
+
+TEST_F(LimitTest, getNeighborInRange1) {
+    DataSet expected({"name", "start"});
+    expected.emplace_back(Row({Value("School2"), Value(2014)}));
+    expected.emplace_back(Row({Value("School1"), Value(2008)}));
+    LIMIT_RESUTL_CHECK("limit_in_neighbor1", 1, 2, expected);
+}
+
+TEST_F(LimitTest, getNeighborInRange2) {
+    DataSet expected({"name", "start"});
+    expected.emplace_back(Row({Value("School1"), Value(2010)}));
+    expected.emplace_back(Row({Value("School2"), Value(2014)}));
+    expected.emplace_back(Row({Value("School1"), Value(2008)}));
+    expected.emplace_back(Row({Value("School2"), Value(2012)}));
+    LIMIT_RESUTL_CHECK("limit_in_neighbor2", 0, 4, expected);
+}
+
+TEST_F(LimitTest, getNeighborOutRange1) {
+    DataSet expected({"name", "start"});
+    expected.emplace_back(Row({Value("School1"), Value(2010)}));
+    expected.emplace_back(Row({Value("School2"), Value(2014)}));
+    expected.emplace_back(Row({Value("School1"), Value(2008)}));
+    expected.emplace_back(Row({Value("School2"), Value(2012)}));
+    LIMIT_RESUTL_CHECK("limit_out_neighbor1", 0, 5, expected);
+}
+
+TEST_F(LimitTest, getNeighborOutRange2) {
+    DataSet expected({"name", "start"});
+    LIMIT_RESUTL_CHECK("limit_out_neighbor1", 4, 2, expected);
+}
+}  // namespace graph
+}  // namespace nebula

src/exec/query/test/QueryTestBase.h

@@ -18,7 +18,7 @@ class QueryTestBase : public testing::Test {
 protected:
     void SetUp() override {
         qctx_ = std::make_unique<QueryContext>();
-
+        // GetNeighbors
         {
             DataSet dataset({"_vid", "_stats", "_tag:person:name:age",
                              "_edge:+study:_dst:start_year:end_year", "_expr"});
@@ -31,24 +31,16 @@ protected:
                 row.values.emplace_back(Value::kEmpty);
 
                 // vertices props
-                List vertices;
-                vertices.values.emplace_back("Ann");
-                vertices.values.emplace_back(18);
-                row.values.emplace_back(vertices);
+                List vertices({"Ann", 18});
+                row.values.emplace_back(std::move(vertices));
 
                 // edge props
                 List edges;
-                List edge1;
-                edge1.values.emplace_back("School1");
-                edge1.values.emplace_back(2010);
-                edge1.values.emplace_back(2014);
+                List edge1({"School1", 2010, 2014});
+                List edge2({"School2", 2014, 2017});
                 edges.values.emplace_back(std::move(edge1));
-                List edge2;
-                edge2.values.emplace_back("School2");
-                edge2.values.emplace_back(2014);
-                edge2.values.emplace_back(2017);
                 edges.values.emplace_back(std::move(edge2));
-                row.values.emplace_back(edges);
+                row.values.emplace_back(std::move(edges));
 
                 // _expr
                 row.values.emplace_back(Value::kEmpty);
@@ -58,29 +50,22 @@ protected:
                 Row row;
                 // vid
                 row.values.emplace_back("Tom");
+
                 // _stats
                 row.values.emplace_back(Value::kEmpty);
+
                 // vertices props
-                List vertices;
-                vertices.values.emplace_back("Tom");
-                vertices.values.emplace_back(18);
-                row.values.emplace_back(vertices);
-                // edges props
-                List edge1;
-                edge1.values.emplace_back("School1");
-                edge1.values.emplace_back(2008);
-                edge1.values.emplace_back(2012);
+                List vertices({"Tom", 18});
+                row.values.emplace_back(std::move(vertices));
 
-                List edge2;
-                edge2.values.emplace_back("School2");
-                edge2.values.emplace_back(2012);
-                edge2.values.emplace_back(2015);
+                // edges props
+                List edge1({"School1", 2008, 2012});
+                List edge2({"School2", 2012, 2015});
 
                 List edges;
                 edges.values.emplace_back(std::move(edge1));
                 edges.values.emplace_back(std::move(edge2));
-
-                row.values.emplace_back(edges);
+                row.values.emplace_back(std::move(edges));
 
                 // _expr
                 row.values.emplace_back(Value::kEmpty);
@@ -91,23 +76,42 @@ protected:
             qctx_->ectx()->setResult("input_neighbor",
                                      ExecResult::buildGetNeighbors(Value(datasetList)));
         }
-
+        // sequential
         {
             DataSet dataset({"vid", "v_name", "v_age", "v_dst", "e_start_year", "e_end_year"});
-            {
-                Row row;
-                row.values = {"Ann", "Ann", 18, "School1", 2010, 2014};
-                dataset.emplace_back(row);
-                row.values = {"Tom", "Tom", 20, "School1", 2008, 2012};
-                dataset.emplace_back(row);
-                row.values = {"Joy", "Joy", 20, "School2", 2008, 2012};
-                dataset.emplace_back(row);
-                row.values = {"Ann", "Ann", 18, "School1", 2010, 2014};
-                dataset.emplace_back(row);
-            }
+            dataset.emplace_back(Row({"Ann", "Ann", 18, "School1", 2010, 2014}));
+            dataset.emplace_back(Row({"Joy", "Joy", Value::kNullValue, "School2", 2009, 2012}));
+            dataset.emplace_back(Row({"Tom", "Tom", 20, "School2", 2008, 2012}));
+            dataset.emplace_back(Row({"Kate", "Kate", 19, "School2", 2009, 2013}));
+            dataset.emplace_back(Row({"Ann", "Ann", 18, "School1", 2010, 2014}));
+            dataset.emplace_back(Row({"Lily", "Lily", 20, "School2", 2009, 2012}));
             qctx_->ectx()->setResult("input_sequential",
                                      ExecResult::buildSequential(Value(dataset)));
         }
+        // sequential init by two sequentialIters
+        {
+            DataSet lds({"vid", "v_name", "v_age", "v_dst", "e_start_year", "e_end_year"});
+            lds.emplace_back(Row({"Ann", "Ann", 18, "School1", 2010, 2014}));
+            lds.emplace_back(Row({"Joy", "Joy", Value::kNullValue, "School2", 2009, 2012}));
+            lds.emplace_back(Row({"Tom", "Tom", 20, "School2", 2008, 2012}));
+            lds.emplace_back(Row({"Kate", "Kate", 19, "School2", 2009, 2013}));
+            qctx_->ectx()->setResult("left_sequential",
+                                     ExecResult::buildSequential(Value(lds)));
+
+            DataSet rds({"vid", "v_name", "v_age", "v_dst", "e_start_year", "e_end_year"});
+            rds.emplace_back(Row({"Ann", "Ann", 18, "School1", 2010, 2014}));
+            rds.emplace_back(Row({"Lily", "Lily", 20, "School2", 2009, 2012}));
+            qctx_->ectx()->setResult("right_sequential",
+                                     ExecResult::buildSequential(Value(rds)));
+
+            auto lIter = qctx_->ectx()->getResult("left_sequential").iter();
+            auto rIter = qctx_->ectx()->getResult("right_sequential").iter();
+            auto result = ExecResult::buildDefault(lIter->valuePtr());
+            auto iter = std::make_unique<SequentialIter>(std::move(lIter), std::move(rIter));
+            result.setIter(std::move(iter));
+            qctx_->ectx()->setResult("union_sequential", std::move(result));
+        }
+        // empty
         {
             DataSet dataset({"_vid", "_stats", "_tag:person:name:age",
                              "_edge:+study:_dst:start_year:end_year", "_expr"});
@@ -132,14 +136,6 @@ protected:
         return yieldSentence->yieldColumns();
     }
 
-    std::vector<Expression*> getExprs(const YieldColumns *yieldColumns) {
-        std::vector<Expression*> exprs;
-        for (auto &col : yieldColumns->columns()) {
-            exprs.emplace_back(col->expr());
-        }
-        return exprs;
-    }
-
     Expression* getYieldFilter(const std::string &query) {
         auto yieldSentence = getYieldSentence(query);
         CHECK(yieldSentence);

src/exec/query/test/SortTest.cpp

+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include <gtest/gtest.h>
+
+#include "context/QueryContext.h"
+#include "planner/Query.h"
+#include "exec/query/SortExecutor.h"
+#include "exec/query/ProjectExecutor.h"
+#include "exec/query/test/QueryTestBase.h"
+
+namespace nebula {
+namespace graph {
+class SortTest : public QueryTestBase {
+};
+
+#define SORT_RESUTL_CHECK(input_name, outputName, multi, factors, expected)           \
+    do {                                                                              \
+        auto* plan = qctx_->plan();                                                   \
+        auto* sortNode = Sort::make(plan, nullptr, factors);                          \
+        sortNode->setInputVar(input_name);                                            \
+        sortNode->setOutputVar(outputName);                                           \
+        auto sortExec = std::make_unique<SortExecutor>(sortNode, qctx_.get());        \
+        EXPECT_TRUE(sortExec->execute().get().ok());                                  \
+        auto& sortResult = qctx_->ectx()->getResult(sortNode->varName());             \
+        EXPECT_EQ(sortResult.state().state(), StateDesc::State::kSuccess);            \
+        std::string sentence;                                                         \
+        std::vector<std::string> colNames;                                            \
+        if (multi) {                                                                  \
+            sentence = "YIELD $-.v_age AS age, $-.e_start_year AS start_year";        \
+            colNames.emplace_back("age");                                             \
+            colNames.emplace_back("start_year");                                      \
+        } else {                                                                      \
+            sentence = "YIELD $-.v_age AS age";                                       \
+            colNames.emplace_back("age");                                             \
+        }                                                                             \
+        auto yieldSentence = getYieldSentence(sentence);                              \
+        auto* project = Project::make(plan, nullptr, yieldSentence->yieldColumns());  \
+        project->setInputVar(sortNode->varName());                                    \
+        project->setColNames(std::move(colNames));                                    \
+        auto proExe = std::make_unique<ProjectExecutor>(project, qctx_.get());        \
+        EXPECT_TRUE(proExe->execute().get().ok());                                    \
+        auto& proResult = qctx_->ectx()->getResult(project->varName());               \
+        EXPECT_EQ(proResult.value().getDataSet(), expected);                          \
+        EXPECT_EQ(proResult.state().state(), StateDesc::State::kSuccess);             \
+    } while (false)
+
+
+TEST_F(SortTest, sortOneColAsc) {
+    DataSet expected({"age"});
+    expected.emplace_back(Row({18}));
+    expected.emplace_back(Row({18}));
+    expected.emplace_back(Row({19}));
+    expected.emplace_back(Row({20}));
+    expected.emplace_back(Row({20}));
+    expected.emplace_back(Row({Value::kNullValue}));
+    std::vector<std::pair<std::string, OrderFactor::OrderType>> factors;
+    factors.emplace_back(std::make_pair("v_age", OrderFactor::OrderType::ASCEND));
+    SORT_RESUTL_CHECK("input_sequential", "sort_one_col_asc", false, factors, expected);
+}
+
+TEST_F(SortTest, sortOneColDes) {
+    DataSet expected({"age"});
+    expected.emplace_back(Row({Value::kNullValue}));
+    expected.emplace_back(Row({20}));
+    expected.emplace_back(Row({20}));
+    expected.emplace_back(Row({19}));
+    expected.emplace_back(Row({18}));
+    expected.emplace_back(Row({18}));
+    std::vector<std::pair<std::string, OrderFactor::OrderType>> factors;
+    factors.emplace_back(std::make_pair("v_age", OrderFactor::OrderType::DESCEND));
+    SORT_RESUTL_CHECK("input_sequential", "sort_one_col_des", false, factors, expected);
+}
+
+TEST_F(SortTest, sortTwoColsAscAsc) {
+    DataSet expected({"age", "start_year"});
+    expected.emplace_back(Row({18, 2010}));
+    expected.emplace_back(Row({18, 2010}));
+    expected.emplace_back(Row({19, 2009}));
+    expected.emplace_back(Row({20, 2008}));
+    expected.emplace_back(Row({20, 2009}));
+    expected.emplace_back(Row({Value::kNullValue, 2009}));
+    std::vector<std::pair<std::string, OrderFactor::OrderType>> factors;
+    factors.emplace_back(std::make_pair("v_age", OrderFactor::OrderType::ASCEND));
+    factors.emplace_back(std::make_pair("e_start_year", OrderFactor::OrderType::ASCEND));
+    SORT_RESUTL_CHECK("input_sequential", "sort_two_cols_asc_asc", true, factors, expected);
+}
+
+TEST_F(SortTest, sortTwoColsAscDes) {
+    DataSet expected({"age", "start_year"});
+    expected.emplace_back(Row({18, 2010}));
+    expected.emplace_back(Row({18, 2010}));
+    expected.emplace_back(Row({19, 2009}));
+    expected.emplace_back(Row({20, 2009}));
+    expected.emplace_back(Row({20, 2008}));
+    expected.emplace_back(Row({Value::kNullValue, 2009}));
+    std::vector<std::pair<std::string, OrderFactor::OrderType>> factors;
+    factors.emplace_back(std::make_pair("v_age", OrderFactor::OrderType::ASCEND));
+    factors.emplace_back(std::make_pair("e_start_year", OrderFactor::OrderType::DESCEND));
+    SORT_RESUTL_CHECK("input_sequential", "sort_two_cols_asc_des", true, factors, expected);
+}
+
+TEST_F(SortTest, sortTwoColDesDes) {
+    DataSet expected({"age", "start_year"});
+    expected.emplace_back(Row({Value::kNullValue, 2009}));
+    expected.emplace_back(Row({20, 2009}));
+    expected.emplace_back(Row({20, 2008}));
+    expected.emplace_back(Row({19, 2009}));
+    expected.emplace_back(Row({18, 2010}));
+    expected.emplace_back(Row({18, 2010}));
+    std::vector<std::pair<std::string, OrderFactor::OrderType>> factors;
+    factors.emplace_back(std::make_pair("v_age", OrderFactor::OrderType::DESCEND));
+    factors.emplace_back(std::make_pair("e_start_year", OrderFactor::OrderType::DESCEND));
+    SORT_RESUTL_CHECK("input_sequential", "sort_two_cols_des_des", true, factors, expected);
+}
+
+TEST_F(SortTest, sortTwoColDesDes_union) {
+    DataSet expected({"age", "start_year"});
+    expected.emplace_back(Row({Value::kNullValue, 2009}));
+    expected.emplace_back(Row({20, 2009}));
+    expected.emplace_back(Row({20, 2008}));
+    expected.emplace_back(Row({19, 2009}));
+    expected.emplace_back(Row({18, 2010}));
+    expected.emplace_back(Row({18, 2010}));
+    std::vector<std::pair<std::string, OrderFactor::OrderType>> factors;
+    factors.emplace_back(std::make_pair("v_age", OrderFactor::OrderType::DESCEND));
+    factors.emplace_back(std::make_pair("e_start_year", OrderFactor::OrderType::DESCEND));
+    SORT_RESUTL_CHECK("union_sequential", "union_sort_two_cols_des_des", true, factors, expected);
+}
+}  // namespace graph
+}  // namespace nebula

src/planner/Query.h

@@ -630,24 +630,26 @@ class Sort final : public SingleInputNode {
 public:
     static Sort* make(ExecutionPlan* plan,
                       PlanNode* input,
-                      OrderFactors* factors) {
-        return new Sort(plan, input, factors);
+                      std::vector<std::pair<std::string, OrderFactor::OrderType>> factors) {
+        return new Sort(plan, input, std::move(factors));
     }
 
-    const OrderFactors* factors() {
+    const std::vector<std::pair<std::string, OrderFactor::OrderType>>& factors() const {
         return factors_;
     }
 
     std::string explain() const override;
 
 private:
-    Sort(ExecutionPlan* plan, PlanNode* input, OrderFactors* factors)
-      : SingleInputNode(plan, Kind::kSort, input) {
-        factors_ = factors;
+    Sort(ExecutionPlan* plan,
+         PlanNode* input,
+         std::vector<std::pair<std::string, OrderFactor::OrderType>> factors)
+        : SingleInputNode(plan, Kind::kSort, input) {
+        factors_ = std::move(factors);
     }
 
 private:
-    OrderFactors*   factors_{nullptr};
+    std::vector<std::pair<std::string, OrderFactor::OrderType>>   factors_;
 };
 
 /**

src/validator/CMakeLists.txt

@@ -17,6 +17,8 @@ nebula_add_library(
     AdminValidator.cpp
     MaintainValidator.cpp
     MutateValidator.cpp
+    LimitValidator.cpp
+    OrderByValidator.cpp
 )
 
 add_subdirectory(test)

src/validator/GoValidator.cpp

@@ -44,13 +44,13 @@ Status GoValidator::validateImpl() {
 
         if (!inputProps_.empty() && fromType_ != kPipe) {
             status = Status::Error("$- must be referred in FROM "
-                                    "before used in WHERE or YIELD");
+                                   "before used in WHERE or YIELD");
             break;
         }
 
         if (!varProps_.empty() && fromType_ != kVariable) {
             status = Status::Error("A variable must be referred in FROM "
-                                    "before used in WHERE or YIELD");
+                                   "before used in WHERE or YIELD");
             break;
         }
 
@@ -121,7 +121,7 @@ Status GoValidator::validateFrom(const FromClause* from) {
             if (type.value() != Value::Type::STRING) {
                 std::stringstream ss;
                 ss << "`" << src->toString() << "', the srcs should be type of string, "
-                    << "but was`" << type.value() << "'";
+                   << "but was`" << type.value() << "'";
                 return Status::Error(ss.str());
             }
             src_ = src;
@@ -184,7 +184,7 @@ Status GoValidator::validateWhere(const WhereClause* where) {
     if (type != Value::Type::BOOL && type != Value::Type::NULLVALUE) {
         std::stringstream ss;
         ss << "`" << filter_->toString() << "', Filter only accpet bool/null value, "
-            << "but was `" << type << "'";
+           << "but was `" << type << "'";
         return Status::Error(ss.str());
     }
 
@@ -465,3 +465,4 @@ GetNeighbors::EdgeProps GoValidator::buildEdgeProps() {
 }
 }  // namespace graph
 }  // namespace nebula
+

src/validator/LimitValidator.cpp

+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include "validator/LimitValidator.h"
+#include "parser/TraverseSentences.h"
+#include "planner/Query.h"
+
+namespace nebula {
+namespace graph {
+Status LimitValidator::validateImpl() {
+    auto limitSentence = static_cast<LimitSentence*>(sentence_);
+    offset_ = limitSentence->offset();
+    count_ = limitSentence->count();
+    if (offset_ < 0) {
+        return Status::SyntaxError("skip `%ld' is illegal", offset_);
+    }
+    if (count_ < 0) {
+        return Status::SyntaxError("count `%ld' is illegal", count_);
+    }
+
+    outputs_ = inputs();
+    return Status::OK();
+}
+
+Status LimitValidator::toPlan() {
+    auto* plan = qctx_->plan();
+    auto *limitNode = Limit::make(plan, plan->root(), offset_, count_);
+    std::vector<std::string> colNames;
+    for (auto &col : outputs_) {
+        colNames.emplace_back(col.first);
+    }
+    limitNode->setColNames(std::move(colNames));
+    root_ = limitNode;
+    tail_ = root_;
+    return Status::OK();
+}
+}  // namespace graph
+}  // namespace nebula

src/validator/LimitValidator.h

+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#ifndef VALIDATOR_LIMITVALIDATOR_H_
+#define VALIDATOR_LIMITVALIDATOR_H_
+
+#include "common/base/Base.h"
+#include "validator/Validator.h"
+
+namespace nebula {
+namespace graph {
+class LimitValidator final : public Validator {
+public:
+    LimitValidator(Sentence* sentence, QueryContext* context)
+        : Validator(sentence, context) {}
+
+private:
+    Status validateImpl() override;
+
+    Status toPlan() override;
+
+private:
+    int64_t                    offset_{-1};
+    int64_t                    count_{-1};
+};
+}  // namespace graph
+}  // namespace nebula
+#endif  // VALIDATOR_LIMITVALIDATOR_H_

src/validator/OrderByValidator.cpp

+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include "validator/OrderByValidator.h"
+#include "parser/TraverseSentences.h"
+#include "planner/Query.h"
+
+namespace nebula {
+namespace graph {
+Status OrderByValidator::validateImpl() {
+    auto sentence = static_cast<OrderBySentence*>(sentence_);
+    auto inputColNames = inputs();
+    auto factors = sentence->factors();
+    for (auto &factor : factors) {
+        if (factor->expr()->kind() != Expression::Kind::kInputProperty) {
+            return Status::Error("Wrong expression");
+        }
+        auto expr = static_cast<InputPropertyExpression*>(factor->expr());
+        auto name = *expr->prop();
+        // Check factor in input node's colNames
+        auto find = std::find_if(inputColNames.begin(), inputColNames.end(),
+                                 [&name] (const auto& colDef) {
+                                     return colDef.first == name;
+                                 });
+        if (find == inputColNames.end()) {
+            LOG(ERROR) << "Order BY on factor `" << name <<"` is not exist";
+            return Status::Error("Order BY on factor `%s` is not exist", name.c_str());
+        }
+        colOrderTypes_.emplace_back(std::make_pair(name, factor->orderType()));
+    }
+
+    outputs_ = inputs();
+    return Status::OK();
+}
+
+Status OrderByValidator::toPlan() {
+    auto* plan = qctx_->plan();
+    auto *sortNode = Sort::make(plan, plan->root(), std::move(colOrderTypes_));
+    std::vector<std::string> colNames;
+    for (auto &col : outputs_) {
+        colNames.emplace_back(col.first);
+    }
+    sortNode->setColNames(std::move(colNames));
+    root_ = sortNode;
+    tail_ = root_;
+    return Status::OK();
+}
+}  // namespace graph
+}  // namespace nebula

src/validator/OrderByValidator.h

+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#ifndef VALIDATOR_ORDERBYVALIDATOR_H_
+#define VALIDATOR_ORDERBYVALIDATOR_H_
+
+#include "common/base/Base.h"
+#include "validator/Validator.h"
+
+namespace nebula {
+namespace graph {
+class OrderByValidator final : public Validator {
+public:
+    OrderByValidator(Sentence* sentence, QueryContext* context)
+        : Validator(sentence, context) {}
+
+private:
+    Status validateImpl() override;
+
+    Status toPlan() override;
+
+private:
+    std::vector<std::pair<std::string, OrderFactor::OrderType>>     colOrderTypes_;
+};
+}  // namespace graph
+}  // namespace nebula
+#endif   // VALIDATOR_ORDERBYVALIDATOR_H_