| 1802 | } |
| 1803 | |
| 1804 | void Scheduler::AssignmentCtx::RecordScanRangeAssignment( |
| 1805 | const BackendDescriptorPB& executor, PlanNodeId node_id, |
| 1806 | const vector<TNetworkAddress>& host_list, |
| 1807 | const TScanRangeLocationList& scan_range_locations, |
| 1808 | FragmentScanRangeAssignment* assignment) { |
| 1809 | if (scan_range_locations.scan_range.__isset.is_system_scan && |
| 1810 | scan_range_locations.scan_range.is_system_scan) { |
| 1811 | // Assigned to a coordinator. |
| 1812 | PerNodeScanRanges* scan_ranges = |
| 1813 | FindOrInsert(assignment, executor.address(), PerNodeScanRanges()); |
| 1814 | vector<ScanRangeParamsPB>* scan_range_params_list = |
| 1815 | FindOrInsert(scan_ranges, node_id, vector<ScanRangeParamsPB>()); |
| 1816 | ScanRangeParamsPB scan_range_params; |
| 1817 | TScanRangeToScanRangePB( |
| 1818 | scan_range_locations.scan_range, scan_range_params.mutable_scan_range()); |
| 1819 | scan_range_params_list->push_back(scan_range_params); |
| 1820 | |
| 1821 | VLOG_FILE << "Scheduler assignment of system table scan to coordinator: " |
| 1822 | << executor.address(); |
| 1823 | return; |
| 1824 | } |
| 1825 | |
| 1826 | int64_t scan_range_length = 0; |
| 1827 | if (scan_range_locations.scan_range.__isset.hdfs_file_split) { |
| 1828 | scan_range_length = scan_range_locations.scan_range.hdfs_file_split.length; |
| 1829 | } else if (scan_range_locations.scan_range.__isset.kudu_scan_token) { |
| 1830 | // Hack so that kudu ranges are well distributed. |
| 1831 | // TODO: KUDU-1133 Use the tablet size instead. |
| 1832 | scan_range_length = 1000; |
| 1833 | } |
| 1834 | |
| 1835 | IpAddr executor_ip; |
| 1836 | bool ret = |
| 1837 | executor_group_.LookUpExecutorIp(executor.address().hostname(), &executor_ip); |
| 1838 | DCHECK(ret); |
| 1839 | DCHECK(!executor_ip.empty()); |
| 1840 | assignment_heap_.InsertOrUpdate( |
| 1841 | executor_ip, scan_range_length, GetExecutorRank(executor_ip)); |
| 1842 | |
| 1843 | // See if the read will be remote. This is not the case if the impalad runs on one of |
| 1844 | // the replica's datanodes. |
| 1845 | bool remote_read = true; |
| 1846 | // For local reads we can set volume_id and try_hdfs_cache. For remote reads HDFS will |
| 1847 | // decide which replica to use so we keep those at default values. |
| 1848 | int volume_id = -1; |
| 1849 | bool try_hdfs_cache = false; |
| 1850 | for (const TScanRangeLocation& location : scan_range_locations.locations) { |
| 1851 | const TNetworkAddress& replica_host = host_list[location.host_idx]; |
| 1852 | IpAddr replica_ip; |
| 1853 | if (executor_group_.LookUpExecutorIp(replica_host.hostname, &replica_ip) |
| 1854 | && executor_ip == replica_ip) { |
| 1855 | remote_read = false; |
| 1856 | volume_id = location.volume_id; |
| 1857 | try_hdfs_cache = location.is_cached; |
| 1858 | break; |
| 1859 | } |
| 1860 | } |
| 1861 |
no test coverage detected