fix

Author: 宋光璠

.idea/vcs.xml

@@ -597,6 +597,8 @@ DEFINE_mInt32(slave_replica_writer_rpc_timeout_sec, "60");
 // Whether to enable stream load record function, the default is false.
 // False: disable stream load record
 DEFINE_mBool(enable_stream_load_record, "false");
+// false: disables http_stream load support
+DEFINE_mBool(enable_http_stream_load, "false");
 // batch size of stream load record reported to FE
 DEFINE_mInt32(stream_load_record_batch_size, "50");
 // expire time of stream load record in rocksdb.

be/src/common/config.cpp

@@ -632,6 +632,8 @@ DECLARE_mInt32(slave_replica_writer_rpc_timeout_sec);
 // Whether to enable stream load record function, the default is false.
 // False: disable stream load record
 DECLARE_mBool(enable_stream_load_record);
+// Whether to enable http_stream load support. Default is false.
+DECLARE_mBool(enable_http_stream_load);
 // batch size of stream load record reported to FE
 DECLARE_mInt32(stream_load_record_batch_size);
 // expire time of stream load record in rocksdb.

be/src/common/config.h

@@ -724,6 +724,48 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req,
         request.__set_cloud_cluster(http_req->header(HTTP_CLOUD_CLUSTER));
     }
 
+    if (config::enable_http_stream_load) {
+        std::string table_name = ctx->table;
+        std::string database_name = request.db;
+
+        std::string columns = http_req->header(HTTP_COLUMNS);
+        if (columns.empty()) {
+            return Status::InvalidArgument("Missing required HTTP header: columns");
+        }
+
+        std::string column_separator = http_req->header(HTTP_COLUMN_SEPARATOR).empty()
+                                               ? ","
+                                               : http_req->header(HTTP_COLUMN_SEPARATOR);
+
+        std::string format =
+                http_req->header("format").empty() ? "csv" : http_req->header("format");
+
+        std::vector<std::string> column_vector = split(columns, std::string(","));
+        std::string column_list = "(" + join(column_vector, ", ") + ")";
+
+        std::vector<std::string> sql_parts;
+        sql_parts.emplace_back("SELECT " + join(column_vector, ", "));
+
+        std::string http_stream_params = "http_stream('format' = '" + format +
+                                         "', 'column_separator' = '" + column_separator + "')";
+
+        sql_parts.emplace_back("FROM " + http_stream_params);
+
+        if (!http_req->header(HTTP_WHERE).empty()) {
+            sql_parts.emplace_back("WHERE " + http_req->header(HTTP_WHERE));
+        }
+
+        std::string full_sql =
+                "INSERT INTO " + database_name + "." + table_name + " " + column_list + " ";
+        for (size_t i = 0; i < sql_parts.size(); i++) {
+            full_sql += sql_parts[i];
+            if (i != sql_parts.size() - 1) full_sql += " ";
+        }
+
+        ctx->sql_str = full_sql;
+        LOG(INFO) << "Generated SQL: " << full_sql << ctx->to_json();
+    }
+
 #ifndef BE_TEST
     // plan this load
     TNetworkAddress master_addr = _exec_env->cluster_info()->master_fe_addr;