| 1828 | } |
| 1829 | |
| 1830 | Status DoWrite(const void* data, int64_t nbytes, |
| 1831 | std::shared_ptr<Buffer> owned_buffer = nullptr) { |
| 1832 | if (closed_) { |
| 1833 | return Status::Invalid("Operation on closed stream"); |
| 1834 | } |
| 1835 | |
| 1836 | const int8_t* data_ptr = reinterpret_cast<const int8_t*>(data); |
| 1837 | auto advance_ptr = [&data_ptr, &nbytes](const int64_t offset) { |
| 1838 | data_ptr += offset; |
| 1839 | nbytes -= offset; |
| 1840 | }; |
| 1841 | |
| 1842 | // Handle case where we have some bytes buffered from prior calls. |
| 1843 | if (current_part_size_ > 0) { |
| 1844 | // Try to fill current buffer |
| 1845 | const int64_t to_copy = std::min(nbytes, kPartUploadSize - current_part_size_); |
| 1846 | RETURN_NOT_OK(current_part_->Write(data_ptr, to_copy)); |
| 1847 | current_part_size_ += to_copy; |
| 1848 | advance_ptr(to_copy); |
| 1849 | pos_ += to_copy; |
| 1850 | |
| 1851 | // If buffer isn't full, break |
| 1852 | if (current_part_size_ < kPartUploadSize) { |
| 1853 | return Status::OK(); |
| 1854 | } |
| 1855 | |
| 1856 | // Upload current buffer. We're only reaching this point if we have accumulated |
| 1857 | // enough data to upload. |
| 1858 | RETURN_NOT_OK(CommitCurrentPart()); |
| 1859 | } |
| 1860 | |
| 1861 | // We can upload chunks without copying them into a buffer |
| 1862 | while (nbytes >= kPartUploadSize) { |
| 1863 | RETURN_NOT_OK(UploadPart(data_ptr, kPartUploadSize)); |
| 1864 | advance_ptr(kPartUploadSize); |
| 1865 | pos_ += kPartUploadSize; |
| 1866 | } |
| 1867 | |
| 1868 | // Buffer remaining bytes |
| 1869 | if (nbytes > 0) { |
| 1870 | current_part_size_ = nbytes; |
| 1871 | ARROW_ASSIGN_OR_RAISE(current_part_, io::BufferOutputStream::Create( |
| 1872 | kPartUploadSize, io_context_.pool())); |
| 1873 | RETURN_NOT_OK(current_part_->Write(data_ptr, current_part_size_)); |
| 1874 | pos_ += current_part_size_; |
| 1875 | } |
| 1876 | |
| 1877 | return Status::OK(); |
| 1878 | } |
| 1879 | |
| 1880 | Status Flush() override { |
| 1881 | auto fut = FlushAsync(); |
nothing calls this directly
no test coverage detected