chore: refactor runner name & improve tests

cases/query/fail_query.yaml

@@ -49,3 +49,24 @@ cases:
       SELECT 100 + 1s;
     expect:
       success: false
+  - id: 3
+    desc: unsupport join
+    inputs:
+      - name: t1
+        columns: ["c1 string","c2 int","c4 timestamp"]
+        indexs: ["index1:c1:c4"]
+        rows:
+          - ["aa",20,1000]
+          - ["bb",30,1000]
+      - name: t2
+        columns: ["c2 int","c4 timestamp"]
+        indexs: ["index1:c2:c4"]
+        rows:
+          - [20,3000]
+          - [20,2000]
+    sql: |
+      select t1.c1 as id, t2.* from t1 right join t2
+      on t1.c2 = t2.c2
+    expect:
+      success: false
+      msg: unsupport join type RightJoin

cases/query/left_join.yml

@@ -315,7 +315,7 @@ cases:
           t3.c1 as c1r,
           t3.c2 as c2r,
           t4.c3 as c3x
-        from t2 left join t3
+        from t2 left outer join t3
         on t2.c1 = t3.c1 and t2.c2 > t3.c2
         left join t4
         on t2.c1 = t4.c1 and t3.c3 < t4.c3
@@ -354,3 +354,29 @@ cases:
         bb, bb, NULL, NULL, NULL
         cc, cc, cc, 27, NULL
         dd, NULL, NULL, NULL, NULL
+  - id: 5
+    desc: simple left join
+    mode: request-unsupport
+    inputs:
+      - name: t1
+        columns: ["c1 string","c2 int","c4 timestamp"]
+        indexs: ["index1:c1:c4"]
+        rows:
+          - ["aa",20,1000]
+          - ["bb",30,1000]
+      - name: t2
+        columns: ["c2 int","c4 timestamp"]
+        indexs: ["index1:c2:c4"]
+        rows:
+          - [20,3000]
+          - [20,2000]
+    sql: |
+      select t1.c1 as id, t2.* from t1 left join t2
+      on t1.c2 = t2.c2
+    expect:
+      order: c1
+      columns: ["id string", "c2 int","c4 timestamp"]
+      data: |
+        aa, 20, 3000
+        aa, 20, 2000
+        bb, NULL, NULL

hybridse/src/vm/runner.cc

@@ -696,7 +696,7 @@ void WindowAggRunner::RunWindowAggOnKey(
     }
 }
 
