| 2312 | } |
| 2313 | |
| 2314 | void executeHttpQueryInAsyncMode( |
| 2315 | String & query1, |
| 2316 | ASTPtr ast1, |
| 2317 | ContextMutablePtr c, |
| 2318 | WriteBuffer & ostr1, |
| 2319 | ReadBuffer * istr1, |
| 2320 | bool has_query_tail, |
| 2321 | const std::optional<FormatSettings> & f, |
| 2322 | std::function<void(const String &, const String &, const String &, const String &, MPPQueryCoordinatorPtr)> set_result_details) |
| 2323 | { |
| 2324 | const auto * ast_query_with_output1 = dynamic_cast<const ASTQueryWithOutput *>(ast1.get()); |
| 2325 | String format_name1 = ast_query_with_output1 && (ast_query_with_output1->format != nullptr) |
| 2326 | ? getIdentifierName(ast_query_with_output1->format) |
| 2327 | : c->getDefaultFormat(); |
| 2328 | auto query_id_tmp = c->getClientInfo().current_query_id; |
| 2329 | |
| 2330 | c->getAsyncQueryManager()->insertAndRun( |
| 2331 | query1, |
| 2332 | ast1, |
| 2333 | c, |
| 2334 | istr1, |
| 2335 | [c, &ostr1, &f](const String & id) { |
| 2336 | MutableColumnPtr table_column_mut = ColumnString::create(); |
| 2337 | table_column_mut->insert(id); |
| 2338 | Block res; |
| 2339 | res.insert(ColumnWithTypeAndName(std::move(table_column_mut), std::make_shared<DataTypeString>(), "async_query_id")); |
| 2340 | |
| 2341 | auto out = FormatFactory::instance().getOutputFormatParallelIfPossible(c->getDefaultFormat(), ostr1, res, c, false, {}, f); |
| 2342 | |
| 2343 | out->write(res); |
| 2344 | out->flush(); |
| 2345 | }, |
| 2346 | [f, has_query_tail, set_result_details_cp=std::move(set_result_details), query_id=query_id_tmp, format_name1_cp=format_name1](String & query, ASTPtr ast, ContextMutablePtr context, ReadBuffer * istr) { |
| 2347 | ASTPtr ast_output; |
| 2348 | BlockIO streams; |
| 2349 | try |
| 2350 | { |
| 2351 | std::tie(ast_output, streams) = executeQueryImpl( |
| 2352 | query.data(), query.data() + query.size(), ast, context, false, QueryProcessingStage::Complete, has_query_tail, istr); |
| 2353 | auto & pipeline = streams.pipeline; |
| 2354 | if (set_result_details_cp) |
| 2355 | set_result_details_cp(query_id, "text/plain; charset=UTF-8", format_name1_cp, DateLUT::serverTimezoneInstance().getTimeZone(), streams.coordinator); |
| 2356 | if (streams.in) |
| 2357 | { |
| 2358 | const auto * ast_query_with_output = dynamic_cast<const ASTQueryWithOutput *>(ast.get()); |
| 2359 | |
| 2360 | std::shared_ptr<WriteBuffer> out_buf; |
| 2361 | std::optional<String> out_path; |
| 2362 | bool write_to_hdfs = false; |
| 2363 | #if USE_HDFS |
| 2364 | std::unique_ptr<WriteBufferFromHDFS> out_hdfs_raw; |
| 2365 | std::optional<ZlibDeflatingWriteBuffer> out_hdfs_buf; |
| 2366 | #endif |
| 2367 | if (ast_query_with_output && ast_query_with_output->out_file) |
| 2368 | { |
| 2369 | out_path.emplace(typeid_cast<const ASTLiteral &>(*ast_query_with_output->out_file).value.safeGet<std::string>()); |
| 2370 | const Poco::URI out_uri(*out_path); |
| 2371 | const String & scheme = out_uri.getScheme(); |
no test coverage detected