be/src/runtime/io/hdfs-file-reader.cc - impala - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //   http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing,
 // software distributed under the License is distributed on an
 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.

 #include <algorithm>

 #include "gutil/strings/substitute.h"
 #include "runtime/io/disk-io-mgr-internal.h"
 #include "runtime/io/hdfs-file-reader.h"
 #include "runtime/io/request-context.h"
 #include "runtime/io/request-ranges.h"
 #include "util/hdfs-util.h"
 #include "util/impalad-metrics.h"

 #include "common/names.h"

 DEFINE_bool(use_hdfs_pread, false, "Enables using hdfsPread() instead of hdfsRead() "
     "when performing HDFS read operations. This is necessary to use HDFS hedged reads "
     "(assuming the HDFS client is configured to do so).");

 #ifndef NDEBUG
 DECLARE_int32(stress_disk_read_delay_ms);
 #endif

 namespace impala {
 namespace io {

 HdfsFileReader::~HdfsFileReader() {
   DCHECK(exclusive_hdfs_fh_ == nullptr) << "File was not closed.";
   DCHECK(cached_buffer_ == nullptr) << "Cached buffer was not released.";
 }

 Status HdfsFileReader::Open(bool use_file_handle_cache) {
   unique_lock<SpinLock> hdfs_lock(lock_);
   RETURN_IF_ERROR(scan_range_->cancel_status_);

   if (exclusive_hdfs_fh_ != nullptr) return Status::OK();
   // With file handle caching, the reader does not maintain its own
   // hdfs file handle. File handle caching is only used for local files,
   // so s3 and remote filesystems should obtain an exclusive file handle
   // for each scan range.
   if (use_file_handle_cache && expected_local_) return Status::OK();
   auto io_mgr = scan_range_->io_mgr_;
   // Get a new exclusive file handle.
   exclusive_hdfs_fh_ = io_mgr->GetExclusiveHdfsFileHandle(hdfs_fs_,
       scan_range_->file_string(), scan_range_->mtime(), scan_range_->reader_);
   if (exclusive_hdfs_fh_ == nullptr) {
     return Status(TErrorCode::DISK_IO_ERROR,
         GetHdfsErrorMsg("Failed to open HDFS file ", *scan_range_->file_string()));
   }
   if (hdfsSeek(hdfs_fs_, exclusive_hdfs_fh_->file(), scan_range_->offset_) != 0) {
     // Destroy the file handle
     io_mgr->ReleaseExclusiveHdfsFileHandle(exclusive_hdfs_fh_);
     exclusive_hdfs_fh_ = nullptr;
     return Status(TErrorCode::DISK_IO_ERROR,
         Substitute("Error seeking to $0 in file: $1 $2", scan_range_->offset(),
             *scan_range_->file_string(), GetHdfsErrorMsg("")));
   }
   ImpaladMetrics::IO_MGR_NUM_OPEN_FILES->Increment(1L);
   return Status::OK();
 }

 Status HdfsFileReader::ReadFromPos(int64_t file_offset, uint8_t* buffer,
     int64_t bytes_to_read, int64_t* bytes_read, bool* eof) {
   DCHECK(scan_range_->read_in_flight());
   DCHECK_GE(bytes_to_read, 0);
   // Delay before acquiring the lock, to allow triggering IMPALA-6587 race.
 #ifndef NDEBUG
   if (FLAGS_stress_disk_read_delay_ms > 0) {
     SleepForMs(FLAGS_stress_disk_read_delay_ms);
   }
 #endif
   unique_lock<SpinLock> hdfs_lock(lock_);
   RETURN_IF_ERROR(scan_range_->cancel_status_);

   auto io_mgr = scan_range_->io_mgr_;
   auto request_context = scan_range_->reader_;
   *eof = false;
   *bytes_read = 0;

   CachedHdfsFileHandle* borrowed_hdfs_fh = nullptr;
   hdfsFile hdfs_file;

   // If the reader has an exclusive file handle, use it. Otherwise, borrow
   // a file handle from the cache.
   if (exclusive_hdfs_fh_ != nullptr) {
     hdfs_file = exclusive_hdfs_fh_->file();
   } else {
     borrowed_hdfs_fh = io_mgr->GetCachedHdfsFileHandle(hdfs_fs_,
         scan_range_->file_string(),
         scan_range_->mtime(), request_context);
     if (borrowed_hdfs_fh == nullptr) {
       return Status(TErrorCode::DISK_IO_ERROR,
           GetHdfsErrorMsg("Failed to open HDFS file ", *scan_range_->file_string()));
     }
     hdfs_file = borrowed_hdfs_fh->file();
   }

   int64_t max_chunk_size = scan_range_->MaxReadChunkSize();
   Status status = Status::OK();
   {
     ScopedTimer<MonotonicStopWatch> req_context_read_timer(
         scan_range_->reader_->read_timer_);
     while (*bytes_read < bytes_to_read) {
       int chunk_size = min(bytes_to_read - *bytes_read, max_chunk_size);
       DCHECK_GT(chunk_size, 0);
       // The hdfsRead() length argument is an int.
       DCHECK_LE(chunk_size, numeric_limits<int>::max());
       int current_bytes_read = -1;
       // bytes_read_ is only updated after the while loop
       int64_t position_in_file = file_offset + *bytes_read;

       // ReadFromPosInternal() might fail due to a bad file handle.
       // If that was the case, allow for a retry to fix it.
       status = ReadFromPosInternal(hdfs_file, position_in_file,
           borrowed_hdfs_fh != nullptr, buffer + *bytes_read, chunk_size,
           &current_bytes_read);

       // Retry if:
       // - first read was not successful
       // and
       // - used a borrowed file handle
       if (!status.ok() && borrowed_hdfs_fh != nullptr) {
         // The error may be due to a bad file handle. Reopen the file handle and retry.
         // Exclude this time from the read timers.
         req_context_read_timer.Stop();
         RETURN_IF_ERROR(io_mgr->ReopenCachedHdfsFileHandle(hdfs_fs_,
                 scan_range_->file_string(), scan_range_->mtime(),
                 request_context, &borrowed_hdfs_fh));
         hdfs_file = borrowed_hdfs_fh->file();
         req_context_read_timer.Start();
         status = ReadFromPosInternal(hdfs_file, position_in_file,
             borrowed_hdfs_fh != nullptr, buffer + *bytes_read, chunk_size,
             &current_bytes_read);
       }
       if (!status.ok()) {
         break;
       }
       DCHECK_GT(current_bytes_read, -1);
       if (current_bytes_read == 0) {
         // No more bytes in the file. The scan range went past the end.
         *eof = true;
         break;
       }
       *bytes_read += current_bytes_read;

       // Collect and accumulate statistics
       GetHdfsStatistics(hdfs_file);
     }
   }

   if (borrowed_hdfs_fh != nullptr) {
     io_mgr->ReleaseCachedHdfsFileHandle(scan_range_->file_string(), borrowed_hdfs_fh);
   }
   return status;
 }

 Status HdfsFileReader::ReadFromPosInternal(hdfsFile hdfs_file, int64_t position_in_file,
     bool is_borrowed_fh, uint8_t* buffer, int64_t chunk_size, int* bytes_read) {
   // For file handles from the cache, any of the below file operations may fail
   // due to a bad file handle.
   if (FLAGS_use_hdfs_pread) {
     *bytes_read = hdfsPread(hdfs_fs_, hdfs_file, position_in_file, buffer, chunk_size);
     if (*bytes_read == -1) {
       return Status(TErrorCode::DISK_IO_ERROR,
           GetHdfsErrorMsg("Error reading from HDFS file: ",
               *scan_range_->file_string()));
     }
   } else {
     // If the file handle is borrowed, it may not be at the appropriate
     // location. Seek to the appropriate location.
     if (is_borrowed_fh) {
       if (hdfsSeek(hdfs_fs_, hdfs_file, position_in_file) != 0) {
         return Status(TErrorCode::DISK_IO_ERROR,
             Substitute("Error seeking to $0 in file: $1: $2",
                 position_in_file, *scan_range_->file_string(), GetHdfsErrorMsg("")));
       }
     }
     *bytes_read = hdfsRead(hdfs_fs_, hdfs_file, buffer, chunk_size);
     if (*bytes_read == -1) {
       return Status(TErrorCode::DISK_IO_ERROR,
           GetHdfsErrorMsg("Error reading from HDFS file: ",
               *scan_range_->file_string()));
     }
   }
   return Status::OK();
 }

 void HdfsFileReader::CachedFile(uint8_t** data, int64_t* length) {
   {
     unique_lock<SpinLock> hdfs_lock(lock_);
     DCHECK(cached_buffer_ == nullptr);
     DCHECK(exclusive_hdfs_fh_ != nullptr);
     cached_buffer_ = hadoopReadZero(exclusive_hdfs_fh_->file(),
         scan_range_->io_mgr_->cached_read_options(), scan_range_->len());
   }
   if (cached_buffer_ == nullptr) {
     *data = nullptr;
     *length = 0;
     return;
   }
   *data = reinterpret_cast<uint8_t*>(
       const_cast<void*>(hadoopRzBufferGet(cached_buffer_)));
   *length = hadoopRzBufferLength(cached_buffer_);
 }

 void HdfsFileReader::Close() {
   unique_lock<SpinLock> hdfs_lock(lock_);
   if (exclusive_hdfs_fh_ != nullptr) {
     GetHdfsStatistics(exclusive_hdfs_fh_->file());

     if (cached_buffer_ != nullptr) {
       hadoopRzBufferFree(exclusive_hdfs_fh_->file(), cached_buffer_);
       cached_buffer_ = nullptr;
     }

     // Destroy the file handle.
     scan_range_->io_mgr_->ReleaseExclusiveHdfsFileHandle(exclusive_hdfs_fh_);
     exclusive_hdfs_fh_ = nullptr;
     ImpaladMetrics::IO_MGR_NUM_OPEN_FILES->Increment(-1L);
   }

   if (FLAGS_use_hdfs_pread && IsHdfsPath(scan_range_->file())) {
     // Update Hedged Read Metrics.
     // We call it only if the --use_hdfs_pread flag is set, to avoid having the
     // libhdfs client malloc and free a hdfsHedgedReadMetrics object unnecessarily
     // otherwise. 'hedged_metrics' is only set upon success.
     // We also avoid calling hdfsGetHedgedReadMetrics() when the file is not on HDFS
     // (see HDFS-13417).
     struct hdfsHedgedReadMetrics* hedged_metrics;
     int success = hdfsGetHedgedReadMetrics(hdfs_fs_, &hedged_metrics);
     if (success == 0) {
       ImpaladMetrics::HEDGED_READ_OPS->SetValue(hedged_metrics->hedgedReadOps);
       ImpaladMetrics::HEDGED_READ_OPS_WIN->SetValue(
           hedged_metrics->hedgedReadOpsWin);
       hdfsFreeHedgedReadMetrics(hedged_metrics);
     }
   }

   if (num_remote_bytes_ > 0) {
     scan_range_->reader_->num_remote_ranges_.Add(1L);
     if (expected_local_) {
       scan_range_->reader_->unexpected_remote_bytes_.Add(num_remote_bytes_);
       VLOG_FILE << "Unexpected remote HDFS read of "
                    << PrettyPrinter::Print(num_remote_bytes_, TUnit::BYTES)
                    << " for file '" << *scan_range_->file_string() << "'";
     }
   }
 }

 void HdfsFileReader::GetHdfsStatistics(hdfsFile hdfs_file) {
   struct hdfsReadStatistics* stats;
   if (IsHdfsPath(scan_range_->file())) {
     int success = hdfsFileGetReadStatistics(hdfs_file, &stats);
     if (success == 0) {
       scan_range_->reader_->bytes_read_local_.Add(stats->totalLocalBytesRead);
       scan_range_->reader_->bytes_read_short_circuit_.Add(
           stats->totalShortCircuitBytesRead);
       scan_range_->reader_->bytes_read_dn_cache_.Add(stats->totalZeroCopyBytesRead);
       if (stats->totalLocalBytesRead != stats->totalBytesRead) {
         num_remote_bytes_ += stats->totalBytesRead - stats->totalLocalBytesRead;
       }
       hdfsFileFreeReadStatistics(stats);
     }
     hdfsFileClearReadStatistics(hdfs_file);
   }
 }

 void HdfsFileReader::ResetState() {
   FileReader::ResetState();
   num_remote_bytes_ = 0;
 }

 string HdfsFileReader::DebugString() const {
   return FileReader::DebugString() + Substitute(
       " exclusive_hdfs_fh=$0 num_remote_bytes=$1",
       exclusive_hdfs_fh_, num_remote_bytes_);
 }

 }
 }
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing,
	// software distributed under the License is distributed on an
	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	// KIND, either express or implied. See the License for the
	// specific language governing permissions and limitations
	// under the License.

	#include <algorithm>

	#include "gutil/strings/substitute.h"
	#include "runtime/io/disk-io-mgr-internal.h"
	#include "runtime/io/hdfs-file-reader.h"
	#include "runtime/io/request-context.h"
	#include "runtime/io/request-ranges.h"
	#include "util/hdfs-util.h"
	#include "util/impalad-metrics.h"

	#include "common/names.h"

	DEFINE_bool(use_hdfs_pread, false, "Enables using hdfsPread() instead of hdfsRead() "
	"when performing HDFS read operations. This is necessary to use HDFS hedged reads "
	"(assuming the HDFS client is configured to do so).");

	#ifndef NDEBUG
	DECLARE_int32(stress_disk_read_delay_ms);
	#endif

	namespace impala {
	namespace io {

	HdfsFileReader::~HdfsFileReader() {
	DCHECK(exclusive_hdfs_fh_ == nullptr) << "File was not closed.";
	DCHECK(cached_buffer_ == nullptr) << "Cached buffer was not released.";
	}

	Status HdfsFileReader::Open(bool use_file_handle_cache) {
	unique_lock<SpinLock> hdfs_lock(lock_);
	RETURN_IF_ERROR(scan_range_->cancel_status_);

	if (exclusive_hdfs_fh_ != nullptr) return Status::OK();
	// With file handle caching, the reader does not maintain its own
	// hdfs file handle. File handle caching is only used for local files,
	// so s3 and remote filesystems should obtain an exclusive file handle
	// for each scan range.
	if (use_file_handle_cache && expected_local_) return Status::OK();
	auto io_mgr = scan_range_->io_mgr_;
	// Get a new exclusive file handle.
	exclusive_hdfs_fh_ = io_mgr->GetExclusiveHdfsFileHandle(hdfs_fs_,
	scan_range_->file_string(), scan_range_->mtime(), scan_range_->reader_);
	if (exclusive_hdfs_fh_ == nullptr) {
	return Status(TErrorCode::DISK_IO_ERROR,
	GetHdfsErrorMsg("Failed to open HDFS file ", *scan_range_->file_string()));
	}
	if (hdfsSeek(hdfs_fs_, exclusive_hdfs_fh_->file(), scan_range_->offset_) != 0) {
	// Destroy the file handle
	io_mgr->ReleaseExclusiveHdfsFileHandle(exclusive_hdfs_fh_);
	exclusive_hdfs_fh_ = nullptr;
	return Status(TErrorCode::DISK_IO_ERROR,
	Substitute("Error seeking to $0 in file: $1 $2", scan_range_->offset(),
	*scan_range_->file_string(), GetHdfsErrorMsg("")));
	}
	ImpaladMetrics::IO_MGR_NUM_OPEN_FILES->Increment(1L);
	return Status::OK();
	}

	Status HdfsFileReader::ReadFromPos(int64_t file_offset, uint8_t* buffer,
	int64_t bytes_to_read, int64_t* bytes_read, bool* eof) {
	DCHECK(scan_range_->read_in_flight());
	DCHECK_GE(bytes_to_read, 0);
	// Delay before acquiring the lock, to allow triggering IMPALA-6587 race.
	#ifndef NDEBUG
	if (FLAGS_stress_disk_read_delay_ms > 0) {
	SleepForMs(FLAGS_stress_disk_read_delay_ms);
	}
	#endif
	unique_lock<SpinLock> hdfs_lock(lock_);
	RETURN_IF_ERROR(scan_range_->cancel_status_);

	auto io_mgr = scan_range_->io_mgr_;
	auto request_context = scan_range_->reader_;
	*eof = false;
	*bytes_read = 0;

	CachedHdfsFileHandle* borrowed_hdfs_fh = nullptr;
	hdfsFile hdfs_file;

	// If the reader has an exclusive file handle, use it. Otherwise, borrow
	// a file handle from the cache.
	if (exclusive_hdfs_fh_ != nullptr) {
	hdfs_file = exclusive_hdfs_fh_->file();
	} else {
	borrowed_hdfs_fh = io_mgr->GetCachedHdfsFileHandle(hdfs_fs_,
	scan_range_->file_string(),
	scan_range_->mtime(), request_context);
	if (borrowed_hdfs_fh == nullptr) {
	return Status(TErrorCode::DISK_IO_ERROR,
	GetHdfsErrorMsg("Failed to open HDFS file ", *scan_range_->file_string()));
	}
	hdfs_file = borrowed_hdfs_fh->file();
	}

	int64_t max_chunk_size = scan_range_->MaxReadChunkSize();
	Status status = Status::OK();
	{
	ScopedTimer<MonotonicStopWatch> req_context_read_timer(
	scan_range_->reader_->read_timer_);
	while (*bytes_read < bytes_to_read) {
	int chunk_size = min(bytes_to_read - *bytes_read, max_chunk_size);
	DCHECK_GT(chunk_size, 0);
	// The hdfsRead() length argument is an int.
	DCHECK_LE(chunk_size, numeric_limits<int>::max());
	int current_bytes_read = -1;
	// bytes_read_ is only updated after the while loop
	int64_t position_in_file = file_offset + *bytes_read;

	// ReadFromPosInternal() might fail due to a bad file handle.
	// If that was the case, allow for a retry to fix it.
	status = ReadFromPosInternal(hdfs_file, position_in_file,
	borrowed_hdfs_fh != nullptr, buffer + *bytes_read, chunk_size,
	&current_bytes_read);

	// Retry if:
	// - first read was not successful
	// and
	// - used a borrowed file handle
	if (!status.ok() && borrowed_hdfs_fh != nullptr) {
	// The error may be due to a bad file handle. Reopen the file handle and retry.
	// Exclude this time from the read timers.
	req_context_read_timer.Stop();
	RETURN_IF_ERROR(io_mgr->ReopenCachedHdfsFileHandle(hdfs_fs_,
	scan_range_->file_string(), scan_range_->mtime(),
	request_context, &borrowed_hdfs_fh));
	hdfs_file = borrowed_hdfs_fh->file();
	req_context_read_timer.Start();
	status = ReadFromPosInternal(hdfs_file, position_in_file,
	borrowed_hdfs_fh != nullptr, buffer + *bytes_read, chunk_size,
	&current_bytes_read);
	}
	if (!status.ok()) {
	break;
	}
	DCHECK_GT(current_bytes_read, -1);
	if (current_bytes_read == 0) {
	// No more bytes in the file. The scan range went past the end.
	*eof = true;
	break;
	}
	*bytes_read += current_bytes_read;

	// Collect and accumulate statistics
	GetHdfsStatistics(hdfs_file);
	}
	}

	if (borrowed_hdfs_fh != nullptr) {
	io_mgr->ReleaseCachedHdfsFileHandle(scan_range_->file_string(), borrowed_hdfs_fh);
	}
	return status;
	}

	Status HdfsFileReader::ReadFromPosInternal(hdfsFile hdfs_file, int64_t position_in_file,
	bool is_borrowed_fh, uint8_t* buffer, int64_t chunk_size, int* bytes_read) {
	// For file handles from the cache, any of the below file operations may fail
	// due to a bad file handle.
	if (FLAGS_use_hdfs_pread) {
	*bytes_read = hdfsPread(hdfs_fs_, hdfs_file, position_in_file, buffer, chunk_size);
	if (*bytes_read == -1) {
	return Status(TErrorCode::DISK_IO_ERROR,
	GetHdfsErrorMsg("Error reading from HDFS file: ",
	*scan_range_->file_string()));
	}
	} else {
	// If the file handle is borrowed, it may not be at the appropriate
	// location. Seek to the appropriate location.
	if (is_borrowed_fh) {
	if (hdfsSeek(hdfs_fs_, hdfs_file, position_in_file) != 0) {
	return Status(TErrorCode::DISK_IO_ERROR,
	Substitute("Error seeking to $0 in file: $1: $2",
	position_in_file, *scan_range_->file_string(), GetHdfsErrorMsg("")));
	}
	}
	*bytes_read = hdfsRead(hdfs_fs_, hdfs_file, buffer, chunk_size);
	if (*bytes_read == -1) {
	return Status(TErrorCode::DISK_IO_ERROR,
	GetHdfsErrorMsg("Error reading from HDFS file: ",
	*scan_range_->file_string()));
	}
	}
	return Status::OK();
	}

	void HdfsFileReader::CachedFile(uint8_t** data, int64_t* length) {
	{
	unique_lock<SpinLock> hdfs_lock(lock_);
	DCHECK(cached_buffer_ == nullptr);
	DCHECK(exclusive_hdfs_fh_ != nullptr);
	cached_buffer_ = hadoopReadZero(exclusive_hdfs_fh_->file(),
	scan_range_->io_mgr_->cached_read_options(), scan_range_->len());
	}
	if (cached_buffer_ == nullptr) {
	*data = nullptr;
	*length = 0;
	return;
	}
	data = reinterpret_cast<uint8_t>(
	const_cast<void*>(hadoopRzBufferGet(cached_buffer_)));
	*length = hadoopRzBufferLength(cached_buffer_);
	}

	void HdfsFileReader::Close() {
	unique_lock<SpinLock> hdfs_lock(lock_);
	if (exclusive_hdfs_fh_ != nullptr) {
	GetHdfsStatistics(exclusive_hdfs_fh_->file());

	if (cached_buffer_ != nullptr) {
	hadoopRzBufferFree(exclusive_hdfs_fh_->file(), cached_buffer_);
	cached_buffer_ = nullptr;
	}

	// Destroy the file handle.
	scan_range_->io_mgr_->ReleaseExclusiveHdfsFileHandle(exclusive_hdfs_fh_);
	exclusive_hdfs_fh_ = nullptr;
	ImpaladMetrics::IO_MGR_NUM_OPEN_FILES->Increment(-1L);
	}

	if (FLAGS_use_hdfs_pread && IsHdfsPath(scan_range_->file())) {
	// Update Hedged Read Metrics.
	// We call it only if the --use_hdfs_pread flag is set, to avoid having the
	// libhdfs client malloc and free a hdfsHedgedReadMetrics object unnecessarily
	// otherwise. 'hedged_metrics' is only set upon success.
	// We also avoid calling hdfsGetHedgedReadMetrics() when the file is not on HDFS
	// (see HDFS-13417).
	struct hdfsHedgedReadMetrics* hedged_metrics;
	int success = hdfsGetHedgedReadMetrics(hdfs_fs_, &hedged_metrics);
	if (success == 0) {
	ImpaladMetrics::HEDGED_READ_OPS->SetValue(hedged_metrics->hedgedReadOps);
	ImpaladMetrics::HEDGED_READ_OPS_WIN->SetValue(
	hedged_metrics->hedgedReadOpsWin);
	hdfsFreeHedgedReadMetrics(hedged_metrics);
	}
	}

	if (num_remote_bytes_ > 0) {
	scan_range_->reader_->num_remote_ranges_.Add(1L);
	if (expected_local_) {
	scan_range_->reader_->unexpected_remote_bytes_.Add(num_remote_bytes_);
	VLOG_FILE << "Unexpected remote HDFS read of "
	<< PrettyPrinter::Print(num_remote_bytes_, TUnit::BYTES)
	<< " for file '" << *scan_range_->file_string() << "'";
	}
	}
	}

	void HdfsFileReader::GetHdfsStatistics(hdfsFile hdfs_file) {
	struct hdfsReadStatistics* stats;
	if (IsHdfsPath(scan_range_->file())) {
	int success = hdfsFileGetReadStatistics(hdfs_file, &stats);
	if (success == 0) {
	scan_range_->reader_->bytes_read_local_.Add(stats->totalLocalBytesRead);
	scan_range_->reader_->bytes_read_short_circuit_.Add(
	stats->totalShortCircuitBytesRead);
	scan_range_->reader_->bytes_read_dn_cache_.Add(stats->totalZeroCopyBytesRead);
	if (stats->totalLocalBytesRead != stats->totalBytesRead) {
	num_remote_bytes_ += stats->totalBytesRead - stats->totalLocalBytesRead;
	}
	hdfsFileFreeReadStatistics(stats);
	}
	hdfsFileClearReadStatistics(hdfs_file);
	}
	}

	void HdfsFileReader::ResetState() {
	FileReader::ResetState();
	num_remote_bytes_ = 0;
	}

	string HdfsFileReader::DebugString() const {
	return FileReader::DebugString() + Substitute(
	" exclusive_hdfs_fh=$0 num_remote_bytes=$1",
	exclusive_hdfs_fh_, num_remote_bytes_);
	}

	}
	}