-std::shared_ptr<DataHandler> RequestLastJoinRunner::Run(
+std::shared_ptr<DataHandler> RequestJoinRunner::Run(
     RunnerContext& ctx,
     const std::vector<std::shared_ptr<DataHandler>>& inputs) {  // NOLINT
     auto fail_ptr = std::shared_ptr<DataHandler>();
@@ -721,8 +721,10 @@ std::shared_ptr<DataHandler> RequestLastJoinRunner::Run(
                 return std::shared_ptr<RowHandler>(
                     new MemRowHandler(join_gen_->RowLastJoin(left_row, right, parameter)));
             }
+        } else if (join_gen_->join_type_ == node::kJoinTypeLeft) {
+            return join_gen_->LazyJoin(left, right, ctx.GetParameterRow());
         } else {
-            // fix later
+            LOG(WARNING) << "unsupport join type " << node::JoinTypeName(join_gen_->join_type_);
             return {};
         }
     } else if (kPartitionHandler == left->GetHandlerType() && right->GetHandlerType() == kPartitionHandler) {
@@ -735,8 +737,8 @@ std::shared_ptr<DataHandler> RequestLastJoinRunner::Run(
     return std::shared_ptr<DataHandler>();
 }
 
-std::shared_ptr<DataHandler> LastJoinRunner::Run(RunnerContext& ctx,
-                                                 const std::vector<std::shared_ptr<DataHandler>>& inputs) {
+std::shared_ptr<DataHandler> JoinRunner::Run(RunnerContext& ctx,
+                                             const std::vector<std::shared_ptr<DataHandler>>& inputs) {
     auto fail_ptr = std::shared_ptr<DataHandler>();
     if (inputs.size() < 2) {
         LOG(WARNING) << "inputs size < 2";

hybridse/src/vm/runner.h

@@ -68,10 +68,10 @@ enum RunnerType {
     kRunnerRequestAggUnion,
     kRunnerPostRequestUnion,
     kRunnerIndexSeek,
-    kRunnerLastJoin,
+    kRunnerJoin,
     kRunnerConcat,
     kRunnerRequestRunProxy,
-    kRunnerRequestLastJoin,
+    kRunnerRequestJoin,
     kRunnerBatchRequestRunProxy,
     kRunnerLimit,
     kRunnerUnknow,
@@ -114,12 +114,12 @@ inline const std::string RunnerTypeName(const RunnerType& type) {
             return "POST_REQUEST_UNION";
         case kRunnerIndexSeek:
             return "INDEX_SEEK";
-        case kRunnerLastJoin:
-            return "LASTJOIN";
+        case kRunnerJoin:
+            return "JOIN";
         case kRunnerConcat:
             return "CONCAT";
-        case kRunnerRequestLastJoin:
-            return "REQUEST_LASTJOIN";
+        case kRunnerRequestJoin:
+            return "REQUEST_JOIN";
         case kRunnerLimit:
             return "LIMIT";
         case kRunnerRequestRunProxy:
@@ -699,30 +699,30 @@ class PostRequestUnionRunner : public Runner {
     OrderGenerator request_ts_gen_;
 };
 
-class LastJoinRunner : public Runner {
+class JoinRunner : public Runner {
  public:
-    LastJoinRunner(const int32_t id, const SchemasContext* schema, const std::optional<int32_t> limit_cnt,
-                   const Join& join, size_t left_slices, size_t right_slices)
-        : Runner(id, kRunnerLastJoin, schema, limit_cnt) {
+    JoinRunner(const int32_t id, const SchemasContext* schema, const std::optional<int32_t> limit_cnt, const Join& join,
+               size_t left_slices, size_t right_slices)
+        : Runner(id, kRunnerJoin, schema, limit_cnt) {
         join_gen_ = JoinGenerator::Create(join, left_slices, right_slices);
     }
-    ~LastJoinRunner() {}
+    ~JoinRunner() {}
     std::shared_ptr<DataHandler> Run(
         RunnerContext& ctx,  // NOLINT
         const std::vector<std::shared_ptr<DataHandler>>& inputs)
         override;  // NOLINT
 
     std::shared_ptr<JoinGenerator> join_gen_;
 };
-class RequestLastJoinRunner : public Runner {
+class RequestJoinRunner : public Runner {
  public:
-    RequestLastJoinRunner(const int32_t id, const SchemasContext* schema, const std::optional<int32_t> limit_cnt,
-                          const Join& join, const size_t left_slices, const size_t right_slices,
-                          const bool output_right_only)
-        : Runner(id, kRunnerRequestLastJoin, schema, limit_cnt), output_right_only_(output_right_only) {
+    RequestJoinRunner(const int32_t id, const SchemasContext* schema, const std::optional<int32_t> limit_cnt,
+                      const Join& join, const size_t left_slices, const size_t right_slices,
+                      const bool output_right_only)
+        : Runner(id, kRunnerRequestJoin, schema, limit_cnt), output_right_only_(output_right_only) {
         join_gen_ = JoinGenerator::Create(join, left_slices, right_slices);
     }
-    ~RequestLastJoinRunner() {}
+    ~RequestJoinRunner() {}
 
     std::shared_ptr<DataHandler> Run(
         RunnerContext& ctx,                                        // NOLINT

hybridse/src/vm/runner_builder.cc

@@ -332,7 +332,7 @@ ClusterTask RunnerBuilder::Build(PhysicalOpNode* node, Status& status) {
             switch (op->join().join_type()) {
                 case node::kJoinTypeLast:
                 case node::kJoinTypeLeft: {
-                    RequestLastJoinRunner* runner = CreateRunner<RequestLastJoinRunner>(
+                    RequestJoinRunner* runner = CreateRunner<RequestJoinRunner>(
                         id_++, node->schemas_ctx(), op->GetLimitCnt(), op->join_,
                         left->output_schemas()->GetSchemaSourceSize(), right->output_schemas()->GetSchemaSourceSize(),
                         op->output_right_only());
@@ -414,15 +414,15 @@ ClusterTask RunnerBuilder::Build(PhysicalOpNode* node, Status& status) {
                     // TableLastJoin convert to Batch Request RequestLastJoin
                     if (support_cluster_optimized_) {
                         // looks strange, join op won't run for batch-cluster mode
-                        RequestLastJoinRunner* runner = CreateRunner<RequestLastJoinRunner>(
+                        RequestJoinRunner* runner = CreateRunner<RequestJoinRunner>(
                             id_++, node->schemas_ctx(), op->GetLimitCnt(), op->join_,
                             left->output_schemas()->GetSchemaSourceSize(),
                             right->output_schemas()->GetSchemaSourceSize(), op->output_right_only_);
                         return RegisterTask(
                             node, BinaryInherit(left_task, right_task, runner, op->join().index_key(), kLeftBias));
                     } else {
-                        LastJoinRunner* runner =
-                            CreateRunner<LastJoinRunner>(id_++, node->schemas_ctx(), op->GetLimitCnt(), op->join_,
+                        JoinRunner* runner =
+                            CreateRunner<JoinRunner>(id_++, node->schemas_ctx(), op->GetLimitCnt(), op->join_,
                                                          left->output_schemas()->GetSchemaSourceSize(),
                                                          right->output_schemas()->GetSchemaSourceSize());
                         return RegisterTask(node, BinaryInherit(left_task, right_task, runner, Key(), kLeftBias));