diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 63a7797e396ef..5b62574286612 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1661,11 +1661,7 @@ config_namespace! { /// pattern of the form `Filter(row_number() = 1)` over a `PARTITION BY` window into an /// `Aggregate(first_value(... ORDER BY ...) GROUP BY partition)`. /// This avoids buffering / fully sorting the input, which is especially beneficial on wide payloads. - /// Disabled by default because on nested/wide value types the aggregate only becomes memory - /// efficient once the columnar nested-type `GroupsAccumulator` fast - /// path is available; enabling it without that support can route wide - /// payloads onto the slow per-group accumulator path. - pub enable_row_number_to_aggregate: bool, default = false + pub enable_row_number_to_aggregate: bool, default = true /// When set to true, the optimizer will push TopK (Sort with fetch) /// below hash repartition when the partition key is a prefix of the diff --git a/datafusion/optimizer/src/replace_filter_top1.rs b/datafusion/optimizer/src/replace_filter_top1.rs index 4505bee753c6b..b91d831a217b9 100644 --- a/datafusion/optimizer/src/replace_filter_top1.rs +++ b/datafusion/optimizer/src/replace_filter_top1.rs @@ -49,7 +49,7 @@ use datafusion_expr::{ExprFunctionExt, LogicalPlanBuilder, lit}; /// - the window function must be `row_number` /// - filter predicate must be "top-1" (rn = 1, <= 1, < 2) /// - window has a `PARTITION BY` clause -/// - gated behind the `optimizer.enable_row_number_to_aggregate` config option (off by default) +/// - gated behind the `optimizer.enable_row_number_to_aggregate` config option (on by default) /// - with duplicate `ORDER BY` keys, `row_number()` and this rewrite with `first_value` may pick different (but equally valid) tied rows - each path breaks ties deterministically, but no guarantee the choices will be the same #[derive(Default, Debug)] pub struct ReplaceFilterTop1 {} diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index c1aabc2946ea6..9e57497821cf0 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -310,7 +310,7 @@ datafusion.optimizer.enable_leaf_expression_pushdown true datafusion.optimizer.enable_physical_uncorrelated_scalar_subquery true datafusion.optimizer.enable_piecewise_merge_join false datafusion.optimizer.enable_round_robin_repartition true -datafusion.optimizer.enable_row_number_to_aggregate false +datafusion.optimizer.enable_row_number_to_aggregate true datafusion.optimizer.enable_sort_pushdown true datafusion.optimizer.enable_topk_aggregation true datafusion.optimizer.enable_topk_dynamic_filter_pushdown true @@ -479,7 +479,7 @@ datafusion.optimizer.enable_leaf_expression_pushdown true When set to true, the datafusion.optimizer.enable_physical_uncorrelated_scalar_subquery true When set to true, uncorrelated scalar subqueries are left in the logical plan and executed by `ScalarSubqueryExec` during physical execution. When set to false, all scalar subqueries (including uncorrelated ones) are rewritten to left joins by the `ScalarSubqueryToJoin` optimizer rule. Note disabling this option is not recommended. It restores pre behavior, which silently produces incorrect results for multi-row subqueries and does not support scalar subqueries in ORDER BY / JOIN ON / aggregate-function arguments. This option is intended as a temporary escape hatch for distributed execution frameworks and is planned to be removed in a future DataFusion release. datafusion.optimizer.enable_piecewise_merge_join false When set to true, piecewise merge join is enabled. PiecewiseMergeJoin is currently experimental. Physical planner will opt for PiecewiseMergeJoin when there is only one range filter. 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.enable_row_number_to_aggregate false When set to true, the optimizer will rewrite a "top-1 per group" pattern of the form `Filter(row_number() = 1)` over a `PARTITION BY` window into an `Aggregate(first_value(... ORDER BY ...) GROUP BY partition)`. This avoids buffering / fully sorting the input, which is especially beneficial on wide payloads. Disabled by default because on nested/wide value types the aggregate only becomes memory efficient once the columnar nested-type `GroupsAccumulator` fast path is available; enabling it without that support can route wide payloads onto the slow per-group accumulator path. +datafusion.optimizer.enable_row_number_to_aggregate true When set to true, the optimizer will rewrite a "top-1 per group" pattern of the form `Filter(row_number() = 1)` over a `PARTITION BY` window into an `Aggregate(first_value(... ORDER BY ...) GROUP BY partition)`. This avoids buffering / fully sorting the input, which is especially beneficial on wide payloads. datafusion.optimizer.enable_sort_pushdown true Enable sort pushdown optimization. When enabled, attempts to push sort requirements down to data sources that can natively handle them (e.g., by reversing file/row group read order). Returns **inexact ordering**: Sort operator is kept for correctness, but optimized input enables early termination for TopK queries (ORDER BY ... LIMIT N), providing significant speedup. Memory: No additional overhead (only changes read order). Future: Will add option to detect perfectly sorted data and eliminate Sort completely. Default: true datafusion.optimizer.enable_topk_aggregation true When set to true, the optimizer will attempt to perform limit operations during aggregations, if possible datafusion.optimizer.enable_topk_dynamic_filter_pushdown true When set to true, the optimizer will attempt to push down TopK dynamic filters into the file scan phase. diff --git a/datafusion/sqllogictest/test_files/row_number_to_aggregate.slt b/datafusion/sqllogictest/test_files/row_number_to_aggregate.slt index 1cbd866077776..06da26947b25f 100644 --- a/datafusion/sqllogictest/test_files/row_number_to_aggregate.slt +++ b/datafusion/sqllogictest/test_files/row_number_to_aggregate.slt @@ -14,7 +14,7 @@ CREATE TABLE t (p INT, o INT, payload VARCHAR) AS VALUES (2, 30, 'd'); # --------------------------------------------------------------------------- -# Rule disabled (the default): the plan keeps the Window + Filter shape. +# Rule disabled via config: the plan keeps the Window + Filter shape. # --------------------------------------------------------------------------- statement ok diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 2c3ca887d4a38..116ece08a088d 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -143,7 +143,7 @@ The following configuration settings are available: | datafusion.optimizer.enable_topk_aggregation | true | When set to true, the optimizer will attempt to perform limit operations during aggregations, if possible | | datafusion.optimizer.enable_window_limits | true | When set to true, the optimizer will attempt to push limit operations past window functions, if possible | | datafusion.optimizer.enable_window_topn | false | When set to true, the optimizer will replace Filter(rn<=K) → Window(ROW_NUMBER) → Sort patterns with a PartitionedTopKExec that maintains per-partition heaps, avoiding a full sort of the input. When the window partition key has low cardinality, enabling this optimization can improve performance. However, for high cardinality keys, it may cause regressions in both memory usage and runtime. | -| datafusion.optimizer.enable_row_number_to_aggregate | false | When set to true, the optimizer will rewrite a "top-1 per group" pattern of the form `Filter(row_number() = 1)` over a `PARTITION BY` window into an `Aggregate(first_value(... ORDER BY ...) GROUP BY partition)`. This avoids buffering / fully sorting the input, which is especially beneficial on wide payloads. Disabled by default because on nested/wide value types the aggregate only becomes memory efficient once the columnar nested-type `GroupsAccumulator` fast path is available; enabling it without that support can route wide payloads onto the slow per-group accumulator path. | +| datafusion.optimizer.enable_row_number_to_aggregate | true | When set to true, the optimizer will rewrite a "top-1 per group" pattern of the form `Filter(row_number() = 1)` over a `PARTITION BY` window into an `Aggregate(first_value(... ORDER BY ...) GROUP BY partition)`. This avoids buffering / fully sorting the input, which is especially beneficial on wide payloads. | | datafusion.optimizer.enable_topk_repartition | true | When set to true, the optimizer will push TopK (Sort with fetch) below hash repartition when the partition key is a prefix of the sort key, reducing data volume before the shuffle. | | datafusion.optimizer.enable_topk_dynamic_filter_pushdown | true | When set to true, the optimizer will attempt to push down TopK dynamic filters into the file scan phase. | | datafusion.optimizer.enable_physical_uncorrelated_scalar_subquery | true | When set to true, uncorrelated scalar subqueries are left in the logical plan and executed by `ScalarSubqueryExec` during physical execution. When set to false, all scalar subqueries (including uncorrelated ones) are rewritten to left joins by the `ScalarSubqueryToJoin` optimizer rule. Note disabling this option is not recommended. It restores pre behavior, which silently produces incorrect results for multi-row subqueries and does not support scalar subqueries in ORDER BY / JOIN ON / aggregate-function arguments. This option is intended as a temporary escape hatch for distributed execution frameworks and is planned to be removed in a future DataFusion release. |