Test upgrade to arrow 7.x with new prost/tonic

[alamb]

Rationale for this change

Test what would happen if we upgraded prost and tonic dependencies in the arrow-rs 6.x line (aka what would happen if we merged apache/arrow-rs#945)?

What changes are included in this PR?

temporarily override arrow to use apache/arrow-rs#945

Are there any user-facing changes?

No

[alamb]

When upgrading arrow only, the build fails like this

   Compiling datafusion v5.1.0 (/Users/alamb/Software/arrow-datafusion/datafusion)
error[E0308]: mismatched types
  --> ballista/rust/core/src/client.rs:75:19
   |
75 |         Ok(Self { flight_client })
   |                   ^^^^^^^^^^^^^ expected struct `Channel`, found struct `tonic::transport::channel::Channel`
   |
   = note: expected struct `FlightServiceClient<Channel>`
              found struct `FlightServiceClient<tonic::transport::channel::Channel>`
   = note: perhaps two different versions of crate `tonic` are being used?

error[E0599]: the method `do_get` exists for struct `FlightServiceClient<Channel>`, but its trait bounds were not satisfied
   --> ballista/rust/core/src/client.rs:112:14
    |
112 |             .do_get(request)
    |              ^^^^^^ method cannot be called on `FlightServiceClient<Channel>` due to unsatisfied trait bounds
    |
   ::: /Users/alamb/.cargo/registry/src/github.com-1ecc6299db9ec823/tonic-0.5.2/src/transport/channel/mod.rs:68:1
    |
68  | pub struct Channel {
    | ------------------
    | |
    | doesn't satisfy `<_ as Service<tonic::codegen::http::Request<http_body::combinators::box_body::UnsyncBoxBody<prost::bytes::Bytes, tonic::status::Status>>>>::Response = tonic::codegen::http::Response<_>`
    | doesn't satisfy `_: Service<tonic::codegen::http::Request<http_body::combinators::box_body::UnsyncBoxBody<prost::bytes::Bytes, tonic::status::Status>>>`
    | doesn't satisfy `_: tonic::client::service::GrpcService<http_body::combinators::box_body::UnsyncBoxBody<prost::bytes::Bytes, tonic::status::Status>>`
    |
    = note: the following trait bounds were not satisfied:
            `Channel: tonic::client::service::GrpcService<http_body::combinators::box_body::UnsyncBoxBody<prost::bytes::Bytes, tonic::status::Status>>`
            `<Channel as Service<tonic::codegen::http::Request<http_body::combinators::box_body::UnsyncBoxBody<prost::bytes::Bytes, tonic::status::Status>>>>::Response = tonic::codegen::http::Response<_>`
            `Channel: Service<tonic::codegen::http::Request<http_body::combinators::box_body::UnsyncBoxBody<prost::bytes::Bytes, tonic::status::Status>>>`

[alamb]

When I also update prost and tonic in the Cargo.toml` in the ballista projects, now it fails like this:

cd /Users/alamb/Software/arrow-datafusion && CARGO_TARGET_DIR=/Volumes/RAMDisk/df-target cargo test --all
   Compiling ballista-core v0.6.0 (/Users/alamb/Software/arrow-datafusion/ballista/rust/core)
error[E0277]: `(dyn tonic::codec::Decoder<Item = FlightData, Error = tonic::Status> + std::marker::Send + 'static)` cannot be shared between threads safely
   --> ballista/rust/core/src/client.rs:128:20
    |
128 |                 Ok(Box::pin(FlightDataStream::new(stream, schema)))
    |                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `(dyn tonic::codec::Decoder<Item = FlightData, Error = tonic::Status> + std::marker::Send + 'static)` cannot be shared between threads safely
    |
    = help: the trait `Sync` is not implemented for `(dyn tonic::codec::Decoder<Item = FlightData, Error = tonic::Status> + std::marker::Send + 'static)`
    = note: required because of the requirements on the impl of `Sync` for `std::ptr::Unique<(dyn tonic::codec::Decoder<Item = FlightData, Error = tonic::Status> + std::marker::Send + 'static)>`
    = note: required because it appears within the type `Box<(dyn tonic::codec::Decoder<Item = FlightData, Error = tonic::Status> + std::marker::Send + 'static)>`
    = note: required because it appears within the type `Streaming<FlightData>`
note: required because it appears within the type `FlightDataStream`
   --> ballista/rust/core/src/client.rs:137:8
    |
137 | struct FlightDataStream {
    |        ^^^^^^^^^^^^^^^^
    = note: required for the cast to the object type `dyn RecordBatchStream<Item = std::result::Result<datafusion::arrow::record_batch::RecordBatch, datafusion::arrow::error::ArrowError>> + Sync + std::marker::Send`

error[E0277]: `(dyn tonic::codegen::Body<Error = tonic::Status, Data = prost::bytes::Bytes> + std::marker::Send + 'static)` cannot be shared between threads safely
   --> ballista/rust/core/src/client.rs:128:20
    |
128 |                 Ok(Box::pin(FlightDataStream::new(stream, schema)))
    |                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `(dyn tonic::codegen::Body<Error = tonic::Status, Data = prost::bytes::Bytes> + std::marker::Send + 'static)` cannot be shared between threads safely
    |
    = help: the trait `Sync` is not implemented for `(dyn tonic::codegen::Body<Error = tonic::Status, Data = prost::bytes::Bytes> + std::marker::Send + 'static)`
    = note: required because of the requirements on the impl of `Sync` for `std::ptr::Unique<(dyn tonic::codegen::Body<Error = tonic::Status, Data = prost::bytes::Bytes> + std::marker::Send + 'static)>`
    = note: required because it appears within the type `Box<(dyn tonic::codegen::Body<Error = tonic::Status, Data = prost::bytes::Bytes> + std::marker::Send + 'static)>`
    = note: required because it appears within the type `Pin<Box<(dyn tonic::codegen::Body<Error = tonic::Status, Data = prost::bytes::Bytes> + std::marker::Send + 'static)>>`
    = note: required because it appears within the type `http_body::combinators::box_body::UnsyncBoxBody<prost::bytes::Bytes, tonic::Status>`
    = note: required because it appears within the type `Streaming<FlightData>`
note: required because it appears within the type `FlightDataStream`
   --> ballista/rust/core/src/client.rs:137:8
    |
137 | struct FlightDataStream {
    |        ^^^^^^^^^^^^^^^^
    = note: required for the cast to the object type `dyn RecordBatchStream<Item = std::result::Result<datafusion::arrow::record_batch::RecordBatch, datafusion::arrow::error::ArrowError>> + Sync + std::marker::Send`

For more information about this error, try `rustc --explain E0277`.
error: could not compile `ballista-core` due to 2 previous errors
warning: build failed, waiting for other jobs to finish...
error: build failed

(namely because FlightData is no longer marked Sync for some reason)

@tustvold any ideas?

"Fixed" (worked around) in f2cfa38

[alamb]

this is needed because after tonic/prost upgrade FlightData is no longer Sync. There are perhaps better ways to handle this

[houqp]

@alamb let us know if you need help with the ballista change.

[alamb]

@alamb let us know if you need help with the ballista change.

Thanks @houqp -- I think this PR is likely to be part of the "upgrade datafusion to use arrow 7.0.0" when that is released

If anyone has any good ideas about how to avoid the use of Mutex in FlightDataStream that would be awesome

[yjshen]

Facing with the same issue while updating #68 with arrow2 version 0.8 and arrow-format 0.3, since the same breaking change was introduced in DataEngineeringLabs/arrow-format#2.

Thanks @alamb I use your mutex way here as a workaround, please let me know if you have further great solutions.😄

[tustvold]

I did some poking around a while back and was going to write up an issue, but got pulled away to other things. The problem is that SendableRecordBatchStream has a Sync constraint on it. I think this is just a mistake in DataFusion, I can't see a compelling reason that being able to share immutable references to streams across threads is important, especially when stream consumption requires &mut.

This would, however, be a fairly annoying breaking change to DataFusion and I wanted to provide more context before filing a ticket, but ran out of time

[alamb]

Included in #1523