| 150 | : src_(src), remaining_(limit), pool_(pool) {} |
| 151 | |
| 152 | ::arrow::Result<int64_t> Read(int64_t nbytes, void* out) override { |
| 153 | if (closed_) |
| 154 | return ::arrow::Status::IOError("read on closed"); |
| 155 | if (nbytes <= 0 || remaining_ == 0) |
| 156 | return int64_t{0}; |
| 157 | |
| 158 | int64_t toRead = std::min<int64_t>(nbytes, remaining_); |
| 159 | auto* p = static_cast<uint8_t*>(out); |
| 160 | int64_t copied = 0; |
| 161 | |
| 162 | while (toRead > 0) { |
| 163 | auto view = src_->nextView(static_cast<int32_t>(toRead)); |
| 164 | if (view.size() > 0) { |
| 165 | std::memcpy(p, view.data(), view.size()); |
| 166 | p += view.size(); |
| 167 | copied += view.size(); |
| 168 | toRead -= view.size(); |
| 169 | pos_ += view.size(); |
| 170 | remaining_ -= view.size(); |
| 171 | continue; |
| 172 | } |
| 173 | |
| 174 | if (src_->atEnd()) { |
| 175 | break; |
| 176 | } |
| 177 | |
| 178 | int64_t chunk = std::min<int64_t>(toRead, 64 * 1024); |
| 179 | src_->readBytes(p, static_cast<int32_t>(chunk)); |
| 180 | p += chunk; |
| 181 | copied += chunk; |
| 182 | toRead -= chunk; |
| 183 | pos_ += chunk; |
| 184 | remaining_ -= chunk; |
| 185 | } |
| 186 | |
| 187 | return copied; |
| 188 | } |
| 189 | |
| 190 | ::arrow::Result<std::shared_ptr<::arrow::Buffer>> Read( |
| 191 | int64_t nbytes) override { |