Skip to content

remove sync constraint of SendableRecordBatchStream #1884

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alamb merged 2 commits into from
Mar 2, 2022
Merged

remove sync constraint of SendableRecordBatchStream #1884

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9884594
remove sync constraint of SendableRecordBatchStream
doki23 Feb 25, 2022
d5b7af8
Merge branch 'apache:master' into master
doki23 Mar 2, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ballista/rust/core/src/execution_plans/shuffle_reader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ fn stats_for_partitions(

async fn fetch_partition(
location: &PartitionLocation,
) -> Result<Pin<Box<dyn RecordBatchStream + Send + Sync>>> {
) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
let metadata = &location.executor_meta;
let partition_id = &location.partition_id;
let mut ballista_client =
Expand Down
8 changes: 4 additions & 4 deletions ballista/rust/core/src/utils.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ use futures::{Stream, StreamExt};
/// Stream data to disk in Arrow IPC format

pub async fn write_stream_to_disk(
stream: &mut Pin<Box<dyn RecordBatchStream + Send + Sync>>,
stream: &mut Pin<Box<dyn RecordBatchStream + Send>>,
path: &str,
disk_write_metric: &metrics::Time,
) -> Result<PartitionStats> {
Expand Down Expand Up @@ -98,7 +98,7 @@ pub async fn write_stream_to_disk(
}

pub async fn collect_stream(
stream: &mut Pin<Box<dyn RecordBatchStream + Send + Sync>>,
stream: &mut Pin<Box<dyn RecordBatchStream + Send>>,
) -> Result<Vec<RecordBatch>> {
let mut batches = vec![];
while let Some(batch) = stream.next().await {
Expand Down Expand Up @@ -310,13 +310,13 @@ impl<T: 'static + AsLogicalPlan> QueryPlanner for BallistaQueryPlanner<T> {
}

pub struct WrappedStream {
stream: Pin<Box<dyn Stream<Item = ArrowResult<RecordBatch>> + Send + Sync>>,
stream: Pin<Box<dyn Stream<Item = ArrowResult<RecordBatch>> + Send>>,
schema: SchemaRef,
}

impl WrappedStream {
pub fn new(
stream: Pin<Box<dyn Stream<Item = ArrowResult<RecordBatch>> + Send + Sync>>,
stream: Pin<Box<dyn Stream<Item = ArrowResult<RecordBatch>> + Send>>,
schema: SchemaRef,
) -> Self {
Self { stream, schema }
Expand Down
2 changes: 1 addition & 1 deletion datafusion/src/physical_plan/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ pub trait RecordBatchStream: Stream<Item = ArrowResult<RecordBatch>> {
}

/// Trait for a stream of record batches.
pub type SendableRecordBatchStream = Pin<Box<dyn RecordBatchStream + Send + Sync>>;
pub type SendableRecordBatchStream = Pin<Box<dyn RecordBatchStream + Send>>;

/// EmptyRecordBatchStream can be used to create a RecordBatchStream
/// that will produce no results
Expand Down