| 242 | } |
| 243 | |
| 244 | Status HdfsSequenceScanner::ProcessDecompressedBlock(RowBatch* row_batch) { |
| 245 | int64_t max_tuples = row_batch->capacity() - row_batch->num_rows(); |
| 246 | int num_to_process = min(max_tuples, num_buffered_records_in_compressed_block_); |
| 247 | num_buffered_records_in_compressed_block_ -= num_to_process; |
| 248 | |
| 249 | TupleRow* tuple_row = row_batch->GetRow(row_batch->AddRow()); |
| 250 | if (scan_node_->materialized_slots().empty()) { |
| 251 | // Handle case where there are no slots to materialize (e.g. count(*)) |
| 252 | num_to_process = WriteTemplateTuples(tuple_row, num_to_process); |
| 253 | COUNTER_ADD(scan_node_->rows_read_counter(), num_to_process); |
| 254 | RETURN_IF_ERROR(CommitRows(num_to_process, row_batch)); |
| 255 | return Status::OK(); |
| 256 | } |
| 257 | |
| 258 | // Parse record starts and lengths |
| 259 | int field_location_offset = 0; |
| 260 | for (int i = 0; i < num_to_process; ++i) { |
| 261 | if (i >= record_locations_.size() || record_locations_[i].len < 0 |
| 262 | || next_record_in_compressed_block_ > data_buffer_end_) { |
| 263 | stringstream ss; |
| 264 | ss << stream_->filename() << " Invalid compressed block"; |
| 265 | return Status(ss.str()); |
| 266 | } |
| 267 | int bytes_read = ReadWriteUtil::GetVLong(next_record_in_compressed_block_, |
| 268 | &record_locations_[i].len, next_record_in_compressed_block_len_); |
| 269 | if (UNLIKELY(bytes_read == -1)) { |
| 270 | stringstream ss; |
| 271 | ss << stream_->filename() << " Invalid compressed block"; |
| 272 | return Status(ss.str()); |
| 273 | } |
| 274 | next_record_in_compressed_block_ += bytes_read; |
| 275 | next_record_in_compressed_block_len_ -= bytes_read; |
| 276 | if (next_record_in_compressed_block_len_ <= 0) { |
| 277 | stringstream ss; |
| 278 | ss << stream_->filename() << " Invalid compressed block"; |
| 279 | return Status(ss.str()); |
| 280 | } |
| 281 | record_locations_[i].record = next_record_in_compressed_block_; |
| 282 | next_record_in_compressed_block_ += record_locations_[i].len; |
| 283 | next_record_in_compressed_block_len_ -= record_locations_[i].len; |
| 284 | if (next_record_in_compressed_block_len_ < 0) { |
| 285 | stringstream ss; |
| 286 | ss << stream_->filename() << " Invalid compressed block"; |
| 287 | return Status(ss.str()); |
| 288 | } |
| 289 | } |
| 290 | |
| 291 | // Parse records to find field locations. |
| 292 | for (int i = 0; i < num_to_process; ++i) { |
| 293 | int num_fields = 0; |
| 294 | if (delimited_text_parser_->escape_char() == '\0') { |
| 295 | RETURN_IF_ERROR(delimited_text_parser_->ParseSingleTuple<false>( |
| 296 | record_locations_[i].len, |
| 297 | reinterpret_cast<char*>(record_locations_[i].record), |
| 298 | &field_locations_[field_location_offset], &num_fields)); |
| 299 | } else { |
| 300 | RETURN_IF_ERROR(delimited_text_parser_->ParseSingleTuple<true>( |
| 301 | record_locations_[i].len, |
nothing calls this directly
no test coverage detected