| 470 | } |
| 471 | |
| 472 | void HdfsScanNode::ProcessSplit(const vector<FilterContext>& filter_ctxs, |
| 473 | MemPool* expr_results_pool, ScanRange* scan_range, |
| 474 | int64_t* scanner_thread_reservation) { |
| 475 | DCHECK(scan_range != nullptr); |
| 476 | ScanRangeMetadata* metadata = static_cast<ScanRangeMetadata*>(scan_range->meta_data()); |
| 477 | int64_t partition_id = metadata->partition_id; |
| 478 | HdfsPartitionDescriptor* partition = hdfs_table_->GetPartition(partition_id); |
| 479 | DCHECK(partition != nullptr) << "table_id=" << hdfs_table_->id() |
| 480 | << " partition_id=" << partition_id |
| 481 | << "\n" << PrintThrift(runtime_state_->instance_ctx()); |
| 482 | ScannerContext context(runtime_state_, this, buffer_pool_client(), |
| 483 | *scanner_thread_reservation, partition, filter_ctxs, expr_results_pool); |
| 484 | context.AddStream(scan_range, *scanner_thread_reservation); |
| 485 | scoped_ptr<HdfsScanner> scanner; |
| 486 | Status status = CreateAndOpenScannerHelper(partition, &context, &scanner); |
| 487 | if (!status.ok()) { |
| 488 | // If preparation fails, avoid leaking unread buffers in the scan_range. |
| 489 | scan_range->Cancel(status); |
| 490 | |
| 491 | if (VLOG_QUERY_IS_ON) { |
| 492 | stringstream ss; |
| 493 | ss << "Error preparing scanner for scan range " << scan_range->file() << |
| 494 | "(" << scan_range->offset() << ":" << scan_range->len() << "). "; |
| 495 | ss << status.msg().msg() << endl << runtime_state_->ErrorLog(); |
| 496 | VLOG_QUERY << ss.str(); |
| 497 | } |
| 498 | |
| 499 | // Ensure that the error is propagated before marking a range as complete (The |
| 500 | // scanner->Close() call marks a scan range as complete). |
| 501 | SetError(status); |
| 502 | if (scanner != nullptr) scanner->Close(); |
| 503 | return; |
| 504 | } |
| 505 | |
| 506 | status = scanner->ProcessSplit(); |
| 507 | if (!status.ok()) { |
| 508 | if (VLOG_QUERY_IS_ON && !status.IsCancelled()) { |
| 509 | // This thread hit an error, record it and bail |
| 510 | stringstream ss; |
| 511 | ss << "Scan node (id=" << id() << ") ran into a parse error for scan range " |
| 512 | << scan_range->file() << "(" << scan_range->offset() << ":" << scan_range->len() |
| 513 | << ")."; |
| 514 | // Parquet doesn't read the range end to end so the current offset isn't useful. |
| 515 | // TODO: make sure the parquet reader is outputting as much diagnostic |
| 516 | // information as possible. |
| 517 | if (partition->file_format() != THdfsFileFormat::PARQUET) { |
| 518 | ScannerContext::Stream* stream = context.GetStream(); |
| 519 | ss << " Processed " << stream->total_bytes_returned() << " bytes."; |
| 520 | } |
| 521 | VLOG_QUERY << ss.str(); |
| 522 | } |
| 523 | |
| 524 | // If status is the first non-ok status returned by any scanner thread, update |
| 525 | // HdfsScanNodeBase::status_ variable and notify other scanner threads. Ensure that |
| 526 | // status_ is updated before marking a range as complete (The scanner->Close() call |
| 527 | // marks a scan range as complete). |
| 528 | SetError(status); |
| 529 | } |
nothing calls this directly
no test coverage detected