| 1704 | return ProcessBatchRequestQuery(ctrl, request, response, buf); |
| 1705 | } |
| 1706 | void TabletImpl::ProcessBatchRequestQuery( |
| 1707 | RpcController* ctrl, const fedb::api::SQLBatchRequestQueryRequest* request, |
| 1708 | fedb::api::SQLBatchRequestQueryResponse* response, butil::IOBuf& buf) { |
| 1709 | ::hybridse::base::Status status; |
| 1710 | ::hybridse::vm::BatchRequestRunSession session; |
| 1711 | // run session |
| 1712 | if (request->is_debug()) { |
| 1713 | session.EnableDebug(); |
| 1714 | } |
| 1715 | bool is_procedure = request->is_procedure(); |
| 1716 | if (is_procedure) { |
| 1717 | std::shared_ptr<hybridse::vm::CompileInfo> request_compile_info; |
| 1718 | { |
| 1719 | hybridse::base::Status status; |
| 1720 | request_compile_info = sp_cache_->GetBatchRequestInfo( |
| 1721 | request->db(), request->sp_name(), status); |
| 1722 | if (!status.isOK()) { |
| 1723 | response->set_code( |
| 1724 | ::fedb::base::ReturnCode::kProcedureNotFound); |
| 1725 | response->set_msg(status.msg); |
| 1726 | PDLOG(WARNING, status.msg.c_str()); |
| 1727 | return; |
| 1728 | } |
| 1729 | session.SetCompileInfo(request_compile_info); |
| 1730 | session.SetSpName(request->sp_name()); |
| 1731 | } |
| 1732 | } else { |
| 1733 | size_t common_column_num = request->common_column_indices().size(); |
| 1734 | for (size_t i = 0; i < common_column_num; ++i) { |
| 1735 | auto col_idx = request->common_column_indices().Get(i); |
| 1736 | session.AddCommonColumnIdx(col_idx); |
| 1737 | } |
| 1738 | bool ok = engine_->Get(request->sql(), request->db(), session, status); |
| 1739 | if (!ok || session.GetCompileInfo() == nullptr) { |
| 1740 | response->set_msg(status.msg); |
| 1741 | response->set_code(::fedb::base::kSQLCompileError); |
| 1742 | DLOG(WARNING) << "fail to get sql engine: \n" |
| 1743 | << request->sql() << "\n" << status.str(); |
| 1744 | return; |
| 1745 | } |
| 1746 | } |
| 1747 | |
| 1748 | // fill input data |
| 1749 | auto compile_info = session.GetCompileInfo(); |
| 1750 | if (compile_info == nullptr) { |
| 1751 | response->set_msg("compile info is null, should never happen"); |
| 1752 | response->set_code(::fedb::base::kSQLCompileError); |
| 1753 | return; |
| 1754 | } |
| 1755 | const auto& batch_request_info = |
| 1756 | compile_info->GetBatchRequestInfo(); |
| 1757 | size_t common_column_num = batch_request_info.common_column_indices.size(); |
| 1758 | bool has_common_and_uncommon_row = |
| 1759 | !request->has_task_id() && common_column_num > 0 && |
| 1760 | common_column_num < |
| 1761 | static_cast<size_t>(session.GetRequestSchema().size()); |
| 1762 | size_t input_row_num = request->row_sizes().size(); |
| 1763 | if (request->common_slices() > 0 && input_row_num > 0) { |
nothing calls this directly
no test coverage detected