feat: avoid some data copies on writes

Author: roeap

Cargo.lock

@@ -1,5 +1,2 @@
 [workspace]
 members = ["object-store"]
-
-[patch.crates-io]
-object_store = { git = "https://github.com/apache/arrow-rs", rev = "28a04db03cea376991e7efb04c3cf4f71f6d05bf" }

Cargo.toml

@@ -10,7 +10,7 @@ async-trait = "0.1.57"
 bytes = "1.2.1"
 futures = "0.3"
 once_cell = "1.12.0"
-object_store = { version = "0.5.2", features = [
+object_store = { version = "0.5.3", features = [
   "azure",
   "aws",
   "gcp",

object-store/Cargo.toml

@@ -396,15 +396,14 @@ impl ObjectInputFile {
         };
         let nbytes = (range.end - range.start) as i64;
         self.pos += nbytes;
-        let obj = if nbytes > 0 {
+        let data = if nbytes > 0 {
             self.rt
                 .block_on(self.store.get_range(&self.path, range))
                 .map_err(ObjectStoreError::from)?
-                .to_vec()
         } else {
-            Vec::new()
+            "".into()
         };
-        Python::with_gil(|py| Ok(PyBytes::new(py, &obj).into_py(py)))
+        Python::with_gil(|py| Ok(PyBytes::new(py, data.as_ref()).into_py(py)))
     }
 
     fn fileno(&self) -> PyResult<()> {
@@ -525,10 +524,10 @@ impl ObjectOutputStream {
         Err(PyNotImplementedError::new_err("'read' not implemented"))
     }
 
-    fn write(&mut self, data: Vec<u8>) -> PyResult<i64> {
+    fn write(&mut self, data: &PyBytes) -> PyResult<i64> {
         self.check_closed()?;
-        let len = data.len() as i64;
-        match self.rt.block_on(self.writer.write_all(&data)) {
+        let len = data.as_bytes().len() as i64;
+        match self.rt.block_on(self.writer.write_all(data.as_bytes())) {
             Ok(_) => Ok(len),
             Err(err) => {
                 self.rt