feat: request mode in raw SQL (#3874)

##  request mode in RAW SQL
```sql
SELECT ....
CONFIG ( execute_mode = 'request', values = (expr1, expr2 , ...) )

SELECT ....
CONFIG ( execute_mode = 'request', values = [(expr1, expr2 , ...), (....)] )
```

## call procedure 

```sql
call dp (expr1, expr2, ...)
```

## TODOs
- docs
- support `request` value in system variable `execute_mode`
- take care const query

cases/plan/cmd.yaml

@@ -754,3 +754,11 @@ cases:
         +-node[CMD]
           +-cmd_type: show create table
           +-args: [db1, t1]
+
+  - id: callstmt
+    sql: call sp(1, cast("ab" as string))
+    expect:
+      plan_tree_str: |
+        +-[kPlanTypeCallStmt]
+          +-procedure_name: [sp]
+          +-arguments: 1, string(ab)

cases/query/fail_query.yaml

@@ -70,3 +70,53 @@ cases:
     expect:
       success: false
       msg: unsupport join type RightJoin
+  - id: 4
+    mode: batch-unsupport
+    inputs:
+      - name: t1
+        columns: ["c1 string","c2 int","c4 timestamp"]
+        indexs: ["index1:c1:c4"]
+        rows:
+          - ["aa",20,1000]
+          - ["bb",30,1000]
+    sql: |
+      SELECT * from t1
+      CONFIG (execute_mode = 'request', values = [])
+    expect:
+      success: false
+      msg: |
+        FAILED_PRECONDITION: element number of the request values must not empty
+  - id: 5
+    mode: batch-unsupport
+    inputs:
+      - name: t1
+        columns: ["c1 string","c2 int","c4 timestamp"]
+        indexs: ["index1:c1:c4"]
+        rows:
+          - ["aa",20,1000]
+          - ["bb",30,1000]
+    sql: |
+      SELECT * from t1
+      CONFIG (execute_mode = 'request', values = ("c1"))
+    expect:
+      success: false
+      msg: |
+        INTERNAL: pass in expr number do not match, expect 3 but got 1: (c1)
+
+  # TODO: open this case after #3847
+  # - id: 6
+  #   mode: batch-unsupport
+  #   inputs:
+  #     - name: t1
+  #       columns: ["c1 string","c2 int","c4 date"]
+  #       indexs: ["index1:c1"]
+  #       rows:
+  #         - ["aa",20,1000]
+  #         - ["bb",30,1000]
+  #   sql: |
+  #     SELECT * from t1
+  #     CONFIG (execute_mode = 'request', values = ("c1", 19, 12) )
+  #   expect:
+  #     success: false
+  #     msg: |
+  #       xx

cases/query/simple_query.yaml

@@ -985,3 +985,29 @@ cases:
       data: |
         1, 10, 1000
         2, 30, 2000
+
+  - id: 108
+    mode: request-unsupport
+    inputs:
+      # t1 as request table, only one request row concerned
+      # from SQL config options
+      - name: t1
+        columns: ["id int32", "b int32", "ts int64"]
+        indexs: ["idx2:b:ts"]
+        data: |
+          1, 10, 1000
+      - name: t2
+        columns: ["id int32", "b int32", "ts int64"]
+        indexs: ["idx2:b:ts"]
+        data: |
+          3, 10, 1000
+          4, 30, 2000
+    sql: |
+      select t1.*, t2.id as ix from t1
+      last join t2 on t1.b = t2.b
+      config (execute_mode = 'request', values = (5, 10, timestamp(4000)))
+    expect:
+      columns: ["id int32", "b int32", "ts int64", "ix int32"]
+      order: id
+      data: |
+        5, 10, 4000, 3

hybridse/examples/toydb/src/cmd/toydb_run_engine.cc

@@ -14,9 +14,10 @@
  * limitations under the License.
  */
 
-#include "absl/strings/match.h"
 #include "testing/toydb_engine_test_base.h"
 
+#include "gflags/gflags.h"
+
 DEFINE_string(yaml_path, "", "Yaml filepath to load cases from");
 DEFINE_string(runner_mode, "batch",
               "Specify runner mode, can be batch or request");
@@ -82,14 +83,15 @@ int RunSingle(const std::string& yaml_path) {
         if (!FLAGS_case_id.empty() && FLAGS_case_id != sql_case.id()) {
             continue;
         }
-        EngineMode mode;
+        EngineMode default_mode;
         if (absl::EqualsIgnoreCase(FLAGS_runner_mode, "batch")) {
-            mode = kBatchMode;
+            default_mode = kBatchMode;
         } else if (absl::EqualsIgnoreCase(FLAGS_runner_mode, "request")) {
-            mode = kRequestMode;
+            default_mode = kRequestMode;
         } else {
-            mode = kBatchRequestMode;
+            default_mode = kBatchRequestMode;
         }
+        auto mode = Engine::TryDetermineEngineMode(sql_case.sql_str_, default_mode);
         int ret = DoRunEngine(sql_case, options, mode);
         if (ret != ENGINE_TEST_RET_SUCCESS) {
             return ret;

hybridse/examples/toydb/src/testing/toydb_engine_test_base.cc

@@ -194,6 +194,7 @@ void BatchRequestEngineCheck(const SqlCase& sql_case,
 
 void EngineCheck(const SqlCase& sql_case, const EngineOptions& options,
                  EngineMode engine_mode) {
+    engine_mode = hybridse::vm::Engine::TryDetermineEngineMode(sql_case.sql_str(), engine_mode);
     if (engine_mode == kBatchMode) {
         ToydbBatchEngineTestRunner engine_test(sql_case, options);
         engine_test.RunCheck();

hybridse/include/node/node_enum.h

@@ -98,6 +98,7 @@ enum SqlNodeType {
     kColumnSchema,
     kCreateUserStmt,
     kAlterUserStmt,
+    kCallStmt,
     kSqlNodeTypeLast,  // debug type
 };
 
@@ -143,8 +144,9 @@ enum ExprType {
     kExprIn,
     kExprEscaped,
     kExprArray,
-    kExprArrayElement,  // extract value from a array or map, with `[]` operator
-    kExprFake,          // not a real one
+    kExprArrayElement,      // extract value from a array or map, with `[]` operator
+    kExprStructCtorParens,  // (expr1, expr2, ...)
+    kExprFake,              // not a real one
     kExprLast = kExprFake,
 };
 
@@ -344,6 +346,7 @@ enum PlanType {
     kPlanTypeShow,
     kPlanTypeCreateUser,
     kPlanTypeAlterUser,
+    kPlanTypeCallStmt,
     kUnknowPlan = -1,
 };
 

hybridse/include/node/plan_node.h

@@ -809,6 +809,22 @@ class AlterTableStmtPlanNode : public LeafPlanNode {
     std::vector<const AlterActionBase *> actions_;
 };
 
+class CallStmtPlan : public LeafPlanNode {
+ public:
+    CallStmtPlan(const std::vector<std::string> names, const std::vector<ExprNode *> args)
+        : LeafPlanNode(kPlanTypeCallStmt), procedure_name_(names), arguments_(args) {}
+    ~CallStmtPlan() override {}
+
+    const std::vector<std::string> &procedure_name() const { return procedure_name_; }
+    const std::vector<ExprNode *> &arguments() const { return arguments_; }
+
+    void Print(std::ostream &output, const std::string &org_tab) const override;
+
+ private:
+    const std::vector<std::string> procedure_name_;
+    const std::vector<ExprNode*> arguments_;
+};
+
 bool PlanEquals(const PlanNode *left, const PlanNode *right);
 bool PlanListEquals(const std::vector<PlanNode *> &list1, const std::vector<PlanNode *> &list2);
 void PrintPlanVector(std::ostream &output, const std::string &tab, PlanNodeList vec, const std::string vector_name,

hybridse/include/node/sql_node.h

@@ -45,10 +45,10 @@ class LlvmUdfGenBase;
 namespace hybridse {
 namespace node {
 
-class ConstNode;
+class ExprNode;
 class WithClauseEntry;
 
-typedef std::unordered_map<std::string, const ConstNode*> OptionsMap;
+typedef std::unordered_map<std::string, const ExprNode*> OptionsMap;
 
 // Global methods
 std::string NameOfSqlNodeType(const SqlNodeType &type);
@@ -2578,6 +2578,7 @@ class FnReturnStmt : public FnNode {
     ExprNode *return_expr_;
 };
 
+// DEPRECATED!
 class StructExpr : public ExprNode {
  public:
     explicit StructExpr(const std::string &name) : ExprNode(kExprStruct), class_name_(name) {}
@@ -2598,6 +2599,26 @@ class StructExpr : public ExprNode {
     FnNodeList *methods_;
 };
 
+// (expr1, expr2, ...)
+class StructCtorWithParens : public ExprNode {
+ public:
+    explicit StructCtorWithParens(absl::Span<ExprNode *const> fields)
+        : ExprNode(kExprStructCtorParens) {
+        for (auto e : fields) {
+            AddChild(e);
+        }
+    }
+    ~StructCtorWithParens() override {}
+
+    absl::Span<ExprNode *const> fields() const { return children_; }
+
+    // LOW priority
+    // void Print(std::ostream &output, const std::string &org_tab) const override;
+    const std::string GetExprString() const override;
+    StructCtorWithParens *ShadowCopy(NodeManager *nm) const override;
+    Status InferAttr(ExprAnalysisContext *ctx) override;
+};
+
 class ExternalFnDefNode : public FnDefNode {
  public:
     ExternalFnDefNode(const std::string &name, void *fn_ptr, const node::TypeNode *ret_type, bool ret_nullable,
@@ -3005,6 +3026,20 @@ class InputParameterNode : public SqlNode {
     bool is_constant_;
 };
 
+class CallStmt : public SqlNode {
+ public:
+    CallStmt(const std::vector<std::string> names, const std::vector<ExprNode *> args)
+        : SqlNode(kCallStmt, 0, 0), procedure_name_(names), arguments_(args) {}
+    ~CallStmt() override {}
+
+    const std::vector<std::string> &procedure_name() const { return procedure_name_; }
+    const std::vector<ExprNode *> &arguments() const { return arguments_; }
+
+ private:
+    const std::vector<std::string> procedure_name_;
+    const std::vector<ExprNode*> arguments_;
+};
+
 std::string ExprString(const ExprNode *expr);
 std::string MakeExprWithTable(const ExprNode *expr, const std::string db);
 const bool IsNullPrimary(const ExprNode *expr);

hybridse/include/plan/plan_api.h

@@ -16,6 +16,7 @@
 #ifndef HYBRIDSE_INCLUDE_PLAN_PLAN_API_H_
 #define HYBRIDSE_INCLUDE_PLAN_PLAN_API_H_
 
+#include <memory>
 #include <string>
 #include <unordered_map>
 
@@ -49,6 +50,8 @@ class PlanAPI {
     static const std::string GenerateName(const std::string prefix, int id);
 };
 
+absl::Status ParseStatement(absl::string_view, std::unique_ptr<zetasql::ParserOutput>*);
+
 // Parse the input str and SQL type and convert to TypeNode representation
 //
 // unimplemnted, reserved for later usage

hybridse/include/vm/engine.h

@@ -20,9 +20,10 @@
 #include <memory>
 #include <set>
 #include <string>
+#include <unordered_map>
 #include <utility>
 #include <vector>
-#include <unordered_map>
+
 #include "base/spin_lock.h"
 #include "codec/fe_row_codec.h"
 #include "vm/catalog.h"
@@ -36,6 +37,7 @@ using ::hybridse::codec::Row;
 
 inline constexpr const char* LONG_WINDOWS = "long_windows";
 
+class SqlContext;
 class Engine;
 /// \brief An options class for controlling engine behaviour.
 class EngineOptions {
@@ -148,7 +150,7 @@ class RunSession {
     }
 
     /// Return query related compile information.
-    virtual std::shared_ptr<hybridse::vm::CompileInfo> GetCompileInfo() {
+    virtual std::shared_ptr<hybridse::vm::CompileInfo> GetCompileInfo() const {
         return compile_info_;
     }
 
@@ -352,6 +354,8 @@ class Engine {
     /// \brief Create an Engine with a specific Catalog object.
     explicit Engine(const std::shared_ptr<Catalog>& cl);
 
+    ~Engine();
+
     /// \brief Create an Engine a specific Catalog object, configuring it with EngineOptions
     Engine(const std::shared_ptr<Catalog>& cl, const EngineOptions& options);
 
@@ -360,7 +364,9 @@ class Engine {
 
     static void InitializeUnsafeRowOptFlag(bool isUnsafeRowOpt);
 
-    ~Engine();
+    /// determine engine mode for `sql`, `sql` may contains option defining
+    /// execute_mode, `default_mode` used if not or error.
+    static EngineMode TryDetermineEngineMode(absl::string_view sql, EngineMode default_mode);
 
     /// \brief Compile sql in db and stored the results in the session
     bool Get(const std::string& sql, const std::string& db,
@@ -418,6 +424,13 @@ class Engine {
     EngineOptions GetEngineOptions();
 
  private:
+    /// extract request rows info in SQL.
+    /// A SQL e.g 'SELECT ... FROM t1 options (execute_mode = "request", values = ...)'
+    /// request row info exists in 'values' option, as a format of:
+    /// 1. [(col1_expr, col2_expr, ... ), (...), ...]
+    /// 2. (col1_expr, col2_expr, ... )
+    static absl::Status ExtractRequestRowsInSQL(SqlContext* ctx);
+
     std::shared_ptr<CompileInfo> GetCacheLocked(const std::string& db,
                                                 const std::string& sql,
                                                 EngineMode engine_mode);

hybridse/include/vm/engine_context.h

@@ -24,8 +24,9 @@
 namespace hybridse {
 namespace vm {
 
-enum EngineMode { kBatchMode, kRequestMode, kMockRequestMode, kBatchRequestMode };
+enum EngineMode { kBatchMode, kRequestMode, kMockRequestMode, kBatchRequestMode, kOffline };
 std::string EngineModeName(EngineMode mode);
+absl::StatusOr<EngineMode> UnparseEngineMode(absl::string_view);
 
 struct BatchRequestInfo {
     // common column indices in batch request mode

hybridse/include/vm/physical_op.h

@@ -1835,14 +1835,14 @@ class PhysicalSelectIntoNode : public PhysicalUnaryNode {
             return nullptr;
         }
         auto it = options_->find(option);
-        return it == options_->end() ? nullptr : it->second;
+        return it == options_->end() ? nullptr : it->second->GetAsOrNull<hybridse::node::ConstNode>();
     }
     const hybridse::node::ConstNode *GetConfigOption(const std::string &option) const {
         if (!config_options_) {
             return nullptr;
         }
         auto it = config_options_->find(option);
-        return it == config_options_->end() ? nullptr : it->second;
+        return it == config_options_->end() ? nullptr : it->second->GetAsOrNull<hybridse::node::ConstNode>();
     }
 
     std::string query_str_, out_file_;
@@ -1878,7 +1878,7 @@ class PhysicalLoadDataNode : public PhysicalOpNode {
             return nullptr;
         }
         auto it = options_->find(option);
-        return it == options_->end() ? nullptr : it->second;
+        return it == options_->end() ? nullptr : it->second->GetAsOrNull<hybridse::node::ConstNode>();
     }
 
     std::string file_;