| 250 | #endif |
| 251 | |
| 252 | Status Scheduler::ComputeScanRangeAssignment( |
| 253 | const ExecutorConfig& executor_config, ScheduleState* state) { |
| 254 | RuntimeProfile::Counter* total_assignment_timer = |
| 255 | ADD_TIMER(state->summary_profile(), "ComputeScanRangeAssignmentTimer"); |
| 256 | const TQueryExecRequest& exec_request = state->request(); |
| 257 | for (const TPlanExecInfo& plan_exec_info : exec_request.plan_exec_info) { |
| 258 | for (const auto& entry : plan_exec_info.per_node_scan_ranges) { |
| 259 | const TPlanNodeId node_id = entry.first; |
| 260 | const TPlanFragment& fragment = state->GetContainingFragment(node_id); |
| 261 | bool exec_at_coord = (fragment.partition.type == TPartitionType::UNPARTITIONED); |
| 262 | DCHECK(executor_config.group.NumExecutors() > 0 || exec_at_coord); |
| 263 | |
| 264 | const TPlanNode& node = state->GetNode(node_id); |
| 265 | DCHECK_EQ(node.node_id, node_id); |
| 266 | |
| 267 | bool has_preference = |
| 268 | node.__isset.hdfs_scan_node && node.hdfs_scan_node.__isset.replica_preference; |
| 269 | const TReplicaPreference::type* node_replica_preference = has_preference ? |
| 270 | &node.hdfs_scan_node.replica_preference : |
| 271 | nullptr; |
| 272 | bool node_random_replica = node.__isset.hdfs_scan_node |
| 273 | && node.hdfs_scan_node.__isset.random_replica |
| 274 | && node.hdfs_scan_node.random_replica; |
| 275 | bool node_schedule_oldest_to_newest = node.__isset.hdfs_scan_node |
| 276 | && node.hdfs_scan_node.__isset.schedule_scanranges_oldest_to_newest |
| 277 | && node.hdfs_scan_node.schedule_scanranges_oldest_to_newest; |
| 278 | |
| 279 | FragmentScanRangeAssignment* assignment = |
| 280 | &state->GetFragmentScheduleState(fragment.idx)->scan_range_assignment; |
| 281 | |
| 282 | const vector<TScanRangeLocationList>* locations = &entry.second.concrete_ranges; |
| 283 | vector<TScanRangeLocationList> expanded_locations; |
| 284 | // Copy the ranges to a separate vector if: |
| 285 | // 1. There are split specs to union with the concrete ranges |
| 286 | // 2. We're scheduling oldest to newest and need to sort the ranges without |
| 287 | // changing the original vector |
| 288 | if (!entry.second.split_specs.empty() || node_schedule_oldest_to_newest) { |
| 289 | locations = &expanded_locations; |
| 290 | expanded_locations.insert(expanded_locations.end(), |
| 291 | entry.second.concrete_ranges.begin(), entry.second.concrete_ranges.end()); |
| 292 | // union concrete ranges and expanded specs |
| 293 | if (!entry.second.split_specs.empty()) { |
| 294 | RETURN_IF_ERROR( |
| 295 | GenerateScanRanges(entry.second.split_specs, &expanded_locations)); |
| 296 | } |
| 297 | } |
| 298 | if (node_schedule_oldest_to_newest) { |
| 299 | DCHECK_GE(expanded_locations.size(), |
| 300 | entry.second.concrete_ranges.size() + entry.second.split_specs.size()); |
| 301 | // This only makes sense for HDFS scan nodes |
| 302 | DCHECK(node.__isset.hdfs_scan_node); |
| 303 | // Sort the scan ranges by modification time ascending. In debug mode, do |
| 304 | // additional validation of the ordering. |
| 305 | #ifndef NDEBUG |
| 306 | std::sort(expanded_locations.begin(), expanded_locations.end(), |
| 307 | ScanRangeOldestToNewestComparatorWithValidation); |
| 308 | #else |
| 309 | std::sort(expanded_locations.begin(), expanded_locations.end(), |