| 1043 | } |
| 1044 | |
| 1045 | void Call::processInput() |
| 1046 | { |
| 1047 | if (!io.pipeline.pushing()) |
| 1048 | return; |
| 1049 | |
| 1050 | bool has_data_to_insert = (insert_query && insert_query->data) |
| 1051 | || !query_info.input_data().empty() || query_info.next_query_info(); |
| 1052 | if (!has_data_to_insert) |
| 1053 | { |
| 1054 | if (!insert_query) |
| 1055 | throw Exception(ErrorCodes::NO_DATA_TO_INSERT, "Query requires data to insert, but it is not an INSERT query"); |
| 1056 | |
| 1057 | const auto & settings = query_context->getSettingsRef(); |
| 1058 | if (settings[Setting::throw_if_no_data_to_insert]) |
| 1059 | throw Exception(ErrorCodes::NO_DATA_TO_INSERT, "No data to insert"); |
| 1060 | |
| 1061 | return; |
| 1062 | } |
| 1063 | |
| 1064 | /// This is significant, because parallel parsing may be used. |
| 1065 | /// So we mustn't touch the input stream from other thread. |
| 1066 | initializePipeline(io.pipeline.getHeader()); |
| 1067 | |
| 1068 | PushingPipelineExecutor executor(io.pipeline); |
| 1069 | executor.start(); |
| 1070 | |
| 1071 | Block block; |
| 1072 | while (pipeline_executor->pull(block)) |
| 1073 | { |
| 1074 | if (!block.empty()) |
| 1075 | executor.push(block); |
| 1076 | } |
| 1077 | |
| 1078 | if (isQueryCancelled()) |
| 1079 | executor.cancel(); |
| 1080 | else |
| 1081 | executor.finish(); |
| 1082 | } |
| 1083 | |
| 1084 | void Call::initializePipeline(const Block & header) |
| 1085 | { |