Struct arrow_flight::sql::server::PeekableFlightDataStream
source · pub struct PeekableFlightDataStream { /* private fields */ }
Expand description
A wrapper around Streaming<FlightData>
that allows “peeking” at the
message at the front of the stream without consuming it.
This is needed because sometimes the first message in the stream will contain
a FlightDescriptor
in addition to potentially any data, and the dispatch logic
must inspect this information.
§Example
PeekableFlightDataStream::peek
can be used to peek at the first message without
discarding it; otherwise, PeekableFlightDataStream
can be used as a regular stream.
See the following example:
use arrow_array::RecordBatch;
use arrow_flight::decode::FlightRecordBatchStream;
use arrow_flight::FlightDescriptor;
use arrow_flight::error::FlightError;
use arrow_flight::sql::server::PeekableFlightDataStream;
use tonic::{Request, Status};
use futures::TryStreamExt;
#[tokio::main]
async fn main() -> Result<(), Status> {
let request: Request<PeekableFlightDataStream> = todo!();
let stream: PeekableFlightDataStream = request.into_inner();
// The first message contains the flight descriptor and the schema.
// Read the flight descriptor without discarding the schema:
let flight_descriptor: FlightDescriptor = stream
.peek()
.await
.cloned()
.transpose()?
.and_then(|data| data.flight_descriptor)
.expect("first message should contain flight descriptor");
// Pass the stream through a decoder
let batches: Vec<RecordBatch> = FlightRecordBatchStream::new_from_flight_data(
request.into_inner().map_err(|e| e.into()),
)
.try_collect()
.await?;
}
Implementations§
source§impl PeekableFlightDataStream
impl PeekableFlightDataStream
sourcepub fn into_inner(self) -> Streaming<FlightData>
pub fn into_inner(self) -> Streaming<FlightData>
Convert this stream into a Streaming<FlightData>
.
Any messages observed through Self::peek
will be lost
after the conversion.
sourcepub fn into_peekable(self) -> Peekable<Streaming<FlightData>>
pub fn into_peekable(self) -> Peekable<Streaming<FlightData>>
Convert this stream into a Peekable<Streaming<FlightData>>
.
Preserves the state of the stream, so that calls to Self::peek
and Self::poll_next
are the same.
Trait Implementations§
source§impl Stream for PeekableFlightDataStream
impl Stream for PeekableFlightDataStream
§type Item = Result<FlightData, Status>
type Item = Result<FlightData, Status>
Auto Trait Implementations§
impl !Freeze for PeekableFlightDataStream
impl !RefUnwindSafe for PeekableFlightDataStream
impl Send for PeekableFlightDataStream
impl !Sync for PeekableFlightDataStream
impl Unpin for PeekableFlightDataStream
impl !UnwindSafe for PeekableFlightDataStream
Blanket Implementations§
source§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
impl<T> BorrowMut<T> for Twhere
T: ?Sized,
source§fn borrow_mut(&mut self) -> &mut T
fn borrow_mut(&mut self) -> &mut T
source§impl<T> Instrument for T
impl<T> Instrument for T
source§fn instrument(self, span: Span) -> Instrumented<Self>
fn instrument(self, span: Span) -> Instrumented<Self>
source§fn in_current_span(self) -> Instrumented<Self>
fn in_current_span(self) -> Instrumented<Self>
source§impl<T> IntoRequest<T> for T
impl<T> IntoRequest<T> for T
source§fn into_request(self) -> Request<T>
fn into_request(self) -> Request<T>
T
in a tonic::Request
source§impl<T> IntoStreamingRequest for T
impl<T> IntoStreamingRequest for T
source§impl<St> StreamExt for St
impl<St> StreamExt for St
source§fn next(&mut self) -> Next<'_, Self>where
Self: Unpin,
fn next(&mut self) -> Next<'_, Self>where
Self: Unpin,
None
if the
stream is finished. Read moresource§fn try_next<T, E>(&mut self) -> TryNext<'_, Self>
fn try_next<T, E>(&mut self) -> TryNext<'_, Self>
source§fn map<T, F>(self, f: F) -> Map<Self, F>
fn map<T, F>(self, f: F) -> Map<Self, F>
source§fn map_while<T, F>(self, f: F) -> MapWhile<Self, F>
fn map_while<T, F>(self, f: F) -> MapWhile<Self, F>
None
. Read moresource§fn then<F, Fut>(self, f: F) -> Then<Self, Fut, F>
fn then<F, Fut>(self, f: F) -> Then<Self, Fut, F>
source§fn merge<U>(self, other: U) -> Merge<Self, U>
fn merge<U>(self, other: U) -> Merge<Self, U>
source§fn filter<F>(self, f: F) -> Filter<Self, F>
fn filter<F>(self, f: F) -> Filter<Self, F>
source§fn filter_map<T, F>(self, f: F) -> FilterMap<Self, F>
fn filter_map<T, F>(self, f: F) -> FilterMap<Self, F>
source§fn fuse(self) -> Fuse<Self>where
Self: Sized,
fn fuse(self) -> Fuse<Self>where
Self: Sized,
None
. Read moresource§fn take(self, n: usize) -> Take<Self>where
Self: Sized,
fn take(self, n: usize) -> Take<Self>where
Self: Sized,
n
items of the underlying stream. Read moresource§fn take_while<F>(self, f: F) -> TakeWhile<Self, F>
fn take_while<F>(self, f: F) -> TakeWhile<Self, F>
true
. Read moresource§fn skip(self, n: usize) -> Skip<Self>where
Self: Sized,
fn skip(self, n: usize) -> Skip<Self>where
Self: Sized,
n
first items of the
underlying stream. Read moresource§fn skip_while<F>(self, f: F) -> SkipWhile<Self, F>
fn skip_while<F>(self, f: F) -> SkipWhile<Self, F>
true
. Read moresource§fn all<F>(&mut self, f: F) -> AllFuture<'_, Self, F>
fn all<F>(&mut self, f: F) -> AllFuture<'_, Self, F>
source§fn any<F>(&mut self, f: F) -> AnyFuture<'_, Self, F>
fn any<F>(&mut self, f: F) -> AnyFuture<'_, Self, F>
source§fn chain<U>(self, other: U) -> Chain<Self, U>
fn chain<U>(self, other: U) -> Chain<Self, U>
source§fn fold<B, F>(self, init: B, f: F) -> FoldFuture<Self, B, F>
fn fold<B, F>(self, init: B, f: F) -> FoldFuture<Self, B, F>
source§fn collect<T>(self) -> Collect<Self, T>
fn collect<T>(self) -> Collect<Self, T>
source§fn timeout(self, duration: Duration) -> Timeout<Self>where
Self: Sized,
fn timeout(self, duration: Duration) -> Timeout<Self>where
Self: Sized,
source§fn timeout_repeating(self, interval: Interval) -> TimeoutRepeating<Self>where
Self: Sized,
fn timeout_repeating(self, interval: Interval) -> TimeoutRepeating<Self>where
Self: Sized,
source§fn throttle(self, duration: Duration) -> Throttle<Self>where
Self: Sized,
fn throttle(self, duration: Duration) -> Throttle<Self>where
Self: Sized,
source§impl<T> StreamExt for T
impl<T> StreamExt for T
source§fn next(&mut self) -> Next<'_, Self>where
Self: Unpin,
fn next(&mut self) -> Next<'_, Self>where
Self: Unpin,
source§fn into_future(self) -> StreamFuture<Self>
fn into_future(self) -> StreamFuture<Self>
source§fn map<T, F>(self, f: F) -> Map<Self, F>
fn map<T, F>(self, f: F) -> Map<Self, F>
source§fn enumerate(self) -> Enumerate<Self>where
Self: Sized,
fn enumerate(self) -> Enumerate<Self>where
Self: Sized,
source§fn filter<Fut, F>(self, f: F) -> Filter<Self, Fut, F>
fn filter<Fut, F>(self, f: F) -> Filter<Self, Fut, F>
source§fn filter_map<Fut, T, F>(self, f: F) -> FilterMap<Self, Fut, F>
fn filter_map<Fut, T, F>(self, f: F) -> FilterMap<Self, Fut, F>
source§fn then<Fut, F>(self, f: F) -> Then<Self, Fut, F>
fn then<Fut, F>(self, f: F) -> Then<Self, Fut, F>
source§fn collect<C>(self) -> Collect<Self, C>
fn collect<C>(self) -> Collect<Self, C>
source§fn unzip<A, B, FromA, FromB>(self) -> Unzip<Self, FromA, FromB>
fn unzip<A, B, FromA, FromB>(self) -> Unzip<Self, FromA, FromB>
source§fn concat(self) -> Concat<Self>
fn concat(self) -> Concat<Self>
source§fn count(self) -> Count<Self>where
Self: Sized,
fn count(self) -> Count<Self>where
Self: Sized,
source§fn fold<T, Fut, F>(self, init: T, f: F) -> Fold<Self, Fut, T, F>
fn fold<T, Fut, F>(self, init: T, f: F) -> Fold<Self, Fut, T, F>
source§fn any<Fut, F>(self, f: F) -> Any<Self, Fut, F>
fn any<Fut, F>(self, f: F) -> Any<Self, Fut, F>
true
if any element in stream satisfied a predicate. Read moresource§fn all<Fut, F>(self, f: F) -> All<Self, Fut, F>
fn all<Fut, F>(self, f: F) -> All<Self, Fut, F>
true
if all element in stream satisfied a predicate. Read moresource§fn flatten(self) -> Flatten<Self>
fn flatten(self) -> Flatten<Self>
source§fn flatten_unordered(
self,
limit: impl Into<Option<usize>>,
) -> FlattenUnorderedWithFlowController<Self, ()>
fn flatten_unordered( self, limit: impl Into<Option<usize>>, ) -> FlattenUnorderedWithFlowController<Self, ()>
source§fn flat_map_unordered<U, F>(
self,
limit: impl Into<Option<usize>>,
f: F,
) -> FlatMapUnordered<Self, U, F>
fn flat_map_unordered<U, F>( self, limit: impl Into<Option<usize>>, f: F, ) -> FlatMapUnordered<Self, U, F>
StreamExt::map
but flattens nested Stream
s
and polls them concurrently, yielding items in any order, as they made
available. Read moresource§fn scan<S, B, Fut, F>(self, initial_state: S, f: F) -> Scan<Self, S, Fut, F>
fn scan<S, B, Fut, F>(self, initial_state: S, f: F) -> Scan<Self, S, Fut, F>
StreamExt::fold
that holds internal state
and produces a new stream. Read moresource§fn skip_while<Fut, F>(self, f: F) -> SkipWhile<Self, Fut, F>
fn skip_while<Fut, F>(self, f: F) -> SkipWhile<Self, Fut, F>
true
. Read moresource§fn take_while<Fut, F>(self, f: F) -> TakeWhile<Self, Fut, F>
fn take_while<Fut, F>(self, f: F) -> TakeWhile<Self, Fut, F>
true
. Read moresource§fn take_until<Fut>(self, fut: Fut) -> TakeUntil<Self, Fut>
fn take_until<Fut>(self, fut: Fut) -> TakeUntil<Self, Fut>
source§fn for_each<Fut, F>(self, f: F) -> ForEach<Self, Fut, F>
fn for_each<Fut, F>(self, f: F) -> ForEach<Self, Fut, F>
source§fn for_each_concurrent<Fut, F>(
self,
limit: impl Into<Option<usize>>,
f: F,
) -> ForEachConcurrent<Self, Fut, F>
fn for_each_concurrent<Fut, F>( self, limit: impl Into<Option<usize>>, f: F, ) -> ForEachConcurrent<Self, Fut, F>
source§fn take(self, n: usize) -> Take<Self>where
Self: Sized,
fn take(self, n: usize) -> Take<Self>where
Self: Sized,
n
items of the underlying stream. Read moresource§fn skip(self, n: usize) -> Skip<Self>where
Self: Sized,
fn skip(self, n: usize) -> Skip<Self>where
Self: Sized,
n
items of the underlying stream. Read moresource§fn boxed<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + Send + 'a>>
fn boxed<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + Send + 'a>>
source§fn boxed_local<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + 'a>>where
Self: Sized + 'a,
fn boxed_local<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + 'a>>where
Self: Sized + 'a,
source§fn buffered(self, n: usize) -> Buffered<Self>
fn buffered(self, n: usize) -> Buffered<Self>
source§fn buffer_unordered(self, n: usize) -> BufferUnordered<Self>
fn buffer_unordered(self, n: usize) -> BufferUnordered<Self>
source§fn zip<St>(self, other: St) -> Zip<Self, St>
fn zip<St>(self, other: St) -> Zip<Self, St>
source§fn peekable(self) -> Peekable<Self>where
Self: Sized,
fn peekable(self) -> Peekable<Self>where
Self: Sized,
peek
method. Read moresource§fn chunks(self, capacity: usize) -> Chunks<Self>where
Self: Sized,
fn chunks(self, capacity: usize) -> Chunks<Self>where
Self: Sized,
source§fn ready_chunks(self, capacity: usize) -> ReadyChunks<Self>where
Self: Sized,
fn ready_chunks(self, capacity: usize) -> ReadyChunks<Self>where
Self: Sized,
source§fn forward<S>(self, sink: S) -> Forward<Self, S>
fn forward<S>(self, sink: S) -> Forward<Self, S>
source§fn split<Item>(self) -> (SplitSink<Self, Item>, SplitStream<Self>)
fn split<Item>(self) -> (SplitSink<Self, Item>, SplitStream<Self>)
source§fn inspect<F>(self, f: F) -> Inspect<Self, F>
fn inspect<F>(self, f: F) -> Inspect<Self, F>
source§fn left_stream<B>(self) -> Either<Self, B>
fn left_stream<B>(self) -> Either<Self, B>
source§fn right_stream<B>(self) -> Either<B, Self>
fn right_stream<B>(self) -> Either<B, Self>
source§fn poll_next_unpin(&mut self, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>where
Self: Unpin,
fn poll_next_unpin(&mut self, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>where
Self: Unpin,
Stream::poll_next
on Unpin
stream types.source§fn select_next_some(&mut self) -> SelectNextSome<'_, Self>where
Self: Unpin + FusedStream,
fn select_next_some(&mut self) -> SelectNextSome<'_, Self>where
Self: Unpin + FusedStream,
source§impl<S> TryStreamExt for S
impl<S> TryStreamExt for S
source§fn err_into<E>(self) -> ErrInto<Self, E>
fn err_into<E>(self) -> ErrInto<Self, E>
source§fn map_ok<T, F>(self, f: F) -> MapOk<Self, F>
fn map_ok<T, F>(self, f: F) -> MapOk<Self, F>
source§fn map_err<E, F>(self, f: F) -> MapErr<Self, F>
fn map_err<E, F>(self, f: F) -> MapErr<Self, F>
source§fn and_then<Fut, F>(self, f: F) -> AndThen<Self, Fut, F>
fn and_then<Fut, F>(self, f: F) -> AndThen<Self, Fut, F>
f
. Read moresource§fn or_else<Fut, F>(self, f: F) -> OrElse<Self, Fut, F>
fn or_else<Fut, F>(self, f: F) -> OrElse<Self, Fut, F>
f
. Read moresource§fn inspect_ok<F>(self, f: F) -> InspectOk<Self, F>
fn inspect_ok<F>(self, f: F) -> InspectOk<Self, F>
source§fn inspect_err<F>(self, f: F) -> InspectErr<Self, F>
fn inspect_err<F>(self, f: F) -> InspectErr<Self, F>
source§fn into_stream(self) -> IntoStream<Self>where
Self: Sized,
fn into_stream(self) -> IntoStream<Self>where
Self: Sized,
source§fn try_next(&mut self) -> TryNext<'_, Self>where
Self: Unpin,
fn try_next(&mut self) -> TryNext<'_, Self>where
Self: Unpin,
source§fn try_for_each<Fut, F>(self, f: F) -> TryForEach<Self, Fut, F>
fn try_for_each<Fut, F>(self, f: F) -> TryForEach<Self, Fut, F>
source§fn try_skip_while<Fut, F>(self, f: F) -> TrySkipWhile<Self, Fut, F>
fn try_skip_while<Fut, F>(self, f: F) -> TrySkipWhile<Self, Fut, F>
true
. Read moresource§fn try_take_while<Fut, F>(self, f: F) -> TryTakeWhile<Self, Fut, F>
fn try_take_while<Fut, F>(self, f: F) -> TryTakeWhile<Self, Fut, F>
true
. Read moresource§fn try_for_each_concurrent<Fut, F>(
self,
limit: impl Into<Option<usize>>,
f: F,
) -> TryForEachConcurrent<Self, Fut, F>
fn try_for_each_concurrent<Fut, F>( self, limit: impl Into<Option<usize>>, f: F, ) -> TryForEachConcurrent<Self, Fut, F>
source§fn try_collect<C>(self) -> TryCollect<Self, C>
fn try_collect<C>(self) -> TryCollect<Self, C>
source§fn try_chunks(self, capacity: usize) -> TryChunks<Self>where
Self: Sized,
fn try_chunks(self, capacity: usize) -> TryChunks<Self>where
Self: Sized,
source§fn try_ready_chunks(self, capacity: usize) -> TryReadyChunks<Self>where
Self: Sized,
fn try_ready_chunks(self, capacity: usize) -> TryReadyChunks<Self>where
Self: Sized,
source§fn try_filter<Fut, F>(self, f: F) -> TryFilter<Self, Fut, F>
fn try_filter<Fut, F>(self, f: F) -> TryFilter<Self, Fut, F>
source§fn try_filter_map<Fut, F, T>(self, f: F) -> TryFilterMap<Self, Fut, F>
fn try_filter_map<Fut, F, T>(self, f: F) -> TryFilterMap<Self, Fut, F>
source§fn try_flatten_unordered(
self,
limit: impl Into<Option<usize>>,
) -> TryFlattenUnordered<Self>
fn try_flatten_unordered( self, limit: impl Into<Option<usize>>, ) -> TryFlattenUnordered<Self>
source§fn try_flatten(self) -> TryFlatten<Self>
fn try_flatten(self) -> TryFlatten<Self>
source§fn try_fold<T, Fut, F>(self, init: T, f: F) -> TryFold<Self, Fut, T, F>
fn try_fold<T, Fut, F>(self, init: T, f: F) -> TryFold<Self, Fut, T, F>
source§fn try_concat(self) -> TryConcat<Self>
fn try_concat(self) -> TryConcat<Self>
source§fn try_buffer_unordered(self, n: usize) -> TryBufferUnordered<Self>
fn try_buffer_unordered(self, n: usize) -> TryBufferUnordered<Self>
source§fn try_buffered(self, n: usize) -> TryBuffered<Self>
fn try_buffered(self, n: usize) -> TryBuffered<Self>
source§fn try_poll_next_unpin(
&mut self,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Ok, Self::Error>>>where
Self: Unpin,
fn try_poll_next_unpin(
&mut self,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Ok, Self::Error>>>where
Self: Unpin,
TryStream::try_poll_next
on Unpin
stream types.source§fn try_all<Fut, F>(self, f: F) -> TryAll<Self, Fut, F>
fn try_all<Fut, F>(self, f: F) -> TryAll<Self, Fut, F>
Err
is encountered or if an Ok
item is found
that does not satisfy the predicate. Read more