| 126 | } |
| 127 | |
| 128 | void HDFSRemoteFSReader::connectIfNeeded() { |
| 129 | if (hdfs_fs_ == nullptr) { |
| 130 | HDFSBuilderPtr builder = hdfs_params_.createBuilder(Poco::URI(hdfs_path_)); |
| 131 | hdfs_fs_ = createHDFSFS(builder.get()); |
| 132 | if (hdfs_fs_ == nullptr) { |
| 133 | throw Exception("Unable to connect to hdfs with " + hdfs_params_.toString() |
| 134 | + ", Error: " + String(hdfsGetLastError()), ErrorCodes::HDFS_ERROR); |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | if (hdfs_file_ == nullptr) { |
| 139 | for (int retry = 0; retry < 2 && hdfs_file_ == nullptr; ++retry) { |
| 140 | ProfileEvents::increment(ProfileEvents::HdfsFileOpen); |
| 141 | hdfs_file_ = hdfsOpenFile(hdfs_fs_.get(), hdfs_path_.c_str(), O_RDONLY, 0, 0, 0, ReadBufferFromHdfsCallBack); |
| 142 | } |
| 143 | |
| 144 | if (hdfs_file_ == nullptr) { |
| 145 | throw Exception("Unable to open file " + hdfs_path_ + ", Error: " |
| 146 | + String(hdfsGetLastError()), ErrorCodes::HDFS_ERROR); |
| 147 | } |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | std::unique_ptr<RemoteFSReader> HDFSRemoteFSReaderOpts::create(const String& path) { |
| 152 | return std::make_unique<HDFSRemoteFSReader>(path, pread_, hdfs_params_); |
nothing calls this directly
no test coverage detected