This is an automated email from the ASF dual-hosted git repository.
yjshen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 6e8f91b41a feat: make threshold for using scalar update in aggregate
configurable (#6166)
6e8f91b41a is described below
commit 6e8f91b41a6ec6a2680357b95b2489d87af33571
Author: Yijie Shen <[email protected]>
AuthorDate: Mon May 1 22:14:07 2023 +0800
feat: make threshold for using scalar update in aggregate configurable
(#6166)
* feat: make threshold for using scalar update in aggregate configurable
* fix
* fix test
---
datafusion/common/src/config.rs | 20 ++++++
.../aggregates/bounded_aggregate_stream.rs | 7 +-
.../core/src/physical_plan/aggregates/mod.rs | 3 +
.../core/src/physical_plan/aggregates/row_hash.rs | 7 +-
.../test_files/information_schema.slt | 1 +
datafusion/execution/src/config.rs | 13 ++++
docs/source/user-guide/configs.md | 79 +++++++++++-----------
7 files changed, 89 insertions(+), 41 deletions(-)
diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index 35afabf307..a89bc2f8fd 100644
--- a/datafusion/common/src/config.rs
+++ b/datafusion/common/src/config.rs
@@ -223,6 +223,9 @@ config_namespace! {
/// Parquet options
pub parquet: ParquetOptions, default = Default::default()
+
+ /// Aggregate options
+ pub aggregate: AggregateOptions, default = Default::default()
}
}
@@ -260,6 +263,23 @@ config_namespace! {
}
}
+config_namespace! {
+ /// Options related to aggregate execution
+ pub struct AggregateOptions {
+ /// Specifies the threshold for using `ScalarValue`s to update
+ /// accumulators during high-cardinality aggregations for each input
batch.
+ ///
+ /// The aggregation is considered high-cardinality if the number of
affected groups
+ /// is greater than or equal to `batch_size / scalar_update_factor`.
In such cases,
+ /// `ScalarValue`s are utilized for updating accumulators, rather than
the default
+ /// batch-slice approach. This can lead to performance improvements.
+ ///
+ /// By adjusting the `scalar_update_factor`, you can balance the
trade-off between
+ /// more efficient accumulator updates and the number of groups
affected.
+ pub scalar_update_factor: usize, default = 10
+ }
+}
+
config_namespace! {
/// Options related to query optimization
pub struct OptimizerOptions {
diff --git
a/datafusion/core/src/physical_plan/aggregates/bounded_aggregate_stream.rs
b/datafusion/core/src/physical_plan/aggregates/bounded_aggregate_stream.rs
index a2ee627eec..41eca116bd 100644
--- a/datafusion/core/src/physical_plan/aggregates/bounded_aggregate_stream.rs
+++ b/datafusion/core/src/physical_plan/aggregates/bounded_aggregate_stream.rs
@@ -103,6 +103,9 @@ pub(crate) struct BoundedAggregateStream {
random_state: RandomState,
/// size to be used for resulting RecordBatches
batch_size: usize,
+ /// threshold for using `ScalarValue`s to update
+ /// accumulators during high-cardinality aggregations for each input batch.
+ scalar_update_factor: usize,
/// if the result is chunked into batches,
/// last offset is preserved for continuation.
row_group_skip_position: usize,
@@ -126,6 +129,7 @@ impl BoundedAggregateStream {
input: SendableRecordBatchStream,
baseline_metrics: BaselineMetrics,
batch_size: usize,
+ scalar_update_factor: usize,
context: Arc<TaskContext>,
partition: usize,
// Stores algorithm mode and output ordering
@@ -228,6 +232,7 @@ impl BoundedAggregateStream {
baseline_metrics,
random_state: Default::default(),
batch_size,
+ scalar_update_factor,
row_group_skip_position: 0,
indices: [normal_agg_indices, row_agg_indices],
is_end: false,
@@ -747,7 +752,7 @@ impl BoundedAggregateStream {
if matches!(self.mode, AggregateMode::Partial |
AggregateMode::Single)
&& normal_aggr_input_values.is_empty()
&& normal_filter_values.is_empty()
- && groups_with_rows.len() >= batch.num_rows() / 10
+ && groups_with_rows.len() >= batch.num_rows() /
self.scalar_update_factor
{
self.update_accumulators_using_scalar(
&groups_with_rows,
diff --git a/datafusion/core/src/physical_plan/aggregates/mod.rs
b/datafusion/core/src/physical_plan/aggregates/mod.rs
index e1f0bfb752..5c77f3b74a 100644
--- a/datafusion/core/src/physical_plan/aggregates/mod.rs
+++ b/datafusion/core/src/physical_plan/aggregates/mod.rs
@@ -424,6 +424,7 @@ impl AggregateExec {
context: Arc<TaskContext>,
) -> Result<StreamType> {
let batch_size = context.session_config().batch_size();
+ let scalar_update_factor =
context.session_config().agg_scalar_update_factor();
let input = self.input.execute(partition, Arc::clone(&context))?;
let baseline_metrics = BaselineMetrics::new(&self.metrics, partition);
@@ -448,6 +449,7 @@ impl AggregateExec {
input,
baseline_metrics,
batch_size,
+ scalar_update_factor,
context,
partition,
aggregation_ordering.clone(),
@@ -463,6 +465,7 @@ impl AggregateExec {
input,
baseline_metrics,
batch_size,
+ scalar_update_factor,
context,
partition,
)?,
diff --git a/datafusion/core/src/physical_plan/aggregates/row_hash.rs
b/datafusion/core/src/physical_plan/aggregates/row_hash.rs
index 3f42a0171b..0b98513cce 100644
--- a/datafusion/core/src/physical_plan/aggregates/row_hash.rs
+++ b/datafusion/core/src/physical_plan/aggregates/row_hash.rs
@@ -98,6 +98,9 @@ pub(crate) struct GroupedHashAggregateStream {
random_state: RandomState,
/// size to be used for resulting RecordBatches
batch_size: usize,
+ /// threshold for using `ScalarValue`s to update
+ /// accumulators during high-cardinality aggregations for each input batch.
+ scalar_update_factor: usize,
/// if the result is chunked into batches,
/// last offset is preserved for continuation.
row_group_skip_position: usize,
@@ -119,6 +122,7 @@ impl GroupedHashAggregateStream {
input: SendableRecordBatchStream,
baseline_metrics: BaselineMetrics,
batch_size: usize,
+ scalar_update_factor: usize,
context: Arc<TaskContext>,
partition: usize,
) -> Result<Self> {
@@ -219,6 +223,7 @@ impl GroupedHashAggregateStream {
baseline_metrics,
random_state: Default::default(),
batch_size,
+ scalar_update_factor,
row_group_skip_position: 0,
indices: [normal_agg_indices, row_agg_indices],
})
@@ -555,7 +560,7 @@ impl GroupedHashAggregateStream {
if matches!(self.mode, AggregateMode::Partial |
AggregateMode::Single)
&& normal_aggr_input_values.is_empty()
&& normal_filter_values.is_empty()
- && groups_with_rows.len() >= batch.num_rows() / 10
+ && groups_with_rows.len() >= batch.num_rows() /
self.scalar_update_factor
{
self.update_accumulators_using_scalar(
&groups_with_rows,
diff --git
a/datafusion/core/tests/sqllogictests/test_files/information_schema.slt
b/datafusion/core/tests/sqllogictests/test_files/information_schema.slt
index 3ea42583c8..290f7182c8 100644
--- a/datafusion/core/tests/sqllogictests/test_files/information_schema.slt
+++ b/datafusion/core/tests/sqllogictests/test_files/information_schema.slt
@@ -136,6 +136,7 @@ datafusion.catalog.format NULL
datafusion.catalog.has_header false
datafusion.catalog.information_schema true
datafusion.catalog.location NULL
+datafusion.execution.aggregate.scalar_update_factor 10
datafusion.execution.batch_size 8192
datafusion.execution.coalesce_batches true
datafusion.execution.collect_statistics false
diff --git a/datafusion/execution/src/config.rs
b/datafusion/execution/src/config.rs
index 730a767826..2867e7bd7c 100644
--- a/datafusion/execution/src/config.rs
+++ b/datafusion/execution/src/config.rs
@@ -237,6 +237,19 @@ impl SessionConfig {
self.options.execution.batch_size
}
+ /// Get the currently configured scalar_update_factor for aggregate
+ pub fn agg_scalar_update_factor(&self) -> usize {
+ self.options.execution.aggregate.scalar_update_factor
+ }
+
+ /// Customize scalar_update_factor for aggregate
+ pub fn with_agg_scalar_update_factor(mut self, n: usize) -> Self {
+ // scalar update factor must be greater than zero
+ assert!(n > 0);
+ self.options.execution.aggregate.scalar_update_factor = n;
+ self
+ }
+
/// Convert configuration options to name-value pairs with values
/// converted to strings.
///
diff --git a/docs/source/user-guide/configs.md
b/docs/source/user-guide/configs.md
index 4b754bcd8a..decab6a719 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -35,42 +35,43 @@ Values are parsed according to the [same rules used in
casts from Utf8](https://
If the value in the environment variable cannot be cast to the type of the
configuration option, the default value will be used instead and a warning
emitted.
Environment variables are read during `SessionConfig` initialisation so they
must be set beforehand and will not affect running sessions.
-| key | default |
description
[...]
-| ---------------------------------------------------------- | ---------- |
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| datafusion.catalog.create_default_catalog_and_schema | true |
Whether the default catalog and schema should be created automatically.
[...]
-| datafusion.catalog.default_catalog | datafusion |
The default catalog name - this impacts what SQL queries use if not specified
[...]
-| datafusion.catalog.default_schema | public |
The default schema name - this impacts what SQL queries use if not specified
[...]
-| datafusion.catalog.information_schema | false |
Should DataFusion provide access to `information_schema` virtual tables for
displaying schema information
[...]
-| datafusion.catalog.location | NULL |
Location scanned to load tables for `default` schema
[...]
-| datafusion.catalog.format | NULL |
Type of `TableProvider` to use when loading `default` schema
[...]
-| datafusion.catalog.has_header | false | If
the file has a header
[...]
-| datafusion.execution.batch_size | 8192 |
Default batch size while creating new batches, it's especially useful for
buffer-in-memory batches since creating tiny batches would result in too much
metadata memory consumption
[...]
-| datafusion.execution.coalesce_batches | true |
When set to true, record batches will be examined between each operator and
small batches will be coalesced into larger batches. This is helpful when there
are highly selective filters or joins that could produce tiny output batches.
The target batch size is determined by the configuration setting
[...]
-| datafusion.execution.collect_statistics | false |
Should DataFusion collect statistics after listing files
[...]
-| datafusion.execution.target_partitions | 0 |
Number of partitions for query execution. Increasing partitions can increase
concurrency. Defaults to the number of CPU cores on the system
[...]
-| datafusion.execution.time_zone | +00:00 |
The default time zone Some functions, e.g. `EXTRACT(HOUR from SOME_TIME)`,
shift the underlying datetime according to this time zone, and then extract the
hour
[...]
-| datafusion.execution.parquet.enable_page_index | false | If
true, uses parquet data page level metadata (Page Index) statistics to reduce
the number of rows decoded.
[...]
-| datafusion.execution.parquet.pruning | true | If
true, the parquet reader attempts to skip entire row groups based on the
predicate in the query and the metadata (min/max values) stored in the parquet
file
[...]
-| datafusion.execution.parquet.skip_metadata | true | If
true, the parquet reader skip the optional embedded metadata that may be in the
file Schema. This setting can help avoid schema conflicts when querying
multiple parquet files with schemas containing compatible types but different
metadata
[...]
-| datafusion.execution.parquet.metadata_size_hint | NULL | If
specified, the parquet reader will try and fetch the last `size_hint` bytes of
the parquet file optimistically. If not specified, two reads are required: One
read to fetch the 8-byte parquet footer and another to fetch the metadata
length encoded in the footer
[...]
-| datafusion.execution.parquet.pushdown_filters | false | If
true, filter expressions are be applied during the parquet decoding operation
to reduce the number of rows decoded
[...]
-| datafusion.execution.parquet.reorder_filters | false | If
true, filter expressions evaluated during the parquet decoding operation will
be reordered heuristically to minimize the cost of evaluation. If false, the
filters are applied in the same order as written in the query
[...]
-| datafusion.optimizer.enable_round_robin_repartition | true |
When set to true, the physical plan optimizer will try to add round robin
repartitioning to increase parallelism to leverage more CPU cores
[...]
-| datafusion.optimizer.filter_null_join_keys | false |
When set to true, the optimizer will insert filters before a join between a
nullable and non-nullable column to filter out nulls on the nullable side. This
filter can add additional overhead when the file format does not fully support
predicate push down.
[...]
-| datafusion.optimizer.repartition_aggregations | true |
Should DataFusion repartition data using the aggregate keys to execute
aggregates in parallel using the provided `target_partitions` level
[...]
-| datafusion.optimizer.repartition_file_min_size | 10485760 |
Minimum total files size in bytes to perform file scan repartitioning.
[...]
-| datafusion.optimizer.repartition_joins | true |
Should DataFusion repartition data using the join keys to execute joins in
parallel using the provided `target_partitions` level
[...]
-| datafusion.optimizer.allow_symmetric_joins_without_pruning | true |
Should DataFusion allow symmetric hash joins for unbounded data sources even
when its inputs do not have any ordering or filtering If the flag is not
enabled, the SymmetricHashJoin operator will be unable to prune its internal
buffers, resulting in certain join types - such as Full, Left, LeftAnti,
LeftSemi, Right, RightAnti, and RightSemi - being produced only at the end of
the execution. This is not typical in [...]
-| datafusion.optimizer.repartition_file_scans | true |
When set to true, file groups will be repartitioned to achieve maximum
parallelism. Currently supported only for Parquet format in which case multiple
row groups from the same file may be read concurrently. If false then each row
group is read serially, though different files may be read in parallel.
[...]
-| datafusion.optimizer.repartition_windows | true |
Should DataFusion repartition data using the partitions keys to execute window
functions in parallel using the provided `target_partitions` level
[...]
-| datafusion.optimizer.repartition_sorts | true |
Should DataFusion execute sorts in a per-partition fashion and merge afterwards
instead of coalescing first and sorting globally. With this flag is enabled,
plans in the form below `text "SortExec: [a@0 ASC]", " CoalescePartitionsExec",
" RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", `
would turn into the plan below which performs better in multithreaded
environments `text "SortPreserving [...]
-| datafusion.optimizer.skip_failed_rules | true |
When set to true, the logical plan optimizer will produce warning messages if
any optimization rules produce errors and then proceed to the next rule. When
set to false, any rules that produce errors will cause the query to fail
[...]
-| datafusion.optimizer.max_passes | 3 |
Number of times that the optimizer will attempt to optimize the plan
[...]
-| datafusion.optimizer.top_down_join_key_reordering | true |
When set to true, the physical plan optimizer will run a top down process to
reorder the join keys
[...]
-| datafusion.optimizer.prefer_hash_join | true |
When set to true, the physical plan optimizer will prefer HashJoin over
SortMergeJoin. HashJoin can work more efficiently than SortMergeJoin but
consumes more memory
[...]
-| datafusion.optimizer.hash_join_single_partition_threshold | 1048576 |
The maximum estimated size in bytes for one input side of a HashJoin will be
collected into a single partition
[...]
-| datafusion.explain.logical_plan_only | false |
When set to true, the explain statement will only print logical plans
[...]
-| datafusion.explain.physical_plan_only | false |
When set to true, the explain statement will only print physical plans
[...]
-| datafusion.sql_parser.parse_float_as_decimal | false |
When set to true, SQL parser will parse float as decimal type
[...]
-| datafusion.sql_parser.enable_ident_normalization | true |
When set to true, SQL parser will normalize ident (convert ident to lowercase
when not quoted)
[...]
-| datafusion.sql_parser.dialect | generic |
Configure the SQL dialect used by DataFusion's parser; supported values
include: Generic, MySQL, PostgreSQL, Hive, SQLite, Snowflake, Redshift, MsSQL,
ClickHouse, BigQuery, and Ansi.
[...]
+| key | default |
description
[...]
+| ---------------------------------------------------------- | ---------- |
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| datafusion.catalog.create_default_catalog_and_schema | true |
Whether the default catalog and schema should be created automatically.
[...]
+| datafusion.catalog.default_catalog | datafusion |
The default catalog name - this impacts what SQL queries use if not specified
[...]
+| datafusion.catalog.default_schema | public |
The default schema name - this impacts what SQL queries use if not specified
[...]
+| datafusion.catalog.information_schema | false |
Should DataFusion provide access to `information_schema` virtual tables for
displaying schema information
[...]
+| datafusion.catalog.location | NULL |
Location scanned to load tables for `default` schema
[...]
+| datafusion.catalog.format | NULL |
Type of `TableProvider` to use when loading `default` schema
[...]
+| datafusion.catalog.has_header | false | If
the file has a header
[...]
+| datafusion.execution.batch_size | 8192 |
Default batch size while creating new batches, it's especially useful for
buffer-in-memory batches since creating tiny batches would result in too much
metadata memory consumption
[...]
+| datafusion.execution.coalesce_batches | true |
When set to true, record batches will be examined between each operator and
small batches will be coalesced into larger batches. This is helpful when there
are highly selective filters or joins that could produce tiny output batches.
The target batch size is determined by the configuration setting
[...]
+| datafusion.execution.collect_statistics | false |
Should DataFusion collect statistics after listing files
[...]
+| datafusion.execution.target_partitions | 0 |
Number of partitions for query execution. Increasing partitions can increase
concurrency. Defaults to the number of CPU cores on the system
[...]
+| datafusion.execution.time_zone | +00:00 |
The default time zone Some functions, e.g. `EXTRACT(HOUR from SOME_TIME)`,
shift the underlying datetime according to this time zone, and then extract the
hour
[...]
+| datafusion.execution.parquet.enable_page_index | false | If
true, uses parquet data page level metadata (Page Index) statistics to reduce
the number of rows decoded.
[...]
+| datafusion.execution.parquet.pruning | true | If
true, the parquet reader attempts to skip entire row groups based on the
predicate in the query and the metadata (min/max values) stored in the parquet
file
[...]
+| datafusion.execution.parquet.skip_metadata | true | If
true, the parquet reader skip the optional embedded metadata that may be in the
file Schema. This setting can help avoid schema conflicts when querying
multiple parquet files with schemas containing compatible types but different
metadata
[...]
+| datafusion.execution.parquet.metadata_size_hint | NULL | If
specified, the parquet reader will try and fetch the last `size_hint` bytes of
the parquet file optimistically. If not specified, two reads are required: One
read to fetch the 8-byte parquet footer and another to fetch the metadata
length encoded in the footer
[...]
+| datafusion.execution.parquet.pushdown_filters | false | If
true, filter expressions are be applied during the parquet decoding operation
to reduce the number of rows decoded
[...]
+| datafusion.execution.parquet.reorder_filters | false | If
true, filter expressions evaluated during the parquet decoding operation will
be reordered heuristically to minimize the cost of evaluation. If false, the
filters are applied in the same order as written in the query
[...]
+| datafusion.execution.aggregate.scalar_update_factor | 10 |
Specifies the threshold for using `ScalarValue`s to update accumulators during
high-cardinality aggregations for each input batch. The aggregation is
considered high-cardinality if the number of affected groups is greater than or
equal to `batch_size / scalar_update_factor`. In such cases, `ScalarValue`s are
utilized for updating accumulators, rather than the default batch-slice
approach. This can lead to perform [...]
+| datafusion.optimizer.enable_round_robin_repartition | true |
When set to true, the physical plan optimizer will try to add round robin
repartitioning to increase parallelism to leverage more CPU cores
[...]
+| datafusion.optimizer.filter_null_join_keys | false |
When set to true, the optimizer will insert filters before a join between a
nullable and non-nullable column to filter out nulls on the nullable side. This
filter can add additional overhead when the file format does not fully support
predicate push down.
[...]
+| datafusion.optimizer.repartition_aggregations | true |
Should DataFusion repartition data using the aggregate keys to execute
aggregates in parallel using the provided `target_partitions` level
[...]
+| datafusion.optimizer.repartition_file_min_size | 10485760 |
Minimum total files size in bytes to perform file scan repartitioning.
[...]
+| datafusion.optimizer.repartition_joins | true |
Should DataFusion repartition data using the join keys to execute joins in
parallel using the provided `target_partitions` level
[...]
+| datafusion.optimizer.allow_symmetric_joins_without_pruning | true |
Should DataFusion allow symmetric hash joins for unbounded data sources even
when its inputs do not have any ordering or filtering If the flag is not
enabled, the SymmetricHashJoin operator will be unable to prune its internal
buffers, resulting in certain join types - such as Full, Left, LeftAnti,
LeftSemi, Right, RightAnti, and RightSemi - being produced only at the end of
the execution. This is not typical in [...]
+| datafusion.optimizer.repartition_file_scans | true |
When set to true, file groups will be repartitioned to achieve maximum
parallelism. Currently supported only for Parquet format in which case multiple
row groups from the same file may be read concurrently. If false then each row
group is read serially, though different files may be read in parallel.
[...]
+| datafusion.optimizer.repartition_windows | true |
Should DataFusion repartition data using the partitions keys to execute window
functions in parallel using the provided `target_partitions` level
[...]
+| datafusion.optimizer.repartition_sorts | true |
Should DataFusion execute sorts in a per-partition fashion and merge afterwards
instead of coalescing first and sorting globally. With this flag is enabled,
plans in the form below `text "SortExec: [a@0 ASC]", " CoalescePartitionsExec",
" RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", `
would turn into the plan below which performs better in multithreaded
environments `text "SortPreserving [...]
+| datafusion.optimizer.skip_failed_rules | true |
When set to true, the logical plan optimizer will produce warning messages if
any optimization rules produce errors and then proceed to the next rule. When
set to false, any rules that produce errors will cause the query to fail
[...]
+| datafusion.optimizer.max_passes | 3 |
Number of times that the optimizer will attempt to optimize the plan
[...]
+| datafusion.optimizer.top_down_join_key_reordering | true |
When set to true, the physical plan optimizer will run a top down process to
reorder the join keys
[...]
+| datafusion.optimizer.prefer_hash_join | true |
When set to true, the physical plan optimizer will prefer HashJoin over
SortMergeJoin. HashJoin can work more efficiently than SortMergeJoin but
consumes more memory
[...]
+| datafusion.optimizer.hash_join_single_partition_threshold | 1048576 |
The maximum estimated size in bytes for one input side of a HashJoin will be
collected into a single partition
[...]
+| datafusion.explain.logical_plan_only | false |
When set to true, the explain statement will only print logical plans
[...]
+| datafusion.explain.physical_plan_only | false |
When set to true, the explain statement will only print physical plans
[...]
+| datafusion.sql_parser.parse_float_as_decimal | false |
When set to true, SQL parser will parse float as decimal type
[...]
+| datafusion.sql_parser.enable_ident_normalization | true |
When set to true, SQL parser will normalize ident (convert ident to lowercase
when not quoted)
[...]
+| datafusion.sql_parser.dialect | generic |
Configure the SQL dialect used by DataFusion's parser; supported values
include: Generic, MySQL, PostgreSQL, Hive, SQLite, Snowflake, Redshift, MsSQL,
ClickHouse, BigQuery, and Ansi.
[...]
