Upgrade delta-kernel-rs to 0.5

[PatrickJin-db]

Existing python tests should be sufficient for correctness. Is any performance testing necessary?

[PatrickJin-db]

Tried to avoid doing this but ran into lifetime issues without the collect().into_iter(). Though I don't think this should be a performance regression either since the execute returned a Vec in the previous delta-kernel-rs version so it was already copying.

[zachschuermann]

what's the error with just results.map(...)?

[OussamaSaoudi-db]

So the way we currently do scan::execute is it takes a reference to the engine, and returns an iterator with the same lifetime as the engine. They share lifetimes because you need a reference to the engine for the duration of iteration for all the processing.

One solution is to map, then collect at the very end:

        let result_iter = self.0.execute(engine_interface.0.as_ref())?;
        let record_batches: Vec<_> = result_iter
            .map(|res| {
                let scan_res = res.and_then(|res| Ok((res.full_mask(), res.raw_data?)));
                let (mask, data) =
                    scan_res.map_err(|e| ArrowError::from_external_error(Box::new(e)))?;
                let record_batch: RecordBatch = data
                    .into_any()
                    .downcast::<ArrowEngineData>()
                    .map_err(|_| {
                        ArrowError::CastError("Couldn't cast to ArrowEngineData".to_string())
                    })?
                    .into();
                if let Some(mask) = mask {
                    let filtered_batch = filter_record_batch(&record_batch, &mask.into())?;
                    Ok(filtered_batch)
                } else {
                    Ok(record_batch)
                }
            })
            .collect();
        let record_batch_iter = RecordBatchIterator::new(record_batches, result_schema);

[OussamaSaoudi-db]

I'd encourage you to submit an issue with delta-kernel-rs on the lifetime issue. I think it'll important to address eventually :)

[PatrickJin-db]

Thanks for the explanation and suggestion. The exact error was

error: lifetime may not live long enough
   --> src/lib.rs:115:9
    |
87  |         &self,
    |         - let's call the lifetime of this reference `'1`
...
115 |         Ok(PyArrowType(Box::new(record_batch_iter)))
    |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ returning this value requires that `'1` must outlive `'static`
    |
help: to declare that the trait object captures data from argument `self`, you can add an explicit `'_` lifetime bound
    |
89  |     ) -> DeltaPyResult<PyArrowType<Box<dyn RecordBatchReader + Send + '_>>> {
    |                                                                     ++++

error: lifetime may not live long enough
   --> src/lib.rs:115:9
    |
88  |         engine_interface: &PythonInterface,
    |                           - let's call the lifetime of this reference `'2`
...
115 |         Ok(PyArrowType(Box::new(record_batch_iter)))
    |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ returning this value requires that `'2` must outlive `'static`
    |
help: to declare that the trait object captures data from argument `engine_interface`, you can add an explicit `'_` lifetime bound
    |
89  |     ) -> DeltaPyResult<PyArrowType<Box<dyn RecordBatchReader + Send + '_>>> {
    |                                                                     ++++

error: could not compile `delta-kernel-rust-sharing-wrapper` (lib) due to 2 previous errors

[zachschuermann]

Hey @PatrickJin-db thanks for looping me in! happy to help. just wanted to flag we've just released version 0.5 so let's change to that and also do a minimal amount of debug on that iterator bit

[zachschuermann]

i think as a consumer it makes more sense picking a specific version

Suggested change

	arrow = { version = ">=53, <54", features = ["pyarrow"] }
	arrow = { version = "53.3.0", features = ["pyarrow"] }

[zachschuermann]

also FYI the ^ is implied

Suggested change

	delta_kernel = {version = "^0.4", features = ["cloud", "default", "default-engine"]}
	delta_kernel = {version = "0.5", features = ["cloud", "default", "default-engine"]}

[zachschuermann]

what's the error with just results.map(...)?

[zachschuermann]

LGTM with a follow-up to optimize that collect then into_iter()!

[zachschuermann]

re-stamping the last didn't seem to say stamped?

LGTM with a follow-up to optimize that collect then into_iter()!