| 95 | } |
| 96 | |
| 97 | Status BaseSequenceScanner::Open(ScannerContext* context) { |
| 98 | RETURN_IF_ERROR(HdfsScanner::Open(context)); |
| 99 | stream_->set_read_past_size_cb(bind(&BaseSequenceScanner::ReadPastSize, this, _1)); |
| 100 | bytes_skipped_counter_ = ADD_COUNTER( |
| 101 | scan_node_->runtime_profile(), "BytesSkipped", TUnit::BYTES); |
| 102 | |
| 103 | header_ = reinterpret_cast<FileHeader*>( |
| 104 | scan_node_->GetFileMetadata( |
| 105 | context->partition_descriptor()->id(), stream_->filename())); |
| 106 | if (header_ == nullptr) { |
| 107 | only_parsing_header_ = true; |
| 108 | return Status::OK(); |
| 109 | } |
| 110 | RETURN_IF_ERROR(InitNewRange()); |
| 111 | |
| 112 | // Skip to the first record |
| 113 | if (stream_->file_offset() < header_->header_size) { |
| 114 | // If the scan range starts within the header, skip to the end of the header so we |
| 115 | // don't accidentally skip to an extra sync within the header |
| 116 | RETURN_IF_FALSE(stream_->SkipBytes( |
| 117 | header_->header_size - stream_->file_offset(), &parse_status_)); |
| 118 | } |
| 119 | RETURN_IF_ERROR(SkipToSync(header_->sync, SYNC_HASH_SIZE)); |
| 120 | return Status::OK(); |
| 121 | } |
| 122 | |
| 123 | void BaseSequenceScanner::Close(RowBatch* row_batch) { |
| 124 | DCHECK(!is_closed_); |
nothing calls this directly
no test coverage detected