Use DataFusionError instead of ArrowError in SendableRecordBatchStream #5101

comphead · 2023-01-28T20:03:15Z

Which issue does this PR close?

Closes #5039.

Rationale for this change

See #5039

What changes are included in this PR?

Replace ArrowError with DataFusionError in DF context

Are these changes tested?

Yes

Are there any user-facing changes?

User will get DatafusionError::ArrowError instead arrow::ArrowError for datafusion contexts

comphead · 2023-01-28T20:04:00Z

@tustvold please check whenever you have time, the code rotting fast :)

tustvold

This looks good to me, mostly just some suggestions to use ? instead of map_err(Into::into) in a couple of places.

Given the intrusiveness of this change, I think we should leave this open for a few days in case there are any objections.

@alamb @andygrove

tustvold · 2023-01-28T21:41:32Z

datafusion/core/src/physical_plan/joins/nested_loop_join.rs

@@ -400,12 +399,9 @@ impl NestedLoopJoinStream {
 let mut left_indices_builder = UInt64Builder::new();
 let mut right_indices_builder = UInt32Builder::new();
 let left_right_indices = match indices_result {
- Err(_) => {
- // TODO why the type of result stream is `Result<T, ArrowError>`, and not the `DataFusionError`


tustvold · 2023-01-28T21:42:20Z

datafusion/core/src/physical_plan/joins/sort_merge_join.rs

+ .collect::<Result<Vec<_>, ArrowError>>()
+ .map_err(Into::<DataFusionError>::into)?;


Suggested change

.collect::<Result<Vec<_>, ArrowError>>()

.map_err(Into::<DataFusionError>::into)?;

.collect::<Result<Vec<_>, ArrowError>>()?;

tustvold · 2023-01-28T21:42:30Z

datafusion/core/src/physical_plan/joins/sort_merge_join.rs

+ .collect::<Result<Vec<_>, ArrowError>>()
+ .map_err(Into::<DataFusionError>::into)?


Suggested change

.collect::<Result<Vec<_>, ArrowError>>()

.map_err(Into::<DataFusionError>::into)?

.collect::<Result<Vec<_>, ArrowError>>()?

tustvold · 2023-01-28T21:44:46Z

datafusion/core/src/physical_plan/joins/utils.rs

@@ -821,7 +820,7 @@ pub(crate) fn build_batch_from_indices(
 };
 columns.push(array);
 }
- RecordBatch::try_new(Arc::new(schema.clone()), columns)
+ RecordBatch::try_new(Arc::new(schema.clone()), columns).map_err(Into::into)


Suggested change

RecordBatch::try_new(Arc::new(schema.clone()), columns).map_err(Into::into)

Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?)

tustvold · 2023-01-28T21:47:20Z

datafusion/core/src/physical_plan/file_format/file_stream.rs

+ .and_then(|b| {
+ self.pc_projector
+ .project(b, partition_values)
+ .map_err(|e| ArrowError::ExternalError(e.into()))


Is this needed, project already returns ArrowError AFAICT?

This is project from DataFusion, it returns DataFusionError

tustvold · 2023-01-28T21:49:27Z

datafusion/core/src/physical_plan/filter.rs

 .and_then(|array| {
 Ok(as_boolean_array(&array)?)
 // apply filter array to record batch
- .and_then(|filter_array| filter_record_batch(batch, filter_array))
+ .and_then(|filter_array| {
+ filter_record_batch(batch, filter_array).map_err(Into::into)


Suggested change

filter_record_batch(batch, filter_array).map_err(Into::into)

Ok(filter_record_batch(batch, filter_array)?)

tustvold · 2023-01-28T21:50:07Z

datafusion/core/src/physical_plan/sorts/sort.rs

+ )
+ .map_err(Into::<DataFusionError>::into)?;


Suggested change

)

.map_err(Into::<DataFusionError>::into)?;

)?;

tustvold · 2023-01-28T21:50:22Z

datafusion/core/src/physical_plan/windows/bounded_window_agg_exec.rs

@@ -462,7 +461,7 @@ impl SortedPartitionByBoundedWindowStream {
 if let Some(columns_to_show) = columns_to_show {
 let n_generated = columns_to_show[0].len();
 self.prune_state(n_generated)?;
- RecordBatch::try_new(schema, columns_to_show)
+ RecordBatch::try_new(schema, columns_to_show).map_err(Into::into)


Suggested change

RecordBatch::try_new(schema, columns_to_show).map_err(Into::into)

Ok(RecordBatch::try_new(schema, columns_to_show)?)

tustvold · 2023-01-28T21:50:37Z

datafusion/core/src/physical_plan/windows/window_agg_exec.rs

+ .collect::<Result<Vec<ArrayRef>, ArrowError>>()
+ .map_err(Into::<DataFusionError>::into)?;


Suggested change

.collect::<Result<Vec<ArrayRef>, ArrowError>>()

.map_err(Into::<DataFusionError>::into)?;

.collect::<Result<Vec<ArrayRef>, ArrowError>>()?;

tustvold · 2023-01-28T21:50:46Z

datafusion/core/src/physical_plan/windows/window_agg_exec.rs


 // combine with the original cols
 // note the setup of window aggregates is that they newly calculated window
 // expression results are always appended to the columns
 let mut batch_columns = batch.columns().to_vec();
 // calculate window cols
 batch_columns.extend_from_slice(&columns);
- RecordBatch::try_new(self.schema.clone(), batch_columns)
+ RecordBatch::try_new(self.schema.clone(), batch_columns).map_err(Into::into)


Suggested change

RecordBatch::try_new(self.schema.clone(), batch_columns).map_err(Into::into)

Ok(RecordBatch::try_new(self.schema.clone(), batch_columns)?)

comphead · 2023-01-29T00:38:21Z

Thanks @tustvold for your review. Fixed all the comments

jackwener · 2023-01-29T07:45:24Z

cc @liukun4515 , You may also be interested in this.

alamb · 2023-01-30T14:30:46Z

Also @yahoNanJing and the Ballista team, this may result in chrun

ursabot · 2023-01-30T15:33:54Z

Benchmark runs are scheduled for baseline = 9c8bdfe and contender = 74b05fa. 74b05fa is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] ec2-t3-xlarge-us-east-2
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] test-mac-arm
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] ursa-i9-9960x
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] ursa-thinkcentre-m75q
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java

Replace ArrowError with DataFusionError in DF context

3338b9c

github-actions bot added the core Core datafusion crate label Jan 28, 2023

tustvold approved these changes Jan 28, 2023

View reviewed changes

tustvold added the api change Changes the API exposed to users of the crate label Jan 28, 2023

tustvold changed the title ~~Replace ArrowError with DataFusionError in DF context~~ Use DataFusionError instead of ArrowError in SendableRecordBatchStream Jan 28, 2023

fix comments

a1f4b5a

alamb approved these changes Jan 30, 2023

View reviewed changes

tustvold merged commit 74b05fa into apache:master Jan 30, 2023

crepererum mentioned this pull request Jan 31, 2023

Error type in RecordBatchStream #4172

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use DataFusionError instead of ArrowError in SendableRecordBatchStream #5101

Use DataFusionError instead of ArrowError in SendableRecordBatchStream #5101

comphead commented Jan 28, 2023

comphead commented Jan 28, 2023

tustvold left a comment

tustvold Jan 28, 2023

tustvold Jan 28, 2023

tustvold Jan 28, 2023

tustvold Jan 28, 2023

tustvold Jan 28, 2023

comphead Jan 29, 2023

tustvold Jan 28, 2023

tustvold Jan 28, 2023

tustvold Jan 28, 2023

tustvold Jan 28, 2023

tustvold Jan 28, 2023

comphead commented Jan 29, 2023

jackwener commented Jan 29, 2023

alamb commented Jan 30, 2023

ursabot commented Jan 30, 2023

		.collect::<Result<Vec<_>, ArrowError>>()
		.map_err(Into::<DataFusionError>::into)?;

	RecordBatch::try_new(Arc::new(schema.clone()), columns).map_err(Into::into)
	Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?)

	filter_record_batch(batch, filter_array).map_err(Into::into)
	Ok(filter_record_batch(batch, filter_array)?)

	RecordBatch::try_new(schema, columns_to_show).map_err(Into::into)
	Ok(RecordBatch::try_new(schema, columns_to_show)?)

		.collect::<Result<Vec<ArrayRef>, ArrowError>>()
		.map_err(Into::<DataFusionError>::into)?;

	RecordBatch::try_new(self.schema.clone(), batch_columns).map_err(Into::into)
	Ok(RecordBatch::try_new(self.schema.clone(), batch_columns)?)

Use DataFusionError instead of ArrowError in SendableRecordBatchStream #5101

Use DataFusionError instead of ArrowError in SendableRecordBatchStream #5101

Conversation

comphead commented Jan 28, 2023

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

comphead commented Jan 28, 2023

tustvold left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

comphead commented Jan 29, 2023

jackwener commented Jan 29, 2023

alamb commented Jan 30, 2023

ursabot commented Jan 30, 2023