| 124 | } |
| 125 | |
| 126 | Status ScannerContext::Stream::GetNextBuffer(int64_t read_past_size) { |
| 127 | DCHECK_EQ(0, io_buffer_bytes_left_); |
| 128 | DiskIoMgr* io_mgr = ExecEnv::GetInstance()->disk_io_mgr(); |
| 129 | if (UNLIKELY(parent_->cancelled())) return CONTEXT_CANCELLED; |
| 130 | if (io_buffer_ != nullptr) ReturnIoBuffer(); |
| 131 | |
| 132 | // Nothing to do if we're at the end of the file - return leaving io_buffer_ == nullptr. |
| 133 | int64_t offset = file_offset() + boundary_buffer_bytes_left_; |
| 134 | int64_t file_bytes_remaining = file_desc()->file_length - offset; |
| 135 | if (file_bytes_remaining == 0) return Status::OK(); |
| 136 | |
| 137 | if (!scan_range_eosr_) { |
| 138 | // Get the next buffer from 'scan_range_'. |
| 139 | SCOPED_TIMER2(parent_->state_->total_storage_wait_timer(), |
| 140 | parent_->scan_node_->scanner_io_wait_time()); |
| 141 | Status status = scan_range_->GetNext(&io_buffer_); |
| 142 | DCHECK(!status.ok() || io_buffer_ != nullptr); |
| 143 | RETURN_IF_ERROR(status); |
| 144 | scan_range_eosr_ = io_buffer_->eosr(); |
| 145 | } else { |
| 146 | // Already got all buffers from 'scan_range_' - reading past end. |
| 147 | SCOPED_TIMER2(parent_->state_->total_storage_wait_timer(), |
| 148 | parent_->scan_node_->scanner_io_wait_time()); |
| 149 | |
| 150 | int64_t read_past_buffer_size = 0; |
| 151 | int64_t max_buffer_size = io_mgr->max_buffer_size(); |
| 152 | if (!read_past_size_cb_.empty()) read_past_buffer_size = read_past_size_cb_(offset); |
| 153 | if (read_past_buffer_size <= 0) { |
| 154 | // Either no callback was set or the callback did not return an estimate. Use |
| 155 | // the default doubling strategy. |
| 156 | read_past_buffer_size = next_read_past_size_bytes_; |
| 157 | next_read_past_size_bytes_ = |
| 158 | min<int64_t>(next_read_past_size_bytes_ * 2, max_buffer_size); |
| 159 | } |
| 160 | read_past_buffer_size = ::max(read_past_buffer_size, read_past_size); |
| 161 | read_past_buffer_size = ::min(read_past_buffer_size, file_bytes_remaining); |
| 162 | read_past_buffer_size = ::min(read_past_buffer_size, max_buffer_size); |
| 163 | read_past_buffer_size = ::min(read_past_buffer_size, reservation_); |
| 164 | // We're reading past the scan range. Be careful not to read past the end of file. |
| 165 | DCHECK_GE(read_past_buffer_size, 0); |
| 166 | if (read_past_buffer_size == 0) { |
| 167 | io_buffer_bytes_left_ = 0; |
| 168 | return Status::OK(); |
| 169 | } |
| 170 | int64_t partition_id = parent_->partition_descriptor()->id(); |
| 171 | bool expected_local = false; |
| 172 | // Disable HDFS caching as we are reading past the end. |
| 173 | int cache_options = scan_range_->cache_options() & ~BufferOpts::USE_HDFS_CACHE; |
| 174 | ScanRange* range = parent_->scan_node_->AllocateScanRange( |
| 175 | scan_range_->GetFileInfo(), read_past_buffer_size, offset, partition_id, |
| 176 | scan_range_->disk_id(), expected_local, BufferOpts(cache_options)); |
| 177 | bool needs_buffers; |
| 178 | RETURN_IF_ERROR( |
| 179 | parent_->scan_node_->reader_context()->StartScanRange(range, &needs_buffers)); |
| 180 | if (needs_buffers) { |
| 181 | // Allocate fresh buffers. The buffers for 'scan_range_' should be released now |
| 182 | // since we hit EOS. |
| 183 | if (reservation_ < io_mgr->min_buffer_size()) { |
nothing calls this directly
no test coverage detected