Rust writer in operations makes a lot of data copies

[wjones127]

Environment

We keep the Parquet file being written as a Vec<u8>. The write_batch method seems to clone the vec each time it is called, and the batch size for many operations is quite small, defaulting at 1024 rows.

delta-rs/rust/src/operations/writer.rs

Lines 326 to 343 in 930d16e

	fn write_batch(&mut self, batch: &RecordBatch) -> DeltaResult<()> {
	// copy current cursor bytes so we can recover from failures
	// TODO is copying this something we should be doing?
	let buffer_bytes = self.buffer.to_vec();
	match self.arrow_writer.write(batch) {
	Ok(_) => {
	apply_null_counts(&batch.clone().into(), &mut self.null_counts, 0);
	Ok(())
	}
	Err(err) => {
	// if a write fails we need to reset the state of the PartitionWriter
	warn!("error writing to arrow buffer, resetting writer state.");
	self.replace_arrow_buffer(buffer_bytes)?;
	Err(err.into())
	}
	}
	}

@roeap Do you remember why we do this? I'm wondering whether we can just rip out this error handling, or if we actually do some sort of retry with this.

[roeap]

I do remember wondering the same thing (hence the TODO 😆). This just got carried over from previous implementations and I think originated somewhere in kafaka-delta ingest. I guess back then the write API exposed was much lower level - i.e. user would have to use the writer structs directly.

In our codebase here I think we are not retrying anything. Also did a quick dive into what errors we may actually see there, and I think we either have an IO issue (where we do retry in object store) or somehow malformed data (did not go too deep into the parquet crate). So my vote would be for ripping it out.

[wjones127]

Okay. I'm going to refactor that module now.

[roeap]

I just opened #1396, which may be relevant in that context, as it also affects the write path.