| 584 | } |
| 585 | |
| 586 | Status HdfsScanNodeBase::Open(RuntimeState* state) { |
| 587 | RETURN_IF_ERROR(ScanNode::Open(state)); |
| 588 | |
| 589 | // Open collection conjuncts |
| 590 | for (auto& entry: conjunct_evals_map_) { |
| 591 | // conjuncts_ are already opened in ExecNode::Open() |
| 592 | if (entry.first == tuple_id_) continue; |
| 593 | RETURN_IF_ERROR(ScalarExprEvaluator::Open(entry.second, state)); |
| 594 | } |
| 595 | |
| 596 | // Open stats conjuncts |
| 597 | RETURN_IF_ERROR(ScalarExprEvaluator::Open(stats_conjunct_evals_, state)); |
| 598 | |
| 599 | RETURN_IF_ERROR(ClaimBufferReservation(state)); |
| 600 | reader_context_ = ExecEnv::GetInstance()->disk_io_mgr()->RegisterContext(); |
| 601 | |
| 602 | // Initialize HdfsScanNode specific counters |
| 603 | hdfs_read_timer_ = PROFILE_TotalRawHdfsReadTime.Instantiate(runtime_profile()); |
| 604 | hdfs_open_file_timer_ = |
| 605 | PROFILE_TotalRawHdfsOpenFileTime.Instantiate(runtime_profile()); |
| 606 | per_read_thread_throughput_counter_ = |
| 607 | PROFILE_PerReadThreadRawHdfsThroughput.Instantiate(runtime_profile(), |
| 608 | bind<int64_t>(&RuntimeProfile::UnitsPerSecond, bytes_read_counter_, |
| 609 | hdfs_read_timer_)); |
| 610 | scan_ranges_complete_counter_ = |
| 611 | PROFILE_ScanRangesComplete.Instantiate(runtime_profile()); |
| 612 | collection_items_read_counter_ = |
| 613 | PROFILE_CollectionItemsRead.Instantiate(runtime_profile()); |
| 614 | if (DiskInfo::num_disks() < 64) { |
| 615 | num_disks_accessed_counter_ = |
| 616 | PROFILE_NumDisksAccessed.Instantiate(runtime_profile()); |
| 617 | } else { |
| 618 | num_disks_accessed_counter_ = NULL; |
| 619 | } |
| 620 | |
| 621 | data_cache_hit_count_ = PROFILE_DataCacheHitCount.Instantiate(runtime_profile()); |
| 622 | data_cache_partial_hit_count_ = |
| 623 | PROFILE_DataCachePartialHitCount.Instantiate(runtime_profile()); |
| 624 | data_cache_miss_count_ = PROFILE_DataCacheMissCount.Instantiate(runtime_profile()); |
| 625 | data_cache_hit_bytes_ = PROFILE_DataCacheHitBytes.Instantiate(runtime_profile()); |
| 626 | data_cache_miss_bytes_ = PROFILE_DataCacheMissBytes.Instantiate(runtime_profile()); |
| 627 | |
| 628 | reader_context_->set_bytes_read_counter(bytes_read_counter()); |
| 629 | reader_context_->set_read_timer(hdfs_read_timer_); |
| 630 | reader_context_->set_open_file_timer(hdfs_open_file_timer_); |
| 631 | reader_context_->set_active_read_thread_counter(&active_hdfs_read_thread_counter_); |
| 632 | reader_context_->set_disks_accessed_bitmap(&disks_accessed_bitmap_); |
| 633 | reader_context_->set_data_cache_hit_counter(data_cache_hit_count_); |
| 634 | reader_context_->set_data_cache_partial_hit_counter(data_cache_partial_hit_count_); |
| 635 | reader_context_->set_data_cache_miss_counter(data_cache_miss_count_); |
| 636 | reader_context_->set_data_cache_hit_bytes_counter(data_cache_hit_bytes_); |
| 637 | reader_context_->set_data_cache_miss_bytes_counter(data_cache_miss_bytes_); |
| 638 | |
| 639 | average_hdfs_read_thread_concurrency_ = |
| 640 | PROFILE_AverageHdfsReadThreadConcurrency.Instantiate(runtime_profile(), |
| 641 | &active_hdfs_read_thread_counter_); |
| 642 | |
| 643 | initial_range_ideal_reservation_stats_ = |
no test coverage detected