adriangb commented on code in PR #25655:
URL: https://github.com/apache/datafusion/pull/25655#discussion_r4167783055
##########
datafusion/common/src/functional_dependencies.rs:
##########
@@ -458,71 +458,81 @@ pub fn aggregate_functional_dependencies(
aggr_schema: &DFSchema,
) -> FunctionalDependencies {
let mut aggregate_func_dependencies = vec![];
- let aggr_input_fields = aggr_input_schema.field_names();
let aggr_fields = aggr_schema.fields();
// Association covers the whole table:
let target_indices = (0..aggr_schema.fields().len()).collect::<Vec<_>>();
// Get functional dependencies of the schema:
let func_dependencies = aggr_input_schema.functional_dependencies();
- for FunctionalDependence {
- source_indices,
- nullable,
- null_equality,
- mode,
- ..
- } in &func_dependencies.deps
- {
- // Keep source indices in a `HashSet` to prevent duplicate entries:
- let mut new_source_indices = vec![];
- let mut new_source_field_names = vec![];
- let source_field_names = source_indices
- .iter()
- .map(|&idx| &aggr_input_fields[idx])
- .collect::<Vec<_>>();
-
- for (idx, group_by_expr_name) in
group_by_expr_names.iter().enumerate() {
- // When one of the input determinant expressions matches with
- // the GROUP BY expression, add the index of the GROUP BY
- // expression as a new determinant key:
- if source_field_names.contains(&group_by_expr_name) {
- new_source_indices.push(idx);
- new_source_field_names.push(group_by_expr_name.clone());
- }
- }
+ // If the input carries no functional dependencies, the loop below can
+ // never turn one into an aggregate dependency (it only re-expresses
+ // dependencies that already exist on the input), so skip building the
+ // input field names and resolving target indices for it entirely. The
+ // GROUP BY-key dependency added after this block does not depend on the
+ // input's functional dependencies, so it still runs unconditionally.
Review Comment:
Small nit: can we make this comment shorter?
```suggestion
// The loop below only re-expresses input dependencies. Skip it when the
// input has none. The GROUP BY-key dependency below always runs.
```
##########
datafusion/common/src/functional_dependencies.rs:
##########
@@ -458,71 +458,81 @@ pub fn aggregate_functional_dependencies(
aggr_schema: &DFSchema,
) -> FunctionalDependencies {
let mut aggregate_func_dependencies = vec![];
- let aggr_input_fields = aggr_input_schema.field_names();
let aggr_fields = aggr_schema.fields();
// Association covers the whole table:
let target_indices = (0..aggr_schema.fields().len()).collect::<Vec<_>>();
// Get functional dependencies of the schema:
let func_dependencies = aggr_input_schema.functional_dependencies();
- for FunctionalDependence {
- source_indices,
- nullable,
- null_equality,
- mode,
- ..
- } in &func_dependencies.deps
- {
- // Keep source indices in a `HashSet` to prevent duplicate entries:
- let mut new_source_indices = vec![];
- let mut new_source_field_names = vec![];
- let source_field_names = source_indices
- .iter()
- .map(|&idx| &aggr_input_fields[idx])
- .collect::<Vec<_>>();
-
- for (idx, group_by_expr_name) in
group_by_expr_names.iter().enumerate() {
- // When one of the input determinant expressions matches with
- // the GROUP BY expression, add the index of the GROUP BY
- // expression as a new determinant key:
- if source_field_names.contains(&group_by_expr_name) {
- new_source_indices.push(idx);
- new_source_field_names.push(group_by_expr_name.clone());
- }
- }
+ // If the input carries no functional dependencies, the loop below can
+ // never turn one into an aggregate dependency (it only re-expresses
+ // dependencies that already exist on the input), so skip building the
+ // input field names and resolving target indices for it entirely. The
+ // GROUP BY-key dependency added after this block does not depend on the
+ // input's functional dependencies, so it still runs unconditionally.
+ if !func_dependencies.is_empty() {
+ let aggr_input_fields = aggr_input_schema.field_names();
+ // Loop-invariant: does not depend on the per-dependence loop
+ // variables, so compute it once instead of on every iteration.
Review Comment:
```suggestion
// Compute once: this does not change in the loop.
```
##########
datafusion/common/src/functional_dependencies.rs:
##########
@@ -458,71 +458,81 @@ pub fn aggregate_functional_dependencies(
aggr_schema: &DFSchema,
) -> FunctionalDependencies {
let mut aggregate_func_dependencies = vec![];
- let aggr_input_fields = aggr_input_schema.field_names();
let aggr_fields = aggr_schema.fields();
// Association covers the whole table:
let target_indices = (0..aggr_schema.fields().len()).collect::<Vec<_>>();
// Get functional dependencies of the schema:
let func_dependencies = aggr_input_schema.functional_dependencies();
- for FunctionalDependence {
- source_indices,
- nullable,
- null_equality,
- mode,
- ..
- } in &func_dependencies.deps
- {
- // Keep source indices in a `HashSet` to prevent duplicate entries:
- let mut new_source_indices = vec![];
- let mut new_source_field_names = vec![];
- let source_field_names = source_indices
- .iter()
- .map(|&idx| &aggr_input_fields[idx])
- .collect::<Vec<_>>();
-
- for (idx, group_by_expr_name) in
group_by_expr_names.iter().enumerate() {
- // When one of the input determinant expressions matches with
- // the GROUP BY expression, add the index of the GROUP BY
- // expression as a new determinant key:
- if source_field_names.contains(&group_by_expr_name) {
- new_source_indices.push(idx);
- new_source_field_names.push(group_by_expr_name.clone());
- }
- }
+ // If the input carries no functional dependencies, the loop below can
+ // never turn one into an aggregate dependency (it only re-expresses
+ // dependencies that already exist on the input), so skip building the
+ // input field names and resolving target indices for it entirely. The
+ // GROUP BY-key dependency added after this block does not depend on the
+ // input's functional dependencies, so it still runs unconditionally.
+ if !func_dependencies.is_empty() {
Review Comment:
The same "no dependencies, skip" check is now at 3 sites (here,
`add_group_by_exprs_from_dependencies`, and
`calc_func_dependencies_for_project`). Could
`get_target_functional_dependencies` also return `None` early, before it calls
`schema.field_names()`? Then future callers get the fast path too:
```rust
let dependencies = schema.functional_dependencies();
if dependencies.is_empty() {
return None;
}
```
##########
datafusion/common/src/functional_dependencies.rs:
##########
@@ -458,71 +458,81 @@ pub fn aggregate_functional_dependencies(
aggr_schema: &DFSchema,
) -> FunctionalDependencies {
let mut aggregate_func_dependencies = vec![];
- let aggr_input_fields = aggr_input_schema.field_names();
let aggr_fields = aggr_schema.fields();
// Association covers the whole table:
let target_indices = (0..aggr_schema.fields().len()).collect::<Vec<_>>();
// Get functional dependencies of the schema:
let func_dependencies = aggr_input_schema.functional_dependencies();
- for FunctionalDependence {
- source_indices,
- nullable,
- null_equality,
- mode,
- ..
- } in &func_dependencies.deps
- {
- // Keep source indices in a `HashSet` to prevent duplicate entries:
- let mut new_source_indices = vec![];
- let mut new_source_field_names = vec![];
- let source_field_names = source_indices
- .iter()
- .map(|&idx| &aggr_input_fields[idx])
- .collect::<Vec<_>>();
-
- for (idx, group_by_expr_name) in
group_by_expr_names.iter().enumerate() {
- // When one of the input determinant expressions matches with
- // the GROUP BY expression, add the index of the GROUP BY
- // expression as a new determinant key:
- if source_field_names.contains(&group_by_expr_name) {
- new_source_indices.push(idx);
- new_source_field_names.push(group_by_expr_name.clone());
- }
- }
+ // If the input carries no functional dependencies, the loop below can
+ // never turn one into an aggregate dependency (it only re-expresses
+ // dependencies that already exist on the input), so skip building the
+ // input field names and resolving target indices for it entirely. The
+ // GROUP BY-key dependency added after this block does not depend on the
+ // input's functional dependencies, so it still runs unconditionally.
+ if !func_dependencies.is_empty() {
+ let aggr_input_fields = aggr_input_schema.field_names();
+ // Loop-invariant: does not depend on the per-dependence loop
+ // variables, so compute it once instead of on every iteration.
let existing_target_indices =
get_target_functional_dependencies(aggr_input_schema,
group_by_expr_names);
- let new_target_indices = get_target_functional_dependencies(
- aggr_input_schema,
- &new_source_field_names,
- );
- let mode = if existing_target_indices == new_target_indices
- && new_target_indices.is_some()
+ for FunctionalDependence {
+ source_indices,
+ nullable,
+ null_equality,
+ mode,
+ ..
+ } in &func_dependencies.deps
{
- // If dependency covers all GROUP BY expressions, mode will be
`Single`:
- Dependency::Single
- } else {
- // Otherwise, existing mode is preserved:
- *mode
- };
- // All of the composite indices occur in the GROUP BY expression:
- if new_source_indices.len() == source_indices.len() {
- // GROUP BY treats NULLs as equal: a determinant covering the
- // complete grouping key gets at most one output row per NULL too.
- let output_null_equality =
- if new_source_indices.len() == group_by_expr_names.len() {
- NullEquality::NullEqualsNull
- } else {
- *null_equality
- };
- aggregate_func_dependencies.push(
- FunctionalDependence::new(
- new_source_indices,
- target_indices.clone(),
- *nullable,
- )
- .with_mode(mode)
- .with_null_equality(output_null_equality),
+ // Keep source indices in a `HashSet` to prevent duplicate entries:
Review Comment:
This comment was already wrong on `main`, but since the line moves here: the
code uses a `Vec`, not a `HashSet`.
```suggestion
// Indices into the GROUP BY list for this determinant:
```
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -4381,7 +4381,29 @@ fn calc_func_dependencies_for_project(
// Sentinel for projection outputs that do not map back to any input field.
const COMPUTED_EXPR_INDEX: usize = usize::MAX;
+ let input_func_dependencies = input.schema().functional_dependencies();
+ // Projecting an empty set of dependencies always yields an empty set, so
+ // skip resolving projection expressions against the input fields. This is
+ // the common case because table sources carry no constraints by default.
+ if input_func_dependencies.is_empty() {
Review Comment:
Question: with this early return, the `Expr::Wildcard` branch no longer
calls `exprlist_to_fields(...)?` when the input has no dependencies. So an
error from that call does not come from here anymore. I think projection schema
construction gives the same error before this point. Can you confirm?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
