You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/12 19:19:20 UTC
(datafusion) branch gh-readonly-queue/main/pr-26192-39be5f279717206a96ab78c43ea9a6f07d58a1bd created (now 9551051f13)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-26192-39be5f279717206a96ab78c43ea9a6f07d58a1bd
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 9551051f13 perf: pre-size collections and reuse per-iteration buffers (#26192)
This branch includes the following new commits:
new 9551051f13 perf: pre-size collections and reuse per-iteration buffers (#26192)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: perf: pre-size collections and reuse per-iteration buffers (#26192)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26192-39be5f279717206a96ab78c43ea9a6f07d58a1bd
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 9551051f13ee321eb0d41bf437f3718a138e12a2
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Mon Oct 12 19:18:57 2026 +0000
perf: pre-size collections and reuse per-iteration buffers (#26192)
## Which issue does this PR close?
- N/A
## Rationale for this change
Many execution paths allocate more than they need. Collections start
empty although their final size is known, and some temporaries are
allocated again on every row, batch, partition or column. Each case is
small, but together they add avoidable allocator traffic on hot paths.
## What changes are included in this PR?
Two mechanical rules applied across the workspace:
- `with_capacity` where the size is known or tightly bounded by
in-memory data.
- Per-iteration temporaries hoisted out of loops and reused with
`clear()` when they don't escape the iteration.
For example, `RepartitionExec` no longer allocates a `Vec` per batch per
output partition, `TDigest` merging reuses one buffer instead of a
full-size `Vec` per merge step, and hive-partitioned writes no longer
build a `Vec<String>` key per row.
## What is the testing strategy for this PR?
No behavior change, so existing tests cover it. The workspace `--lib
--tests --bins` suite and sqllogictest pass locally. Criterion benches
against `main` (`regexp_replace`, `reverse`, `array_slice`,
`bounded_window`) are neutral to a few percent faster.
## Are there any user-facing changes?
No.
---
datafusion-cli/src/functions.rs | 89 ++++++++++++----------
datafusion-cli/src/helper.rs | 2 +-
datafusion/catalog-listing/src/helpers.rs | 2 +-
datafusion/catalog/src/information_schema.rs | 24 +++---
datafusion/catalog/src/memory/table.rs | 4 +-
datafusion/common/src/config.rs | 9 ++-
datafusion/common/src/cse.rs | 14 ++--
datafusion/common/src/dfschema.rs | 6 +-
datafusion/common/src/error.rs | 5 +-
datafusion/common/src/functional_dependencies.rs | 6 +-
datafusion/common/src/hash_utils.rs | 6 +-
datafusion/common/src/scalar/mod.rs | 5 +-
datafusion/common/src/stats.rs | 11 ++-
datafusion/common/src/utils/mod.rs | 12 ++-
datafusion/core/src/dataframe/mod.rs | 3 +-
.../core/src/datasource/listing_table_factory.rs | 2 +-
datafusion/core/src/execution/context/mod.rs | 4 +-
datafusion/core/src/execution/session_state.rs | 20 +++--
datafusion/core/src/physical_planner.rs | 14 ++--
datafusion/datasource-arrow/src/file_format.rs | 2 +-
datafusion/datasource-avro/src/file_format.rs | 2 +-
datafusion/datasource-parquet/src/access_plan.rs | 3 +-
datafusion/datasource-parquet/src/page_filter.rs | 16 ++--
.../datasource-parquet/src/row_group_filter.rs | 3 +-
datafusion/datasource-parquet/src/sort.rs | 2 +-
datafusion/datasource-parquet/src/source.rs | 3 +-
.../datasource/src/file_scan_config/proto.rs | 4 +-
datafusion/datasource/src/memory.rs | 3 +-
datafusion/datasource/src/projection.rs | 3 +-
datafusion/datasource/src/write/demux.rs | 24 +++---
datafusion/datasource/src/write/orchestration.rs | 2 +-
datafusion/execution/src/config.rs | 5 +-
datafusion/expr-common/src/signature.rs | 2 +-
datafusion/expr-common/src/type_coercion/binary.rs | 13 ++--
datafusion/expr/src/expr_rewriter/order_by.rs | 19 +++--
datafusion/expr/src/logical_plan/builder.rs | 13 ++--
datafusion/expr/src/logical_plan/plan.rs | 22 +++---
datafusion/expr/src/utils.rs | 15 ++--
datafusion/expr/src/window_state.rs | 20 ++---
datafusion/ffi/src/config/mod.rs | 12 ++-
datafusion/ffi/src/udaf/groups_accumulator.rs | 2 -
.../src/aggregate/count_distinct/groups.rs | 5 +-
.../src/aggregate/groups_accumulator.rs | 10 ++-
.../functions-aggregate-common/src/merge_arrays.rs | 7 +-
.../functions-aggregate-common/src/tdigest.rs | 14 +++-
.../src/approx_percentile_cont_with_weight.rs | 2 +-
datafusion/functions-aggregate/src/array_agg.rs | 27 +++++--
datafusion/functions-aggregate/src/first_last.rs | 12 +--
datafusion/functions-aggregate/src/nth_value.rs | 2 +-
.../functions-aggregate/src/percentile_cont.rs | 14 ++--
datafusion/functions-nested/src/concat.rs | 3 +-
datafusion/functions-nested/src/extract.rs | 30 +++++---
datafusion/functions-nested/src/make_array.rs | 2 +-
datafusion/functions-nested/src/map.rs | 7 +-
datafusion/functions-nested/src/range.rs | 3 +-
datafusion/functions-nested/src/resize.rs | 3 +-
datafusion/functions-nested/src/reverse.rs | 3 +-
datafusion/functions-nested/src/utils.rs | 2 +-
datafusion/functions-table/src/generate_series.rs | 2 +-
datafusion/functions/src/core/planner.rs | 2 +-
datafusion/functions/src/crypto/basic.rs | 4 +-
datafusion/functions/src/regex/regexplike.rs | 4 +-
datafusion/functions/src/regex/regexpreplace.rs | 35 +++++----
datafusion/functions/src/string/common.rs | 18 ++++-
datafusion/functions/src/unicode/reverse.rs | 29 +++++--
.../src/analyzer/resolve_grouping_function.rs | 2 +-
.../optimizer/src/common_subexpr_eliminate.rs | 4 +-
datafusion/optimizer/src/decorrelate.rs | 13 ++--
.../optimizer/src/decorrelate_lateral_join.rs | 4 +-
.../src/decorrelate_predicate_subquery.rs | 2 +-
datafusion/optimizer/src/eliminate_outer_join.rs | 2 +-
.../optimizer/src/extract_leaf_expressions.rs | 9 ++-
.../optimizer/src/optimize_projections/mod.rs | 5 +-
datafusion/optimizer/src/push_down_filter.rs | 14 ++--
.../optimizer/src/scalar_subquery_to_join.rs | 2 +-
.../src/simplify_expressions/expr_simplifier.rs | 3 +-
.../simplify_expressions/simplify_predicates.rs | 2 +-
datafusion/optimizer/src/unions_to_filter.rs | 2 +-
datafusion/physical-expr-common/src/sort_expr.rs | 9 ++-
datafusion/physical-expr/src/analysis.rs | 3 +-
datafusion/physical-expr/src/equivalence/class.rs | 3 +-
.../src/equivalence/properties/mod.rs | 7 +-
datafusion/physical-expr/src/expressions/case.rs | 2 +-
.../physical-expr/src/expressions/in_list.rs | 3 +-
.../src/expressions/in_list/array_static_filter.rs | 2 +-
.../physical-expr/src/intervals/cp_solver.rs | 10 ++-
datafusion/physical-expr/src/physical_expr.rs | 10 +--
datafusion/physical-expr/src/projection.rs | 10 ++-
datafusion/physical-expr/src/utils/guarantee.rs | 6 +-
datafusion/physical-expr/src/window/standard.rs | 5 +-
datafusion/physical-expr/src/window/window_expr.rs | 3 +-
.../physical-optimizer/src/aggregate_statistics.rs | 2 +-
.../ensure_requirements/enforce_distribution.rs | 6 +-
.../src/ensure_requirements/enforce_sorting/mod.rs | 4 +-
.../physical-optimizer/src/filter_pushdown.rs | 8 +-
.../physical-optimizer/src/output_requirements.rs | 4 +-
.../physical-optimizer/src/projection_pushdown.rs | 5 +-
.../src/aggregates/aggregate_hash_table/common.rs | 3 +-
.../group_values/single_group_by/bytes.rs | 2 +-
.../group_values/single_group_by/bytes_view.rs | 2 +-
.../src/aggregates/grouped_hash_stream.rs | 17 +++--
datafusion/physical-plan/src/aggregates/mod.rs | 5 +-
datafusion/physical-plan/src/display.rs | 2 +-
datafusion/physical-plan/src/execution_plan.rs | 4 +-
datafusion/physical-plan/src/filter.rs | 2 +-
datafusion/physical-plan/src/joins/asof_join.rs | 23 +++---
datafusion/physical-plan/src/joins/cross_join.rs | 20 ++---
.../src/joins/hash_join/shared_bounds.rs | 2 +-
.../physical-plan/src/joins/nested_loop_join.rs | 4 +-
.../src/joins/piecewise_merge_join/classic_join.rs | 7 +-
.../piecewise_merge_join/right_existence_join.rs | 3 +-
.../joins/sort_merge_join/materializing_stream.rs | 12 +--
.../physical-plan/src/joins/stream_join_utils.rs | 8 +-
datafusion/physical-plan/src/projection.rs | 6 +-
datafusion/physical-plan/src/repartition/mod.rs | 29 ++++---
datafusion/physical-plan/src/sorts/builder.rs | 14 ++--
datafusion/physical-plan/src/sorts/partial_sort.rs | 26 ++++---
datafusion/physical-plan/src/spill/mod.rs | 3 +-
datafusion/physical-plan/src/streaming.rs | 2 +-
datafusion/physical-plan/src/topk/mod.rs | 11 ++-
datafusion/physical-plan/src/union.rs | 2 +-
datafusion/physical-plan/src/unnest.rs | 5 +-
.../src/windows/bounded_window_agg_exec.rs | 16 ++--
datafusion/physical-plan/src/windows/mod.rs | 4 +-
.../physical-plan/src/windows/window_agg_exec.rs | 2 +-
datafusion/proto-common/src/from_proto/mod.rs | 5 +-
datafusion/proto/src/logical_plan/mod.rs | 11 ++-
datafusion/pruning/src/pruning_predicate.rs | 4 +-
datafusion/spark/src/function/array/shuffle.rs | 16 ++--
datafusion/spark/src/function/hash/sha1.rs | 24 +++---
datafusion/spark/src/function/hash/utils.rs | 6 +-
datafusion/spark/src/function/json/json_tuple.rs | 5 +-
datafusion/spark/src/function/map/utils.rs | 4 +-
datafusion/spark/src/function/string/encode.rs | 52 ++++++-------
.../spark/src/function/string/format_string.rs | 36 ++++-----
datafusion/sql/src/expr/function.rs | 2 +-
datafusion/sql/src/expr/mod.rs | 6 +-
datafusion/sql/src/select.rs | 9 ++-
datafusion/sql/src/statement.rs | 8 +-
datafusion/sql/src/unparser/expr.rs | 2 +-
datafusion/sql/src/unparser/plan.rs | 5 +-
datafusion/sql/src/unparser/rewrite.rs | 2 +-
.../sqllogictest/test_files/array/array_slice.slt | 20 +++++
datafusion/substrait/src/extensions.rs | 6 +-
.../consumer/expr/function_arguments.rs | 2 +-
.../src/logical_plan/consumer/expr/if_then.rs | 3 +-
.../src/logical_plan/consumer/expr/mod.rs | 2 +-
.../src/logical_plan/consumer/rel/aggregate_rel.rs | 6 +-
.../src/logical_plan/consumer/rel/exchange_rel.rs | 2 +-
.../substrait/src/logical_plan/consumer/rel/mod.rs | 4 +-
.../src/logical_plan/consumer/rel/project_rel.rs | 9 +--
.../src/logical_plan/consumer/rel/read_rel.rs | 4 +-
.../substrait/src/logical_plan/consumer/types.rs | 2 +-
.../substrait/src/logical_plan/consumer/utils.rs | 2 +-
.../producer/expr/aggregate_function.rs | 2 +-
.../logical_plan/producer/expr/scalar_function.rs | 2 +-
.../logical_plan/producer/expr/window_function.rs | 2 +-
.../src/logical_plan/producer/rel/aggregate_rel.rs | 4 +-
datafusion/substrait/src/physical_plan/producer.rs | 4 +-
159 files changed, 779 insertions(+), 564 deletions(-)
diff --git a/datafusion-cli/src/functions.rs b/datafusion-cli/src/functions.rs
index 514eda4001..d5889bd2a5 100644
--- a/datafusion-cli/src/functions.rs
+++ b/datafusion-cli/src/functions.rs
@@ -366,29 +366,32 @@ impl TableFunctionImpl for ParquetMetadataFunc {
]));
// construct record batch from metadata
- let mut filename_arr = vec![];
- let mut row_group_id_arr = vec![];
- let mut row_group_num_rows_arr = vec![];
- let mut row_group_num_columns_arr = vec![];
- let mut row_group_bytes_arr = vec![];
- let mut column_id_arr = vec![];
- let mut file_offset_arr = vec![];
- let mut num_values_arr = vec![];
- let mut path_in_schema_arr = vec![];
- let mut type_arr = vec![];
- let mut stats_min_arr = vec![];
- let mut stats_max_arr = vec![];
- let mut stats_null_count_arr = vec![];
- let mut stats_distinct_count_arr = vec![];
- let mut stats_min_value_arr = vec![];
- let mut stats_max_value_arr = vec![];
- let mut compression_arr = vec![];
- let mut encodings_arr = vec![];
- let mut index_page_offset_arr = vec![];
- let mut dictionary_page_offset_arr = vec![];
- let mut data_page_offset_arr = vec![];
- let mut total_compressed_size_arr = vec![];
- let mut total_uncompressed_size_arr = vec![];
+ let num_column_chunks = metadata.num_row_groups()
+ * metadata.file_metadata().schema_descr().num_columns();
+ let mut filename_arr = Vec::with_capacity(num_column_chunks);
+ let mut row_group_id_arr = Vec::with_capacity(num_column_chunks);
+ let mut row_group_num_rows_arr = Vec::with_capacity(num_column_chunks);
+ let mut row_group_num_columns_arr = Vec::with_capacity(num_column_chunks);
+ let mut row_group_bytes_arr = Vec::with_capacity(num_column_chunks);
+ let mut column_id_arr = Vec::with_capacity(num_column_chunks);
+ let mut file_offset_arr = Vec::with_capacity(num_column_chunks);
+ let mut num_values_arr = Vec::with_capacity(num_column_chunks);
+ let mut path_in_schema_arr = Vec::with_capacity(num_column_chunks);
+ let mut type_arr = Vec::with_capacity(num_column_chunks);
+ let mut stats_min_arr = Vec::with_capacity(num_column_chunks);
+ let mut stats_max_arr = Vec::with_capacity(num_column_chunks);
+ let mut stats_null_count_arr = Vec::with_capacity(num_column_chunks);
+ let mut stats_distinct_count_arr = Vec::with_capacity(num_column_chunks);
+ let mut stats_min_value_arr = Vec::with_capacity(num_column_chunks);
+ let mut stats_max_value_arr = Vec::with_capacity(num_column_chunks);
+ let mut compression_arr = Vec::with_capacity(num_column_chunks);
+ let mut encodings_arr = Vec::with_capacity(num_column_chunks);
+ let mut index_page_offset_arr = Vec::with_capacity(num_column_chunks);
+ let mut dictionary_page_offset_arr = Vec::with_capacity(num_column_chunks);
+ let mut data_page_offset_arr = Vec::with_capacity(num_column_chunks);
+ let mut total_compressed_size_arr = Vec::with_capacity(num_column_chunks);
+ let mut total_uncompressed_size_arr = Vec::with_capacity(num_column_chunks);
+ let mut encodings = Vec::new();
for (rg_idx, row_group) in metadata.row_groups().iter().enumerate() {
for (col_idx, column) in row_group.columns().iter().enumerate() {
filename_arr.push(filename);
@@ -423,7 +426,8 @@ impl TableFunctionImpl for ParquetMetadataFunc {
}
compression_arr.push(format!("{:?}", column.compression()));
// need to collect into Vec to format
- let encodings: Vec<_> = column.encodings().collect();
+ encodings.clear();
+ encodings.extend(column.encodings());
encodings_arr.push(format!("{encodings:?}"));
index_page_offset_arr.push(column.index_page_offset());
dictionary_page_offset_arr.push(column.dictionary_page_offset());
@@ -710,18 +714,19 @@ impl TableFunctionImpl for MetadataCacheFunc {
Field::new("extra", DataType::Utf8, true),
]));
- // construct record batch from metadata
- let mut path_arr = vec![];
- let mut file_modified_arr = vec![];
- let mut file_size_bytes_arr = vec![];
- let mut e_tag_arr = vec![];
- let mut version_arr = vec![];
- let mut metadata_size_bytes = vec![];
- let mut hits_arr = vec![];
- let mut extra_arr = vec![];
-
let cached_entries = self.cache_manager.get_file_metadata_cache().list_entries();
+ // construct record batch from metadata
+ let mut path_arr = Vec::with_capacity(cached_entries.len());
+ let mut file_modified_arr = Vec::with_capacity(cached_entries.len());
+ let mut file_size_bytes_arr = Vec::with_capacity(cached_entries.len());
+ let mut e_tag_arr = Vec::with_capacity(cached_entries.len());
+ let mut version_arr = Vec::with_capacity(cached_entries.len());
+ let mut metadata_size_bytes = Vec::with_capacity(cached_entries.len());
+ let mut hits_arr = Vec::with_capacity(cached_entries.len());
+ let mut extra_arr = Vec::with_capacity(cached_entries.len());
+ let mut extra = Vec::new();
+
for (path, entry) in cached_entries {
path_arr.push(path.to_string());
file_modified_arr
@@ -732,13 +737,15 @@ impl TableFunctionImpl for MetadataCacheFunc {
metadata_size_bytes.push(entry.size_bytes as u64);
hits_arr.push(entry.hits as u64);
- let mut extra = entry
- .value
- .file_metadata
- .extra_info()
- .iter()
- .map(|(k, v)| format!("{k}={v}"))
- .collect::<Vec<_>>();
+ extra.clear();
+ extra.extend(
+ entry
+ .value
+ .file_metadata
+ .extra_info()
+ .iter()
+ .map(|(k, v)| format!("{k}={v}")),
+ );
extra.sort();
extra_arr.push(extra.join(" "));
}
diff --git a/datafusion-cli/src/helper.rs b/datafusion-cli/src/helper.rs
index c167e083d8..8b7d95797c 100644
--- a/datafusion-cli/src/helper.rs
+++ b/datafusion-cli/src/helper.rs
@@ -180,7 +180,7 @@ impl Helper for CliHelper {}
/// Splits a string which consists of multiple queries.
pub(crate) fn split_from_semicolon(sql: &str) -> Vec<String> {
let mut commands = Vec::new();
- let mut current_command = String::new();
+ let mut current_command = String::with_capacity(sql.len());
let mut in_single_quote = false;
let mut in_double_quote = false;
diff --git a/datafusion/catalog-listing/src/helpers.rs b/datafusion/catalog-listing/src/helpers.rs
index cf0d170809..bb0c372357 100644
--- a/datafusion/catalog-listing/src/helpers.rs
+++ b/datafusion/catalog-listing/src/helpers.rs
@@ -272,7 +272,7 @@ pub fn evaluate_partition_prefix<'a>(
return None;
}
- let mut parts = vec![];
+ let mut parts = Vec::with_capacity(partition_cols.len());
for (p, _) in partition_cols {
match partition_values.get(p.as_str()) {
Some(PartitionValue::Single(val)) => {
diff --git a/datafusion/catalog/src/information_schema.rs b/datafusion/catalog/src/information_schema.rs
index 85222bc2f6..9e3a1d8743 100644
--- a/datafusion/catalog/src/information_schema.rs
+++ b/datafusion/catalog/src/information_schema.rs
@@ -273,8 +273,8 @@ impl InformationSchemaConfig {
Self::is_deterministic(udf.signature()),
return_type.as_ref(),
"SCALAR",
- udf.documentation().map(|d| d.description.to_string()),
- udf.documentation().map(|d| d.syntax_example.to_string()),
+ udf.documentation().map(|d| d.description.as_str()),
+ udf.documentation().map(|d| d.syntax_example.as_str()),
)
}
}
@@ -293,8 +293,8 @@ impl InformationSchemaConfig {
Self::is_deterministic(udaf.signature()),
return_type.as_ref(),
"AGGREGATE",
- udaf.documentation().map(|d| d.description.to_string()),
- udaf.documentation().map(|d| d.syntax_example.to_string()),
+ udaf.documentation().map(|d| d.description.as_str()),
+ udaf.documentation().map(|d| d.syntax_example.as_str()),
)
}
}
@@ -313,8 +313,8 @@ impl InformationSchemaConfig {
Self::is_deterministic(udwf.signature()),
return_type.as_ref(),
"WINDOW",
- udwf.documentation().map(|d| d.description.to_string()),
- udwf.documentation().map(|d| d.syntax_example.to_string()),
+ udwf.documentation().map(|d| d.description.as_str()),
+ udwf.documentation().map(|d| d.syntax_example.as_str()),
)
}
}
@@ -393,12 +393,12 @@ impl InformationSchemaConfig {
};
for (func_name, udf) in udfs {
- let args = udf.documentation().and_then(|d| d.arguments.clone());
+ let args = udf.documentation().and_then(|d| d.arguments.as_ref());
let combinations = get_udf_args_and_return_types(udf)?;
for (rid, (arg_types, return_type)) in combinations.into_iter().enumerate() {
add_parameters(
func_name,
- args.as_ref(),
+ args,
arg_types,
return_type,
Self::is_variadic(udf.signature()),
@@ -408,12 +408,12 @@ impl InformationSchemaConfig {
}
for (func_name, udaf) in udafs {
- let args = udaf.documentation().and_then(|d| d.arguments.clone());
+ let args = udaf.documentation().and_then(|d| d.arguments.as_ref());
let combinations = get_udaf_args_and_return_types(udaf)?;
for (rid, (arg_types, return_type)) in combinations.into_iter().enumerate() {
add_parameters(
func_name,
- args.as_ref(),
+ args,
arg_types,
return_type,
Self::is_variadic(udaf.signature()),
@@ -423,12 +423,12 @@ impl InformationSchemaConfig {
}
for (func_name, udwf) in udwfs {
- let args = udwf.documentation().and_then(|d| d.arguments.clone());
+ let args = udwf.documentation().and_then(|d| d.arguments.as_ref());
let combinations = get_udwf_args_and_return_types(udwf)?;
for (rid, (arg_types, return_type)) in combinations.into_iter().enumerate() {
add_parameters(
func_name,
- args.as_ref(),
+ args,
arg_types,
return_type,
Self::is_variadic(udwf.signature()),
diff --git a/datafusion/catalog/src/memory/table.rs b/datafusion/catalog/src/memory/table.rs
index 6fbff7dd5d..0fbdd1aab8 100644
--- a/datafusion/catalog/src/memory/table.rs
+++ b/datafusion/catalog/src/memory/table.rs
@@ -300,7 +300,7 @@ impl MemTable {
_filters: &[Expr],
_limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>> {
- let mut partitions = vec![];
+ let mut partitions = Vec::with_capacity(self.batches.len());
for arc_inner_vec in self.batches.iter() {
let inner_vec = arc_inner_vec.read().await;
partitions.push(inner_vec.clone())
@@ -321,7 +321,7 @@ impl MemTable {
let df_schema = DFSchema::try_from(Arc::clone(&self.schema))?;
let eqp = state.execution_props();
- let mut file_sort_order = vec![];
+ let mut file_sort_order = Vec::with_capacity(sort_order.len());
for sort_exprs in sort_order.iter() {
let physical_exprs = create_physical_sort_exprs(
sort_exprs,
diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index d5e11c12d9..6790f935c2 100644
--- a/datafusion/common/src/config.rs
+++ b/datafusion/common/src/config.rs
@@ -3675,7 +3675,7 @@ impl From<&Arc<FileEncryptionProperties>> for ConfigFileEncryptionProperties {
let mut column_encryption_properties: HashMap<
String,
ColumnEncryptionProperties,
- > = HashMap::new();
+ > = HashMap::with_capacity(column_names_vec.len());
for (i, column_name) in column_names_vec.iter().enumerate() {
let column_key_as_hex = hex::encode(&column_keys_vec[i]);
@@ -3783,8 +3783,9 @@ impl TryFrom<ConfigFileDecryptionProperties> for FileDecryptionProperties {
type Error = DataFusionError;
fn try_from(val: ConfigFileDecryptionProperties) -> Result<Self> {
- let mut column_names: Vec<&str> = Vec::new();
- let mut column_keys: Vec<Vec<u8>> = Vec::new();
+ let num_columns = val.column_decryption_properties.len();
+ let mut column_names: Vec<&str> = Vec::with_capacity(num_columns);
+ let mut column_keys: Vec<Vec<u8>> = Vec::with_capacity(num_columns);
for (col_name, decryption_properties) in val.column_decryption_properties.iter() {
let column_key = hex::decode(&decryption_properties.column_key_as_hex).map_err(|e| {
@@ -3856,7 +3857,7 @@ impl TryFrom<&Arc<FileDecryptionProperties>> for ConfigFileDecryptionProperties
let mut column_decryption_properties: HashMap<
String,
ColumnDecryptionProperties,
- > = HashMap::new();
+ > = HashMap::with_capacity(column_names_vec.len());
for (i, column_name) in column_names_vec.iter().enumerate() {
let props = ColumnDecryptionProperties {
column_key_as_hex: hex::encode(column_keys_vec[i].clone()),
diff --git a/datafusion/common/src/cse.rs b/datafusion/common/src/cse.rs
index 7469563bd6..f349058118 100644
--- a/datafusion/common/src/cse.rs
+++ b/datafusion/common/src/cse.rs
@@ -260,6 +260,9 @@ where
/// inner states
visit_stack: Vec<VisitRecord<'n, N>>,
+ /// buffer reused by `pop_enter_mark` to collect the identifiers of children
+ node_ids: Vec<Identifier<'n, N>>,
+
/// preorder index, start from 0.
down_index: usize,
@@ -319,22 +322,22 @@ where
&mut self,
can_normalize: bool,
) -> (usize, Option<Identifier<'n, N>>, bool) {
- let mut node_ids: Vec<Identifier<'n, N>> = vec![];
let mut is_valid = true;
while let Some(item) = self.visit_stack.pop() {
match item {
VisitRecord::EnterMark(down_index) => {
if can_normalize {
- node_ids.sort_by_key(|i| i.hash);
+ self.node_ids.sort_by_key(|i| i.hash);
}
- let node_id = node_ids
- .into_iter()
+ let node_id = self
+ .node_ids
+ .drain(..)
.fold(None, |accum, item| Some(item.combine(accum)));
return (down_index, node_id, is_valid);
}
VisitRecord::NodeItem(sub_node_id, sub_node_is_valid) => {
- node_ids.push(sub_node_id);
+ self.node_ids.push(sub_node_id);
is_valid &= sub_node_is_valid;
}
}
@@ -551,6 +554,7 @@ where
node_stats,
id_array,
visit_stack: vec![],
+ node_ids: vec![],
down_index: 0,
up_index: 0,
random_state: &self.random_state,
diff --git a/datafusion/common/src/dfschema.rs b/datafusion/common/src/dfschema.rs
index d9234dae3e..1c072aaa25 100644
--- a/datafusion/common/src/dfschema.rs
+++ b/datafusion/common/src/dfschema.rs
@@ -286,7 +286,9 @@ impl DFSchema {
/// Create a new schema that contains the fields from this schema followed by the fields
/// from the supplied schema. An error will be returned if there are duplicate field names.
pub fn join(&self, schema: &DFSchema) -> Result<Self> {
- let mut schema_builder = SchemaBuilder::new();
+ let mut schema_builder = SchemaBuilder::with_capacity(
+ self.inner.fields().len() + schema.fields().len(),
+ );
schema_builder.extend(self.inner.fields().iter().cloned());
schema_builder.extend(schema.fields().iter().cloned());
let new_schema = schema_builder.finish();
@@ -338,7 +340,7 @@ impl DFSchema {
.collect();
let mut schema_builder = SchemaBuilder::from(self.inner.fields.clone());
- let mut qualifiers = Vec::new();
+ let mut qualifiers = Vec::with_capacity(other_schema.inner.fields.len());
for (qualifier, field) in other_schema.iter() {
// skip duplicate columns
let duplicated_field = match qualifier {
diff --git a/datafusion/common/src/error.rs b/datafusion/common/src/error.rs
index a103563424..9fceaba3e6 100644
--- a/datafusion/common/src/error.rs
+++ b/datafusion/common/src/error.rs
@@ -45,7 +45,7 @@ use std::io;
use std::result;
use std::sync::Arc;
-use crate::utils::datafusion_strsim::{levenshtein, normalized_levenshtein};
+use crate::utils::datafusion_strsim::{levenshtein_with_buffer, normalized_levenshtein};
use crate::utils::quote_identifier;
use crate::{Column, DFSchema, Diagnostic, TableReference};
use arrow::error::ArrowError;
@@ -236,6 +236,7 @@ fn closest_valid_field<'a>(
];
let mut best_match: Option<(usize, usize, usize, &Column)> = None;
+ let mut cache = Vec::new();
for (index, valid_field) in valid_fields.iter().enumerate() {
let valid_names = [
valid_field.name().to_lowercase(),
@@ -243,7 +244,7 @@ fn closest_valid_field<'a>(
];
for target in &target_names {
for valid_name in &valid_names {
- let distance = levenshtein(target, valid_name);
+ let distance = levenshtein_with_buffer(target, valid_name, &mut cache);
let max_len = target.chars().count().max(valid_name.chars().count());
// If there are no shared characters, or we would have to edit
// more than half of the longer name, don't suggest a potential match.
diff --git a/datafusion/common/src/functional_dependencies.rs b/datafusion/common/src/functional_dependencies.rs
index 5ce3ed3b2b..8bcfdbe97c 100644
--- a/datafusion/common/src/functional_dependencies.rs
+++ b/datafusion/common/src/functional_dependencies.rs
@@ -324,7 +324,7 @@ impl FunctionalDependencies {
// to correctly associate a `Single`-mode dependence with the whole table.
n_out: usize,
) -> FunctionalDependencies {
- let mut projected_func_dependencies = vec![];
+ let mut projected_func_dependencies = Vec::with_capacity(self.deps.len());
for FunctionalDependence {
source_indices,
target_indices,
@@ -658,8 +658,8 @@ pub fn get_required_sort_exprs_indices(
let dependencies = schema.functional_dependencies();
let field_names = schema.field_names();
- let mut known_field_indices = HashSet::new();
- let mut required_sort_expr_indices = Vec::new();
+ let mut known_field_indices = HashSet::with_capacity(sort_expr_names.len());
+ let mut required_sort_expr_indices = Vec::with_capacity(sort_expr_names.len());
for (sort_expr_idx, sort_expr_name) in sort_expr_names.iter().enumerate() {
// If the sort expression doesn't correspond to a known schema field
diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs
index 46b86d1a05..f7a8b037fb 100644
--- a/datafusion/common/src/hash_utils.rs
+++ b/datafusion/common/src/hash_utils.rs
@@ -903,6 +903,7 @@ fn hash_sparse_union_array(
child_hashing.create_hashes([array.child(type_id)], &mut child_hashes)?;
} else {
let (starts, rows) = group_rows_by_type_id(type_ids);
+ let mut taken_hashes: Vec<u64> = Vec::new();
for (type_id, _) in union_fields.iter() {
let (start, end) = (starts[type_id as usize], starts[type_id as usize + 1]);
if start == end {
@@ -931,10 +932,11 @@ fn hash_sparse_union_array(
child_hashing.create_hashes([masked], &mut child_hashes)?;
} else {
let indices = UInt32Array::new(rows.slice(start, end - start), None);
- let mut taken_hashes = vec![0; indices.len()];
+ taken_hashes.clear();
+ taken_hashes.resize(indices.len(), 0);
child_hashing
.create_hashes([take(child, &indices, None)?], &mut taken_hashes)?;
- for (&row, hash) in indices.values().iter().zip(taken_hashes) {
+ for (&row, &hash) in indices.values().iter().zip(&taken_hashes) {
child_hashes[row as usize] = hash;
}
}
diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs
index a0b42b4e42..ca9fc654b7 100644
--- a/datafusion/common/src/scalar/mod.rs
+++ b/datafusion/common/src/scalar/mod.rs
@@ -3745,7 +3745,6 @@ impl ScalarValue {
),
ScalarValue::Union(value, fields, mode) => match value {
Some((v_id, value)) => {
- let mut new_fields = Vec::with_capacity(fields.len());
let mut child_arrays = Vec::<ArrayRef>::with_capacity(fields.len());
for (f_id, field) in fields.iter() {
let ar = if f_id == *v_id {
@@ -3759,9 +3758,7 @@ impl ScalarValue {
UnionMode::Dense => new_null_array(dt, 0),
}
};
- let field = (**field).clone();
child_arrays.push(ar);
- new_fields.push(field.clone());
}
let type_ids = repeat_n(*v_id, size);
let type_ids = ScalarBuffer::<i8>::from_iter(type_ids);
@@ -5720,7 +5717,7 @@ impl fmt::Display for ScalarValue {
.iter()
.map(|struct_array| {
if let Some(arr) = struct_array {
- let mut buffer = VecDeque::new();
+ let mut buffer = VecDeque::with_capacity(arr.len());
for i in 0..arr.len() {
let key =
array_value_to_string(arr.column(0), i).unwrap();
diff --git a/datafusion/common/src/stats.rs b/datafusion/common/src/stats.rs
index 1df71e15a2..9a81ce3a3d 100644
--- a/datafusion/common/src/stats.rs
+++ b/datafusion/common/src/stats.rs
@@ -524,10 +524,13 @@ impl Statistics {
}
// Convert to Vec<Slot> so we can avoid copying the statistics
- let mut columns: Vec<_> = std::mem::take(&mut self.column_statistics)
- .into_iter()
- .map(Slot::Present)
- .collect();
+ let mut columns: Vec<_> = std::mem::replace(
+ &mut self.column_statistics,
+ Vec::with_capacity(projection.len()),
+ )
+ .into_iter()
+ .map(Slot::Present)
+ .collect();
for idx in projection.iter() {
let next_idx = self.column_statistics.len();
diff --git a/datafusion/common/src/utils/mod.rs b/datafusion/common/src/utils/mod.rs
index 107226b42b..60c7767f9c 100644
--- a/datafusion/common/src/utils/mod.rs
+++ b/datafusion/common/src/utils/mod.rs
@@ -184,9 +184,10 @@ pub fn find_bisect_point<F>(
where
F: Fn(&[ScalarValue], &[ScalarValue]) -> Result<bool>,
{
+ let mut val = Vec::with_capacity(item_columns.len());
while low < high {
let mid = ((high - low) / 2) + low;
- let val = get_row_at_idx(item_columns, mid)?;
+ extract_row_at_idx_to_buf(item_columns, mid, &mut val)?;
if compare_fn(&val, target)? {
low = mid + 1;
} else {
@@ -231,8 +232,9 @@ pub fn search_in_slice<F>(
where
F: Fn(&[ScalarValue], &[ScalarValue]) -> Result<bool>,
{
+ let mut val = Vec::new();
while low < high {
- let val = get_row_at_idx(item_columns, low)?;
+ extract_row_at_idx_to_buf(item_columns, low, &mut val)?;
if !compare_fn(&val, target)? {
break;
}
@@ -356,7 +358,7 @@ pub(crate) fn parse_identifiers_normalized(s: &str, ignore_case: bool) -> Vec<St
false
};
if is_double_quoted {
- id[1..id.len() - 1].to_string().replace("\"\"", "\"")
+ id[1..id.len() - 1].replace("\"\"", "\"")
} else if ignore_case {
id
} else {
@@ -1070,7 +1072,9 @@ pub fn transpose<T>(original: Vec<Vec<T>>) -> Vec<Vec<T>> {
match original.as_slice() {
[] => vec![],
[first, ..] => {
- let mut result = (0..first.len()).map(|_| vec![]).collect::<Vec<_>>();
+ let mut result = (0..first.len())
+ .map(|_| Vec::with_capacity(original.len()))
+ .collect::<Vec<_>>();
for row in original {
for (item, transposed_row) in row.into_iter().zip(&mut result) {
transposed_row.push(item);
diff --git a/datafusion/core/src/dataframe/mod.rs b/datafusion/core/src/dataframe/mod.rs
index 55d3695a68..0cf86aa969 100644
--- a/datafusion/core/src/dataframe/mod.rs
+++ b/datafusion/core/src/dataframe/mod.rs
@@ -1107,8 +1107,9 @@ impl DataFrame {
let mut array_ref_vec: Vec<ArrayRef> = vec![Arc::new(StringArray::from(
supported_describe_functions.clone(),
))];
+ let mut array_datas = Vec::with_capacity(describe_record_batch.len());
for field in original_schema_fields {
- let mut array_datas = vec![];
+ array_datas.clear();
for result in describe_record_batch.iter() {
let array_ref = match result {
Ok(df) => {
diff --git a/datafusion/core/src/datasource/listing_table_factory.rs b/datafusion/core/src/datasource/listing_table_factory.rs
index 1e597e38fb..8b86d7b370 100644
--- a/datafusion/core/src/datasource/listing_table_factory.rs
+++ b/datafusion/core/src/datasource/listing_table_factory.rs
@@ -205,7 +205,7 @@ impl ListingTableFactory {
// exclude partition columns to support creating partitioned external table
// with a specified column definition like
// `create external table a(c0 int, c1 int) stored as csv partitioned by (c1)...`
- let mut project_idx = Vec::new();
+ let mut project_idx = Vec::with_capacity(schema.fields().len());
for i in 0..schema.fields().len() {
if !cmd.table_partition_cols.contains(schema.field(i).name()) {
project_idx.push(i);
diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs
index 1b1174e535..13281eb4d2 100644
--- a/datafusion/core/src/execution/context/mod.rs
+++ b/datafusion/core/src/execution/context/mod.rs
@@ -1813,9 +1813,7 @@ impl SessionContext {
// check if the file extension matches the expected extension
for path in &table_paths {
let file_path = path.as_str();
- if !file_path.ends_with(option_extension.clone().as_str())
- && !path.is_collection()
- {
+ if !file_path.ends_with(option_extension.as_str()) && !path.is_collection() {
return exec_err!(
"File path '{file_path}' does not match the expected extension '{option_extension}'"
);
diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs
index 040a90a244..1a895aa6b1 100644
--- a/datafusion/core/src/execution/session_state.rs
+++ b/datafusion/core/src/execution/session_state.rs
@@ -1733,14 +1733,24 @@ impl SessionStateBuilder {
Arc::new(MemoryCatalogProviderList::new()) as Arc<dyn CatalogProviderList>
}),
table_functions: table_functions.unwrap_or_default(),
- scalar_functions: HashMap::new(),
- higher_order_functions: HashMap::new(),
- aggregate_functions: HashMap::new(),
- window_functions: HashMap::new(),
+ scalar_functions: HashMap::with_capacity(
+ scalar_functions.as_ref().map_or(0, Vec::len),
+ ),
+ higher_order_functions: HashMap::with_capacity(
+ higher_order_functions.as_ref().map_or(0, Vec::len),
+ ),
+ aggregate_functions: HashMap::with_capacity(
+ aggregate_functions.as_ref().map_or(0, Vec::len),
+ ),
+ window_functions: HashMap::with_capacity(
+ window_functions.as_ref().map_or(0, Vec::len),
+ ),
extension_types: Arc::new(MemoryExtensionTypeRegistry::default()),
serializer_registry: serializer_registry
.unwrap_or_else(|| Arc::new(EmptySerializerRegistry)),
- file_formats: HashMap::new(),
+ file_formats: HashMap::with_capacity(
+ file_formats.as_ref().map_or(0, Vec::len),
+ ),
table_options: table_options.unwrap_or_else(|| {
TableOptions::default_from_session_config(config.options())
}),
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index b48bc19a55..2637d428bf 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -1888,12 +1888,12 @@ impl DefaultPhysicalPlanner {
LogicalPlan::Extension(Extension { node }) => {
let mut maybe_plan = None;
let children = children.vec();
+ let logical_input = node.inputs();
for planner in &self.extension_planners {
if maybe_plan.is_some() {
break;
}
- let logical_input = node.inputs();
maybe_plan = planner
.plan_extension(
self,
@@ -2519,10 +2519,11 @@ fn extract_dml_filters(
//
// Deduplication is necessary because filters may appear in both Filter nodes
// and TableScan.filters when the optimizer performs partial (Inexact) pushdown.
- let mut seen_filters = HashSet::new();
- filters
- .into_iter()
- .try_fold(Vec::new(), |mut deduped, filter| {
+ let num_filters = filters.len();
+ let mut seen_filters = HashSet::with_capacity(num_filters);
+ filters.into_iter().try_fold(
+ Vec::with_capacity(num_filters),
+ |mut deduped, filter| {
let unqualified = strip_column_qualifiers(filter).map_err(|e| {
e.context(format!(
"Failed to strip column qualifiers for DML filter on table '{target}'"
@@ -2532,7 +2533,8 @@ fn extract_dml_filters(
deduped.push(unqualified);
}
Ok(deduped)
- })
+ },
+ )
}
/// Determine whether a predicate references only columns from the target table
diff --git a/datafusion/datasource-arrow/src/file_format.rs b/datafusion/datasource-arrow/src/file_format.rs
index b874466e2f..d6ad147850 100644
--- a/datafusion/datasource-arrow/src/file_format.rs
+++ b/datafusion/datasource-arrow/src/file_format.rs
@@ -139,7 +139,7 @@ impl FileFormat for ArrowFormat {
store: &Arc<dyn ObjectStore>,
objects: &[ObjectMeta],
) -> Result<SchemaRef> {
- let mut schemas = vec![];
+ let mut schemas = Vec::with_capacity(objects.len());
for object in objects {
let r = store.as_ref().get(&object.location).await?;
let schema = match r.payload {
diff --git a/datafusion/datasource-avro/src/file_format.rs b/datafusion/datasource-avro/src/file_format.rs
index 0c081b0ee9..6aa28667f4 100644
--- a/datafusion/datasource-avro/src/file_format.rs
+++ b/datafusion/datasource-avro/src/file_format.rs
@@ -134,7 +134,7 @@ impl FileFormat for AvroFormat {
store: &Arc<dyn ObjectStore>,
objects: &[ObjectMeta],
) -> Result<SchemaRef> {
- let mut schemas = vec![];
+ let mut schemas = Vec::with_capacity(objects.len());
for object in objects {
let r = store.as_ref().get(&object.location).await?;
let schema = match r.payload {
diff --git a/datafusion/datasource-parquet/src/access_plan.rs b/datafusion/datasource-parquet/src/access_plan.rs
index 961a91c0b4..c1bf0d6d15 100644
--- a/datafusion/datasource-parquet/src/access_plan.rs
+++ b/datafusion/datasource-parquet/src/access_plan.rs
@@ -777,7 +777,8 @@ impl PreparedAccessPlan {
// expression that isn't a plain `Column` in the file schema —
// stats for later columns can't refine the order once an
// unresolvable key sits between them and the resolved prefix.
- let mut sort_columns: Vec<arrow::compute::SortColumn> = Vec::new();
+ let mut sort_columns: Vec<arrow::compute::SortColumn> =
+ Vec::with_capacity(sort_order.len());
for (i, sort_expr) in sort_order.iter().enumerate() {
let column: &Column = match sort_expr.expr.downcast_ref::<Column>() {
Some(col) => col,
diff --git a/datafusion/datasource-parquet/src/page_filter.rs b/datafusion/datasource-parquet/src/page_filter.rs
index 7190a60bda..30952095ae 100644
--- a/datafusion/datasource-parquet/src/page_filter.rs
+++ b/datafusion/datasource-parquet/src/page_filter.rs
@@ -296,6 +296,9 @@ impl PagePruningAccessPlanFilter {
// for each row group specified in the access plan
let row_group_indexes = access_plan.row_group_indexes();
+ // Page index sets, reused across row groups
+ let mut matched_pages_in_group: HashSet<usize> = HashSet::new();
+ let mut matched_pages_indexes: HashSet<usize> = HashSet::new();
for row_group_index in row_group_indexes {
// Skip page pruning for fully matched row groups: all rows are
// known to satisfy the predicate, so page-level pruning is wasted work.
@@ -318,8 +321,8 @@ impl PagePruningAccessPlanFilter {
.and_then(|page_index| page_index.offset_index(row_group_index, 0))
.map_or(0, |offset_index| offset_index.page_locations().len());
// stores the indexes of the matched pages
- let mut matched_pages_in_group: HashSet<usize> =
- HashSet::from_iter(0..total_pages_in_group);
+ matched_pages_in_group.clear();
+ matched_pages_in_group.extend(0..total_pages_in_group);
for predicate in page_index_predicates {
let selection = prune_pages_for_predicate(
@@ -342,12 +345,9 @@ impl PagePruningAccessPlanFilter {
predicate.predicate_expr(),
);
- let matched_pages_indexes: HashSet<_> = pages
- .into_iter()
- .enumerate()
- .filter(|x| x.1)
- .map(|x| x.0)
- .collect();
+ matched_pages_indexes.clear();
+ matched_pages_indexes
+ .extend(pages.into_iter().enumerate().filter(|x| x.1).map(|x| x.0));
matched_pages_in_group.retain(|x| matched_pages_indexes.contains(x));
overall_selection = update_selection(overall_selection, selection);
diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs
index b3feac2d3c..02fc2700d1 100644
--- a/datafusion/datasource-parquet/src/row_group_filter.rs
+++ b/datafusion/datasource-parquet/src/row_group_filter.rs
@@ -367,7 +367,8 @@ impl RowGroupAccessPlanFilter {
// try to prune the row groups in a single call
match predicate.prune(&pruning_stats) {
Ok(values) => {
- let mut fully_contained_candidates_original_idx: Vec<usize> = Vec::new();
+ let mut fully_contained_candidates_original_idx: Vec<usize> =
+ Vec::with_capacity(row_group_indexes.len());
for (idx, &value) in row_group_indexes.iter().zip(values.iter()) {
if !value {
self.access_plan.skip(*idx);
diff --git a/datafusion/datasource-parquet/src/sort.rs b/datafusion/datasource-parquet/src/sort.rs
index 4e683d7fcd..44c55614b3 100644
--- a/datafusion/datasource-parquet/src/sort.rs
+++ b/datafusion/datasource-parquet/src/sort.rs
@@ -119,7 +119,7 @@ fn extract_topk_sort_info(
return vec![];
};
let leading_descending = sort_order.first().options.descending;
- let mut keys = Vec::new();
+ let mut keys = Vec::with_capacity(sort_order.len());
for sort_expr in sort_order.iter() {
let Some(col) = sort_expr.expr.downcast_ref::<Column>() else {
break;
diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs
index 601e3bb2dd..da7d295847 100644
--- a/datafusion/datasource-parquet/src/source.rs
+++ b/datafusion/datasource-parquet/src/source.rs
@@ -1003,8 +1003,7 @@ impl FileSource for ParquetSource {
// is already in prefix order the stats-based reorder is mostly
// a no-op anyway, so the trade-off is plainly bad.
for prefix_len in 1..order.len() {
- let prefix = order[..prefix_len].to_vec();
- if eq_properties.ordering_satisfy(prefix.iter().cloned())? {
+ if eq_properties.ordering_satisfy(order[..prefix_len].iter().cloned())? {
return Ok(SortOrderPushdownResult::Unsupported);
}
}
diff --git a/datafusion/datasource/src/file_scan_config/proto.rs b/datafusion/datasource/src/file_scan_config/proto.rs
index d2dbb5f475..9370a0cfb7 100644
--- a/datafusion/datasource/src/file_scan_config/proto.rs
+++ b/datafusion/datasource/src/file_scan_config/proto.rs
@@ -106,7 +106,7 @@ impl FileScanConfig {
.map(TryInto::try_into)
.collect::<Result<Vec<_>>>()?;
- let mut proto_output_ordering = vec![];
+ let mut proto_output_ordering = Vec::with_capacity(output_ordering.len());
for order in output_ordering {
let nodes = sort_exprs_try_to_proto(order.iter(), &ctx.expr_ctx())?;
proto_output_ordering.push(protobuf::PhysicalSortExprNodeCollection {
@@ -242,7 +242,7 @@ impl FileScanConfig {
true => ObjectStoreUrl::local_filesystem(),
};
- let mut decoded_output_ordering = vec![];
+ let mut decoded_output_ordering = Vec::with_capacity(output_ordering.len());
for node_collection in output_ordering {
let protobuf::PhysicalSortExprNodeCollection {
physical_sort_expr_nodes,
diff --git a/datafusion/datasource/src/memory.rs b/datafusion/datasource/src/memory.rs
index 0d0d02a37f..43f0a05f33 100644
--- a/datafusion/datasource/src/memory.rs
+++ b/datafusion/datasource/src/memory.rs
@@ -756,7 +756,8 @@ impl MemorySourceConfig {
// Stored sort information has already been projected by
// `try_with_sort_information`; decode it against that same schema and
// do not project it a second time.
- let mut decoded_sort_information = vec![];
+ let mut decoded_sort_information =
+ Vec::with_capacity(proto_sort_information.len());
for ordering in proto_sort_information {
let protobuf::PhysicalSortExprNodeCollection {
physical_sort_expr_nodes,
diff --git a/datafusion/datasource/src/projection.rs b/datafusion/datasource/src/projection.rs
index 3cf4f29a77..20292075ff 100644
--- a/datafusion/datasource/src/projection.rs
+++ b/datafusion/datasource/src/projection.rs
@@ -225,7 +225,8 @@ impl SplitProjection {
// Separate file and partition columns, assigning final indices
// Pre-create all remapped columns to avoid duplicate Arc'd expressions
- let mut column_mapping = std::collections::HashMap::new();
+ let mut column_mapping =
+ std::collections::HashMap::with_capacity(sorted_columns.len());
let mut file_idx = 0;
let mut partition_idx = 0;
diff --git a/datafusion/datasource/src/write/demux.rs b/datafusion/datasource/src/write/demux.rs
index cbd0a9747b..7cc2dcc008 100644
--- a/datafusion/datasource/src/write/demux.rs
+++ b/datafusion/datasource/src/write/demux.rs
@@ -344,12 +344,12 @@ async fn hive_style_partitions_demuxer(
// Next compute how the batch should be split up to take each distinct key to its own batch
let take_map = compute_take_arrays(&rb, &all_partition_values);
+ let struct_array: StructArray = rb.clone().into();
// Divide up the batch into distinct partition key batches and send each batch
for (part_key, mut builder) in take_map.into_iter() {
// Take method adapted from https://github.com/lancedb/lance/pull/1337/files
// TODO: upstream RecordBatch::take to arrow-rs
let take_indices = builder.finish();
- let struct_array: StructArray = rb.clone().into();
let parted_batch = RecordBatch::from(
arrow::compute::take(&struct_array, &take_indices, None)?.as_struct(),
);
@@ -408,7 +408,7 @@ fn compute_partition_keys_by_row<'a>(
rb: &'a RecordBatch,
partition_by: &'a [(String, DataType)],
) -> Result<Vec<Vec<Cow<'a, str>>>> {
- let mut all_partition_values = vec![];
+ let mut all_partition_values = Vec::with_capacity(partition_by.len());
const EPOCH_DAYS_FROM_CE: i32 = 719_163;
@@ -419,7 +419,7 @@ fn compute_partition_keys_by_row<'a>(
// batch schema instead.
let schema = rb.schema();
for (col, _) in partition_by.iter() {
- let mut partition_values = vec![];
+ let mut partition_values = Vec::with_capacity(rb.num_rows());
let dtype = schema.field_with_name(col)?.data_type();
let col_array = rb.column_by_name(col).ok_or(exec_datafusion_err!(
@@ -583,13 +583,18 @@ fn compute_take_arrays(
rb: &RecordBatch,
all_partition_values: &[Vec<Cow<str>>],
) -> HashMap<Vec<String>, UInt64Builder> {
- let mut take_map = HashMap::new();
+ let mut take_map: HashMap<Vec<String>, UInt64Builder> = HashMap::new();
+ // Reused for every row, cloned only when a new partition key is seen
+ let mut part_key = vec![String::new(); all_partition_values.len()];
for i in 0..rb.num_rows() {
- let mut part_key = vec![];
- for vals in all_partition_values.iter() {
- part_key.push(vals[i].clone().into());
+ for (key, vals) in part_key.iter_mut().zip(all_partition_values) {
+ key.clear();
+ key.push_str(&vals[i]);
}
- let builder = take_map.entry(part_key).or_insert_with(UInt64Builder::new);
+ let builder = match take_map.get_mut(&part_key) {
+ Some(builder) => builder,
+ None => take_map.entry(part_key.clone()).or_default(),
+ };
builder.append_value(i as u64);
}
take_map
@@ -599,13 +604,12 @@ fn remove_partition_by_columns(
parted_batch: &RecordBatch,
partition_by: &[(String, DataType)],
) -> Result<RecordBatch> {
- let partition_names: Vec<_> = partition_by.iter().map(|(s, _)| s).collect();
let (non_part_cols, non_part_fields): (Vec<_>, Vec<_>) = parted_batch
.columns()
.iter()
.zip(parted_batch.schema().fields())
.filter_map(|(a, f)| {
- if !partition_names.contains(&f.name()) {
+ if !partition_by.iter().any(|(s, _)| s == f.name()) {
Some((Arc::clone(a), (**f).clone()))
} else {
None
diff --git a/datafusion/datasource/src/write/orchestration.rs b/datafusion/datasource/src/write/orchestration.rs
index 8975d0f85e..4f1a10149d 100644
--- a/datafusion/datasource/src/write/orchestration.rs
+++ b/datafusion/datasource/src/write/orchestration.rs
@@ -177,7 +177,7 @@ pub(crate) async fn stateless_serialize_and_write_files(
serialize_rb_stream_to_object_store(data_rx, serializer, writer).await
});
}
- let mut finished_writers = Vec::new();
+ let mut finished_writers = Vec::with_capacity(join_set.len());
while let Some(result) = join_set.join_next().await {
match result {
Ok(res) => match res {
diff --git a/datafusion/execution/src/config.rs b/datafusion/execution/src/config.rs
index 80658477c9..dbfd9bf1d8 100644
--- a/datafusion/execution/src/config.rs
+++ b/datafusion/execution/src/config.rs
@@ -569,9 +569,10 @@ impl SessionConfig {
///
/// [`options`]: Self::options
pub fn to_props(&self) -> HashMap<String, String> {
- let mut map = HashMap::new();
+ let entries = self.options.entries();
+ let mut map = HashMap::with_capacity(entries.len());
// copy configs from config_options
- for entry in self.options.entries() {
+ for entry in entries {
map.insert(entry.key, entry.value.unwrap_or_default());
}
diff --git a/datafusion/expr-common/src/signature.rs b/datafusion/expr-common/src/signature.rs
index cc3c178f56..e9b7067016 100644
--- a/datafusion/expr-common/src/signature.rs
+++ b/datafusion/expr-common/src/signature.rs
@@ -1684,7 +1684,7 @@ impl Signature {
}
}
- let mut seen = std::collections::HashSet::new();
+ let mut seen = std::collections::HashSet::with_capacity(names.len());
for name in names {
if !seen.insert(name) {
return plan_err!("Duplicate parameter name: '{}'", name);
diff --git a/datafusion/expr-common/src/type_coercion/binary.rs b/datafusion/expr-common/src/type_coercion/binary.rs
index c39fcaa416..5355cebd75 100644
--- a/datafusion/expr-common/src/type_coercion/binary.rs
+++ b/datafusion/expr-common/src/type_coercion/binary.rs
@@ -903,7 +903,7 @@ pub fn try_type_union_resolution_with_struct(
let mut keys_string: Option<String> = None;
for data_type in data_types {
if let DataType::Struct(fields) = data_type {
- let keys = fields.iter().map(|f| f.name().to_owned()).join(",");
+ let keys = fields.iter().map(|f| f.name()).join(",");
if let Some(ref k) = keys_string {
if *k != keys {
return exec_err!(
@@ -931,11 +931,10 @@ pub fn try_type_union_resolution_with_struct(
for data_type in data_types.iter().skip(1) {
if let DataType::Struct(fields) = data_type {
- let incoming_struct_types: Vec<DataType> =
- fields.iter().map(|f| f.data_type().to_owned()).collect();
// The order of field is verified above
- for (lhs_type, rhs_type) in
- struct_types.iter_mut().zip(incoming_struct_types.iter())
+ for (lhs_type, rhs_type) in struct_types
+ .iter_mut()
+ .zip(fields.iter().map(|f| f.data_type()))
{
if let Some(coerced_type) =
type_union_resolution_coercion(lhs_type, rhs_type)
@@ -954,9 +953,9 @@ pub fn try_type_union_resolution_with_struct(
}
}
- let mut final_struct_types = vec![];
+ let mut final_struct_types = Vec::with_capacity(data_types.len());
for s in data_types {
- let mut new_fields = vec![];
+ let mut new_fields = Vec::with_capacity(struct_types.len());
if let DataType::Struct(fields) = s {
for (i, f) in fields.iter().enumerate() {
let field = Arc::unwrap_or_clone(Arc::clone(f))
diff --git a/datafusion/expr/src/expr_rewriter/order_by.rs b/datafusion/expr/src/expr_rewriter/order_by.rs
index 5ac17480cb..6945770f26 100644
--- a/datafusion/expr/src/expr_rewriter/order_by.rs
+++ b/datafusion/expr/src/expr_rewriter/order_by.rs
@@ -30,12 +30,18 @@ pub fn rewrite_sort_cols_by_aggs(
sorts: impl IntoIterator<Item = impl Into<Sort>>,
plan: &LogicalPlan,
) -> Result<Vec<Sort>> {
+ let plan_inputs = plan.inputs();
+ let proj_exprs = if plan_inputs.len() == 1 {
+ plan.expressions()
+ } else {
+ vec![]
+ };
sorts
.into_iter()
.map(|e| {
let sort = e.into();
Ok(Sort::new(
- rewrite_sort_col_by_aggs(sort.expr, plan)?,
+ rewrite_sort_col_by_aggs(sort.expr, &plan_inputs, &proj_exprs)?,
sort.asc,
sort.nulls_first,
))
@@ -43,14 +49,15 @@ pub fn rewrite_sort_cols_by_aggs(
.collect()
}
-fn rewrite_sort_col_by_aggs(expr: Expr, plan: &LogicalPlan) -> Result<Expr> {
- let plan_inputs = plan.inputs();
-
+fn rewrite_sort_col_by_aggs(
+ expr: Expr,
+ plan_inputs: &[&LogicalPlan],
+ proj_exprs: &[Expr],
+) -> Result<Expr> {
// Joins, and Unions are not yet handled (should have a projection
// on top of them)
if plan_inputs.len() == 1 {
- let proj_exprs = plan.expressions();
- rewrite_in_terms_of_projection(expr, &proj_exprs, plan_inputs[0])
+ rewrite_in_terms_of_projection(expr, proj_exprs, plan_inputs[0])
} else {
Ok(expr)
}
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index 798b9ea3e9..345180808e 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -331,7 +331,7 @@ impl LogicalPlanBuilder {
if let Some(prev_type) = common_type {
// Widen the running type so that it can hold both the
// previously seen rows and this row's value.
- let data_types = vec![prev_type.clone(), data_type.clone()];
+ let data_types = [prev_type.clone(), data_type.clone()];
let Some(new_type) = type_union_resolution(&data_types) else {
return plan_err!(
"Inconsistent data type across values list at row {i} column {j}. Was {prev_type} but found {data_type}"
@@ -1304,7 +1304,7 @@ impl LogicalPlanBuilder {
.collect::<Result<_>>()?;
let on: Vec<(_, _)> = left_keys.into_iter().zip(right_keys).collect();
- let mut join_on: Vec<(Expr, Expr)> = vec![];
+ let mut join_on: Vec<(Expr, Expr)> = Vec::with_capacity(on.len());
let mut filters: Option<Expr> = None;
for (l, r) in &on {
if self.plan.schema().has_column(l)
@@ -1726,9 +1726,9 @@ pub fn unique_field_aliases(fields: &Fields) -> Vec<Option<String>> {
// That's why we need the `seen` set, so the fields are always unique.
// Tracks a mapping between a field name and the number of appearances of that field.
- let mut name_map = HashMap::<&str, usize>::new();
+ let mut name_map = HashMap::<&str, usize>::with_capacity(fields.len());
// Tracks all the fields and aliases that were previously seen.
- let mut seen = HashSet::<Cow<String>>::new();
+ let mut seen = HashSet::<Cow<String>>::with_capacity(fields.len());
fields
.iter()
@@ -2015,9 +2015,10 @@ pub fn validate_unique_names<'a>(
node_name: &str,
expressions: impl IntoIterator<Item = &'a Expr>,
) -> Result<()> {
- let mut unique_names = HashMap::new();
+ let expressions = expressions.into_iter();
+ let mut unique_names = HashMap::with_capacity(expressions.size_hint().0);
- expressions.into_iter().enumerate().try_for_each(|(position, expr)| {
+ expressions.enumerate().try_for_each(|(position, expr)| {
let name = expr.schema_name().to_string();
match unique_names.get(&name) {
None => {
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index 9d360d052f..e0dd008945 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -522,8 +522,9 @@ impl LogicalPlan {
};
if let Some(on) = on {
// The join keys in using-join must be columns.
- let columns =
- on.iter().try_fold(HashSet::new(), |mut accumu, (l, r)| {
+ let columns = on.iter().try_fold(
+ HashSet::with_capacity(on.len() * 2),
+ |mut accumu, (l, r)| {
let Some(l) = l.get_as_join_column() else {
return internal_err!(
"Invalid join key. Expected column, found {l:?}"
@@ -537,7 +538,8 @@ impl LogicalPlan {
accumu.insert(l.to_owned());
accumu.insert(r.to_owned());
Result::<_, DataFusionError>::Ok(accumu)
- })?;
+ },
+ )?;
using_columns.push(columns);
}
Ok(TreeNodeRecursion::Continue)
@@ -3596,13 +3598,13 @@ impl Union {
}
}
- let mut name_counts: HashMap<String, usize> = HashMap::new();
+ let mut name_counts: HashMap<String, usize> =
+ HashMap::with_capacity(fields_count);
+ let mut fields: Vec<&FieldRef> = Vec::with_capacity(inputs.len());
let union_fields = (0..fields_count)
.map(|i| {
- let fields = inputs
- .iter()
- .map(|input| input.schema().field(i))
- .collect::<Vec<_>>();
+ fields.clear();
+ fields.extend(inputs.iter().map(|input| input.schema().field(i)));
let first_field = fields[0];
let base_name = first_field.name().to_string();
@@ -4352,7 +4354,7 @@ impl PartialOrd for Aggregate {
#[allow(clippy::allow_attributes, clippy::mutable_key_type)] // Expr contains Arc with interior mutability but is intentionally used as hash key
fn max_grouping_set_duplicate_ordinal(group_expr: &[Expr]) -> usize {
if let Some(Expr::GroupingSet(GroupingSet::GroupingSets(sets))) = group_expr.first() {
- let mut counts: HashMap<&[Expr], usize> = HashMap::new();
+ let mut counts: HashMap<&[Expr], usize> = HashMap::with_capacity(sets.len());
for set in sets {
*counts.entry(set).or_insert(0) += 1;
}
@@ -5162,7 +5164,7 @@ impl Unnest {
let input_schema = input.schema();
- let mut dependency_indices = vec![];
+ let mut dependency_indices = Vec::with_capacity(input_schema.fields().len());
// Transform input schema into new schema
// Given this comprehensive example
//
diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs
index 238025b00a..771571f489 100644
--- a/datafusion/expr/src/utils.rs
+++ b/datafusion/expr/src/utils.rs
@@ -70,8 +70,8 @@ pub fn grouping_set_expr_count(group_expr: &[Expr]) -> Result<usize> {
/// Returns an iterator of index vectors, where each vector contains the indices
/// of elements to include in that subset.
fn powerset_indices(len: usize) -> impl Iterator<Item = Vec<usize>> {
- (0..(1 << len)).map(move |mask| {
- let mut indices = vec![];
+ (0..(1 << len)).map(move |mask: u64| {
+ let mut indices = Vec::with_capacity(mask.count_ones() as usize);
let mut bitset = mask;
while bitset > 0 {
let rightmost: u64 = bitset & !(bitset - 1);
@@ -373,7 +373,7 @@ fn get_excluded_columns(
return plan_err!("EXCLUDE or EXCEPT contains duplicate column names");
}
- let mut result = vec![];
+ let mut result = Vec::with_capacity(unique_idents.len());
for ident in unique_idents.into_iter() {
let col_name = ident.value.as_str();
let (qualifier, field) = schema.qualified_field_with_name(qualifier, col_name)?;
@@ -525,8 +525,9 @@ pub fn generate_sort_key(
})
.collect::<Vec<_>>();
- let mut final_sort_keys = vec![];
- let mut is_partition_flag = vec![];
+ let num_keys = partition_by.len() + order_by.len();
+ let mut final_sort_keys = Vec::with_capacity(num_keys);
+ let mut is_partition_flag = Vec::with_capacity(num_keys);
for e in partition_by {
// By default, create sort key with ASC is true and NULLS LAST to be consistent with
// PostgreSQL's rule: https://www.postgresql.org/docs/current/queries-order.html
@@ -1543,14 +1544,12 @@ pub fn collect_subquery_cols(
subquery_schema: &DFSchema,
) -> Result<BTreeSet<Column>> {
exprs.iter().try_fold(BTreeSet::new(), |mut cols, expr| {
- let mut using_cols: Vec<Column> = vec![];
for col in expr.column_refs().into_iter() {
if subquery_schema.has_column(col) {
- using_cols.push(col.clone());
+ cols.insert(col.clone());
}
}
- cols.extend(using_cols);
Result::<_>::Ok(cols)
})
}
diff --git a/datafusion/expr/src/window_state.rs b/datafusion/expr/src/window_state.rs
index b5083a3b77..acba75268b 100644
--- a/datafusion/expr/src/window_state.rs
+++ b/datafusion/expr/src/window_state.rs
@@ -440,13 +440,13 @@ impl WindowFrameStateRange {
delta: Option<&ScalarValue>,
length: usize,
) -> Result<usize> {
- let current_row_values = get_row_at_idx(range_columns, idx)?;
+ let mut end_range = get_row_at_idx(range_columns, idx)?;
let search_start = if SIDE {
last_range.start
} else {
last_range.end
};
- let end_range = if let Some(delta) = delta {
+ if let Some(delta) = delta {
let is_descending: bool = self
.sort_options
.first()
@@ -463,10 +463,9 @@ impl WindowFrameStateRange {
// wrapped-around target: PRECEDING searches reach `search_start`,
// FOLLOWING searches reach `length`.
let unbounded_edge = if SEARCH_SIDE { search_start } else { length };
- let mut targets = Vec::with_capacity(current_row_values.len());
- for value in ¤t_row_values {
+ // Turn the current row values into search targets in place
+ for value in &mut end_range {
if value.is_null() {
- targets.push(value.clone());
continue;
}
let target = if SEARCH_SIDE == is_descending {
@@ -474,23 +473,20 @@ impl WindowFrameStateRange {
Ok(v) => v,
Err(_) => return Ok(unbounded_edge),
}
- } else if value.is_unsigned() && value < delta {
+ } else if value.is_unsigned() && *value < *delta {
// NOTE: This gets a polymorphic zero without having long coercion code for ScalarValue.
// If we decide to implement a "default" construction mechanism for ScalarValue,
// change the following statement to use that.
- value.sub(value)?
+ value.sub(&*value)?
} else {
match value.sub_checked(delta) {
Ok(v) => v,
Err(_) => return Ok(unbounded_edge),
}
};
- targets.push(target);
+ *value = target;
}
- targets
- } else {
- current_row_values
- };
+ }
let compare_fn = |current: &[ScalarValue], target: &[ScalarValue]| {
let cmp = compare_rows(current, target, &self.sort_options)?;
Ok(if SIDE { cmp.is_lt() } else { cmp.is_le() })
diff --git a/datafusion/ffi/src/config/mod.rs b/datafusion/ffi/src/config/mod.rs
index 4c834b0ac0..14e4ca8dea 100644
--- a/datafusion/ffi/src/config/mod.rs
+++ b/datafusion/ffi/src/config/mod.rs
@@ -17,6 +17,8 @@
pub mod extension_options;
+use std::fmt::Write;
+
use datafusion_common::config::{
ConfigExtension, ConfigOptions, ExtensionOptions, TableOptions,
};
@@ -49,11 +51,14 @@ impl From<&ConfigOptions> for FFI_ConfigOptions {
.collect();
let mut extensions = FFI_ExtensionOptions::default();
+ let mut key = String::new();
for (extension_name, extension) in options.extensions.iter() {
for entry in extension.entries().iter() {
if let Some(value) = entry.value.as_ref() {
+ key.clear();
+ write!(key, "{extension_name}.{}", entry.key).unwrap();
extensions
- .set(format!("{extension_name}.{}", entry.key).as_str(), value)
+ .set(key.as_str(), value)
.expect("FFI_ExtensionOptions set should always return Ok");
}
}
@@ -137,11 +142,14 @@ impl From<&TableOptions> for FFI_TableOptions {
.collect();
let mut extensions = FFI_ExtensionOptions::default();
+ let mut key = String::new();
for (extension_name, extension) in options.extensions.iter() {
for entry in extension.entries().iter() {
if let Some(value) = entry.value.as_ref() {
+ key.clear();
+ write!(key, "{extension_name}.{}", entry.key).unwrap();
extensions
- .set(format!("{extension_name}.{}", entry.key).as_str(), value)
+ .set(key.as_str(), value)
.expect("FFI_ExtensionOptions set should always return Ok");
}
}
diff --git a/datafusion/ffi/src/udaf/groups_accumulator.rs b/datafusion/ffi/src/udaf/groups_accumulator.rs
index a23d613150..a60340f5d0 100644
--- a/datafusion/ffi/src/udaf/groups_accumulator.rs
+++ b/datafusion/ffi/src/udaf/groups_accumulator.rs
@@ -138,7 +138,6 @@ unsafe extern "C" fn update_batch_fn_wrapper(
unsafe {
let accumulator = accumulator.inner_mut();
let values = sresult_return!(process_values(values));
- let group_indices: Vec<usize> = group_indices.into_iter().collect();
let opt_filter = sresult_return!(process_opt_filter(opt_filter));
sresult!(accumulator.update_batch(
@@ -196,7 +195,6 @@ unsafe extern "C" fn merge_batch_fn_wrapper(
unsafe {
let accumulator = accumulator.inner_mut();
let values = sresult_return!(process_values(values));
- let group_indices: Vec<usize> = group_indices.into_iter().collect();
sresult!(accumulator.merge_batch(&values, &group_indices, total_num_groups))
}
diff --git a/datafusion/functions-aggregate-common/src/aggregate/count_distinct/groups.rs b/datafusion/functions-aggregate-common/src/aggregate/count_distinct/groups.rs
index 38f6d26200..52e3f2d82e 100644
--- a/datafusion/functions-aggregate-common/src/aggregate/count_distinct/groups.rs
+++ b/datafusion/functions-aggregate-common/src/aggregate/count_distinct/groups.rs
@@ -152,7 +152,10 @@ where
self.seen = HashSet::default();
self.counts = Vec::new();
} else {
- let mut remaining = HashSet::default();
+ let mut remaining = HashSet::with_capacity_and_hasher(
+ self.seen.len().saturating_sub(total as usize),
+ RandomState::default(),
+ );
for (group_idx, value) in self.seen.drain() {
if group_idx < num_emitted {
let pos = cursors[group_idx] as usize;
diff --git a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs b/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs
index dbf2af512b..a2987be918 100644
--- a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs
+++ b/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs
@@ -609,6 +609,7 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
let vec_size_pre = self.states.allocated_size();
let states = emit_to.take_needed(&mut self.states);
+ let num_states = states.len();
// each accumulator produces a potential vector of values
// which we need to form into columns
@@ -617,7 +618,8 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
for mut state in states {
self.free_state_allocation(&state);
let accumulator_state = state.accumulator.state()?;
- results.resize_with(accumulator_state.len(), Vec::new);
+ results
+ .resize_with(accumulator_state.len(), || Vec::with_capacity(num_states));
for (idx, state_val) in accumulator_state.into_iter().enumerate() {
results[idx].push(state_val);
}
@@ -686,10 +688,12 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
// Each row has its respective group
let mut results = vec![];
let mut metric_recorder = None;
+ let mut prepared =
+ Vec::with_capacity(GROUPED_METRIC_PREPARATION_CHUNK_SIZE.min(num_rows));
for chunk_start in (0..num_rows).step_by(GROUPED_METRIC_PREPARATION_CHUNK_SIZE) {
let chunk_end =
(chunk_start + GROUPED_METRIC_PREPARATION_CHUNK_SIZE).min(num_rows);
- let mut prepared = Vec::with_capacity(chunk_end - chunk_start);
+ prepared.clear();
for row_idx in chunk_start..chunk_end {
// Create the empty accumulator and prepare adapter-owned input
@@ -718,7 +722,7 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
};
update_result?;
- for (mut accumulator, _) in prepared {
+ for (mut accumulator, _) in prepared.drain(..) {
let states = accumulator.state()?;
// Resize results to have enough columns according to the converted states
diff --git a/datafusion/functions-aggregate-common/src/merge_arrays.rs b/datafusion/functions-aggregate-common/src/merge_arrays.rs
index 43ce525d8e..c73dde6fe2 100644
--- a/datafusion/functions-aggregate-common/src/merge_arrays.rs
+++ b/datafusion/functions-aggregate-common/src/merge_arrays.rs
@@ -120,7 +120,7 @@ pub fn merge_ordered_arrays(
sort_options: &[SortOptions],
) -> datafusion_common::Result<(Vec<ScalarValue>, Vec<Vec<ScalarValue>>)> {
// Keep track of the most recent data of each branch, in a binary heap data structure.
- let mut heap = BinaryHeap::<CustomElement>::new();
+ let mut heap = BinaryHeap::<CustomElement>::with_capacity(values.len());
if values.len() != ordering_values.len() {
return exec_err!(
@@ -143,8 +143,9 @@ pub fn merge_ordered_arrays(
);
}
let n_branch = values.len();
- let mut merged_values = vec![];
- let mut merged_orderings = vec![];
+ let total_len: usize = values.iter().map(VecDeque::len).sum();
+ let mut merged_values = Vec::with_capacity(total_len);
+ let mut merged_orderings = Vec::with_capacity(total_len);
// Continue iterating the loop until consuming data of all branches.
loop {
let minimum = if let Some(minimum) = heap.pop() {
diff --git a/datafusion/functions-aggregate-common/src/tdigest.rs b/datafusion/functions-aggregate-common/src/tdigest.rs
index ec9b988df2..2c244b567c 100644
--- a/datafusion/functions-aggregate-common/src/tdigest.rs
+++ b/datafusion/functions-aggregate-common/src/tdigest.rs
@@ -312,8 +312,9 @@ impl TDigest {
first: usize,
middle: usize,
last: usize,
+ result: &mut Vec<Centroid>,
) {
- let mut result: Vec<Centroid> = Vec::with_capacity(centroids.len());
+ result.clear();
let mut i = first;
let mut j = middle;
@@ -346,7 +347,7 @@ impl TDigest {
}
i = first;
- for centroid in result.into_iter() {
+ for centroid in result.drain(..) {
centroids[i] = centroid;
i += 1;
}
@@ -389,6 +390,7 @@ impl TDigest {
return TDigest::default();
}
+ let mut merge_buffer: Vec<Centroid> = Vec::with_capacity(centroids.len());
let mut digests_per_block: usize = 1;
while digests_per_block < starts.len() {
for i in (0..starts.len()).step_by(digests_per_block * 2) {
@@ -402,7 +404,13 @@ impl TDigest {
};
debug_assert!(first <= middle && middle <= last);
- Self::external_merge(&mut centroids, first, middle, last);
+ Self::external_merge(
+ &mut centroids,
+ first,
+ middle,
+ last,
+ &mut merge_buffer,
+ );
}
}
diff --git a/datafusion/functions-aggregate/src/approx_percentile_cont_with_weight.rs b/datafusion/functions-aggregate/src/approx_percentile_cont_with_weight.rs
index 1a810a3c36..6a893fc199 100644
--- a/datafusion/functions-aggregate/src/approx_percentile_cont_with_weight.rs
+++ b/datafusion/functions-aggregate/src/approx_percentile_cont_with_weight.rs
@@ -339,7 +339,7 @@ impl Accumulator for ApproxPercentileWithWeightAccumulator {
);
let means_f64 = ApproxPercentileAccumulator::convert_to_float(&means)?;
let weights_f64 = ApproxPercentileAccumulator::convert_to_float(&weights)?;
- let mut digests: Vec<TDigest> = vec![];
+ let mut digests: Vec<TDigest> = Vec::with_capacity(means_f64.len());
for (mean, weight) in means_f64.iter().zip(weights_f64.iter()) {
digests.push(TDigest::new_with_centroid(
self.approx_percentile_cont_accumulator.max_size(),
diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs
index 70a04e0fe9..0e49abd5bb 100644
--- a/datafusion/functions-aggregate/src/array_agg.rs
+++ b/datafusion/functions-aggregate/src/array_agg.rs
@@ -631,7 +631,12 @@ impl GroupsAccumulator for ArrayAggGroupsAccumulator {
None
};
- let mut entries = Vec::new();
+ // Without a filter or nulls to skip, every row becomes an entry
+ let mut entries = if opt_filter.is_none() && nulls.is_none() {
+ Vec::with_capacity(group_indices.len())
+ } else {
+ Vec::new()
+ };
for (row_idx, &group_idx) in group_indices.iter().enumerate() {
// Skip filtered rows
@@ -757,7 +762,12 @@ impl GroupsAccumulator for ArrayAggGroupsAccumulator {
let list_values = input_list.values();
let list_offsets = input_list.offsets();
- let mut entries = Vec::new();
+ // Without null rows, every value in the offset span becomes an entry
+ let mut entries = if input_list.null_count() == 0 {
+ Vec::with_capacity(offset_span(list_offsets).1)
+ } else {
+ Vec::new()
+ };
for (row_idx, &group_idx) in group_indices.iter().enumerate() {
if input_list.is_null(row_idx) {
@@ -1682,6 +1692,7 @@ impl Accumulator for OrderSensitiveArrayAggAccumulator {
"ordered array_agg ordering state must contain Struct values"
)
})?;
+ let mut ordering_columns = Vec::with_capacity(ordering_struct.num_columns());
for row_idx in 0..array_agg_values.len() {
let payload_len = if array_agg_values.is_null(row_idx) {
0
@@ -1701,11 +1712,13 @@ impl Accumulator for OrderSensitiveArrayAggAccumulator {
let payload = array_agg_values.value(row_idx);
let ordering_start = agg_orderings.value_offsets()[row_idx] as usize;
let len = payload.len();
- let ordering_columns = ordering_struct
- .columns()
- .iter()
- .map(|column| column.slice(ordering_start, len))
- .collect::<Vec<_>>();
+ ordering_columns.clear();
+ ordering_columns.extend(
+ ordering_struct
+ .columns()
+ .iter()
+ .map(|column| column.slice(ordering_start, len)),
+ );
self.append_sorted_run(&payload, &ordering_columns)?;
}
self.can_extend_preordered_run = false;
diff --git a/datafusion/functions-aggregate/src/first_last.rs b/datafusion/functions-aggregate/src/first_last.rs
index 17be6359cf..fba4aa6977 100644
--- a/datafusion/functions-aggregate/src/first_last.rs
+++ b/datafusion/functions-aggregate/src/first_last.rs
@@ -824,14 +824,14 @@ impl<S: ValueState + 'static> GroupsAccumulator for FirstLastGroupsAccumulator<S
fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
let (val_arr, orderings, is_sets) = self.take_state(emit_to)?;
- let mut result = Vec::with_capacity(self.orderings.len() + 2);
+ let mut result = Vec::with_capacity(self.ordering_req.len() + 2);
result.push(val_arr);
let ordering_cols = {
let mut ordering_cols = Vec::with_capacity(self.ordering_req.len());
for _ in 0..self.ordering_req.len() {
- ordering_cols.push(Vec::with_capacity(self.orderings.len()));
+ ordering_cols.push(Vec::with_capacity(orderings.len()));
}
for row in orderings.into_iter() {
debug_assert_eq!(row.len(), self.ordering_req.len());
@@ -1091,7 +1091,8 @@ impl FirstValueAccumulator {
impl Accumulator for FirstValueAccumulator {
fn state(&mut self) -> Result<Vec<ScalarValue>> {
- let mut result = vec![self.first.clone()];
+ let mut result = Vec::with_capacity(self.orderings.len() + 2);
+ result.push(self.first.clone());
result.extend(self.orderings.iter().cloned());
result.push(ScalarValue::from(self.is_set));
Ok(result)
@@ -1484,8 +1485,9 @@ impl LastValueAccumulator {
impl Accumulator for LastValueAccumulator {
fn state(&mut self) -> Result<Vec<ScalarValue>> {
- let mut result = vec![self.last.clone()];
- result.extend(self.orderings.clone());
+ let mut result = Vec::with_capacity(self.orderings.len() + 2);
+ result.push(self.last.clone());
+ result.extend(self.orderings.iter().cloned());
result.push(ScalarValue::from(self.is_set));
Ok(result)
}
diff --git a/datafusion/functions-aggregate/src/nth_value.rs b/datafusion/functions-aggregate/src/nth_value.rs
index d3571bc925..e3ed110712 100644
--- a/datafusion/functions-aggregate/src/nth_value.rs
+++ b/datafusion/functions-aggregate/src/nth_value.rs
@@ -363,8 +363,8 @@ impl NthValueAccumulator {
fn evaluate_orderings(&self) -> Result<ScalarValue> {
let fields = ordering_fields(&self.ordering_req, &self.datatypes[1..]);
- let mut column_wise_ordering_values = vec![];
let num_columns = fields.len();
+ let mut column_wise_ordering_values = Vec::with_capacity(num_columns);
for i in 0..num_columns {
let column_values = self
.ordering_values
diff --git a/datafusion/functions-aggregate/src/percentile_cont.rs b/datafusion/functions-aggregate/src/percentile_cont.rs
index 5b1f6bdc1f..b6e97d5897 100644
--- a/datafusion/functions-aggregate/src/percentile_cont.rs
+++ b/datafusion/functions-aggregate/src/percentile_cont.rs
@@ -526,10 +526,13 @@ where
}
fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ let arr = values[0].as_primitive::<T>();
let mut to_remove: HashMap<Hashable<T::Native>, usize, RandomState> =
- HashMap::default();
+ HashMap::with_capacity_and_hasher(
+ arr.len() - arr.null_count(),
+ RandomState::default(),
+ );
- let arr = values[0].as_primitive::<T>();
if arr.null_count() > 0 {
for value in arr.iter().flatten() {
*to_remove.entry(Hashable(value)).or_default() += 1;
@@ -669,7 +672,7 @@ where
let emit_group_values = emit_to.take_needed(&mut self.group_values);
// Build offsets
- let mut offsets = Vec::with_capacity(self.group_values.len() + 1);
+ let mut offsets = Vec::with_capacity(emit_group_values.len() + 1);
offsets.push(0);
let mut cur_len = 0_i32;
for group_value in &emit_group_values {
@@ -679,8 +682,9 @@ where
let offsets = OffsetBuffer::new(ScalarBuffer::from(offsets));
// Build inner array
- let flatten_group_values =
- emit_group_values.into_iter().flatten().collect::<Vec<_>>();
+ let mut flatten_group_values: Vec<T::Native> =
+ Vec::with_capacity(cur_len as usize);
+ flatten_group_values.extend(emit_group_values.into_iter().flatten());
let group_values_array =
PrimitiveArray::<T>::new(ScalarBuffer::from(flatten_group_values), None)
.with_data_type(self.data_type.clone());
diff --git a/datafusion/functions-nested/src/concat.rs b/datafusion/functions-nested/src/concat.rs
index 4f8c2e7975..60f8fc67c7 100644
--- a/datafusion/functions-nested/src/concat.rs
+++ b/datafusion/functions-nested/src/concat.rs
@@ -603,7 +603,8 @@ fn generic_append_and_prepend<O: OffsetSizeTrait>(
where
i64: TryInto<O>,
{
- let mut offsets = vec![O::usize_as(0)];
+ let mut offsets = Vec::with_capacity(list_array.len() + 1);
+ offsets.push(O::usize_as(0));
let values = list_array.values();
let original_data = values.to_data();
let element_data = element_array.to_data();
diff --git a/datafusion/functions-nested/src/extract.rs b/datafusion/functions-nested/src/extract.rs
index 8f6cc2ce5d..25ce9ecfeb 100644
--- a/datafusion/functions-nested/src/extract.rs
+++ b/datafusion/functions-nested/src/extract.rs
@@ -509,9 +509,9 @@ enum SlicePlan<O: OffsetSizeTrait> {
/// A contiguous run starting at `start` (relative to the row) with `len`
/// elements can be copied in one go.
Contiguous { start: O, len: O },
- /// Arbitrary positions (already relative to the row) must be copied in
- /// sequence.
- Indices(Vec<O>),
+ /// Arbitrary positions (already relative to the row), written to the
+ /// caller's index buffer, must be copied in sequence.
+ Indices,
}
/// Produces a [`SlicePlan`] for the given logical slice parameters.
@@ -520,6 +520,7 @@ fn compute_slice_plan<O: OffsetSizeTrait>(
from_raw: i64,
to_raw: i64,
stride_raw: Option<i64>,
+ indices: &mut Vec<O>,
) -> Result<SlicePlan<O>>
where
i64: TryInto<O>,
@@ -558,22 +559,22 @@ where
let len = to - from + O::usize_as(1);
Ok(SlicePlan::Contiguous { start: from, len })
} else {
- let mut indices = Vec::new();
+ indices.clear();
let mut index = from;
while index <= to {
indices.push(index);
index += stride;
}
- Ok(SlicePlan::Indices(indices))
+ Ok(SlicePlan::Indices)
}
} else {
- let mut indices = Vec::new();
+ indices.clear();
let mut index = from;
while index >= to {
indices.push(index);
index += stride;
}
- Ok(SlicePlan::Indices(indices))
+ Ok(SlicePlan::Indices)
}
}
@@ -618,7 +619,9 @@ where
// We have the slice syntax compatible with DuckDB v0.8.1.
// The rule `adjusted_from_index` and `adjusted_to_index` follows the rule of array_slice in duckdb.
- let mut offsets = vec![O::usize_as(0)];
+ let mut offsets = Vec::with_capacity(array.len() + 1);
+ offsets.push(O::usize_as(0));
+ let mut indices = Vec::new();
let nulls = combine_input_nulls(array, from_array, to_array, stride);
@@ -646,6 +649,7 @@ where
from_array.value(row_index),
to_array.value(row_index),
stride.map(|s| s.value(row_index)),
+ &mut indices,
)?;
match slice_plan {
@@ -659,9 +663,9 @@ where
mutable.try_extend(0, start_index, end_index)?;
offsets.push(offsets[row_index] + slice_len);
}
- SlicePlan::Indices(indices) => {
+ SlicePlan::Indices => {
let count = indices.len();
- for rel_index in indices {
+ for &rel_index in &indices {
let absolute_index = (start + rel_index).to_usize().unwrap();
mutable.try_extend(0, absolute_index, absolute_index + 1)?;
}
@@ -710,6 +714,7 @@ where
let mut offsets = Vec::with_capacity(array.len());
let mut sizes = Vec::with_capacity(array.len());
let mut current_offset = O::usize_as(0);
+ let mut indices = Vec::new();
let nulls = combine_input_nulls(array, from_array, to_array, stride);
@@ -734,6 +739,7 @@ where
from_array.value(row_index),
to_array.value(row_index),
stride.map(|s| s.value(row_index)),
+ &mut indices,
)?;
let start = array.value_offset(row_index);
@@ -753,9 +759,9 @@ where
sizes.push(slice_len);
current_offset += slice_len;
}
- SlicePlan::Indices(indices) => {
+ SlicePlan::Indices => {
let count = indices.len();
- for rel_index in indices {
+ for &rel_index in &indices {
let absolute_index = (start + rel_index).to_usize().unwrap();
mutable.try_extend(0, absolute_index, absolute_index + 1)?;
}
diff --git a/datafusion/functions-nested/src/make_array.rs b/datafusion/functions-nested/src/make_array.rs
index 2b90aca62e..e7b27f04b8 100644
--- a/datafusion/functions-nested/src/make_array.rs
+++ b/datafusion/functions-nested/src/make_array.rs
@@ -205,7 +205,7 @@ pub fn array_array<O: OffsetSizeTrait>(
));
}
- let mut data = vec![];
+ let mut data = Vec::with_capacity(args.len());
let mut total_len = 0;
for arg in args {
let arg_data = if arg.as_any().is::<NullArray>() {
diff --git a/datafusion/functions-nested/src/map.rs b/datafusion/functions-nested/src/map.rs
index 264bbf2922..369c35a820 100644
--- a/datafusion/functions-nested/src/map.rs
+++ b/datafusion/functions-nested/src/map.rs
@@ -594,8 +594,8 @@ fn build_map_array(
return exec_err!("map requires key and value lists to have the same length");
}
- let mut key_array_vec = vec![];
- let mut value_array_vec = vec![];
+ let mut key_array_vec = Vec::with_capacity(keys.len());
+ let mut value_array_vec = Vec::with_capacity(values.len());
for (k, v) in keys.iter().zip(values.iter()) {
key_array_vec.push(k.as_ref());
value_array_vec.push(v.as_ref());
@@ -606,7 +606,8 @@ fn build_map_array(
// If it's not NULL, the offset advances by the number of entries in that map
// NOTE: MapArray always requires i32 offsets, regardless of input list type
let mut running_offset = 0i32;
- let mut offset_buffer = vec![running_offset];
+ let mut offset_buffer = Vec::with_capacity(original_len + 1);
+ offset_buffer.push(running_offset);
let mut non_null_idx = 0;
for i in 0..original_len {
let is_null = nulls_bitmap.as_ref().is_some_and(|nulls| nulls.is_null(i));
diff --git a/datafusion/functions-nested/src/range.rs b/datafusion/functions-nested/src/range.rs
index 264bd70cff..819ec6dd3d 100644
--- a/datafusion/functions-nested/src/range.rs
+++ b/datafusion/functions-nested/src/range.rs
@@ -327,7 +327,8 @@ impl Range {
};
let mut values = vec![];
- let mut offsets = vec![0];
+ let mut offsets = Vec::with_capacity(stop_array.len() + 1);
+ offsets.push(0);
let mut valid = NullBufferBuilder::new(stop_array.len());
for (idx, stop) in stop_array.iter().enumerate() {
match retrieve_range_args(start_array, stop, step_array, idx) {
diff --git a/datafusion/functions-nested/src/resize.rs b/datafusion/functions-nested/src/resize.rs
index 4308b1b1e2..6b682e3ce9 100644
--- a/datafusion/functions-nested/src/resize.rs
+++ b/datafusion/functions-nested/src/resize.rs
@@ -319,7 +319,8 @@ where
F: FnMut(&mut MutableArrayData, usize, usize) -> Result<()>,
{
let capacity = Capacities::Array(output_values_len);
- let mut offsets = vec![O::usize_as(0)];
+ let mut offsets = Vec::with_capacity(array.len() + 1);
+ offsets.push(O::usize_as(0));
let sources = match default_value_data {
Some(default_value_data) => vec![original_data, default_value_data],
None => vec![original_data],
diff --git a/datafusion/functions-nested/src/reverse.rs b/datafusion/functions-nested/src/reverse.rs
index 2587a5ebff..69745c6ee1 100644
--- a/datafusion/functions-nested/src/reverse.rs
+++ b/datafusion/functions-nested/src/reverse.rs
@@ -149,7 +149,8 @@ fn general_array_reverse<O: OffsetSizeTrait>(
field: &FieldRef,
) -> Result<ArrayRef> {
let values = array.values();
- let mut offsets = vec![O::usize_as(0)];
+ let mut offsets = Vec::with_capacity(array.len() + 1);
+ offsets.push(O::usize_as(0));
let mut indices: Vec<O> = Vec::with_capacity(offset_span_len(array.offsets()));
for (row_index, (&start, &end)) in array.offsets().iter().tuple_windows().enumerate()
diff --git a/datafusion/functions-nested/src/utils.rs b/datafusion/functions-nested/src/utils.rs
index def7b3a6bc..cb4bea42aa 100644
--- a/datafusion/functions-nested/src/utils.rs
+++ b/datafusion/functions-nested/src/utils.rs
@@ -168,7 +168,7 @@ pub(crate) fn align_array_dimensions<O: OffsetSizeTrait>(
let mut aligned_array = Arc::clone(&array);
for _ in 0..(max_ndim - ndim) {
let data_type = aligned_array.data_type().to_owned();
- let array_lengths = vec![1; aligned_array.len()];
+ let array_lengths = std::iter::repeat_n(1, aligned_array.len());
let offsets = OffsetBuffer::<O>::from_lengths(array_lengths);
aligned_array = Arc::new(GenericListArray::<O>::try_new(
diff --git a/datafusion/functions-table/src/generate_series.rs b/datafusion/functions-table/src/generate_series.rs
index bca6081ebd..32db9bc5be 100644
--- a/datafusion/functions-table/src/generate_series.rs
+++ b/datafusion/functions-table/src/generate_series.rs
@@ -665,7 +665,7 @@ impl TableFunctionImpl for GenerateSeriesFuncImpl {
impl GenerateSeriesFuncImpl {
fn call_int64(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
- let mut normalize_args = Vec::new();
+ let mut normalize_args = Vec::with_capacity(exprs.len());
for (expr_index, expr) in exprs.iter().enumerate() {
match expr {
Expr::Literal(ScalarValue::Null, _) => {}
diff --git a/datafusion/functions/src/core/planner.rs b/datafusion/functions/src/core/planner.rs
index 22d9b5c428..b15156db4d 100644
--- a/datafusion/functions/src/core/planner.rs
+++ b/datafusion/functions/src/core/planner.rs
@@ -33,7 +33,7 @@ impl ExprPlanner for CoreFunctionPlanner {
expr: RawDictionaryExpr,
_schema: &DFSchema,
) -> Result<PlannerResult<RawDictionaryExpr>> {
- let mut args = vec![];
+ let mut args = Vec::with_capacity(expr.keys.len() * 2);
for (k, v) in expr.keys.into_iter().zip(expr.values) {
args.push(k);
args.push(v);
diff --git a/datafusion/functions/src/crypto/basic.rs b/datafusion/functions/src/crypto/basic.rs
index e848daaed1..67621adcfc 100644
--- a/datafusion/functions/src/crypto/basic.rs
+++ b/datafusion/functions/src/crypto/basic.rs
@@ -143,7 +143,7 @@ impl DigestAlgorithm {
opt.map(|x| {
let mut digest = Blake3::default();
digest.update(x.as_bytes());
- Blake3::finalize(&digest).as_bytes().to_vec()
+ *Blake3::finalize(&digest).as_bytes()
})
})
.collect();
@@ -174,7 +174,7 @@ impl DigestAlgorithm {
opt.map(|x| {
let mut digest = Blake3::default();
digest.update(x);
- Blake3::finalize(&digest).as_bytes().to_vec()
+ *Blake3::finalize(&digest).as_bytes()
})
})
.collect();
diff --git a/datafusion/functions/src/regex/regexplike.rs b/datafusion/functions/src/regex/regexplike.rs
index 546e1dcfb5..632d5da640 100644
--- a/datafusion/functions/src/regex/regexplike.rs
+++ b/datafusion/functions/src/regex/regexplike.rs
@@ -327,12 +327,12 @@ pub fn regexp_like(args: &[ArrayRef]) -> Result<ArrayRef> {
}
_ => {
let string_view_array = flags.as_string_view();
- let string_vec: Vec<Option<String>> = (0..string_view_array.len())
+ let string_vec: Vec<Option<&str>> = (0..string_view_array.len())
.map(|i| {
if string_view_array.is_null(i) {
None
} else {
- Some(string_view_array.value(i).to_string())
+ Some(string_view_array.value(i))
}
})
.collect();
diff --git a/datafusion/functions/src/regex/regexpreplace.rs b/datafusion/functions/src/regex/regexpreplace.rs
index 647abfed59..2c0fad9072 100644
--- a/datafusion/functions/src/regex/regexpreplace.rs
+++ b/datafusion/functions/src/regex/regexpreplace.rs
@@ -399,6 +399,8 @@ where
}
Some(flags_array) => {
let flags_array_iter = ArrayIter::new(flags_array);
+ // Reused across rows to build the pattern cache key
+ let mut pattern_buf = String::new();
let result_iter = string_array_iter
.zip(pattern_array_iter)
@@ -412,26 +414,27 @@ where
// format flags into rust pattern
let replace_all = flags.contains('g');
let flags = flags.replace('g', "");
- let pattern = if flags.is_empty() {
- pattern.to_string()
- } else {
- format!("(?{flags}){pattern}")
- };
+ pattern_buf.clear();
+ if !flags.is_empty() {
+ pattern_buf.push_str("(?");
+ pattern_buf.push_str(&flags);
+ pattern_buf.push(')');
+ }
+ pattern_buf.push_str(pattern);
+ let pattern = pattern_buf.as_str();
// if patterns hashmap already has regexp then use else create and return
- let re = match patterns.get(&pattern) {
+ let re = match patterns.get(pattern) {
Some(re) => Ok(re),
- None => match compile_regex(
- "regexp_replace",
- pattern.as_str(),
- None,
- ) {
- Ok(re) => {
- patterns.insert(pattern.clone(), re);
- Ok(patterns.get(&pattern).unwrap())
+ None => {
+ match compile_regex("regexp_replace", pattern, None) {
+ Ok(re) => {
+ patterns.insert(pattern.to_string(), re);
+ Ok(patterns.get(pattern).unwrap())
+ }
+ Err(err) => Err(err),
}
- Err(err) => Err(err),
- },
+ }
};
Some(re.map(|re| {
diff --git a/datafusion/functions/src/string/common.rs b/datafusion/functions/src/string/common.rs
index f2544fa386..968790738d 100644
--- a/datafusion/functions/src/string/common.rs
+++ b/datafusion/functions/src/string/common.rs
@@ -390,6 +390,19 @@ fn unicode_case(s: &str, lower: bool) -> String {
}
}
+/// Like [`unicode_case`], but uppercases into `buf` so its allocation is reused
+/// across rows. Lowercasing still allocates, see [`write_unicode_case`].
+#[inline]
+fn unicode_case_into<'a>(s: &str, lower: bool, buf: &'a mut String) -> &'a str {
+ if lower {
+ *buf = s.to_lowercase();
+ } else {
+ buf.clear();
+ buf.extend(s.chars().flat_map(char::to_uppercase));
+ }
+ buf
+}
+
/// Writes the case-converted form of `s` directly into `w`.
///
/// Uppercasing is a context-free character mapping, so each character is
@@ -483,6 +496,7 @@ fn case_conversion_utf8view(array: &ArrayRef, lower: bool) -> Result<ArrayRef> {
// Null-preserving: reuse the input null buffer as the output null buffer.
let nulls = string_array.nulls().cloned();
let mut builder = StringViewArrayBuilder::with_capacity(item_len);
+ let mut buf = String::new();
if let Some(ref n) = nulls {
for i in 0..item_len {
@@ -491,14 +505,14 @@ fn case_conversion_utf8view(array: &ArrayRef, lower: bool) -> Result<ArrayRef> {
} else {
// SAFETY: `n.is_null(i)` was false in the branch above.
let s = unsafe { string_array.value_unchecked(i) };
- builder.try_append_value(&unicode_case(s, lower))?;
+ builder.try_append_value(unicode_case_into(s, lower, &mut buf))?;
}
}
} else {
for i in 0..item_len {
// SAFETY: no null buffer means every index is valid.
let s = unsafe { string_array.value_unchecked(i) };
- builder.try_append_value(&unicode_case(s, lower))?;
+ builder.try_append_value(unicode_case_into(s, lower, &mut buf))?;
}
}
diff --git a/datafusion/functions/src/unicode/reverse.rs b/datafusion/functions/src/unicode/reverse.rs
index 9815e22cab..b05d8f847d 100644
--- a/datafusion/functions/src/unicode/reverse.rs
+++ b/datafusion/functions/src/unicode/reverse.rs
@@ -26,6 +26,7 @@ use arrow::array::{Array, ArrayRef, AsArray, StringArrayType};
use arrow::datatypes::DataType;
use datafusion_common::Result;
use datafusion_common::types::logical_string;
+use datafusion_common::utils::offset_span_len;
use datafusion_expr::{
Coercion, ColumnarValue, Documentation, EncodingPreservation, ScalarFunctionArgs,
ScalarUDFImpl, Signature, TypeSignatureClass, Volatility,
@@ -102,14 +103,26 @@ fn reverse(args: &[ArrayRef]) -> Result<ArrayRef> {
let len = args[0].len();
match args[0].data_type() {
- LargeUtf8 => reverse_impl(
- &args[0].as_string::<i64>(),
- GenericStringArrayBuilder::<i64>::with_capacity(len, 1024),
- ),
- Utf8 => reverse_impl(
- &args[0].as_string::<i32>(),
- GenericStringArrayBuilder::<i32>::with_capacity(len, 1024),
- ),
+ LargeUtf8 => {
+ let string_array = args[0].as_string::<i64>();
+ reverse_impl(
+ &string_array,
+ GenericStringArrayBuilder::<i64>::with_capacity(
+ len,
+ offset_span_len(string_array.offsets()),
+ ),
+ )
+ }
+ Utf8 => {
+ let string_array = args[0].as_string::<i32>();
+ reverse_impl(
+ &string_array,
+ GenericStringArrayBuilder::<i32>::with_capacity(
+ len,
+ offset_span_len(string_array.offsets()),
+ ),
+ )
+ }
Utf8View => reverse_impl(
&args[0].as_string_view(),
StringViewArrayBuilder::with_capacity(len),
diff --git a/datafusion/optimizer/src/analyzer/resolve_grouping_function.rs b/datafusion/optimizer/src/analyzer/resolve_grouping_function.rs
index 8b1077c6b0..2dbc08cfdd 100644
--- a/datafusion/optimizer/src/analyzer/resolve_grouping_function.rs
+++ b/datafusion/optimizer/src/analyzer/resolve_grouping_function.rs
@@ -84,9 +84,9 @@ fn replace_grouping_exprs(
let group_expr_to_bitmap_index = group_expr_to_bitmap_index(&group_expr)?;
let columns = schema.columns();
let mut new_agg_expr = Vec::new();
- let mut projection_exprs = Vec::new();
let grouping_id_len = usize::from(is_grouping_set);
let group_expr_len = columns.len() - aggr_expr.len() - grouping_id_len;
+ let mut projection_exprs = Vec::with_capacity(group_expr_len + aggr_expr.len());
projection_exprs.extend(
columns
.iter()
diff --git a/datafusion/optimizer/src/common_subexpr_eliminate.rs b/datafusion/optimizer/src/common_subexpr_eliminate.rs
index 48501f4258..e0d43d7c15 100644
--- a/datafusion/optimizer/src/common_subexpr_eliminate.rs
+++ b/datafusion/optimizer/src/common_subexpr_eliminate.rs
@@ -336,7 +336,9 @@ impl CommonSubexprEliminate {
.map(|(expr, expr_alias)| expr.alias(expr_alias))
.collect::<Vec<_>>();
- let mut proj_exprs = vec![];
+ let mut proj_exprs = Vec::with_capacity(
+ new_group_expr.len() + rewritten_aggr_expr.len(),
+ );
for expr in &new_group_expr {
extract_expressions(expr, &mut proj_exprs)
}
diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs
index e4577d6eb0..43037b258f 100644
--- a/datafusion/optimizer/src/decorrelate.rs
+++ b/datafusion/optimizer/src/decorrelate.rs
@@ -681,7 +681,8 @@ impl PullUpCorrelatedExpr {
exprs: &[Expr],
correlated_subquery_cols: &BTreeSet<Column>,
) -> Result<Vec<Expr>> {
- let mut missing_exprs = vec![];
+ let mut missing_exprs =
+ Vec::with_capacity(exprs.len() + correlated_subquery_cols.len());
for expr in exprs {
if !missing_exprs.contains(expr) {
missing_exprs.push(expr.clone())
@@ -835,7 +836,7 @@ fn collect_local_correlated_cols(
) {
for child in plan.inputs() {
if let Some(cols) = all_cols_map.get(child) {
- local_cols.extend(cols.clone());
+ local_cols.extend(cols.iter().cloned());
}
// SubqueryAlias is treated as the leaf node
if !matches!(child, LogicalPlan::SubqueryAlias(_)) {
@@ -884,6 +885,10 @@ fn agg_exprs_evaluation_result_on_empty_batch(
schema: &DFSchemaRef,
expr_result_map_for_count_bug: &mut ExprResultMap,
) -> Result<()> {
+ let info = SimplifyContext::builder()
+ .with_schema(Arc::clone(schema))
+ .build();
+ let simplifier = ExprSimplifier::new(info);
for e in agg_expr.iter() {
let result_expr = e
.clone()
@@ -900,10 +905,6 @@ fn agg_exprs_evaluation_result_on_empty_batch(
.data()?;
let result_expr = result_expr.unalias();
- let info = SimplifyContext::builder()
- .with_schema(Arc::clone(schema))
- .build();
- let simplifier = ExprSimplifier::new(info);
let result_expr = simplifier.simplify(result_expr)?;
expr_result_map_for_count_bug.insert(e.schema_name().to_string(), result_expr);
}
diff --git a/datafusion/optimizer/src/decorrelate_lateral_join.rs b/datafusion/optimizer/src/decorrelate_lateral_join.rs
index 0c97c2a1d6..fe484ff737 100644
--- a/datafusion/optimizer/src/decorrelate_lateral_join.rs
+++ b/datafusion/optimizer/src/decorrelate_lateral_join.rs
@@ -232,7 +232,7 @@ fn rewrite_internal(join: Join) -> Result<Transformed<LogicalPlan>> {
let new_plan = if let Some(expr_map) = collected_count_expr_map {
let join_schema = new_plan.schema();
let alias_qualifier = alias.as_ref();
- let mut proj_exprs: Vec<Expr> = vec![];
+ let mut proj_exprs: Vec<Expr> = Vec::with_capacity(join_schema.fields().len());
for (i, (qualifier, field)) in join_schema.iter().enumerate() {
let col = Expr::Column(Column::new(qualifier.cloned(), field.name()));
@@ -283,7 +283,7 @@ fn rewrite_internal(join: Join) -> Result<Transformed<LogicalPlan>> {
// assumes it is deterministic.
let new_plan = if let Some(on_cond) = on_condition_for_projection {
let schema = Arc::clone(new_plan.schema());
- let mut proj_exprs: Vec<Expr> = vec![];
+ let mut proj_exprs: Vec<Expr> = Vec::with_capacity(schema.fields().len());
for (i, (qualifier, field)) in schema.iter().enumerate() {
let col = Expr::Column(Column::new(qualifier.cloned(), field.name()));
diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs
index 0a016cd949..99b57a5494 100644
--- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs
+++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs
@@ -793,7 +793,7 @@ fn build_join(
pull_up
.correlated_subquery_cols_map
.values()
- .for_each(|cols| all_correlated_cols.extend(cols.clone()));
+ .for_each(|cols| all_correlated_cols.extend(cols.iter().cloned()));
// alias the join filter
let join_filter_opt = conjunction(pull_up.join_filters)
diff --git a/datafusion/optimizer/src/eliminate_outer_join.rs b/datafusion/optimizer/src/eliminate_outer_join.rs
index 34feded110..d5e4b47819 100644
--- a/datafusion/optimizer/src/eliminate_outer_join.rs
+++ b/datafusion/optimizer/src/eliminate_outer_join.rs
@@ -190,7 +190,7 @@ fn try_simplify_join(join: &Join, predicate: &Expr) -> Option<LogicalPlan> {
/// purposes). Therefore, function volatility and `MoveTowardsLeafNodes`
/// placement can be ignored here.
fn inline_through_projection(predicate: Expr, p: &Projection) -> Result<Expr> {
- let mut map: HashMap<String, Expr> = HashMap::new();
+ let mut map: HashMap<String, Expr> = HashMap::with_capacity(p.expr.len());
for ((qualifier, field), expr) in p.schema.iter().zip(p.expr.iter()) {
map.insert(
qualified_name(qualifier, field.name()),
diff --git a/datafusion/optimizer/src/extract_leaf_expressions.rs b/datafusion/optimizer/src/extract_leaf_expressions.rs
index 53d6663240..aa3cef881f 100644
--- a/datafusion/optimizer/src/extract_leaf_expressions.rs
+++ b/datafusion/optimizer/src/extract_leaf_expressions.rs
@@ -454,7 +454,7 @@ fn remap_pairs_and_columns(
from_schema: &DFSchema,
to_schema: &DFSchema,
) -> Result<ExtractionTarget> {
- let mut replace_map = HashMap::new();
+ let mut replace_map = HashMap::with_capacity(from_schema.fields().len());
for ((from_q, from_f), (to_q, to_f)) in from_schema.iter().zip(to_schema.iter()) {
replace_map.insert(
qualified_name(from_q, from_f.name()),
@@ -765,7 +765,8 @@ fn build_extraction_projection_impl(
Projection::try_new(proj_exprs, Arc::clone(&existing.input))
} else {
// Build new projection with extracted expressions + all input columns
- let mut proj_exprs = Vec::new();
+ let mut proj_exprs =
+ Vec::with_capacity(extracted_exprs.len() + target_schema.fields().len());
for (expr, alias) in extracted_exprs {
proj_exprs.push(expr.clone().alias(alias));
}
@@ -910,9 +911,9 @@ fn merge_would_duplicate_kept_expr(
// same kept column twice still inlines one copy, so count each pair at most
// once per column.
let mut usage: HashMap<String, usize> = HashMap::new();
+ let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
for (expr, _alias) in pairs {
- let mut seen: std::collections::HashSet<String> =
- std::collections::HashSet::new();
+ seen.clear();
for col in expr.column_refs() {
let name = col.flat_name();
if kept_columns.contains(&name) && seen.insert(name.clone()) {
diff --git a/datafusion/optimizer/src/optimize_projections/mod.rs b/datafusion/optimizer/src/optimize_projections/mod.rs
index 7d47f4514a..c2550f59bb 100644
--- a/datafusion/optimizer/src/optimize_projections/mod.rs
+++ b/datafusion/optimizer/src/optimize_projections/mod.rs
@@ -442,7 +442,8 @@ fn optimize_projections(
RequiredIndices::new().with_plan_exprs(&plan, input.schema())?;
// Add additional required indices from the parent
- let mut additional_necessary_child_indices = Vec::new();
+ let mut additional_necessary_child_indices =
+ Vec::with_capacity(indices.indices().len());
indices.indices().iter().for_each(|idx| {
if let Some(index) = dependency_indices.get(*idx) {
additional_necessary_child_indices.push(*index);
@@ -599,7 +600,7 @@ fn merge_consecutive_projections_one_level(
// Try to rewrite the expressions in the current projection using the
// previous projection as input:
let name_preserver = NamePreserver::new_for_projection();
- let mut original_names = vec![];
+ let mut original_names = Vec::with_capacity(expr.len());
let new_exprs = expr.map_elements(|expr| {
original_names.push(name_preserver.save(&expr));
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index 1f1b1a6973..999cddfb62 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -783,8 +783,9 @@ fn infer_join_predicates_impl<
input_predicates: &[Expr],
inferred_predicates: &mut InferredPredicates,
) -> Result<()> {
+ let mut join_cols_to_replace = HashMap::new();
for predicate in input_predicates {
- let mut join_cols_to_replace = HashMap::new();
+ join_cols_to_replace.clear();
for &col in &predicate.column_refs() {
for (l, r) in join_col_keys.iter() {
@@ -948,7 +949,8 @@ impl OptimizerRule for PushDownFilter {
Ok(Transformed::yes(LogicalPlan::Sort(sort)))
}
LogicalPlan::SubqueryAlias(mut subquery_alias) => {
- let mut replace_map = HashMap::new();
+ let mut replace_map =
+ HashMap::with_capacity(subquery_alias.input.schema().fields().len());
for (i, (qualifier, field)) in
subquery_alias.input.schema().iter().enumerate()
{
@@ -999,9 +1001,10 @@ impl OptimizerRule for PushDownFilter {
}
}
+ let mut accum: HashSet<Column> = HashSet::new();
for predicate in predicates {
// collect all the Expr::Column in predicate recursively
- let mut accum: HashSet<Column> = HashSet::new();
+ accum.clear();
expr_to_columns(&predicate, &mut accum)?;
let contains_list_columns =
@@ -1047,8 +1050,9 @@ impl OptimizerRule for PushDownFilter {
}
LogicalPlan::Union(mut union) => {
let mut inputs = Vec::with_capacity(union.inputs.len());
+ let mut replace_map = HashMap::with_capacity(union.schema.fields().len());
for input in union.inputs {
- let mut replace_map = HashMap::new();
+ replace_map.clear();
for (i, (qualifier, field)) in input.schema().iter().enumerate() {
let (union_qualifier, union_field) =
union.schema.qualified_field(i);
@@ -1085,7 +1089,7 @@ impl OptimizerRule for PushDownFilter {
// As for plan Filter: Column(a+b) > 0 -- Agg: groupby:[Column(a)+Column(b)]
// After push, we need to replace `a+b` with Column(a)+Column(b)
// So we need create a replace_map, add {`a+b` --> Expr(Column(a)+Column(b))}
- let mut replace_map = HashMap::new();
+ let mut replace_map = HashMap::with_capacity(agg.group_expr.len());
for expr in &agg.group_expr {
replace_map.insert(expr.schema_name().to_string(), unalias(expr));
}
diff --git a/datafusion/optimizer/src/scalar_subquery_to_join.rs b/datafusion/optimizer/src/scalar_subquery_to_join.rs
index 745418f3ad..91383a24a6 100644
--- a/datafusion/optimizer/src/scalar_subquery_to_join.rs
+++ b/datafusion/optimizer/src/scalar_subquery_to_join.rs
@@ -215,7 +215,7 @@ impl OptimizerRule for ScalarSubqueryToJoin {
}
}
- let mut proj_exprs = vec![];
+ let mut proj_exprs = Vec::with_capacity(projection.expr.len());
for (expr, new_expr) in projection.expr.iter().zip(rewrite_exprs) {
let old_expr_name = expr.schema_name().to_string();
let new_expr_name = new_expr.schema_name().to_string();
diff --git a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs b/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs
index 6d75eb4352..ad23bb3c25 100644
--- a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs
+++ b/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs
@@ -1816,7 +1816,8 @@ impl TreeNodeRewriter for Simplifier<'_> {
let rhs = to_inlist(*right).unwrap();
#[allow(clippy::allow_attributes, clippy::mutable_key_type)]
// Expr contains Arc with interior mutability but is intentionally used as hash key
- let mut seen: HashSet<Expr> = HashSet::new();
+ let mut seen: HashSet<Expr> =
+ HashSet::with_capacity(lhs.list.len() + rhs.list.len());
let list = lhs
.list
.into_iter()
diff --git a/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs b/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
index 003ccf339b..35f0cc513d 100644
--- a/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
+++ b/datafusion/optimizer/src/simplify_expressions/simplify_predicates.rs
@@ -55,7 +55,7 @@ pub fn simplify_predicates(predicates: Vec<Expr>) -> Result<Vec<Expr>> {
// Group predicates by their column reference
let mut column_predicates: BTreeMap<Column, Vec<Expr>> = BTreeMap::new();
- let mut other_predicates = Vec::new();
+ let mut other_predicates = Vec::with_capacity(predicates.len());
for pred in predicates {
match pred {
diff --git a/datafusion/optimizer/src/unions_to_filter.rs b/datafusion/optimizer/src/unions_to_filter.rs
index 345c28187d..e90634c76f 100644
--- a/datafusion/optimizer/src/unions_to_filter.rs
+++ b/datafusion/optimizer/src/unions_to_filter.rs
@@ -104,7 +104,7 @@ fn try_rewrite_distinct_union(plan: LogicalPlan) -> Result<Option<LogicalPlan>>
// Use a Vec instead of HashMap: union branches are typically 2-10 entries,
// so a linear scan with PartialEq is faster than recursively hashing entire
// LogicalPlan subtrees (O(N * tree_size) hashing for every insert/lookup).
- let mut grouped: Vec<(GroupKey, Vec<Expr>)> = Vec::new();
+ let mut grouped: Vec<(GroupKey, Vec<Expr>)> = Vec::with_capacity(inputs.len());
let mut transformed = false;
for input in inputs {
diff --git a/datafusion/physical-expr-common/src/sort_expr.rs b/datafusion/physical-expr-common/src/sort_expr.rs
index 43073ec36e..ce64d46798 100644
--- a/datafusion/physical-expr-common/src/sort_expr.rs
+++ b/datafusion/physical-expr-common/src/sort_expr.rs
@@ -486,10 +486,11 @@ impl LexOrdering {
/// If the vector is empty, returns `None`.
pub fn new(exprs: impl IntoIterator<Item = PhysicalSortExpr>) -> Option<Self> {
let exprs = exprs.into_iter();
+ let capacity = exprs.size_hint().0;
let mut candidate = Self {
// not valid yet; valid publicly-returned instance must be non-empty
- exprs: Vec::new(),
- set: IndexSet::new(),
+ exprs: Vec::with_capacity(capacity),
+ set: IndexSet::with_capacity(capacity),
};
for expr in exprs {
candidate.push(expr);
@@ -736,9 +737,9 @@ impl LexRequirement {
fn construct(
reqs: impl IntoIterator<Item = PhysicalSortRequirement>,
) -> (bool, Self) {
- let mut set = HashSet::new();
+ let reqs = reqs.into_iter();
+ let mut set = HashSet::with_capacity(reqs.size_hint().0);
let reqs = reqs
- .into_iter()
.filter_map(|r| set.insert(Arc::clone(&r.expr)).then_some(r))
.collect();
(!set.is_empty(), Self { reqs })
diff --git a/datafusion/physical-expr/src/analysis.rs b/datafusion/physical-expr/src/analysis.rs
index ceac0094ed..73bb8a7aac 100644
--- a/datafusion/physical-expr/src/analysis.rs
+++ b/datafusion/physical-expr/src/analysis.rs
@@ -198,8 +198,9 @@ pub fn analyze(
.map(|c| Arc::new(c) as _)
.collect::<Vec<_>>();
- let mut target_indices_and_boundaries = vec![];
let target_expr_and_indices = graph.gather_node_indices(columns.as_slice());
+ let mut target_indices_and_boundaries =
+ Vec::with_capacity(target_expr_and_indices.len());
for (expr, index) in &target_expr_and_indices {
if let Some(column) = expr.downcast_ref::<Column>()
diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index de02662228..b68d20fccf 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -737,7 +737,8 @@ impl EquivalenceGroup {
// The key is the source expression, and the value is the equivalence
// class that contains the corresponding target expression.
let mut new_constants = vec![];
- let mut new_classes = IndexMap::<_, EquivalenceClass>::new();
+ let mut new_classes =
+ IndexMap::<_, EquivalenceClass>::with_capacity(mapping.len());
for (source, targets) in mapping.iter() {
// We need to find equivalent projected expressions. For example,
// consider a table with columns `[a, b, c]` with `a` == `b`, and
diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs
index c68157fecb..1c2d970ccf 100644
--- a/datafusion/physical-expr/src/equivalence/properties/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs
@@ -169,9 +169,10 @@ impl OrderingEquivalenceCache {
pub fn new(
orderings: impl IntoIterator<Item = impl IntoIterator<Item = PhysicalSortExpr>>,
) -> Self {
+ let normal_cls = OrderingEquivalenceClass::new(orderings);
let mut cache = Self {
- normal_cls: OrderingEquivalenceClass::new(orderings),
- leading_map: HashMap::new(),
+ leading_map: HashMap::with_capacity(normal_cls.len()),
+ normal_cls,
};
cache.update_map();
cache
@@ -1431,7 +1432,7 @@ impl EquivalenceProperties {
}
// Rewrite equivalence classes according to the new schema:
- let mut eq_classes = vec![];
+ let mut eq_classes = Vec::with_capacity(self.eq_group.len());
for mut eq_class in self.eq_group {
// Rewrite the expressions in the equivalence class:
eq_class.exprs = eq_class
diff --git a/datafusion/physical-expr/src/expressions/case.rs b/datafusion/physical-expr/src/expressions/case.rs
index dd98029c73..1e4d106837 100644
--- a/datafusion/physical-expr/src/expressions/case.rs
+++ b/datafusion/physical-expr/src/expressions/case.rs
@@ -1348,7 +1348,7 @@ impl PhysicalExpr for CaseExpr {
}
fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
- let mut children = vec![];
+ let mut children = Vec::with_capacity(self.body.when_then_expr.len() * 2 + 2);
if let Some(expr) = &self.body.expr {
children.push(expr)
}
diff --git a/datafusion/physical-expr/src/expressions/in_list.rs b/datafusion/physical-expr/src/expressions/in_list.rs
index 0b3f8067ae..b30a5ba343 100644
--- a/datafusion/physical-expr/src/expressions/in_list.rs
+++ b/datafusion/physical-expr/src/expressions/in_list.rs
@@ -456,7 +456,8 @@ impl PhysicalExpr for InListExpr {
}
fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
- let mut children = vec![&self.expr];
+ let mut children = Vec::with_capacity(self.list.len() + 1);
+ children.push(&self.expr);
children.extend(&self.list);
children
}
diff --git a/datafusion/physical-expr/src/expressions/in_list/array_static_filter.rs b/datafusion/physical-expr/src/expressions/in_list/array_static_filter.rs
index c2d3b47282..24f4dfc69c 100644
--- a/datafusion/physical-expr/src/expressions/in_list/array_static_filter.rs
+++ b/datafusion/physical-expr/src/expressions/in_list/array_static_filter.rs
@@ -67,7 +67,7 @@ impl ArrayStaticFilter {
haystack: &ArrayRef,
state: &RandomState,
) -> Result<HashTable<usize>> {
- let mut table = HashTable::new();
+ let mut table = HashTable::with_capacity(haystack.len() - haystack.null_count());
with_hashes([haystack.as_ref()], state, |hashes| -> Result<()> {
let cmp = make_comparator(haystack, haystack, SortOptions::default())?;
diff --git a/datafusion/physical-expr/src/intervals/cp_solver.rs b/datafusion/physical-expr/src/intervals/cp_solver.rs
index f9d9561fee..b28a42af5b 100644
--- a/datafusion/physical-expr/src/intervals/cp_solver.rs
+++ b/datafusion/physical-expr/src/intervals/cp_solver.rs
@@ -533,7 +533,7 @@ impl ExprIntervalGraph {
/// Returns the set of node indices reachable from the root node via a
/// simple depth-first search.
fn connected_nodes(&self) -> HashSet<NodeIndex> {
- let mut nodes = HashSet::new();
+ let mut nodes = HashSet::with_capacity(self.graph.node_count());
let mut dfs = Dfs::new(&self.graph, self.root);
while let Some(node) = dfs.next(&self.graph) {
nodes.insert(node);
@@ -661,10 +661,12 @@ impl ExprIntervalGraph {
}
let mut bfs = Bfs::new(&self.graph, self.root);
+ let mut children = Vec::new();
while let Some(node) = bfs.next(&self.graph) {
let neighbors = self.graph.neighbors_directed(node, Outgoing);
- let mut children = neighbors.collect::<Vec<_>>();
+ children.clear();
+ children.extend(neighbors);
// If the current expression is a leaf, its range is now final.
// So, just continue with the propagation procedure:
if children.is_empty() {
@@ -691,8 +693,8 @@ impl ExprIntervalGraph {
.expr
.propagate_constraints(node_interval, &children_intervals)?;
if let Some(propagated_intervals) = propagated_intervals {
- for (child, interval) in children.into_iter().zip(propagated_intervals) {
- self.graph[child].interval = interval;
+ for (child, interval) in children.iter().zip(propagated_intervals) {
+ self.graph[*child].interval = interval;
}
} else {
// The constraint is infeasible, report:
diff --git a/datafusion/physical-expr/src/physical_expr.rs b/datafusion/physical-expr/src/physical_expr.rs
index d45d0fe149..0904cb9bd8 100644
--- a/datafusion/physical-expr/src/physical_expr.rs
+++ b/datafusion/physical-expr/src/physical_expr.rs
@@ -79,8 +79,8 @@ pub fn physical_exprs_bag_equal(
lhs: &[Arc<dyn PhysicalExpr>],
rhs: &[Arc<dyn PhysicalExpr>],
) -> bool {
- let mut multi_set_lhs: HashMap<_, usize> = HashMap::new();
- let mut multi_set_rhs: HashMap<_, usize> = HashMap::new();
+ let mut multi_set_lhs: HashMap<_, usize> = HashMap::with_capacity(lhs.len());
+ let mut multi_set_rhs: HashMap<_, usize> = HashMap::with_capacity(rhs.len());
for expr in lhs {
*multi_set_lhs.entry(expr).or_insert(0) += 1;
}
@@ -139,11 +139,11 @@ pub fn create_ordering(
schema: &Schema,
sort_order: &[Vec<SortExpr>],
) -> Result<Vec<LexOrdering>> {
- let mut all_sort_orders = vec![];
+ let mut all_sort_orders = Vec::with_capacity(sort_order.len());
for (group_idx, exprs) in sort_order.iter().enumerate() {
// Construct PhysicalSortExpr objects from Expr objects:
- let mut sort_exprs = vec![];
+ let mut sort_exprs = Vec::with_capacity(exprs.len());
for (expr_idx, sort) in exprs.iter().enumerate() {
match &sort.expr {
Expr::Column(col) => match expressions::col(&col.name, schema) {
@@ -184,7 +184,7 @@ pub fn create_lex_ordering(
let df_schema = DFSchema::try_from(Arc::clone(schema))?;
- let mut all_sort_orders = vec![];
+ let mut all_sort_orders = Vec::with_capacity(sort_order.len());
for exprs in sort_order.iter() {
all_sort_orders.extend(LexOrdering::new(create_physical_sort_exprs(
diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs
index 7fe6789699..fefb7eaf09 100644
--- a/datafusion/physical-expr/src/projection.rs
+++ b/datafusion/physical-expr/src/projection.rs
@@ -1294,8 +1294,10 @@ impl ProjectionMapping {
input_schema: &SchemaRef,
) -> Result<Self> {
// Construct a map from the input expressions to the output expression of the projection:
- let mut map = IndexMap::<_, ProjectionTargets>::new();
- for (expr_idx, (expr, name)) in expr.into_iter().enumerate() {
+ let exprs = expr.into_iter();
+ let mut map =
+ IndexMap::<_, ProjectionTargets>::with_capacity(exprs.size_hint().0);
+ for (expr_idx, (expr, name)) in exprs.enumerate() {
let target_expr = Arc::new(Column::new(&name, expr_idx)) as _;
let source_expr = expr.transform_down(|e| match e.downcast_ref::<Column>() {
Some(col) => {
@@ -1429,7 +1431,7 @@ pub fn project_orderings(
orderings: &[LexOrdering],
schema: &SchemaRef,
) -> Vec<LexOrdering> {
- let mut projected_orderings = vec![];
+ let mut projected_orderings = Vec::with_capacity(orderings.len());
for ordering in orderings {
projected_orderings.extend(project_ordering(ordering, schema));
@@ -1471,7 +1473,7 @@ pub fn project_ordering(
ordering: &LexOrdering,
schema: &SchemaRef,
) -> Option<LexOrdering> {
- let mut projected_exprs = vec![];
+ let mut projected_exprs = Vec::with_capacity(ordering.len());
for PhysicalSortExpr { expr, options } in ordering.iter() {
let transformed = Arc::clone(expr).transform_up(|expr| {
let Some(col) = expr.downcast_ref::<Column>() else {
diff --git a/datafusion/physical-expr/src/utils/guarantee.rs b/datafusion/physical-expr/src/utils/guarantee.rs
index 8b870c573d..4224943eb7 100644
--- a/datafusion/physical-expr/src/utils/guarantee.rs
+++ b/datafusion/physical-expr/src/utils/guarantee.rs
@@ -520,12 +520,14 @@ fn find_common_columns<'a>(
}
// Intersect with columns from remaining termsets
+ let mut termset_cols: HashSet<&crate::expressions::Column> = HashSet::new();
for termset in termsets.iter().skip(1) {
- let termset_cols: HashSet<_> = termset.iter().map(|term| term.col()).collect();
+ termset_cols.clear();
+ termset_cols.extend(termset.iter().map(|term| term.col()));
if termset_cols.len() != termset.len() {
return Vec::new();
}
- common_cols = common_cols.intersection(&termset_cols).copied().collect();
+ common_cols.retain(|col| termset_cols.contains(col));
}
common_cols.into_iter().collect()
diff --git a/datafusion/physical-expr/src/window/standard.rs b/datafusion/physical-expr/src/window/standard.rs
index 2a1239793c..97e22f223c 100644
--- a/datafusion/physical-expr/src/window/standard.rs
+++ b/datafusion/physical-expr/src/window/standard.rs
@@ -117,7 +117,7 @@ impl WindowExpr for StandardWindowExpr {
let num_rows = batch.num_rows();
if evaluator.uses_window_frame() {
let sort_options = self.order_by.iter().map(|o| o.options).collect();
- let mut row_wise_results = vec![];
+ let mut row_wise_results = Vec::with_capacity(num_rows);
let mut values = self.evaluate_args(batch)?;
let order_bys = get_orderby_values(self.order_by_columns(batch)?);
@@ -203,7 +203,8 @@ impl WindowExpr for StandardWindowExpr {
// We iterate on each row to perform a running calculation.
let record_batch = &partition_batch_state.record_batch;
let num_rows = record_batch.num_rows();
- let mut row_wise_results: Vec<ScalarValue> = vec![];
+ let mut row_wise_results: Vec<ScalarValue> =
+ Vec::with_capacity(num_rows.saturating_sub(state.last_calculated_index));
let is_causal = if evaluator.uses_window_frame() {
self.window_frame.is_causal()
} else {
diff --git a/datafusion/physical-expr/src/window/window_expr.rs b/datafusion/physical-expr/src/window/window_expr.rs
index 0af4f0a4eb..690731f4a9 100644
--- a/datafusion/physical-expr/src/window/window_expr.rs
+++ b/datafusion/physical-expr/src/window/window_expr.rs
@@ -350,7 +350,8 @@ pub trait AggregateWindowExpr: WindowExpr {
// We iterate on each row to perform a running calculation.
let length = values[0].len();
- let mut row_wise_results: Vec<ScalarValue> = vec![];
+ let mut row_wise_results: Vec<ScalarValue> =
+ Vec::with_capacity(length.saturating_sub(idx));
let is_causal = self.get_window_frame().is_causal();
while idx < length {
// Start search from the last_range. This squeezes searched range.
diff --git a/datafusion/physical-optimizer/src/aggregate_statistics.rs b/datafusion/physical-optimizer/src/aggregate_statistics.rs
index 43b1abb4b6..38d618ca22 100644
--- a/datafusion/physical-optimizer/src/aggregate_statistics.rs
+++ b/datafusion/physical-optimizer/src/aggregate_statistics.rs
@@ -75,7 +75,7 @@ impl PhysicalOptimizerRule for AggregateStatistics {
{
projections.push(ProjectionExpr {
expr: expressions::lit(optimizable_statistic),
- alias: name.to_owned(),
+ alias: name,
});
} else {
// TODO: we need all aggr_expr to be resolved (cf TODO fullres)
diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
index 3c2426a5ef..e490c959eb 100644
--- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
@@ -605,8 +605,8 @@ fn try_reorder(
return (join_keys, None);
};
- let mut new_left_keys = vec![];
- let mut new_right_keys = vec![];
+ let mut new_left_keys = Vec::with_capacity(positions.len());
+ let mut new_right_keys = Vec::with_capacity(positions.len());
for pos in positions.iter() {
new_left_keys.push(Arc::clone(&join_keys.left_keys[*pos]));
new_right_keys.push(Arc::clone(&join_keys.right_keys[*pos]));
@@ -1003,7 +1003,7 @@ fn get_repartition_requirement_status(
let children = plan.children();
let rr_beneficial = plan.benefits_from_input_partitioning();
let requirements = plan.input_distribution_requirements().into_per_child();
- let mut repartition_status_flags = vec![];
+ let mut repartition_status_flags = Vec::with_capacity(children.len());
for (child, requirement, roundrobin_beneficial) in
izip!(children.into_iter(), requirements, rr_beneficial)
{
diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs
index cd2b5bb2f4..414a1d0938 100644
--- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs
@@ -214,7 +214,7 @@ pub fn replace_with_partial_sort(
let mut common_prefix_length = 0;
while child_eq_properties
- .ordering_satisfy(sort_exprs[0..=common_prefix_length].to_vec())?
+ .ordering_satisfy(sort_exprs[0..=common_prefix_length].iter().cloned())?
{
common_prefix_length += 1;
}
@@ -420,7 +420,7 @@ pub fn ensure_sorting(
};
let plan = &requirements.plan;
- let mut updated_children = vec![];
+ let mut updated_children = Vec::with_capacity(requirements.children.len());
for (idx, (required_ordering, mut child)) in plan
.required_input_ordering()
.into_iter()
diff --git a/datafusion/physical-optimizer/src/filter_pushdown.rs b/datafusion/physical-optimizer/src/filter_pushdown.rs
index 8beadf5d04..2d44c68320 100644
--- a/datafusion/physical-optimizer/src/filter_pushdown.rs
+++ b/datafusion/physical-optimizer/src/filter_pushdown.rs
@@ -445,10 +445,10 @@ fn push_down_filters(
) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
let mut parent_filter_pushdown_supports: Vec<Vec<PushedDown>> =
vec![vec![]; parent_predicates.len()];
- let mut self_filters_pushdown_supports = vec![];
- let mut new_children = Vec::with_capacity(node.children().len());
let children = node.children();
+ let mut self_filters_pushdown_supports = Vec::with_capacity(children.len());
+ let mut new_children = Vec::with_capacity(children.len());
// Filter out expressions that are not allowed for pushdown
let parent_filtered = FilteredVec::new(&parent_predicates, allow_pushdown_for_expr);
@@ -621,8 +621,8 @@ impl<T: Clone> FilteredVec<T> {
where
F: Fn(&T) -> bool,
{
- let mut filtered_items = Vec::new();
- let mut original_indices = Vec::new();
+ let mut filtered_items = Vec::with_capacity(items.len());
+ let mut original_indices = Vec::with_capacity(items.len());
for (idx, item) in items.iter().enumerate() {
if predicate(item) {
diff --git a/datafusion/physical-optimizer/src/output_requirements.rs b/datafusion/physical-optimizer/src/output_requirements.rs
index 83cef91ec4..bed7bd2428 100644
--- a/datafusion/physical-optimizer/src/output_requirements.rs
+++ b/datafusion/physical-optimizer/src/output_requirements.rs
@@ -295,8 +295,8 @@ impl ExecutionPlan for OutputRequirementExec {
let mut requirements = self.required_input_ordering().swap_remove(0);
if let Some(reqs) = requirements {
- let mut updated_reqs = vec![];
let (lexes, soft) = reqs.into_alternatives();
+ let mut updated_reqs = Vec::with_capacity(lexes.len());
for lex in lexes.into_iter() {
let Some(updated_lex) = update_ordering_requirement(lex, proj_exprs)?
else {
@@ -313,7 +313,7 @@ impl ExecutionPlan for OutputRequirementExec {
Distribution::HashPartitioned(exprs)
| Distribution::KeyPartitioned(exprs),
) => {
- let mut updated_exprs = vec![];
+ let mut updated_exprs = Vec::with_capacity(exprs.len());
for expr in exprs {
let Some(new_expr) = update_expr(expr, projection.expr(), false)?
else {
diff --git a/datafusion/physical-optimizer/src/projection_pushdown.rs b/datafusion/physical-optimizer/src/projection_pushdown.rs
index 97ec7718b8..f38c93a4c1 100644
--- a/datafusion/physical-optimizer/src/projection_pushdown.rs
+++ b/datafusion/physical-optimizer/src/projection_pushdown.rs
@@ -137,7 +137,8 @@ fn try_push_down_join_filter(
None => match join.join_type() {
JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full => {
// Build projections that ignore the newly projected columns.
- let mut projections = Vec::new();
+ let mut projections =
+ Vec::with_capacity(original_lhs_length + original_rhs_length);
projections.extend(0..original_lhs_length);
projections.extend(new_lhs_length..new_lhs_length + original_rhs_length);
projections
@@ -240,7 +241,7 @@ fn minimize_join_filter(
lhs_schema: &Schema,
rhs_schema: &Schema,
) -> JoinFilter {
- let mut used_columns = HashSet::new();
+ let mut used_columns = HashSet::with_capacity(old_column_indices.len());
expr.apply(|expr| {
if let Some(col) = expr.downcast_ref::<Column>() {
used_columns.insert(col.index());
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
index 76e0b0dbe3..729932eb04 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
@@ -451,7 +451,8 @@ impl<AggrMode> AggregateHashTable<AggrMode> {
let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics);
let any_interned = group_by_metrics.time_group_key_preparation(|| {
let max_ordinal = max_duplicate_ordinal(state.group_by.groups());
- let mut ordinals: HashMap<&[bool], usize> = HashMap::new();
+ let mut ordinals: HashMap<&[bool], usize> =
+ HashMap::with_capacity(state.group_by.groups().len());
let group_schema = state.group_by.group_schema(&self.input_schema)?;
let n_expr = state.group_by.expr().len();
let mut any_interned = false;
diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes.rs
index 3c92ce787d..fa357c3c6f 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes.rs
@@ -111,7 +111,7 @@ impl<O: OffsetSizeTrait> GroupValues for GroupValuesBytes<O> {
map_contents.slice(n, map_contents.len() - n);
self.num_groups = 0;
- let mut group_indexes = vec![];
+ let mut group_indexes = Vec::with_capacity(remaining_group_values.len());
self.intern(&[remaining_group_values], &mut group_indexes)?;
// Verify that the group indexes were assigned in the correct order
diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes_view.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes_view.rs
index a2e3573ce8..75fe65e13b 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes_view.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes_view.rs
@@ -113,7 +113,7 @@ impl GroupValues for GroupValuesBytesView {
map_contents.slice(n, map_contents.len() - n);
self.num_groups = 0;
- let mut group_indexes = vec![];
+ let mut group_indexes = Vec::with_capacity(remaining_group_values.len());
self.intern(&[remaining_group_values], &mut group_indexes)?;
// Verify that the group indexes were assigned in the correct order
diff --git a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
index a8c35ff774..fd78ee721e 100644
--- a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
@@ -908,6 +908,14 @@ impl GroupedHashAggregateStream {
Ok::<_, DataFusionError>((input_values, filter_values))
})?;
+ // Convert filters before timing accumulator calls. The timer covers
+ // one interval across all accumulator operations, rather than a
+ // prefix sum charged once per aggregate expression.
+ let filters = filter_values
+ .iter()
+ .map(|filter| filter.as_ref().map(|filter| filter.as_boolean()))
+ .collect::<Vec<_>>();
+
for group_values in &group_by_values {
// Calculate group indices and update ordering information.
let total_num_groups =
@@ -927,13 +935,6 @@ impl GroupedHashAggregateStream {
Ok::<_, DataFusionError>(total_num_groups)
})?;
- // Convert filters before timing accumulator calls. The timer covers
- // one interval across all accumulator operations, rather than a
- // prefix sum charged once per aggregate expression.
- let filters = filter_values
- .iter()
- .map(|filter| filter.as_ref().map(|filter| filter.as_boolean()))
- .collect::<Vec<_>>();
self.group_by_metrics.time_aggregation(|| {
let group_indices = &self.current_group_indices;
let t = self
@@ -1120,7 +1121,7 @@ impl GroupedHashAggregateStream {
let any_interned = group_by_metrics.time_group_key_preparation(|| {
let max_ordinal = max_duplicate_ordinal(self.group_by.groups());
let mut ordinals: std::collections::HashMap<&[bool], usize> =
- std::collections::HashMap::new();
+ std::collections::HashMap::with_capacity(self.group_by.groups().len());
let group_schema = self.group_by.group_schema(&self.input_schema)?;
let n_expr = self.group_by.expr().len();
let mut any_interned = false;
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
index 4284e7e0f9..31ce7899d3 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -3613,7 +3613,7 @@ pub(crate) fn group_id_array(
/// three times the ordinals are 0, 1, 2 and this function returns 2.
/// Returns 0 when no grouping set is duplicated.
pub(crate) fn max_duplicate_ordinal(groups: &[Vec<bool>]) -> usize {
- let mut counts: HashMap<&[bool], usize> = HashMap::new();
+ let mut counts: HashMap<&[bool], usize> = HashMap::with_capacity(groups.len());
for group in groups {
*counts.entry(group).or_insert(0) += 1;
}
@@ -3661,7 +3661,8 @@ pub fn evaluate_group_by(
batch: &RecordBatch,
) -> Result<Vec<Vec<ArrayRef>>> {
let max_ordinal = max_duplicate_ordinal(&group_by.groups);
- let mut ordinal_per_pattern: HashMap<&[bool], usize> = HashMap::new();
+ let mut ordinal_per_pattern: HashMap<&[bool], usize> =
+ HashMap::with_capacity(group_by.groups.len());
let exprs = evaluate_expressions_to_arrays(
group_by.expr.iter().map(|(expr, _)| expr),
batch,
diff --git a/datafusion/physical-plan/src/display.rs b/datafusion/physical-plan/src/display.rs
index f13e66d7d0..91da64528d 100644
--- a/datafusion/physical-plan/src/display.rs
+++ b/datafusion/physical-plan/src/display.rs
@@ -1395,7 +1395,7 @@ impl TreeRenderVisitor<'_, '_> {
let mut splits: Vec<String> = str.split('\n').map(String::from).collect();
if splits.len() > max_lines {
- let mut truncated_splits = Vec::new();
+ let mut truncated_splits = Vec::with_capacity(max_lines + 1);
for split in splits.iter().take(max_lines / 2) {
truncated_splits.push(split.clone());
}
diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 2f5680e659..5d6afc2b9b 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -1615,7 +1615,7 @@ impl<'a> DefaultInvariantChecker<'a> {
&self,
dynamic_expressions: Vec<Arc<dyn PhysicalExpr>>,
) -> Result<()> {
- let mut produced_ids = HashSet::new();
+ let mut produced_ids = HashSet::with_capacity(dynamic_expressions.len());
for expr in dynamic_expressions {
let Some(expression_id) = expr.expression_id() else {
return internal_err!(
@@ -1835,7 +1835,7 @@ pub async fn collect_partitioned(
});
});
- let mut batches = vec![];
+ let mut batches = Vec::with_capacity(join_set.len());
// Note that currently this doesn't identify the thread that panicked
//
// TODO: Replace with [join_next_with_id](https://docs.rs/tokio/latest/tokio/task/struct.JoinSet.html#method.join_next_with_id
diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs
index cc5e0846d3..a9546864c9 100644
--- a/datafusion/physical-plan/src/filter.rs
+++ b/datafusion/physical-plan/src/filter.rs
@@ -1150,7 +1150,7 @@ fn restricted_column(expr: &Arc<dyn PhysicalExpr>) -> Option<(usize, usize)> {
return None;
}
let column = in_list.expr().downcast_ref::<Column>()?;
- let mut values: Vec<&ScalarValue> = vec![];
+ let mut values: Vec<&ScalarValue> = Vec::with_capacity(in_list.list().len());
for expr in in_list.list() {
let value = expr.downcast_ref::<Literal>()?.value();
if !value.is_null() && !values.contains(&value) {
diff --git a/datafusion/physical-plan/src/joins/asof_join.rs b/datafusion/physical-plan/src/joins/asof_join.rs
index cb6e71eea7..bc38c28b43 100644
--- a/datafusion/physical-plan/src/joins/asof_join.rs
+++ b/datafusion/physical-plan/src/joins/asof_join.rs
@@ -1009,6 +1009,7 @@ impl PendingRows {
&self,
source_column: usize,
data_type: &arrow::datatypes::DataType,
+ interleave_indices: &mut Vec<(usize, usize)>,
) -> Result<ArrayRef> {
if self.indices.is_empty() {
return internal_err!("ASOF output materialization has no pending rows");
@@ -1041,15 +1042,12 @@ impl PendingRows {
.map(|batch| batch.column(source_column).as_ref()),
);
let source_offset = usize::from(has_null);
- let interleave_indices = self
- .indices
- .iter()
- .map(|index| match index {
- Some((source, row)) => (source + source_offset, *row),
- None => (0, 0),
- })
- .collect::<Vec<_>>();
- interleave(&source_arrays, &interleave_indices).map_err(Into::into)
+ interleave_indices.clear();
+ interleave_indices.extend(self.indices.iter().map(|index| match index {
+ Some((source, row)) => (source + source_offset, *row),
+ None => (0, 0),
+ }));
+ interleave(&source_arrays, interleave_indices).map_err(Into::into)
}
fn clear(&mut self) {
@@ -1374,6 +1372,7 @@ impl AsOfJoinStream {
let _timer = self.metrics.baseline.elapsed_compute().timer();
let row_count = self.pending_left.len();
let mut arrays = Vec::with_capacity(self.schema.fields().len());
+ let mut interleave_indices = Vec::new();
for (field, column) in self.schema.fields().iter().zip(&self.column_indices) {
let pending = match column.side {
JoinSide::Left => &self.pending_left,
@@ -1382,7 +1381,11 @@ impl AsOfJoinStream {
return internal_err!("ASOF projection cannot contain a mark column");
}
};
- arrays.push(pending.materialize_column(column.index, field.data_type())?);
+ arrays.push(pending.materialize_column(
+ column.index,
+ field.data_type(),
+ &mut interleave_indices,
+ )?);
}
self.pending_left.clear();
self.pending_right.clear();
diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs
index 51142e9d8e..df8a3a8aad 100644
--- a/datafusion/physical-plan/src/joins/cross_join.rs
+++ b/datafusion/physical-plan/src/joins/cross_join.rs
@@ -626,22 +626,16 @@ fn build_batch(
schema: &Schema,
) -> Result<RecordBatch> {
// Repeat value on the left n times
- let arrays = left_data
- .columns()
- .iter()
- .map(|arr| {
- let scalar = ScalarValue::try_from_array(arr, left_index)?;
- scalar.to_array_of_size(batch.num_rows())
- })
- .collect::<Result<Vec<_>>>()?;
+ let mut arrays = Vec::with_capacity(left_data.num_columns() + batch.num_columns());
+ for arr in left_data.columns() {
+ let scalar = ScalarValue::try_from_array(arr, left_index)?;
+ arrays.push(scalar.to_array_of_size(batch.num_rows())?);
+ }
+ arrays.extend_from_slice(batch.columns());
RecordBatch::try_new_with_options(
Arc::new(schema.clone()),
- arrays
- .iter()
- .chain(batch.columns().iter())
- .cloned()
- .collect(),
+ arrays,
&RecordBatchOptions::new().with_row_count(Some(batch.num_rows())),
)
.map_err(Into::into)
diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
index 7d81dc9162..d1a161aa43 100644
--- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
@@ -159,7 +159,7 @@ fn create_bounds_predicate(
on_right: &[PhysicalExprRef],
bounds: &PartitionBounds,
) -> Option<Arc<dyn PhysicalExpr>> {
- let mut column_predicates = Vec::new();
+ let mut column_predicates = Vec::with_capacity(on_right.len());
for (col_idx, right_expr) in on_right.iter().enumerate() {
if let Some(column_bounds) = bounds.get_column_bounds(col_idx) {
diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs
index dfa7eb0d8f..ea1127c0b0 100644
--- a/datafusion/physical-plan/src/joins/nested_loop_join.rs
+++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs
@@ -3441,7 +3441,7 @@ impl NestedLoopJoinStream {
)?
} else {
// No filter: all pairs match
- BooleanArray::from(vec![true; total_rows])
+ BooleanArray::new(BooleanBuffer::new_set(total_rows), None)
};
// Update the global left or right bitmap for matched indices
@@ -3573,7 +3573,7 @@ impl NestedLoopJoinStream {
let cur_right_bitmap = if let Some(filter) = &self.join_filter {
apply_filter_to_row_join_batch(left_row, right_batch, filter)?
} else {
- BooleanArray::from(vec![true; right_row_count])
+ BooleanArray::new(BooleanBuffer::new_set(right_row_count), None)
};
self.update_matched_bitmap(l_index, &cur_right_bitmap)?;
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
index 25e7d6a833..98b32b3151 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
@@ -581,13 +581,12 @@ fn build_matched_indices_and_mark_buffered(
.buffered_data
.batch()
.slice(buffered_range.0, buffered_range.1);
- let mut buffered_columns = new_buffered_batch.columns().to_vec();
+ let mut buffered_columns = Vec::with_capacity(join_schema.fields().len());
+ buffered_columns.extend_from_slice(new_buffered_batch.columns());
let indices = UInt32Array::from_value(streamed_range.0 as u32, streamed_range.1);
let new_stream_batch = take_record_batch(&stream_batch.batch, &indices)?;
- let streamed_columns = new_stream_batch.columns().to_vec();
-
- buffered_columns.extend(streamed_columns);
+ buffered_columns.extend_from_slice(new_stream_batch.columns());
Ok(RecordBatch::try_new(
Arc::clone(join_schema),
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs
index 23cc609009..ecbb160f0d 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs
@@ -355,7 +355,8 @@ impl RightExistencePWMJStream {
Some(matched) => Arc::new(boolean_mask_from_filter(&matched)),
};
- let mut columns = batch.columns().to_vec();
+ let mut columns = Vec::with_capacity(batch.num_columns() + 1);
+ columns.extend_from_slice(batch.columns());
columns.push(mark);
// Right existence joins output the streamed columns (plus `mark` for `RightMark`)
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs
index 6e5193f535..8027d83dbe 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs
@@ -1489,8 +1489,7 @@ impl MaterializingSortMergeJoinStream {
})
.collect::<Vec<_>>();
- let buffered_indices =
- UInt64Array::from_iter_values(not_matched_buffered_indices.iter().copied());
+ let buffered_indices = UInt64Array::from(not_matched_buffered_indices);
if let Some(record_batch) = produce_buffered_null_batch(
&self.schema,
@@ -1515,7 +1514,8 @@ impl MaterializingSortMergeJoinStream {
// Matched chunks are collected and processed together in
// freeze_streamed_matched() to amortize filter evaluation overhead.
fn freeze_streamed(&mut self) -> Result<()> {
- let mut matched_chunks: Vec<(usize, UInt64Array, UInt64Array)> = Vec::new();
+ let mut matched_chunks: Vec<(usize, UInt64Array, UInt64Array)> =
+ Vec::with_capacity(self.streamed_batch.output_indices.len());
let mut total_matched_rows: usize = 0;
for chunk in self.streamed_batch.output_indices.iter_mut() {
@@ -1812,7 +1812,8 @@ impl MaterializingSortMergeJoinStream {
// Sentinel for "no source index assigned to this buffered batch yet".
const UNSEEN: usize = usize::MAX;
let mut source_of_batch = vec![UNSEEN; max_batch_idx - min_batch_idx + 1];
- let mut source_batches: Vec<usize> = Vec::new();
+ let mut source_batches: Vec<usize> =
+ Vec::with_capacity(matched_chunks.len().min(source_of_batch.len()));
let mut interleave_indices: Vec<(usize, usize)> =
Vec::with_capacity(total_matched_rows);
for (batch_idx, _, right) in matched_chunks {
@@ -1886,7 +1887,7 @@ impl MaterializingSortMergeJoinStream {
let record_batch = self.joined_record_batches.concat_batches(&self.schema)?;
let (mut out_indices, mut out_mask, mut batch_ids) =
self.joined_record_batches.filter_metadata.finish_metadata();
- let default_batch_ids = vec![0; record_batch.num_rows()];
+ let default_batch_ids: Vec<usize>;
// If only nulls come in and indices sizes doesn't match with expected record batch count
// generate missing indices
@@ -1896,6 +1897,7 @@ impl MaterializingSortMergeJoinStream {
{
out_mask = BooleanArray::from(vec![None; record_batch.num_rows()]);
out_indices = UInt64Array::from(vec![None; record_batch.num_rows()]);
+ default_batch_ids = vec![0; record_batch.num_rows()];
batch_ids = &default_batch_ids;
}
diff --git a/datafusion/physical-plan/src/joins/stream_join_utils.rs b/datafusion/physical-plan/src/joins/stream_join_utils.rs
index b2a30e1a80..f74c3671d0 100644
--- a/datafusion/physical-plan/src/joins/stream_join_utils.rs
+++ b/datafusion/physical-plan/src/joins/stream_join_utils.rs
@@ -18,6 +18,7 @@
//! This file contains common subroutines for symmetric hash join
//! related functionality, used both in join calculations and optimization rules.
+use std::borrow::Cow;
use std::collections::{HashMap, VecDeque};
use std::mem::size_of;
use std::sync::Arc;
@@ -73,8 +74,11 @@ impl JoinHashMapType for PruningJoinHashMap {
iter: Box<dyn Iterator<Item = (usize, &'a u64)> + 'a>,
deleted_offset: Option<usize>,
) -> (Vec<u32>, Vec<u64>) {
- // Flatten the deque
- let next: Vec<u64> = self.next.iter().copied().collect();
+ // Flatten the deque, unless it is already contiguous
+ let next = match self.next.as_slices() {
+ (head, []) => Cow::Borrowed(head),
+ _ => Cow::Owned(self.next.iter().copied().collect()),
+ };
get_matched_indices::<u64>(&self.map, &next, iter, deleted_offset)
}
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index 061afc6e6e..063ae6d138 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -561,7 +561,7 @@ impl ExecutionPlan for ProjectionExec {
order: &[PhysicalSortExpr],
) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
let child = self.input();
- let mut child_order = Vec::new();
+ let mut child_order = Vec::with_capacity(order.len());
// Check and transform sort expressions
for sort_expr in order {
@@ -1150,7 +1150,7 @@ pub fn update_ordering(
ordering: LexOrdering,
projected_exprs: &[ProjectionExpr],
) -> Result<Option<LexOrdering>> {
- let mut updated_exprs = vec![];
+ let mut updated_exprs = Vec::with_capacity(ordering.len());
for mut sort_expr in ordering.into_iter() {
let Some(updated_expr) = update_expr(&sort_expr.expr, projected_exprs, false)?
else {
@@ -1168,7 +1168,7 @@ pub fn update_ordering_requirement(
reqs: LexRequirement,
projected_exprs: &[ProjectionExpr],
) -> Result<Option<LexRequirement>> {
- let mut updated_exprs = vec![];
+ let mut updated_exprs = Vec::with_capacity(reqs.len());
for mut sort_expr in reqs.into_iter() {
let Some(updated_expr) = update_expr(&sort_expr.expr, projected_exprs, false)?
else {
diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs
index 629846b3de..b4984ab5b2 100644
--- a/datafusion/physical-plan/src/repartition/mod.rs
+++ b/datafusion/physical-plan/src/repartition/mod.rs
@@ -213,10 +213,14 @@ impl PartitionSpillWriters {
}
impl OutputChannel {
- fn coalesce(&mut self, batch: RecordBatch) -> Result<Vec<RecordBatch>> {
+ /// Appends the batches that are ready to send to `out`.
+ fn coalesce(&mut self, batch: RecordBatch, out: &mut Vec<RecordBatch>) -> Result<()> {
match &self.shared_coalescer {
- Some(shared) => Ok(shared.push_and_drain(batch)?),
- None => Ok(vec![batch]),
+ Some(shared) => shared.push_and_drain(batch, out),
+ None => {
+ out.push(batch);
+ Ok(())
+ }
}
}
@@ -288,15 +292,18 @@ impl SharedCoalescer {
}
/// Push `batch` into the coalescer and drain any newly completed
- /// batches. The mutex is held only briefly.
- fn push_and_drain(&self, batch: RecordBatch) -> Result<Vec<RecordBatch>> {
- let mut acc = Vec::new();
+ /// batches into `out`. The mutex is held only briefly.
+ fn push_and_drain(
+ &self,
+ batch: RecordBatch,
+ out: &mut Vec<RecordBatch>,
+ ) -> Result<()> {
let mut c = self.inner.lock();
c.push_batch(batch)?;
while let Some(b) = c.next_completed_batch() {
- acc.push(b);
+ out.push(b);
}
- Ok(acc)
+ Ok(())
}
/// Decrement the active-senders counter. If this caller was the last
@@ -1959,7 +1966,7 @@ impl ExecutionPlan for RepartitionExec {
let new_partitioning = match self.partitioning() {
Partitioning::Hash(partitions, size) => {
- let mut new_partitions = vec![];
+ let mut new_partitions = Vec::with_capacity(partitions.len());
for partition in partitions {
let Some(new_partition) =
update_expr(partition, projection.expr(), false)?
@@ -2300,6 +2307,7 @@ impl RepartitionExec {
// While there are still outputs to send to, keep pulling inputs
let mut batches_until_yield = partitioner.num_partitions();
+ let mut coalesced = Vec::new();
while !output_channels.is_empty() {
// fetch the next batch
let timer = metrics.fetch_time.timer();
@@ -2323,7 +2331,8 @@ impl RepartitionExec {
let timer = metrics.send_time[partition].timer();
// if there is still a receiver, send to it
if let Some(output_channel) = output_channels.get_mut(&partition) {
- for batch in output_channel.coalesce(batch)? {
+ output_channel.coalesce(batch, &mut coalesced)?;
+ for batch in coalesced.drain(..) {
if output_channel.send(batch).await.is_err() {
// If the other end has hung up, it was an early shutdown (e.g. LIMIT)
// so ignore this channel from now on.
diff --git a/datafusion/physical-plan/src/sorts/builder.rs b/datafusion/physical-plan/src/sorts/builder.rs
index 89763efc4d..38df487d64 100644
--- a/datafusion/physical-plan/src/sorts/builder.rs
+++ b/datafusion/physical-plan/src/sorts/builder.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::spill::get_record_batch_memory_size;
-use arrow::array::ArrayRef;
+use arrow::array::{Array, ArrayRef};
use arrow::compute::interleave;
use arrow::datatypes::SchemaRef;
use arrow::error::ArrowError;
@@ -151,13 +151,15 @@ impl BatchBuilder {
&self,
indices: &[(usize, usize)],
) -> Result<Vec<ArrayRef>> {
+ let mut arrays: Vec<&dyn Array> = Vec::with_capacity(self.batches.len());
(0..self.schema.fields.len())
.map(|column_idx| {
- let arrays: Vec<_> = self
- .batches
- .iter()
- .map(|(_, batch)| batch.column(column_idx).as_ref())
- .collect();
+ arrays.clear();
+ arrays.extend(
+ self.batches
+ .iter()
+ .map(|(_, batch)| batch.column(column_idx).as_ref()),
+ );
// Arrow 58.1.0+ returns OffsetOverflowError directly from
// interleave, allowing retry_interleave to shrink the batch.
interleave(&arrays, indices).map_err(Into::into)
diff --git a/datafusion/physical-plan/src/sorts/partial_sort.rs b/datafusion/physical-plan/src/sorts/partial_sort.rs
index 517e7ec5c5..29f0a55bc3 100644
--- a/datafusion/physical-plan/src/sorts/partial_sort.rs
+++ b/datafusion/physical-plan/src/sorts/partial_sort.rs
@@ -854,7 +854,9 @@ impl PartialSortStream {
})
.collect::<Result<Vec<_>>>()?;
let mut remaining_fetch = self.fetch.unwrap_or(usize::MAX);
- let mut interleave_indices = vec![];
+ let total_rows = completed.batch_ends.last().copied().unwrap_or(0);
+ let mut interleave_indices = Vec::with_capacity(remaining_fetch.min(total_rows));
+ let mut sorted_indices: Vec<usize> = Vec::new();
for prefix_range in completed.prefix_ranges() {
if remaining_fetch == 0 {
@@ -863,26 +865,28 @@ impl PartialSortStream {
let row_count = prefix_range.len();
let prefix_fetch = remaining_fetch.min(row_count);
- let sorted_indices = if let Some(suffix_ordering) = &self.suffix_ordering {
+ sorted_indices.clear();
+ if let Some(suffix_ordering) = &self.suffix_ordering {
let sort_columns = completed.gather_sort_columns(
prefix_range.clone(),
&sort_columns_by_batch,
suffix_ordering,
)?;
let fetch = (prefix_fetch < row_count).then_some(prefix_fetch);
- lexsort_to_indices(&sort_columns, fetch)?
- .values()
- .iter()
- .map(|idx| *idx as usize)
- .collect::<Vec<_>>()
+ sorted_indices.extend(
+ lexsort_to_indices(&sort_columns, fetch)?
+ .values()
+ .iter()
+ .map(|idx| *idx as usize),
+ );
} else {
- (0..prefix_fetch).collect()
- };
+ sorted_indices.extend(0..prefix_fetch);
+ }
interleave_indices.extend(
sorted_indices
- .into_iter()
- .map(|index| completed.source_row(prefix_range.start + index)),
+ .iter()
+ .map(|&index| completed.source_row(prefix_range.start + index)),
);
remaining_fetch -= prefix_fetch;
}
diff --git a/datafusion/physical-plan/src/spill/mod.rs b/datafusion/physical-plan/src/spill/mod.rs
index 21e07fa23b..9eaa04389c 100644
--- a/datafusion/physical-plan/src/spill/mod.rs
+++ b/datafusion/physical-plan/src/spill/mod.rs
@@ -891,7 +891,8 @@ fn compact_view_array<T: ByteViewType>(
/// Returns, for each entry of `buffers`, its index in a list without repeated
/// entries of the same buffer, and that list.
fn dedup_buffer_entries(buffers: &[Buffer]) -> (Vec<u32>, Vec<Buffer>) {
- let mut first_index: HashMap<(*const u8, usize), u32> = HashMap::new();
+ let mut first_index: HashMap<(*const u8, usize), u32> =
+ HashMap::with_capacity(buffers.len());
let mut unique: Vec<Buffer> = Vec::with_capacity(buffers.len());
let remap = buffers
.iter()
diff --git a/datafusion/physical-plan/src/streaming.rs b/datafusion/physical-plan/src/streaming.rs
index e58de52ae9..586d7ab1bc 100644
--- a/datafusion/physical-plan/src/streaming.rs
+++ b/datafusion/physical-plan/src/streaming.rs
@@ -347,7 +347,7 @@ impl ExecutionPlan for StreamingTableExec {
.unwrap_or_else(|| (0..self.schema().fields().len()).collect()),
);
- let mut lex_orderings = vec![];
+ let mut lex_orderings = Vec::with_capacity(self.projected_output_ordering.len());
for ordering in self.projected_output_ordering().into_iter() {
let Some(ordering) = update_ordering(ordering, projection.expr())? else {
return Ok(None);
diff --git a/datafusion/physical-plan/src/topk/mod.rs b/datafusion/physical-plan/src/topk/mod.rs
index 653875b8d3..64b31f5e81 100644
--- a/datafusion/physical-plan/src/topk/mod.rs
+++ b/datafusion/physical-plan/src/topk/mod.rs
@@ -276,10 +276,9 @@ impl<'a> TopKHeapBoundary<'a> {
sort_exprs: &[PhysicalSortExpr],
) -> Result<Vec<ScalarValue>> {
let mut scalar_values = Vec::with_capacity(sort_exprs.len());
+ let row = self.batch.slice(self.row.index, 1);
for sort_expr in sort_exprs {
- let value = sort_expr
- .expr
- .evaluate(&self.batch.slice(self.row.index, 1))?;
+ let value = sort_expr.expr.evaluate(&row)?;
let scalar = match value {
ColumnarValue::Scalar(scalar) => scalar,
@@ -2959,6 +2958,7 @@ impl PartitionedTopKDenseRank {
let mut batch_refs: Vec<&RecordBatch> = Vec::new();
let mut batch_id_pos: HashMap<u32, usize> = HashMap::new();
let mut indices: Vec<(usize, usize)> = Vec::with_capacity(batch_size);
+ let mut sorted_obs: Vec<(Vec<u8>, Vec<GroupEntry>)> = Vec::new();
// Chunk at `batch_size` so the operator emits the same batch sizes
// as before and no single `interleave` output exceeds `batch_size`.
@@ -2970,10 +2970,9 @@ impl PartitionedTopKDenseRank {
} = states.remove(&pk).expect("key from states.keys()");
// Sort the <= K distinct ob keys so rows emit ascending
// (byte-comparable encoding == sort order).
- let mut sorted_obs: Vec<(Vec<u8>, Vec<GroupEntry>)> =
- groups.into_iter().collect();
+ sorted_obs.extend(groups);
sorted_obs.sort_by(|a, b| a.0.cmp(&b.0));
- for (_ob, entries) in sorted_obs {
+ for (_ob, entries) in sorted_obs.drain(..) {
for entry in entries {
let batch = &store
.get(entry.batch_id)
diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs
index b764098e7c..bbf8192d21 100644
--- a/datafusion/physical-plan/src/union.rs
+++ b/datafusion/physical-plan/src/union.rs
@@ -846,7 +846,7 @@ impl ExecutionPlan for InterleaveExec {
let elapsed_compute = baseline_metrics.elapsed_compute().clone();
let _timer = elapsed_compute.timer(); // record on drop
- let mut input_stream_vec = vec![];
+ let mut input_stream_vec = Vec::with_capacity(self.inputs.len());
for input in self.inputs.iter() {
if partition < input.output_partitioning().partition_count() {
let stream = input.execute(partition, Arc::clone(&context))?;
diff --git a/datafusion/physical-plan/src/unnest.rs b/datafusion/physical-plan/src/unnest.rs
index 4af27dd140..c58faf496c 100644
--- a/datafusion/physical-plan/src/unnest.rs
+++ b/datafusion/physical-plan/src/unnest.rs
@@ -996,7 +996,8 @@ fn build_batch(
let transformed = match list_type_columns.len() {
0 => flatten_struct_cols(batch.columns(), schema, struct_column_indices),
_ => {
- let mut temp_unnested_result = HashMap::new();
+ let mut temp_unnested_result =
+ HashMap::with_capacity(list_type_columns.len());
let max_recursion = list_type_columns
.iter()
.fold(0, |highest_depth, ListUnnest { depth, .. }| {
@@ -1036,7 +1037,7 @@ fn build_batch(
}
let unnested_array_map: HashMap<usize, Vec<UnnestingResult>> =
temp_unnested_result.into_iter().fold(
- HashMap::new(),
+ HashMap::with_capacity(list_type_columns.len()),
|mut acc,
(
ListUnnest {
diff --git a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs
index 724a722898..fc1263ea90 100644
--- a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs
+++ b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs
@@ -718,6 +718,8 @@ pub struct LinearSearch {
/// Keeps the hash of input buffer calculated from PARTITION BY columns.
/// Its length is equal to the `input_buffer` length.
input_buffer_hashes: VecDeque<u64>,
+ /// Hashes of the latest batch, reused across batches.
+ batch_hashes: Vec<u64>,
/// Used during hash value calculation.
random_state: RandomState,
/// Input ordering and partition by key ordering need not be the same, so
@@ -899,6 +901,7 @@ impl LinearSearch {
fn new(ordered_partition_by_indices: Vec<usize>, input_schema: SchemaRef) -> Self {
LinearSearch {
input_buffer_hashes: VecDeque::new(),
+ batch_hashes: Vec::new(),
random_state: Default::default(),
ordered_partition_by_indices,
row_map_batch: HashTable::with_capacity(256),
@@ -920,9 +923,10 @@ impl LinearSearch {
batch: &RecordBatch,
) -> Result<(Vec<PartitionKey>, Vec<u32>, Vec<usize>)> {
let num_rows = batch.num_rows();
- let mut batch_hashes = vec![0; num_rows];
- create_hashes(columns, &self.random_state, &mut batch_hashes)?;
- self.input_buffer_hashes.extend(&batch_hashes);
+ self.batch_hashes.clear();
+ self.batch_hashes.resize(num_rows, 0);
+ create_hashes(columns, &self.random_state, &mut self.batch_hashes)?;
+ self.input_buffer_hashes.extend(&self.batch_hashes);
// reset row_map for new calculation
self.row_map_batch.clear();
let mut keys: Vec<PartitionKey> = vec![];
@@ -930,7 +934,7 @@ impl LinearSearch {
let mut row_partition_ids = Vec::with_capacity(num_rows);
// Number of rows in each partition:
let mut counts: Vec<usize> = vec![];
- for (hash, row_idx) in batch_hashes.into_iter().zip(0u32..) {
+ for (&hash, row_idx) in self.batch_hashes.iter().zip(0u32..) {
let entry = self.row_map_batch.find_mut(hash, |(_, group_idx)| {
let row = get_row_at_idx(columns, row_idx as usize).unwrap();
// Handle hash collisions with an equality check:
@@ -1114,7 +1118,7 @@ impl SortedSearch {
) -> usize {
// Different window aggregators may produce results at different rates.
// We produce the overall batch result only as fast as the slowest one.
- let mut counts = vec![];
+ let mut counts = Vec::with_capacity(window_agg_states.len());
let out_col_counts = window_agg_states.iter().map(|window_agg_state| {
// Store how many elements are generated for the current
// window expression:
@@ -1559,7 +1563,7 @@ fn get_aggregate_result_out_column(
) -> Result<ArrayRef> {
let mut result = None;
let mut running_length = 0;
- let mut batches_to_concat = vec![];
+ let mut batches_to_concat = Vec::with_capacity(partition_window_agg_states.len());
// We assume that iteration order is according to insertion order
for (
_,
diff --git a/datafusion/physical-plan/src/windows/mod.rs b/datafusion/physical-plan/src/windows/mod.rs
index aeb49010d2..eade0a569a 100644
--- a/datafusion/physical-plan/src/windows/mod.rs
+++ b/datafusion/physical-plan/src/windows/mod.rs
@@ -398,7 +398,7 @@ pub(crate) fn window_equivalence_properties(
// Find "one" valid ordering for partition columns to avoid exponential complexity.
// see https://github.com/apache/datafusion/issues/17401
let mut all_satisfied_lexs = vec![];
- let mut candidate_ordering = vec![];
+ let mut candidate_ordering = Vec::with_capacity(partitioning_exprs.len());
for partition_expr in partitioning_exprs.iter() {
let sort_options =
@@ -518,7 +518,7 @@ pub(crate) fn window_equivalence_properties(
// Find one valid ordering for aggregate arguments instead of
// checking all combinations
let aggregate_exprs = sliding_expr.get_aggregate_expr().expressions();
- let mut candidate_order = vec![];
+ let mut candidate_order = Vec::with_capacity(aggregate_exprs.len());
let mut asc = false;
for (idx, expr) in aggregate_exprs.iter().enumerate() {
diff --git a/datafusion/physical-plan/src/windows/window_agg_exec.rs b/datafusion/physical-plan/src/windows/window_agg_exec.rs
index e00aaa21bb..89fafd054a 100644
--- a/datafusion/physical-plan/src/windows/window_agg_exec.rs
+++ b/datafusion/physical-plan/src/windows/window_agg_exec.rs
@@ -555,7 +555,7 @@ impl WindowAggStream {
let partition_points =
evaluate_partition_ranges(batch.num_rows(), &partition_by_sort_keys)?;
- let mut partition_results = vec![];
+ let mut partition_results = Vec::with_capacity(partition_points.len());
// Calculate window cols
for partition_point in partition_points {
let length = partition_point.end - partition_point.start;
diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs
index 98c734f184..8e10c74c76 100644
--- a/datafusion/proto-common/src/from_proto/mod.rs
+++ b/datafusion/proto-common/src/from_proto/mod.rs
@@ -492,7 +492,8 @@ impl TryFrom<&protobuf::ScalarValue> for ScalarValue {
)
})?;
- let mut dict_by_id: HashMap<i64, ArrayRef> = HashMap::new();
+ let mut dict_by_id: HashMap<i64, ArrayRef> =
+ HashMap::with_capacity(dictionaries.len());
for protobuf::scalar_nested_value::Dictionary {
ipc_message,
arrow_data,
@@ -1371,7 +1372,7 @@ impl TryFrom<&protobuf::TableParquetOptions> for TableParquetOptions {
value: &protobuf::TableParquetOptions,
) -> datafusion_common::Result<Self, Self::Error> {
let mut column_specific_options: HashMap<String, ParquetColumnOptions> =
- HashMap::new();
+ HashMap::with_capacity(value.column_specific_options.len());
for protobuf::ParquetColumnSpecificOptions {
column_name,
options: maybe_options,
diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs
index 2e1811f76b..9ebc21c876 100644
--- a/datafusion/proto/src/logical_plan/mod.rs
+++ b/datafusion/proto/src/logical_plan/mod.rs
@@ -590,7 +590,7 @@ impl AsLogicalPlan for LogicalPlanNode {
let filters =
from_proto::parse_exprs(&scan.filters, ctx, extension_codec)?;
- let mut all_sort_orders = vec![];
+ let mut all_sort_orders = Vec::with_capacity(scan.file_sort_order.len());
for order in &scan.file_sort_order {
all_sort_orders.push(from_proto::parse_sorts(
&order.sort_expr_nodes,
@@ -825,7 +825,8 @@ impl AsLogicalPlan for LogicalPlanNode {
None
};
- let mut order_exprs = vec![];
+ let mut order_exprs =
+ Vec::with_capacity(create_extern_table.order_exprs.len());
for expr in &create_extern_table.order_exprs {
order_exprs.push(from_proto::parse_sorts(
&expr.sort_expr_nodes,
@@ -1525,7 +1526,8 @@ impl AsLogicalPlan for LogicalPlanNode {
let schema: protobuf::Schema = (&schema).try_into()?;
- let mut exprs_vec: Vec<SortExprNodeCollection> = vec![];
+ let mut exprs_vec: Vec<SortExprNodeCollection> =
+ Vec::with_capacity(options.file_sort_order.len());
for order in &options.file_sort_order {
let expr_vec = SortExprNodeCollection {
sort_expr_nodes: serialize_sorts(order, extension_codec)?,
@@ -2035,7 +2037,8 @@ impl AsLogicalPlan for LogicalPlanNode {
column_defaults,
temporary,
} = ce.as_ref();
- let mut converted_order_exprs: Vec<SortExprNodeCollection> = vec![];
+ let mut converted_order_exprs: Vec<SortExprNodeCollection> =
+ Vec::with_capacity(order_exprs.len());
for order in order_exprs {
let temp = SortExprNodeCollection {
sort_expr_nodes: serialize_sorts(order, extension_codec)?,
diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs
index 4021f25d3e..07a24efb0d 100644
--- a/datafusion/pruning/src/pruning_predicate.rs
+++ b/datafusion/pruning/src/pruning_predicate.rs
@@ -765,7 +765,7 @@ impl PruningPredicate {
/// used in the predicate. For example, it can be used to avoid reading
/// unneeded bloom filters (a non trivial operation).
pub fn literal_columns(&self) -> Vec<String> {
- let mut seen = HashSet::new();
+ let mut seen = HashSet::with_capacity(self.literal_guarantees.len());
self.literal_guarantees
.iter()
.map(|e| &e.column.name)
@@ -1063,7 +1063,7 @@ fn build_statistics_record_batch<S: PruningStatistics + ?Sized>(
statistics: &S,
required_columns: &RequiredColumns,
) -> Result<RecordBatch> {
- let mut arrays = Vec::<ArrayRef>::new();
+ let mut arrays = Vec::<ArrayRef>::with_capacity(required_columns.columns.len());
// For each needed statistics column:
for (column, statistics_type, stat_field) in required_columns.iter() {
let column = Column::from_name(column.name());
diff --git a/datafusion/spark/src/function/array/shuffle.rs b/datafusion/spark/src/function/array/shuffle.rs
index cf1bfe4ed8..c86433c763 100644
--- a/datafusion/spark/src/function/array/shuffle.rs
+++ b/datafusion/spark/src/function/array/shuffle.rs
@@ -171,8 +171,9 @@ fn general_array_shuffle<O: OffsetSizeTrait>(
// Each null row emits one placeholder child value.
let capacity =
Capacities::Array(offset_span_len(array.offsets()) + array.null_count());
- let mut offsets = vec![O::usize_as(0)];
- let mut nulls = vec![];
+ let mut offsets = Vec::with_capacity(array.len() + 1);
+ offsets.push(O::usize_as(0));
+ let mut nulls = Vec::with_capacity(array.len());
let mut mutable =
MutableArrayData::with_capacities(vec![&original_data], false, capacity);
let mut rng = if let Some(s) = seed {
@@ -182,6 +183,7 @@ fn general_array_shuffle<O: OffsetSizeTrait>(
let seed = rng().random::<u64>();
StdRng::seed_from_u64(seed)
};
+ let mut indices: Vec<usize> = Vec::new();
for (row_index, offset_window) in array.offsets().windows(2).enumerate() {
// skip the null value
@@ -196,8 +198,8 @@ fn general_array_shuffle<O: OffsetSizeTrait>(
let length = (end - start).to_usize().unwrap();
// Create indices and shuffle them
- let mut indices: Vec<usize> =
- (start.to_usize().unwrap()..end.to_usize().unwrap()).collect();
+ indices.clear();
+ indices.extend(start.to_usize().unwrap()..end.to_usize().unwrap());
indices.shuffle(&mut rng);
// Add shuffled elements
@@ -225,7 +227,7 @@ fn fixed_size_array_shuffle(
let values = array.values();
let original_data = values.to_data();
let capacity = Capacities::Array(original_data.len());
- let mut nulls = vec![];
+ let mut nulls = Vec::with_capacity(array.len());
let mut mutable =
MutableArrayData::with_capacities(vec![&original_data], false, capacity);
let value_length = array.value_length() as usize;
@@ -236,6 +238,7 @@ fn fixed_size_array_shuffle(
let seed = rng().random::<u64>();
StdRng::seed_from_u64(seed)
};
+ let mut indices: Vec<usize> = Vec::with_capacity(value_length);
for row_index in 0..array.len() {
// skip the null value
@@ -250,7 +253,8 @@ fn fixed_size_array_shuffle(
let end = start + value_length;
// Create indices and shuffle them
- let mut indices: Vec<usize> = (start..end).collect();
+ indices.clear();
+ indices.extend(start..end);
indices.shuffle(&mut rng);
// Add shuffled elements
diff --git a/datafusion/spark/src/function/hash/sha1.rs b/datafusion/spark/src/function/hash/sha1.rs
index 64c52a7108..d19e6dc4e3 100644
--- a/datafusion/spark/src/function/hash/sha1.rs
+++ b/datafusion/spark/src/function/hash/sha1.rs
@@ -17,7 +17,7 @@
use std::sync::Arc;
-use arrow::array::{ArrayRef, StringArray};
+use arrow::array::{ArrayRef, StringArray, StringBuilder};
use arrow::datatypes::{DataType, Field, FieldRef};
use datafusion_common::cast::{
as_binary_array, as_binary_view_array, as_fixed_size_binary_array,
@@ -91,18 +91,24 @@ impl ScalarUDFImpl for SparkSha1 {
}
#[inline]
-fn spark_sha1_digest(value: &[u8]) -> String {
- let mut out = Vec::with_capacity(40);
+fn spark_sha1_digest<'b>(value: &[u8], out: &'b mut Vec<u8>) -> &'b str {
+ out.clear();
+ encode_bytes_into(&Sha1::digest(value), HexCase::Lower, out);
// Safe: `encode_bytes_into` only writes ASCII hex digits, which are valid UTF-8.
- encode_bytes_into(&Sha1::digest(value), HexCase::Lower, &mut out);
- unsafe { String::from_utf8_unchecked(out) }
+ unsafe { std::str::from_utf8_unchecked(out) }
}
fn spark_sha1_impl<'a>(input: impl Iterator<Item = Option<&'a [u8]>>) -> ArrayRef {
- let result = input
- .map(|value| value.map(spark_sha1_digest))
- .collect::<StringArray>();
- Arc::new(result)
+ let len = input.size_hint().0;
+ let mut builder = StringBuilder::with_capacity(len, len * 40);
+ let mut out = Vec::with_capacity(40);
+ for value in input {
+ match value {
+ Some(value) => builder.append_value(spark_sha1_digest(value, &mut out)),
+ None => builder.append_null(),
+ }
+ }
+ Arc::new(builder.finish())
}
fn spark_sha1(args: &[ArrayRef]) -> Result<ArrayRef> {
diff --git a/datafusion/spark/src/function/hash/utils.rs b/datafusion/spark/src/function/hash/utils.rs
index 7b6f520ea4..cc88b1d709 100644
--- a/datafusion/spark/src/function/hash/utils.rs
+++ b/datafusion/spark/src/function/hash/utils.rs
@@ -990,21 +990,21 @@ macro_rules! create_hashes_internal {
DataType::Struct(_) => {
let struct_array = col.as_any().downcast_ref::<StructArray>().unwrap();
// Hash each field of the struct - Spark hashes all fields recursively
- let columns: Vec<ArrayRef> = struct_array.columns().to_vec();
+ let columns = struct_array.columns();
if !columns.is_empty() {
match struct_array.nulls().filter(|nulls| nulls.null_count() > 0) {
// A NULL struct leaves the hash unchanged like any other
// NULL, whatever its fields still hold.
Some(nulls) => {
let before = $hashes_buffer.to_vec();
- $recursive_hash_method(&columns, $hashes_buffer)?;
+ $recursive_hash_method(columns, $hashes_buffer)?;
for (i, hash) in $hashes_buffer.iter_mut().enumerate() {
if nulls.is_null(i) {
*hash = before[i];
}
}
}
- None => $recursive_hash_method(&columns, $hashes_buffer)?,
+ None => $recursive_hash_method(columns, $hashes_buffer)?,
}
}
}
diff --git a/datafusion/spark/src/function/json/json_tuple.rs b/datafusion/spark/src/function/json/json_tuple.rs
index 3496f979ff..99df115010 100644
--- a/datafusion/spark/src/function/json/json_tuple.rs
+++ b/datafusion/spark/src/function/json/json_tuple.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::fmt::Write as _;
use std::sync::Arc;
use arrow::array::{Array, ArrayRef, NullBufferBuilder, StringBuilder, StructArray};
@@ -151,7 +152,9 @@ fn json_tuple_inner(args: &[ArrayRef], return_type: &DataType) -> Result<ArrayRe
builder.append_value(s);
}
Some(other) => {
- builder.append_value(other.to_string());
+ // Serialize straight into the builder, then finish the row.
+ write!(builder, "{other}")?;
+ builder.append_value("");
}
None => {
builder.append_null();
diff --git a/datafusion/spark/src/function/map/utils.rs b/datafusion/spark/src/function/map/utils.rs
index 6f6fec5b3b..e450ba19aa 100644
--- a/datafusion/spark/src/function/map/utils.rs
+++ b/datafusion/spark/src/function/map/utils.rs
@@ -194,7 +194,9 @@ fn map_deduplicate_keys(
// in `flat_values` to materialize for each output slot (updated in place
// on overwrite).
let mut keys_mask_builder = BooleanBuilder::new();
- let mut value_indices: Vec<i32> = Vec::new();
+ let mut value_indices: Vec<i32> = Vec::with_capacity(
+ keys_offsets.last().map_or(0, |offset| *offset as usize) - first_keys_offset,
+ );
let mut key_to_output_idx: HashMap<ScalarValue, usize> = HashMap::new();
for (row_idx, (next_keys_offset, next_values_offset)) in keys_offsets
.iter()
diff --git a/datafusion/spark/src/function/string/encode.rs b/datafusion/spark/src/function/string/encode.rs
index 250778be15..1e523e3f0e 100644
--- a/datafusion/spark/src/function/string/encode.rs
+++ b/datafusion/spark/src/function/string/encode.rs
@@ -74,64 +74,56 @@ impl SparkEncode {
}
}
-/// Encodes a single string value using the specified charset.
+/// Encodes a single string value using the specified charset into `bytes`,
+/// replacing its previous contents.
/// Unmappable characters are silently replaced with `?` (Spark 3.5 behavior).
-fn encode_string(s: &str, charset: &str) -> Result<Vec<u8>> {
+fn encode_string(s: &str, charset: &str, bytes: &mut Vec<u8>) -> Result<()> {
+ bytes.clear();
match charset {
- "UTF-8" | "UTF8" => Ok(s.as_bytes().to_vec()),
- "US-ASCII" | "ASCII" => Ok(s
- .chars()
- .map(|c| if c.is_ascii() { c as u8 } else { b'?' })
- .collect()),
- "ISO-8859-1" | "ISO88591" | "LATIN1" => Ok(s
- .chars()
- .map(|c| {
- let cp = c as u32;
- if cp <= 255 { cp as u8 } else { b'?' }
- })
- .collect()),
+ "UTF-8" | "UTF8" => bytes.extend_from_slice(s.as_bytes()),
+ "US-ASCII" | "ASCII" => {
+ bytes.extend(s.chars().map(|c| if c.is_ascii() { c as u8 } else { b'?' }))
+ }
+ "ISO-8859-1" | "ISO88591" | "LATIN1" => bytes.extend(s.chars().map(|c| {
+ let cp = c as u32;
+ if cp <= 255 { cp as u8 } else { b'?' }
+ })),
"UTF-16BE" | "UTF16BE" => {
- let mut bytes = Vec::new();
for code_unit in s.encode_utf16() {
bytes.extend_from_slice(&code_unit.to_be_bytes());
}
- Ok(bytes)
}
"UTF-16LE" | "UTF16LE" => {
- let mut bytes = Vec::new();
for code_unit in s.encode_utf16() {
bytes.extend_from_slice(&code_unit.to_le_bytes());
}
- Ok(bytes)
}
"UTF-16" | "UTF16" => {
// BOM (big-endian marker) followed by UTF-16BE encoded bytes
- let mut bytes = vec![0xFE, 0xFF];
+ bytes.extend_from_slice(&[0xFE, 0xFF]);
for code_unit in s.encode_utf16() {
bytes.extend_from_slice(&code_unit.to_be_bytes());
}
- Ok(bytes)
}
// Spark treats UTF-32 as UTF-32BE (no BOM), matching Spark 3.5 and 4.1.
"UTF-32" | "UTF32" | "UTF-32BE" | "UTF32BE" => {
- let mut bytes = Vec::with_capacity(s.len() * 4);
for c in s.chars() {
bytes.extend_from_slice(&(c as u32).to_be_bytes());
}
- Ok(bytes)
}
"UTF-32LE" | "UTF32LE" => {
- let mut bytes = Vec::with_capacity(s.len() * 4);
for c in s.chars() {
bytes.extend_from_slice(&(c as u32).to_le_bytes());
}
- Ok(bytes)
}
- _ => exec_err!(
- "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
- charset
- ),
+ _ => {
+ return exec_err!(
+ "Unsupported charset for encode: '{}'. Supported: US-ASCII, ISO-8859-1, UTF-8, UTF-16, UTF-16BE, UTF-16LE, UTF-32, UTF-32BE, UTF-32LE",
+ charset
+ );
+ }
}
+ Ok(())
}
/// The charset for each row. `Constant` is a single uppercased charset; `PerRow`
@@ -163,10 +155,12 @@ fn encode_rows<'a>(
charsets: &Charsets,
) -> Result<ArrayRef> {
let mut builder = BinaryBuilder::with_capacity(len, len * 4);
+ let mut bytes = Vec::new();
for i in 0..len {
match (is_null(i), charsets.get(i)) {
(false, Some(charset)) => {
- builder.append_value(&encode_string(&decode(i), &charset)?)
+ encode_string(&decode(i), &charset, &mut bytes)?;
+ builder.append_value(&bytes);
}
_ => builder.append_null(),
}
diff --git a/datafusion/spark/src/function/string/format_string.rs b/datafusion/spark/src/function/string/format_string.rs
index 131d1c14df..0e24c87fc8 100644
--- a/datafusion/spark/src/function/string/format_string.rs
+++ b/datafusion/spark/src/function/string/format_string.rs
@@ -124,11 +124,12 @@ impl ScalarUDFImpl for FormatStringFunc {
| ColumnarValue::Scalar(ScalarValue::Utf8View(Some(fmt))) => {
let formatter = Formatter::parse(fmt, &data_types)?;
let mut result = Vec::with_capacity(len.unwrap_or(1));
+ let mut scalars = Vec::with_capacity(data_types.len());
for i in 0..len.unwrap_or(1) {
- let scalars = args.args[1..]
- .iter()
- .map(|arg| try_to_scalar(arg.clone(), i))
- .collect::<Result<Vec<_>>>()?;
+ scalars.clear();
+ for arg in &args.args[1..] {
+ scalars.push(try_to_scalar(arg.clone(), i)?);
+ }
let formatted = formatter.format(&scalars)?;
result.push(formatted);
}
@@ -158,15 +159,16 @@ impl ScalarUDFImpl for FormatStringFunc {
}
ColumnarValue::Array(fmts) => {
let mut result = Vec::with_capacity(len.unwrap());
+ let mut scalars = Vec::with_capacity(data_types.len());
for i in 0..len.unwrap() {
let fmt = ScalarValue::try_from_array(fmts, i)?;
match fmt.try_as_str() {
Some(Some(fmt)) => {
let formatter = Formatter::parse(fmt, &data_types)?;
- let scalars = args.args[1..]
- .iter()
- .map(|arg| try_to_scalar(arg.clone(), i))
- .collect::<Result<Vec<_>>>()?;
+ scalars.clear();
+ for arg in &args.args[1..] {
+ scalars.push(try_to_scalar(arg.clone(), i)?);
+ }
let formatted = formatter.format(&scalars)?;
result.push(Some(formatted));
}
@@ -1610,8 +1612,8 @@ impl ConversionSpecifier {
}
writer.push_str(suffix);
} else {
- while temp.len() < width as usize {
- temp = " ".to_owned() + &temp;
+ for _ in temp.len()..width as usize {
+ writer.push(' ');
}
writer.push_str(&temp);
}
@@ -1878,11 +1880,11 @@ impl ConversionSpecifier {
}
prefix + &s
} else {
- let mut num_str = prefix + &s;
- while num_str.len() < width as usize {
- num_str = " ".to_owned() + &num_str;
+ let mut num_str = String::new();
+ while num_str.len() + prefix.len() + s.len() < width as usize {
+ num_str.push(' ');
}
- num_str
+ num_str + &prefix + &s
}
} else {
prefix + &s
@@ -2248,9 +2250,9 @@ impl ConversionSpecifier {
writer.push_str(number);
writer.push_str(suffix);
} else {
- let mut full_num = prefix + number + suffix;
- while full_num.len() < width as usize {
- full_num = " ".to_owned() + &full_num;
+ let full_num = prefix + number + suffix;
+ for _ in full_num.len()..width as usize {
+ writer.push(' ');
}
writer.push_str(&full_num);
}
diff --git a/datafusion/sql/src/expr/function.rs b/datafusion/sql/src/expr/function.rs
index 4bbf3834de..f3e4c79740 100644
--- a/datafusion/sql/src/expr/function.rs
+++ b/datafusion/sql/src/expr/function.rs
@@ -608,7 +608,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
if let Expr::Column(col) = &orderby_expr.expr {
let idx = schema.index_of_column(col).ok()?;
return if func_deps.iter().any(|dep| {
- dep.source_indices == vec![idx] && dep.mode == Dependency::Single
+ dep.source_indices == [idx] && dep.mode == Dependency::Single
}) {
Some(true)
} else {
diff --git a/datafusion/sql/src/expr/mod.rs b/datafusion/sql/src/expr/mod.rs
index 1372773d35..40bd8bb71d 100644
--- a/datafusion/sql/src/expr/mod.rs
+++ b/datafusion/sql/src/expr/mod.rs
@@ -996,8 +996,8 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
schema: &DFSchema,
planner_context: &mut PlannerContext,
) -> Result<Expr> {
- let mut keys = vec![];
- let mut values = vec![];
+ let mut keys = Vec::with_capacity(fields.len());
+ let mut values = Vec::with_capacity(fields.len());
for field in fields {
let key = lit(field.key.value);
let value =
@@ -1027,7 +1027,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
) -> Result<Expr> {
let mut exprs: Vec<_> = entries
.into_iter()
- .flat_map(|entry| vec![entry.key, entry.value].into_iter())
+ .flat_map(|entry| [entry.key, entry.value])
.map(|expr| self.sql_expr_to_logical_expr(*expr, schema, planner_context))
.collect::<Result<Vec<_>>>()?;
for planner in self.context_provider.get_expr_planners() {
diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs
index 79ab852d3d..8b7718ca00 100644
--- a/datafusion/sql/src/select.rs
+++ b/datafusion/sql/src/select.rs
@@ -708,7 +708,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
let null_handling = collect_unnest_null_handling(&intermediate_expr_groups)?;
let mut unnest_options =
UnnestOptions::new().with_null_handling(null_handling);
- let mut unnest_col_vec = vec![];
+ let mut unnest_col_vec = Vec::with_capacity(unnest_columns.len());
for (col, maybe_list_unnest) in unnest_columns.into_iter() {
if let Some(list_unnest) = maybe_list_unnest {
@@ -867,7 +867,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
};
projection_exprs.extend(inner_projection_exprs.into_exprs());
- let mut unnest_col_vec = vec![];
+ let mut unnest_col_vec = Vec::with_capacity(unnest_columns.len());
for (col, maybe_list_unnest) in unnest_columns.into_iter() {
if let Some(list_unnest) = maybe_list_unnest {
@@ -1001,7 +1001,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
empty_from: bool,
planner_context: &mut PlannerContext,
) -> Result<Vec<SelectExpr>> {
- let mut prepared_select_exprs = vec![];
+ let mut prepared_select_exprs = Vec::with_capacity(projection.len());
let mut error_builder = DataFusionErrorBuilder::new();
for expr in projection {
@@ -1251,7 +1251,8 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
// combine the original grouping and aggregate expressions into one list (note that
// we do not add the "having" expression since that is not part of the projection)
- let mut aggr_projection_exprs = vec![];
+ let mut aggr_projection_exprs =
+ Vec::with_capacity(group_by_exprs.len() + aggr_exprs.len());
for expr in group_by_exprs {
match expr {
Expr::GroupingSet(GroupingSet::Rollup(exprs)) => {
diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs
index c89c2a622e..f2fe369508 100644
--- a/datafusion/sql/src/statement.rs
+++ b/datafusion/sql/src/statement.rs
@@ -1831,7 +1831,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
return Ok(results);
}
- let mut all_results = vec![];
+ let mut all_results = Vec::with_capacity(order_exprs.len());
for expr in order_exprs {
// Convert each OrderByExpr to a SortExpr:
let expr_vec =
@@ -2090,7 +2090,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
options: Vec<(String, Value)>,
allow_duplicates: bool,
) -> Result<HashMap<String, String>> {
- let mut options_map = HashMap::new();
+ let mut options_map = HashMap::with_capacity(options.len());
for (key, value) in options {
if !allow_duplicates && options_map.contains_key(&key) {
return plan_err!("Option {key} is specified multiple times");
@@ -2721,7 +2721,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok((col_name, value))
})
.collect::<Result<Vec<_>>>()?;
- let mut seen = HashSet::new();
+ let mut seen = HashSet::with_capacity(assignments.len());
for (column, _) in &assignments {
if !seen.insert(column.as_str()) {
return plan_err!("Duplicate column '{column}' in MERGE UPDATE");
@@ -2742,7 +2742,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
.collect::<Result<Vec<_>>>()?;
// Validate: no duplicates, all columns exist in target schema
- let mut seen = HashSet::new();
+ let mut seen = HashSet::with_capacity(columns.len());
for col in &columns {
if !seen.insert(col.as_str()) {
return plan_err!("Duplicate column '{col}' in MERGE INSERT");
diff --git a/datafusion/sql/src/unparser/expr.rs b/datafusion/sql/src/unparser/expr.rs
index c1a8c921a4..7c7fec38bf 100644
--- a/datafusion/sql/src/unparser/expr.rs
+++ b/datafusion/sql/src/unparser/expr.rs
@@ -726,7 +726,7 @@ impl Unparser<'_> {
}
fn scalar_value_list_to_sql(&self, array: &ArrayRef) -> Result<ast::Expr> {
- let mut elem = Vec::new();
+ let mut elem = Vec::with_capacity(array.len());
for i in 0..array.len() {
let value = ScalarValue::try_from_array(&array, i)?;
elem.push(self.scalar_to_sql(&value)?);
diff --git a/datafusion/sql/src/unparser/plan.rs b/datafusion/sql/src/unparser/plan.rs
index 9b1b979e3f..83ced616e0 100644
--- a/datafusion/sql/src/unparser/plan.rs
+++ b/datafusion/sql/src/unparser/plan.rs
@@ -503,7 +503,8 @@ impl Unparser<'_> {
.iter()
.any(|column| column.relation.is_some())
});
- let mut input_names = HashSet::new();
+ let mut input_names =
+ HashSet::with_capacity(projection.input.schema().fields().len());
let unique_input_names = projection
.input
.schema()
@@ -2582,11 +2583,11 @@ impl Unparser<'_> {
if project_vec.is_empty() {
builder = builder.project(self.empty_projection_fallback())?;
} else {
+ let schema = table_scan.source.schema();
let project_columns = project_vec
.iter()
.copied()
.map(|i| {
- let schema = table_scan.source.schema();
let field = schema.field(i);
if alias.is_some() {
Column::new(alias.clone(), field.name().clone())
diff --git a/datafusion/sql/src/unparser/rewrite.rs b/datafusion/sql/src/unparser/rewrite.rs
index 63cdad5fc8..a29ee8c5b2 100644
--- a/datafusion/sql/src/unparser/rewrite.rs
+++ b/datafusion/sql/src/unparser/rewrite.rs
@@ -194,7 +194,7 @@ pub(super) fn rewrite_plan_for_sort_on_non_projected_fields(
return None;
};
- let mut map = HashMap::new();
+ let mut map = HashMap::with_capacity(inner_p.expr.len());
let inner_exprs = inner_p
.expr
.iter()
diff --git a/datafusion/sqllogictest/test_files/array/array_slice.slt b/datafusion/sqllogictest/test_files/array/array_slice.slt
index 76b81b28ef..9f75d927c1 100644
--- a/datafusion/sqllogictest/test_files/array/array_slice.slt
+++ b/datafusion/sqllogictest/test_files/array/array_slice.slt
@@ -450,6 +450,26 @@ NULL
NULL
[1, 3, 5]
+# array_slice switching between strided, contiguous, empty, NULL and
+# negative-stride rows in one batch
+query ??
+select array_slice(column1, column2, column3, column4),
+ array_slice(arrow_cast(column1, 'ListView(Int64)'), column2, column3, column4)
+from (
+ values
+ (make_array(1, 2, 3, 4, 5, 6, 7, 8, 9), 1, 9, 2),
+ (make_array(1, 2, 3, 4, 5), 2, 4, 1),
+ (make_array(), 1, 3, 2),
+ (NULL, 1, 3, 2),
+ (make_array(1, 2, 3, 4, 5), 5, 1, -3)
+) as t(column1, column2, column3, column4);
+----
+[1, 3, 5, 7, 9] [1, 3, 5, 7, 9]
+[2, 3, 4] [2, 3, 4]
+[] []
+NULL NULL
+[5, 2] [5, 2]
+
# maintains inner nullability
query ?T
select array_slice(column1, 2, 3), arrow_typeof(array_slice(column1, 2, 3))
diff --git a/datafusion/substrait/src/extensions.rs b/datafusion/substrait/src/extensions.rs
index 78c357f3b8..d8dfd999a2 100644
--- a/datafusion/substrait/src/extensions.rs
+++ b/datafusion/substrait/src/extensions.rs
@@ -84,7 +84,7 @@ impl TryFrom<&Vec<SimpleExtensionDeclaration>> for Extensions {
fn try_from(
value: &Vec<SimpleExtensionDeclaration>,
) -> datafusion::common::Result<Self> {
- let mut functions = HashMap::new();
+ let mut functions = HashMap::with_capacity(value.len());
let mut types = HashMap::new();
let mut type_variations = HashMap::new();
@@ -114,7 +114,9 @@ impl TryFrom<&Vec<SimpleExtensionDeclaration>> for Extensions {
impl From<Extensions> for Vec<SimpleExtensionDeclaration> {
fn from(val: Extensions) -> Vec<SimpleExtensionDeclaration> {
- let mut extensions = vec![];
+ let mut extensions = Vec::with_capacity(
+ val.functions.len() + val.types.len() + val.type_variations.len(),
+ );
for (f_anchor, f_name) in val.functions {
let function_extension = ExtensionFunction {
extension_urn_reference: u32::MAX,
diff --git a/datafusion/substrait/src/logical_plan/consumer/expr/function_arguments.rs b/datafusion/substrait/src/logical_plan/consumer/expr/function_arguments.rs
index cae5ecb6e5..b91d0c3a35 100644
--- a/datafusion/substrait/src/logical_plan/consumer/expr/function_arguments.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/expr/function_arguments.rs
@@ -27,7 +27,7 @@ pub async fn from_substrait_func_args(
arguments: &Vec<FunctionArgument>,
input_schema: &DFSchema,
) -> datafusion::common::Result<Vec<Expr>> {
- let mut args: Vec<Expr> = vec![];
+ let mut args: Vec<Expr> = Vec::with_capacity(arguments.len());
for arg in arguments {
let arg_expr = match &arg.arg_type {
Some(ArgType::Value(e)) => consumer.consume_expression(e, input_schema).await,
diff --git a/datafusion/substrait/src/logical_plan/consumer/expr/if_then.rs b/datafusion/substrait/src/logical_plan/consumer/expr/if_then.rs
index c4cc6c2fcd..eb97d7e323 100644
--- a/datafusion/substrait/src/logical_plan/consumer/expr/if_then.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/expr/if_then.rs
@@ -27,7 +27,8 @@ pub async fn from_if_then(
) -> datafusion::common::Result<Expr> {
// Parse `ifs`
// If the first element does not have a `then` part, then we can assume it's a base expression
- let mut when_then_expr: Vec<(Box<Expr>, Box<Expr>)> = vec![];
+ let mut when_then_expr: Vec<(Box<Expr>, Box<Expr>)> =
+ Vec::with_capacity(if_then.ifs.len());
let mut expr = None;
for (i, if_expr) in if_then.ifs.iter().enumerate() {
if i == 0 {
diff --git a/datafusion/substrait/src/logical_plan/consumer/expr/mod.rs b/datafusion/substrait/src/logical_plan/consumer/expr/mod.rs
index 623bb715cd..ed0f0966d0 100644
--- a/datafusion/substrait/src/logical_plan/consumer/expr/mod.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/expr/mod.rs
@@ -193,7 +193,7 @@ pub async fn from_substrait_rex_vec(
exprs: &Vec<Expression>,
input_schema: &DFSchema,
) -> datafusion::common::Result<Vec<Expr>> {
- let mut expressions: Vec<Expr> = vec![];
+ let mut expressions: Vec<Expr> = Vec::with_capacity(exprs.len());
for expr in exprs {
let expression = consumer.consume_expression(expr, input_schema).await?;
expressions.push(expression);
diff --git a/datafusion/substrait/src/logical_plan/consumer/rel/aggregate_rel.rs b/datafusion/substrait/src/logical_plan/consumer/rel/aggregate_rel.rs
index 982a87d6d5..b0eac9ce8d 100644
--- a/datafusion/substrait/src/logical_plan/consumer/rel/aggregate_rel.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/rel/aggregate_rel.rs
@@ -32,7 +32,7 @@ pub async fn from_aggregate_rel(
) -> datafusion::common::Result<LogicalPlan> {
if let Some(input) = agg.input.as_ref() {
let input = LogicalPlanBuilder::from(consumer.consume_rel(input).await?);
- let mut ref_group_exprs = vec![];
+ let mut ref_group_exprs = Vec::with_capacity(agg.grouping_expressions.len());
for e in &agg.grouping_expressions {
let x = consumer.consume_expression(e, input.schema()).await?;
@@ -40,7 +40,7 @@ pub async fn from_aggregate_rel(
}
let mut group_exprs = vec![];
- let mut aggr_exprs = vec![];
+ let mut aggr_exprs = Vec::with_capacity(agg.measures.len());
match agg.groupings.len() {
0 => {}
@@ -56,7 +56,7 @@ pub async fn from_aggregate_rel(
);
}
_ => {
- let mut grouping_sets = vec![];
+ let mut grouping_sets = Vec::with_capacity(agg.groupings.len());
for grouping in &agg.groupings {
let grouping_set = from_substrait_grouping(
consumer,
diff --git a/datafusion/substrait/src/logical_plan/consumer/rel/exchange_rel.rs b/datafusion/substrait/src/logical_plan/consumer/rel/exchange_rel.rs
index b275e523f5..56f0e18938 100644
--- a/datafusion/substrait/src/logical_plan/consumer/rel/exchange_rel.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/rel/exchange_rel.rs
@@ -39,7 +39,7 @@ pub async fn from_exchange_rel(
// ref: https://substrait.io/relations/physical_relations/#exchange-types
let partitioning_scheme = match exchange_kind {
ExchangeKind::ScatterByFields(scatter_fields) => {
- let mut partition_columns = vec![];
+ let mut partition_columns = Vec::with_capacity(scatter_fields.fields.len());
let input_schema = input.schema();
for field_ref in &scatter_fields.fields {
let column =
diff --git a/datafusion/substrait/src/logical_plan/consumer/rel/mod.rs b/datafusion/substrait/src/logical_plan/consumer/rel/mod.rs
index f95c2e7dd2..e9b5190606 100644
--- a/datafusion/substrait/src/logical_plan/consumer/rel/mod.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/rel/mod.rs
@@ -100,7 +100,7 @@ fn apply_emit_kind(
// converting a single call of the random() function into multiple calls due to
// duplicate fields in the output_mapping.
LogicalPlan::Projection(proj) if !contains_volatile_expr(&proj) => {
- let mut exprs: Vec<Expr> = vec![];
+ let mut exprs: Vec<Expr> = Vec::with_capacity(output_mapping.len());
for field in output_mapping {
let expr = proj.expr
.get(field as usize)
@@ -118,7 +118,7 @@ fn apply_emit_kind(
_ => {
let input_schema = plan.schema();
- let mut exprs: Vec<Expr> = vec![];
+ let mut exprs: Vec<Expr> = Vec::with_capacity(output_mapping.len());
for index in output_mapping.into_iter() {
let column = Expr::Column(Column::from(
input_schema.qualified_field(index as usize),
diff --git a/datafusion/substrait/src/logical_plan/consumer/rel/project_rel.rs b/datafusion/substrait/src/logical_plan/consumer/rel/project_rel.rs
index 211ccdd848..74c3b497c0 100644
--- a/datafusion/substrait/src/logical_plan/consumer/rel/project_rel.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/rel/project_rel.rs
@@ -53,7 +53,7 @@ pub async fn from_project_rel(
// Substrait Projects it adds an output mapping that excludes all input columns
// leaving only explicit expressions.
- let mut explicit_exprs: Vec<Expr> = vec![];
+ let mut explicit_exprs: Vec<Expr> = Vec::with_capacity(p.expressions.len());
// For WindowFunctions, we need to wrap them in a Window relation. If there are duplicates,
// we can do the window'ing only once, then the project will duplicate the result.
// Order here doesn't matter since LPB::window_plan sorts the expressions.
@@ -61,9 +61,7 @@ pub async fn from_project_rel(
// Expr contains Arc with interior mutability but is intentionally used as hash key
let mut window_exprs: HashSet<Expr> = HashSet::new();
for expr in &p.expressions {
- let e = consumer
- .consume_expression(expr, input.clone().schema())
- .await?;
+ let e = consumer.consume_expression(expr, input.schema()).await?;
// The project's builder uses columnize_expr(..) to transform
// nested window expressions into column references.
window_exprs.extend(find_window_exprs([&e]));
@@ -107,7 +105,8 @@ pub async fn from_project_rel(
input
};
- let mut final_exprs: Vec<Expr> = vec![];
+ let mut final_exprs: Vec<Expr> =
+ Vec::with_capacity(original_schema.fields().len() + explicit_exprs.len());
for index in 0..original_schema.fields().len() {
let e = Expr::Column(Column::from(original_schema.qualified_field(index)));
final_exprs.push(name_tracker.get_uniquely_named_expr(e)?);
diff --git a/datafusion/substrait/src/logical_plan/consumer/rel/read_rel.rs b/datafusion/substrait/src/logical_plan/consumer/rel/read_rel.rs
index f871fd4f4a..9490c10d73 100644
--- a/datafusion/substrait/src/logical_plan/consumer/rel/read_rel.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/rel/read_rel.rs
@@ -139,7 +139,7 @@ pub async fn from_read_rel(
}));
}
- let mut rows = vec![];
+ let mut rows = Vec::with_capacity(vt.expressions.len());
for row in &vt.expressions {
if row.fields.len() != substrait_schema.fields().len() {
return substrait_err!(
@@ -149,7 +149,7 @@ pub async fn from_read_rel(
);
}
- let mut row_exprs = vec![];
+ let mut row_exprs = Vec::with_capacity(row.fields.len());
let mut name_idx = 0;
for expression in &row.fields {
// Top-level names are provided through schema
diff --git a/datafusion/substrait/src/logical_plan/consumer/types.rs b/datafusion/substrait/src/logical_plan/consumer/types.rs
index e588734782..cac64f0f5d 100644
--- a/datafusion/substrait/src/logical_plan/consumer/types.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/types.rs
@@ -374,7 +374,7 @@ fn from_substrait_struct_type(
dfs_names: &[String],
name_idx: &mut usize,
) -> datafusion::common::Result<Fields> {
- let mut fields = vec![];
+ let mut fields = Vec::with_capacity(s.types.len());
for (i, f) in s.types.iter().enumerate() {
let name = next_struct_field_name(i, dfs_names, name_idx)?;
let field = substrait_type_to_field(consumer, f, dfs_names, name_idx)?
diff --git a/datafusion/substrait/src/logical_plan/consumer/utils.rs b/datafusion/substrait/src/logical_plan/consumer/utils.rs
index 04fafdc691..3354374100 100644
--- a/datafusion/substrait/src/logical_plan/consumer/utils.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/utils.rs
@@ -511,7 +511,7 @@ pub async fn from_substrait_sorts(
substrait_sorts: &Vec<SortField>,
input_schema: &DFSchema,
) -> datafusion::common::Result<Vec<Sort>> {
- let mut sorts: Vec<Sort> = vec![];
+ let mut sorts: Vec<Sort> = Vec::with_capacity(substrait_sorts.len());
for s in substrait_sorts {
let expr = consumer
.consume_expression(s.expr.as_ref().unwrap(), input_schema)
diff --git a/datafusion/substrait/src/logical_plan/producer/expr/aggregate_function.rs b/datafusion/substrait/src/logical_plan/producer/expr/aggregate_function.rs
index d2e495c31d..ded5bc681f 100644
--- a/datafusion/substrait/src/logical_plan/producer/expr/aggregate_function.rs
+++ b/datafusion/substrait/src/logical_plan/producer/expr/aggregate_function.rs
@@ -47,7 +47,7 @@ pub fn from_aggregate_function(
.iter()
.map(|expr| to_substrait_sort_field(producer, expr, schema))
.collect::<datafusion::common::Result<Vec<_>>>()?;
- let mut arguments: Vec<FunctionArgument> = vec![];
+ let mut arguments: Vec<FunctionArgument> = Vec::with_capacity(args.len());
for arg in args {
arguments.push(FunctionArgument {
arg_type: Some(ArgType::Value(producer.handle_expr(arg, schema)?)),
diff --git a/datafusion/substrait/src/logical_plan/producer/expr/scalar_function.rs b/datafusion/substrait/src/logical_plan/producer/expr/scalar_function.rs
index 2f4fb5c1ce..827e27a868 100644
--- a/datafusion/substrait/src/logical_plan/producer/expr/scalar_function.rs
+++ b/datafusion/substrait/src/logical_plan/producer/expr/scalar_function.rs
@@ -142,7 +142,7 @@ fn from_function(
output_nullability: bool,
schema: &DFSchemaRef,
) -> datafusion::common::Result<Expression> {
- let mut arguments: Vec<FunctionArgument> = vec![];
+ let mut arguments: Vec<FunctionArgument> = Vec::with_capacity(args.len());
for arg in args {
arguments.push(FunctionArgument {
arg_type: Some(ArgType::Value(producer.handle_expr(arg, schema)?)),
diff --git a/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs b/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs
index 98c8569c74..a262842ae3 100644
--- a/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs
+++ b/datafusion/substrait/src/logical_plan/producer/expr/window_function.rs
@@ -62,7 +62,7 @@ pub fn from_window_function(
// function reference
let function_anchor = producer.register_function(fun.to_string());
// arguments
- let mut arguments: Vec<FunctionArgument> = vec![];
+ let mut arguments: Vec<FunctionArgument> = Vec::with_capacity(args.len());
for arg in args {
arguments.push(FunctionArgument {
arg_type: Some(ArgType::Value(producer.handle_expr(arg, schema)?)),
diff --git a/datafusion/substrait/src/logical_plan/producer/rel/aggregate_rel.rs b/datafusion/substrait/src/logical_plan/producer/rel/aggregate_rel.rs
index 7b6c113cce..6e1af747b6 100644
--- a/datafusion/substrait/src/logical_plan/producer/rel/aggregate_rel.rs
+++ b/datafusion/substrait/src/logical_plan/producer/rel/aggregate_rel.rs
@@ -178,8 +178,8 @@ pub fn parse_flat_grouping_exprs(
schema: &DFSchemaRef,
ref_group_exprs: &mut Vec<Expression>,
) -> datafusion::common::Result<Grouping> {
- let mut expression_references = vec![];
- let mut grouping_expressions = vec![];
+ let mut expression_references = Vec::with_capacity(exprs.len());
+ let mut grouping_expressions = Vec::with_capacity(exprs.len());
for e in exprs {
let rex = producer.handle_expr(e, schema)?;
diff --git a/datafusion/substrait/src/physical_plan/producer.rs b/datafusion/substrait/src/physical_plan/producer.rs
index 21282b9e8b..3cc2abb9c1 100644
--- a/datafusion/substrait/src/physical_plan/producer.rs
+++ b/datafusion/substrait/src/physical_plan/producer.rs
@@ -70,8 +70,8 @@ pub fn to_substrait_rel(
}
}
- let mut names = vec![];
- let mut types = vec![];
+ let mut names = Vec::with_capacity(file_config.file_schema().fields.len());
+ let mut types = Vec::with_capacity(file_config.file_schema().fields.len());
for field in file_config.file_schema().fields.iter() {
let t = to_substrait_type(field.data_type(), field.is_nullable())?;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org