Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
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: 2 additions & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,8 @@ strip = false
# Detects large stack-allocated futures that may cause stack overflow crashes (see threshold in clippy.toml)
large_futures = "warn"
used_underscore_binding = "warn"
or_fun_call = "warn"
unnecessary_lazy_evaluations = "warn"

[workspace.lints.rust]
unexpected_cfgs = { level = "warn", check-cfg = ["cfg(tarpaulin)"] }
Expand Down
2 changes: 1 addition & 1 deletion benchmarks/src/bin/external_aggr.rs
Original file line number Diff line number Diff line change
Expand Up @@ -335,7 +335,7 @@ impl ExternalAggrConfig {
fn partitions(&self) -> usize {
self.common
.partitions
.unwrap_or(get_available_parallelism())
.unwrap_or_else(get_available_parallelism)
}
}

Expand Down
2 changes: 1 addition & 1 deletion benchmarks/src/imdb/run.rs
Original file line number Diff line number Diff line change
Expand Up @@ -471,7 +471,7 @@ impl RunOpt {
fn partitions(&self) -> usize {
self.common
.partitions
.unwrap_or(get_available_parallelism())
.unwrap_or_else(get_available_parallelism)
}
}

Expand Down
2 changes: 1 addition & 1 deletion benchmarks/src/sort.rs
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ impl RunOpt {
let config = SessionConfig::new().with_target_partitions(
self.common
.partitions
.unwrap_or(get_available_parallelism()),
.unwrap_or_else(get_available_parallelism),
);
let ctx = SessionContext::new_with_config(config);
let (rows, elapsed) =
Expand Down
2 changes: 1 addition & 1 deletion benchmarks/src/sort_tpch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -352,6 +352,6 @@ impl RunOpt {
fn partitions(&self) -> usize {
self.common
.partitions
.unwrap_or(get_available_parallelism())
.unwrap_or_else(get_available_parallelism)
}
}
2 changes: 1 addition & 1 deletion benchmarks/src/tpch/run.rs
Original file line number Diff line number Diff line change
Expand Up @@ -313,7 +313,7 @@ impl RunOpt {
fn partitions(&self) -> usize {
self.common
.partitions
.unwrap_or(get_available_parallelism())
.unwrap_or_else(get_available_parallelism)
}
}

Expand Down
15 changes: 7 additions & 8 deletions benchmarks/src/util/options.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ use datafusion::{
};
use datafusion_common::{DataFusionError, Result};
use structopt::StructOpt;
use datafusion::common::utils::get_available_parallelism;

// Common benchmark options (don't use doc comments otherwise this doc
// shows up in help files)
Expand Down Expand Up @@ -70,14 +71,12 @@ impl CommonOpt {
}

/// Modify the existing config appropriately
pub fn update_config(&self, mut config: SessionConfig) -> SessionConfig {
if let Some(batch_size) = self.batch_size {
config = config.with_batch_size(batch_size)
}

if let Some(partitions) = self.partitions {
config = config.with_target_partitions(partitions)
}
pub fn update_config(&self, config: SessionConfig) -> SessionConfig {
let mut config = config
.with_target_partitions(
self.partitions.unwrap_or_else(get_available_parallelism),
)
.with_batch_size(self.batch_size.unwrap_or(8192));

if let Some(sort_spill_reservation_bytes) = self.sort_spill_reservation_bytes {
config =
Expand Down
8 changes: 4 additions & 4 deletions datafusion/common/src/column.rs
Original file line number Diff line number Diff line change
Expand Up @@ -130,8 +130,8 @@ impl Column {
/// where `"foo.BAR"` would be parsed to a reference to column named `foo.BAR`
pub fn from_qualified_name(flat_name: impl Into<String>) -> Self {
let flat_name = flat_name.into();
Self::from_idents(parse_identifiers_normalized(&flat_name, false)).unwrap_or(
Self {
Self::from_idents(parse_identifiers_normalized(&flat_name, false)).unwrap_or_else(
|| Self {
relation: None,
name: flat_name,
spans: Spans::new(),
Expand All @@ -142,8 +142,8 @@ impl Column {
/// Deserialize a fully qualified name string into a column preserving column text case
pub fn from_qualified_name_ignore_case(flat_name: impl Into<String>) -> Self {
let flat_name = flat_name.into();
Self::from_idents(parse_identifiers_normalized(&flat_name, true)).unwrap_or(
Self {
Self::from_idents(parse_identifiers_normalized(&flat_name, true)).unwrap_or_else(
|| Self {
relation: None,
name: flat_name,
spans: Spans::new(),
Expand Down
7 changes: 4 additions & 3 deletions datafusion/common/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -526,7 +526,7 @@ impl DataFusionError {
pub fn message(&self) -> Cow<str> {
match *self {
DataFusionError::ArrowError(ref desc, ref backtrace) => {
let backtrace = backtrace.clone().unwrap_or("".to_owned());
let backtrace = backtrace.clone().unwrap_or_else(|| "".to_owned());
Cow::Owned(format!("{desc}{backtrace}"))
}
#[cfg(feature = "parquet")]
Expand All @@ -535,7 +535,8 @@ impl DataFusionError {
DataFusionError::AvroError(ref desc) => Cow::Owned(desc.to_string()),
DataFusionError::IoError(ref desc) => Cow::Owned(desc.to_string()),
DataFusionError::SQL(ref desc, ref backtrace) => {
let backtrace: String = backtrace.clone().unwrap_or("".to_owned());
let backtrace: String =
backtrace.clone().unwrap_or_else(|| "".to_owned());
Cow::Owned(format!("{desc:?}{backtrace}"))
}
DataFusionError::Configuration(ref desc) => Cow::Owned(desc.to_string()),
Expand All @@ -547,7 +548,7 @@ impl DataFusionError {
DataFusionError::Plan(ref desc) => Cow::Owned(desc.to_string()),
DataFusionError::SchemaError(ref desc, ref backtrace) => {
let backtrace: &str =
&backtrace.as_ref().clone().unwrap_or("".to_owned());
&backtrace.as_ref().clone().unwrap_or_else(|| "".to_owned());
Cow::Owned(format!("{desc}{backtrace}"))
}
DataFusionError::Execution(ref desc) => Cow::Owned(desc.to_string()),
Expand Down
2 changes: 1 addition & 1 deletion datafusion/core/src/datasource/file_format/parquet.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1073,7 +1073,7 @@ mod tests {
let format = state
.get_file_format_factory("parquet")
.map(|factory| factory.create(state, &Default::default()).unwrap())
.unwrap_or(Arc::new(ParquetFormat::new()));
.unwrap_or_else(|| Arc::new(ParquetFormat::new()));

scan_format(
state, &*format, None, &testdata, file_name, projection, limit,
Expand Down
10 changes: 5 additions & 5 deletions datafusion/datasource-csv/src/file_format.rs
Original file line number Diff line number Diff line change
Expand Up @@ -414,11 +414,11 @@ impl FileFormat for CsvFormat {
let has_header = self
.options
.has_header
.unwrap_or(state.config_options().catalog.has_header);
.unwrap_or_else(|| state.config_options().catalog.has_header);
let newlines_in_values = self
.options
.newlines_in_values
.unwrap_or(state.config_options().catalog.newlines_in_values);
.unwrap_or_else(|| state.config_options().catalog.newlines_in_values);

let conf_builder = FileScanConfigBuilder::from(conf)
.with_file_compression_type(self.options.compression.into())
Expand Down Expand Up @@ -454,11 +454,11 @@ impl FileFormat for CsvFormat {
let has_header = self
.options()
.has_header
.unwrap_or(state.config_options().catalog.has_header);
.unwrap_or_else(|| state.config_options().catalog.has_header);
let newlines_in_values = self
.options()
.newlines_in_values
.unwrap_or(state.config_options().catalog.newlines_in_values);
.unwrap_or_else(|| state.config_options().catalog.newlines_in_values);

let options = self
.options()
Expand Down Expand Up @@ -504,7 +504,7 @@ impl CsvFormat {
&& self
.options
.has_header
.unwrap_or(state.config_options().catalog.has_header),
.unwrap_or_else(|| state.config_options().catalog.has_header),
)
.with_delimiter(self.options.delimiter)
.with_quote(self.options.quote);
Expand Down
2 changes: 1 addition & 1 deletion datafusion/datasource/src/file_scan_config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -582,7 +582,7 @@ impl DataSource for FileScanConfig {
&file_scan
.projection
.clone()
.unwrap_or((0..self.file_schema.fields().len()).collect()),
.unwrap_or_else(|| (0..self.file_schema.fields().len()).collect()),
);
DataSourceExec::from_data_source(
FileScanConfigBuilder::from(file_scan)
Expand Down
4 changes: 3 additions & 1 deletion datafusion/datasource/src/statistics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,9 @@ impl MinMaxStatistics {
// Reverse the projection to get the index of the column in the full statistics
// The file statistics contains _every_ column , but the sort column's index()
// refers to the index in projected_schema
let i = projection.map(|p| p[c.index()]).unwrap_or(c.index());
let i = projection
.map(|p| p[c.index()])
.unwrap_or_else(|| c.index());

let (min, max) = get_min_max(i).map_err(|e| {
e.context(format!("get min/max for column: '{}'", c.name()))
Expand Down
2 changes: 1 addition & 1 deletion datafusion/datasource/src/write/demux.rs
Original file line number Diff line number Diff line change
Expand Up @@ -513,7 +513,7 @@ fn compute_take_arrays(
for vals in all_partition_values.iter() {
part_key.push(vals[i].clone().into());
}
let builder = take_map.entry(part_key).or_insert(UInt64Builder::new());
let builder = take_map.entry(part_key).or_insert_with(UInt64Builder::new);
builder.append_value(i as u64);
}
take_map
Expand Down
2 changes: 1 addition & 1 deletion datafusion/expr-common/src/type_coercion/binary.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1138,7 +1138,7 @@ fn dictionary_comparison_coercion(
/// 2. Data type of the other side should be able to cast to string type
fn string_concat_coercion(lhs_type: &DataType, rhs_type: &DataType) -> Option<DataType> {
use arrow::datatypes::DataType::*;
string_coercion(lhs_type, rhs_type).or(match (lhs_type, rhs_type) {
string_coercion(lhs_type, rhs_type).or_else(|| match (lhs_type, rhs_type) {
(Utf8View, from_type) | (from_type, Utf8View) => {
string_concat_internal_coercion(from_type, &Utf8View)
}
Expand Down
2 changes: 1 addition & 1 deletion datafusion/expr/src/expr_fn.rs
Original file line number Diff line number Diff line change
Expand Up @@ -838,7 +838,7 @@ impl ExprFuncBuilder {
partition_by: partition_by.unwrap_or_default(),
order_by: order_by.unwrap_or_default(),
window_frame: window_frame
.unwrap_or(WindowFrame::new(has_order_by)),
.unwrap_or_else(|| WindowFrame::new(has_order_by)),
null_treatment,
},
})
Expand Down
2 changes: 1 addition & 1 deletion datafusion/functions-aggregate/src/array_agg.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1007,7 +1007,7 @@ mod tests {

fn print_nulls(sort: Vec<Option<String>>) -> Vec<String> {
sort.into_iter()
.map(|v| v.unwrap_or("NULL".to_string()))
.map(|v| v.unwrap_or_else(|| "NULL".to_string()))
.collect()
}

Expand Down
2 changes: 1 addition & 1 deletion datafusion/functions-window/src/lead_lag.rs
Original file line number Diff line number Diff line change
Expand Up @@ -321,7 +321,7 @@ fn parse_default_value(
unparsed
.filter(|v| !v.data_type().is_null())
.map(|v| v.cast_to(&expr_type))
.unwrap_or(ScalarValue::try_from(expr_type))
.unwrap_or_else(|| ScalarValue::try_from(expr_type))
}

#[derive(Debug)]
Expand Down
2 changes: 1 addition & 1 deletion datafusion/macros/src/user_doc.rs
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,7 @@ pub fn user_doc(args: TokenStream, input: TokenStream) -> TokenStream {
};
let doc_section_description = doc_section_desc
.map(|desc| quote! { Some(#desc)})
.unwrap_or(quote! { None });
.unwrap_or_else(|| quote! { None });

let sql_example = sql_example.map(|ex| {
quote! {
Expand Down
2 changes: 1 addition & 1 deletion datafusion/optimizer/src/simplify_expressions/regex.rs
Original file line number Diff line number Diff line change
Expand Up @@ -331,7 +331,7 @@ fn lower_simple(mode: &OperatorMode, left: &Expr, hir: &Hir) -> Option<Expr> {
}
HirKind::Concat(inner) => {
if let Some(pattern) = partial_anchored_literal_to_like(inner)
.or(collect_concat_to_like_string(inner))
.or_else(|| collect_concat_to_like_string(inner))
{
return Some(mode.expr(Box::new(left.clone()), pattern));
}
Expand Down
2 changes: 1 addition & 1 deletion datafusion/physical-expr/src/analysis.rs
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,7 @@ impl ExprBoundaries {
.min_value
.get_value()
.cloned()
.unwrap_or(empty_field.clone()),
.unwrap_or_else(|| empty_field.clone()),
col_stats
.max_value
.get_value()
Expand Down
2 changes: 1 addition & 1 deletion datafusion/physical-expr/src/equivalence/properties/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1364,7 +1364,7 @@ impl EquivalenceProperties {
.transform_up(|expr| update_properties(expr, self))
.data()
.map(|node| node.data)
.unwrap_or(ExprProperties::new_unknown())
.unwrap_or_else(|_| ExprProperties::new_unknown())
}

/// Transforms this `EquivalenceProperties` into a new `EquivalenceProperties`
Expand Down
5 changes: 2 additions & 3 deletions datafusion/physical-optimizer/src/enforce_distribution.rs
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,6 @@ use datafusion_physical_expr::utils::map_columns_before_projection;
use datafusion_physical_expr::{
physical_exprs_equal, EquivalenceProperties, PhysicalExpr, PhysicalExprRef,
};
use datafusion_physical_expr_common::sort_expr::LexOrdering;
use datafusion_physical_plan::aggregates::{
AggregateExec, AggregateMode, PhysicalGroupBy,
};
Expand Down Expand Up @@ -953,8 +952,8 @@ fn add_spm_on_top(input: DistributionContext) -> DistributionContext {
input
.plan
.output_ordering()
.unwrap_or(&LexOrdering::default())
.clone(),
.cloned()
.unwrap_or_default(),
Arc::clone(&input.plan),
)) as _
} else {
Expand Down
5 changes: 3 additions & 2 deletions datafusion/physical-optimizer/src/enforce_sorting/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -502,10 +502,11 @@ fn analyze_immediate_sort_removal(
let sort_input = sort_exec.input();
// If this sort is unnecessary, we should remove it:
if sort_input.equivalence_properties().ordering_satisfy(
sort_exec
&sort_exec
.properties()
.output_ordering()
.unwrap_or(LexOrdering::empty()),
.cloned()
.unwrap_or_default(),
) {
node.plan = if !sort_exec.preserve_partitioning()
&& sort_input.output_partitioning().partition_count() > 1
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,10 @@ use crate::utils::{

use datafusion_common::config::ConfigOptions;
use datafusion_common::tree_node::Transformed;
use datafusion_common::{internal_err, Result};
use datafusion_physical_expr_common::sort_expr::LexOrdering;
use datafusion_physical_plan::internal_err;

use datafusion_common::Result;
// use datafusion_physical_expr_common::sort_expr::LexOrdering;
use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec;
use datafusion_physical_plan::execution_plan::EmissionType;
use datafusion_physical_plan::repartition::RepartitionExec;
Expand Down Expand Up @@ -281,10 +283,11 @@ pub fn replace_with_order_preserving_variants(
.plan
.equivalence_properties()
.ordering_satisfy(
requirements
&requirements
.plan
.output_ordering()
.unwrap_or(LexOrdering::empty()),
.cloned()
.unwrap_or_default(),
)
{
for child in alternate_plan.children.iter_mut() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,7 @@ fn pushdown_requirement_to_children(
.properties()
.output_ordering()
.cloned()
.unwrap_or(LexOrdering::default()),
.unwrap_or_else(LexOrdering::default),
);
if sort_exec
.properties()
Expand All @@ -258,7 +258,7 @@ fn pushdown_requirement_to_children(
plan.properties()
.output_ordering()
.cloned()
.unwrap_or(LexOrdering::default()),
.unwrap_or_else(LexOrdering::default),
);
// Push down through operator with fetch when:
// - requirement is aligned with output ordering
Expand Down
4 changes: 2 additions & 2 deletions datafusion/physical-optimizer/src/update_aggr_exprs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ use datafusion_physical_expr::aggregate::AggregateFunctionExpr;
use datafusion_physical_expr::{
reverse_order_bys, EquivalenceProperties, PhysicalSortRequirement,
};
use datafusion_physical_expr::{LexOrdering, LexRequirement};
use datafusion_physical_expr::{LexRequirement};
use datafusion_physical_plan::aggregates::concat_slices;
use datafusion_physical_plan::windows::get_ordered_partition_by_indices;
use datafusion_physical_plan::{
Expand Down Expand Up @@ -159,7 +159,7 @@ fn try_convert_aggregate_if_better(
aggr_exprs
.into_iter()
.map(|aggr_expr| {
let aggr_sort_exprs = aggr_expr.order_bys().unwrap_or(LexOrdering::empty());
let aggr_sort_exprs = &aggr_expr.order_bys().cloned().unwrap_or_default();
let reverse_aggr_sort_exprs = reverse_order_bys(aggr_sort_exprs);
let aggr_sort_reqs = LexRequirement::from(aggr_sort_exprs.clone());
let reverse_aggr_req = LexRequirement::from(reverse_aggr_sort_exprs);
Expand Down
2 changes: 1 addition & 1 deletion datafusion/physical-plan/src/projection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -538,7 +538,7 @@ pub fn remove_unnecessary_projections(
} else {
return Ok(Transformed::no(plan));
};
Ok(maybe_modified.map_or(Transformed::no(plan), Transformed::yes))
Ok(maybe_modified.map_or_else(|| Transformed::no(plan), Transformed::yes))
}

/// Compare the inputs and outputs of the projection. All expressions must be
Expand Down
2 changes: 1 addition & 1 deletion datafusion/physical-plan/src/streaming.rs
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,7 @@ impl ExecutionPlan for StreamingTableExec {
let new_projections = new_projections_for_columns(
projection,
&streaming_table_projections
.unwrap_or((0..self.schema().fields().len()).collect()),
.unwrap_or_else(|| (0..self.schema().fields().len()).collect()),
);

let mut lex_orderings = vec![];
Expand Down
Loading