add plan

be/src/common/config.cpp

@@ -106,7 +106,7 @@ DEFINE_String(memory_mode, "moderate");
 
 DEFINE_mBool(enable_use_cgroup_memory_info, "true");
 
-DEFINE_mBool(enable_stream_load_mysql, "false");
+DEFINE_mBool(enable_streamload_by_httpstream, "false");
 
 // process memory limit specified as number of bytes
 // ('<int>[bB]?'), megabytes ('<float>[mM]'), gigabytes ('<float>[gG]'),

be/src/common/config.h

@@ -226,7 +226,7 @@ DECLARE_mInt64(crash_in_alloc_large_memory_bytes);
 DECLARE_mBool(crash_in_memory_tracker_inaccurate);
 
 // config to enable Stream Load functionality through SQL statements.
-DECLARE_mBool(enable_stream_load_mysql);
+DECLARE_mBool(enable_streamload_by_httpstream);
 
 // default is true. if any memory tracking in Orphan mem tracker will report error.
 // !! not modify the default value of this conf!! otherwise memory errors cannot be detected in time.

be/src/http/action/stream_load.cpp

@@ -445,7 +445,7 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req,
         ctx->body_sink = file_sink;
     }
 
-    if (config::enable_stream_load_mysql) {
+    if (config::enable_streamload_by_httpstream) {
         std::string table_name = ctx->table;
         std::string columns, where_clause, column_separator = ",", format = "csv";
         std::vector<std::string> sql_parts;
@@ -473,6 +473,8 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req,
 
         LOG(INFO) << "Generated SQL: " << full_sql;
         request.__set_load_sql(full_sql);
+        ctx->sql_str = full_sql;
+        return _exec_env->stream_load_executor()->execute_plan_fragment(ctx);
     }
 
     if (!http_req->header(HTTP_COLUMNS).empty()) {