You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "gruuya (via GitHub)" <gi...@apache.org> on 2023/10/30 10:40:59 UTC
[PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
gruuya opened a new pull request, #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981
## Which issue does this PR close?
Closes #7827.
## Rationale for this change
While in principle the output of `DISTINCT ON` can be emulated by combining several aggregation, projection and sorting queries/operators, in practice this is very verbose and impractical for users, unlike `DISTINCT ON` which is quite succinct. On the other hand that does mean that `DISTINCT ON` can be optimized into the equivalent representation under the hood during logical planning, while at the same time providing the compact syntactical interface.
## What changes are included in this PR?
- A new logical plan sub-variety for handling the `DISTINCT ON` from Postgres.
- Parsing and planning for it
- Optimization which converts it into the equivalent representation using the more common logical plan primitives
- Proto serialization
- Tests
## Are these changes tested?
Yes, there are also new SLTs.
## Are there any user-facing changes?
SQL support extends to encompass `DISTINCT ON` queries.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1377481562
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2132,9 +2181,96 @@ pub struct Limit {
/// Removes duplicate rows from the input
#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct Distinct {
+pub enum Distinct {
+ /// Plain `DISTINCT` referencing all selection expressions
+ All(Arc<LogicalPlan>),
+ /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
+ On(DistinctOn),
+}
+
+/// Removes duplicate rows from the input
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct DistinctOn {
+ /// The `DISTINCT ON` clause expression list
+ pub on_expr: Vec<Expr>,
+ /// The selected projection expression list
+ pub select_expr: Vec<Expr>,
+ /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause
+ pub sort_expr: Option<Vec<Expr>>,
/// The logical plan that is being DISTINCT'd
pub input: Arc<LogicalPlan>,
+ /// The schema description of the DISTINCT ON output
+ pub schema: DFSchemaRef,
+}
+
+impl DistinctOn {
+ /// Create a new `DistintOn` struct.
+ pub fn try_new(
+ on_expr: Vec<Expr>,
+ select_expr: Vec<Expr>,
+ sort_expr: Option<Vec<Expr>>,
+ input: Arc<LogicalPlan>,
+ ) -> Result<Self> {
+ let on_expr = normalize_cols(on_expr, input.as_ref())?;
+
+ // Create fields with any qualifier stuffed in the name itself
Review Comment:
This is sort of a hack, but the reason I was forced to do it is so that the plan schema remains unchanged after `replace_distinct_aggregate`. In particular, in that optimizer I alias the final projected expressions with the selection expression display name, but that always results in a schema without qualifiers (with any original qualifier crammed into the field name).
I feel like this case should be handled by `Projection::try_new_with_schema` over at `replace_distinct_aggregate`, but it seems that after #7919 the custom schema doesn't propagate through the optimization chain.
UPDATE: I've opened a related issue for this: https://github.com/apache/arrow-datafusion/issues/8008
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#issuecomment-1789474611
I am sorry -- I have seen this PR but I am somewhat backed up on reviews
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1382416566
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -60,6 +80,62 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
)?);
Ok(Some(aggregate))
}
+ LogicalPlan::Distinct(Distinct::On(DistinctOn {
+ select_expr,
+ on_expr,
+ sort_expr,
+ input,
+ ..
+ })) => {
+ // Construct the aggregation expression to be used to fetch the selected expressions.
+ let aggr_expr = select_expr
+ .iter()
+ .map(|e| {
+ Expr::AggregateFunction(AggregateFunction::new(
+ AggregateFunctionFunc::FirstValue,
+ vec![e.clone()],
+ false,
+ None,
+ sort_expr.clone(),
+ ))
+ })
+ .collect::<Vec<Expr>>();
+
+ // Build the aggregation plan
+ let plan = LogicalPlanBuilder::from(input.as_ref().clone())
+ .aggregate(on_expr.clone(), aggr_expr.to_vec())?
+ .build()?;
+
+ let plan = if let Some(sort_expr) = sort_expr {
+ // While sort expressions were used in the `FIRST_VALUE` aggregation itself above,
+ // this on it's own isn't enough to guarantee the proper output order of the grouping
+ // (`ON`) expression, so we need to sort those as well.
+ LogicalPlanBuilder::from(plan)
+ .sort(sort_expr[..on_expr.len()].to_vec())?
Review Comment:
Yeah, my thinking was that the `first_value` aggregation actually doesn't do any real sorting and is more like top-1, i.e. it only works on the incoming stream and picks up one row per-grouping set (`ON` clause) if it's a better match according to the sorting expressions:
https://github.com/apache/arrow-datafusion/blob/2af326a0bdbb60e7d7420089c7aa9ff22c7319ee/datafusion/physical-expr/src/aggregate/first_last.rs#L236-L244
The subsequent actual sort works on the grouped data, so it should have a much lower cardinality.
Whereas if we sorted prior to aggregation then, I think we would materialize all the data initially (whether in memory or by spilling), only to throw away most of it.
Granted I haven't done any perf testing between the two approaches for now, this is just from my reading of the code.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1379324782
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,8 +238,17 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation epxressions if they have changed
+ // TODO: This should have really been identified above and handled in the `else` branch
+ let aggr_exprs = new_aggr_expr
+ .iter()
+ .zip(aggr_expr.iter())
+ .map(|(new_expr, old_expr)| {
+ new_expr.clone().alias_if_changed(old_expr.display_name()?)
+ })
+ .collect::<Result<Vec<Expr>>>()?;
// Since group_epxr changes, schema changes also. Use try_new method.
- Aggregate::try_new(Arc::new(new_input), new_group_expr, new_aggr_expr)
+ Aggregate::try_new(Arc::new(new_input), new_group_expr, aggr_exprs)
.map(LogicalPlan::Aggregate)
Review Comment:
Ok, I think I understand what is going on:
- in general the root aggregation expressions need not be rewritten/aliased above since they can only appear once (i.e. they're not common), so this is ok
- however, some of their child expressions may get rewritten; an example is the query `select distinct on (c1 + 1) c2 from test order by c1 + 1, c3`, where the initial aggregation of the sole projected field gets rewritten from `FIRST_VALUE(test.column2) ORDER BY [test.column1 + Int64(1) ASC NULLS LAST, test.column3 ASC NULLS LAST]` to `FIRST_VALUE(test.column2) ORDER BY [test.column1 + Int64(1)Int64(1)test.column1 AS test.column1 + Int64(1) ASC NULLS LAST, test.column3 ASC NULLS LAST]` (i.e. `c1 + 1` is a common expression of it and the `ON` expression so it gets rewritten)
- consequently the new schema will be different from the original one
I think this explains the behavior I've encountered here, and that this behavior is not anomalous. The only question is whether there is a better way to align the new aggregation schema with the old one besides aliasing as above.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1382641037
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -60,6 +80,62 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
)?);
Ok(Some(aggregate))
}
+ LogicalPlan::Distinct(Distinct::On(DistinctOn {
+ select_expr,
+ on_expr,
+ sort_expr,
+ input,
+ ..
+ })) => {
+ // Construct the aggregation expression to be used to fetch the selected expressions.
+ let aggr_expr = select_expr
+ .iter()
+ .map(|e| {
+ Expr::AggregateFunction(AggregateFunction::new(
+ AggregateFunctionFunc::FirstValue,
+ vec![e.clone()],
+ false,
+ None,
+ sort_expr.clone(),
+ ))
+ })
+ .collect::<Vec<Expr>>();
+
+ // Build the aggregation plan
+ let plan = LogicalPlanBuilder::from(input.as_ref().clone())
+ .aggregate(on_expr.clone(), aggr_expr.to_vec())?
+ .build()?;
+
+ let plan = if let Some(sort_expr) = sort_expr {
+ // While sort expressions were used in the `FIRST_VALUE` aggregation itself above,
+ // this on it's own isn't enough to guarantee the proper output order of the grouping
+ // (`ON`) expression, so we need to sort those as well.
+ LogicalPlanBuilder::from(plan)
+ .sort(sort_expr[..on_expr.len()].to_vec())?
Review Comment:
I see now that my assumptions were wrong, and that indeed sorting gets pulled down beneath aggregation during physical planning anyway. I do think that in the case of `FIRST_VALUE` with `ORDER BY` this is sub-optimal, since that problem reduces to a partitioned top-1 problem, which doesn't need to work on pre-sorted input I think.
Either way, even when I pull sorting beneath the aggregation, I still find I end up needing another post-aggregation sort to get the correct order for the `ON` clause expressions, since presumably `AggregateExec` doesn't preserve order:
```sql
❯ explain select distinct on (c1) c1, c2 from test order by c1, c3;
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST] AS test.c1, FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST] AS test.c2 |
| | Aggregate: groupBy=[[test.c1]], aggr=[[FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST], FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]]] |
| | Sort: test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST |
| | TableScan: test projection=[c1, c2, c3] |
| physical_plan | ProjectionExec: expr=[FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]@1 as test.c1, FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]@2 as test.c2] |
| | AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1], aggr=[FIRST_VALUE(test.c1), FIRST_VALUE(test.c2)], ordering_mode=Sorted |
| | SortExec: expr=[c1@0 ASC NULLS LAST] |
| | CoalesceBatchesExec: target_batch_size=8192 |
| | RepartitionExec: partitioning=Hash([c1@0], 10), input_partitions=10 |
| | AggregateExec: mode=Partial, gby=[c1@0 as c1], aggr=[FIRST_VALUE(test.c1), FIRST_VALUE(test.c2)], ordering_mode=Sorted |
| | RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1 |
| | SortExec: expr=[c1@0 ASC NULLS LAST,c3@2 ASC NULLS LAST] |
| | CsvExec: file_groups={1 group: [[Users/markogrujic/Splitgraph/arrow-datafusion/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1, c2, c3], has_header=true |
| | |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
```
compare that with the currently generated plans which do get the order correct:
```sql
❯ explain select distinct on (c1) c1, c2 from test order by c1, c3;
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST] AS test.c1, FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST] AS test.c2 |
| | Sort: test.c1 ASC NULLS LAST |
| | Aggregate: groupBy=[[test.c1]], aggr=[[FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST], FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]]] |
| | TableScan: test projection=[c1, c2, c3] |
| physical_plan | ProjectionExec: expr=[FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]@1 as test.c1, FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]@2 as test.c2] |
| | SortPreservingMergeExec: [c1@0 ASC NULLS LAST] |
| | SortExec: expr=[c1@0 ASC NULLS LAST] |
| | AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1], aggr=[FIRST_VALUE(test.c1), FIRST_VALUE(test.c2)] |
| | CoalesceBatchesExec: target_batch_size=8192 |
| | RepartitionExec: partitioning=Hash([c1@0], 10), input_partitions=10 |
| | AggregateExec: mode=Partial, gby=[c1@0 as c1], aggr=[FIRST_VALUE(test.c1), FIRST_VALUE(test.c2)], ordering_mode=Sorted |
| | SortExec: expr=[c1@0 ASC NULLS LAST,c3@2 ASC NULLS LAST] |
| | RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1 |
| | CsvExec: file_groups={1 group: [[Users/markogrujic/Splitgraph/arrow-datafusion/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1, c2, c3], has_header=true |
| | |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
```
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -60,6 +80,62 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
)?);
Ok(Some(aggregate))
}
+ LogicalPlan::Distinct(Distinct::On(DistinctOn {
+ select_expr,
+ on_expr,
+ sort_expr,
+ input,
+ ..
+ })) => {
+ // Construct the aggregation expression to be used to fetch the selected expressions.
+ let aggr_expr = select_expr
+ .iter()
+ .map(|e| {
+ Expr::AggregateFunction(AggregateFunction::new(
+ AggregateFunctionFunc::FirstValue,
+ vec![e.clone()],
+ false,
+ None,
+ sort_expr.clone(),
+ ))
+ })
+ .collect::<Vec<Expr>>();
+
+ // Build the aggregation plan
+ let plan = LogicalPlanBuilder::from(input.as_ref().clone())
+ .aggregate(on_expr.clone(), aggr_expr.to_vec())?
+ .build()?;
+
+ let plan = if let Some(sort_expr) = sort_expr {
+ // While sort expressions were used in the `FIRST_VALUE` aggregation itself above,
+ // this on it's own isn't enough to guarantee the proper output order of the grouping
+ // (`ON`) expression, so we need to sort those as well.
+ LogicalPlanBuilder::from(plan)
+ .sort(sort_expr[..on_expr.len()].to_vec())?
Review Comment:
I see now that my assumptions were wrong, and that indeed sorting gets pulled down beneath aggregation during physical planning anyway. I do think that in the case of `FIRST_VALUE` with `ORDER BY` this is sub-optimal, since that problem reduces to a partitioned top-1 problem, which doesn't need to work on pre-sorted input I think.
Either way, even when I pull sorting beneath the aggregation, I still find I end up needing another post-aggregation sort to get the correct order for the `ON` clause expressions, since presumably `AggregateExec` doesn't preserve order:
```sql
❯ explain select distinct on (c1) c1, c2 from test order by c1, c3;
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST] AS test.c1, FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST] AS test.c2 |
| | Aggregate: groupBy=[[test.c1]], aggr=[[FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST], FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]]] |
| | Sort: test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST |
| | TableScan: test projection=[c1, c2, c3] |
| physical_plan | ProjectionExec: expr=[FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]@1 as test.c1, FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]@2 as test.c2] |
| | AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1], aggr=[FIRST_VALUE(test.c1), FIRST_VALUE(test.c2)], ordering_mode=Sorted |
| | SortExec: expr=[c1@0 ASC NULLS LAST] |
| | CoalesceBatchesExec: target_batch_size=8192 |
| | RepartitionExec: partitioning=Hash([c1@0], 10), input_partitions=10 |
| | AggregateExec: mode=Partial, gby=[c1@0 as c1], aggr=[FIRST_VALUE(test.c1), FIRST_VALUE(test.c2)], ordering_mode=Sorted |
| | RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1 |
| | SortExec: expr=[c1@0 ASC NULLS LAST,c3@2 ASC NULLS LAST] |
| | CsvExec: file_groups={1 group: [[Users/markogrujic/Splitgraph/arrow-datafusion/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1, c2, c3], has_header=true |
| | |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
```
compare that with the currently generated plans which do get the order correct:
```sql
❯ explain select distinct on (c1) c1, c2 from test order by c1, c3;
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST] AS test.c1, FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST] AS test.c2 |
| | Sort: test.c1 ASC NULLS LAST |
| | Aggregate: groupBy=[[test.c1]], aggr=[[FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST], FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]]] |
| | TableScan: test projection=[c1, c2, c3] |
| physical_plan | ProjectionExec: expr=[FIRST_VALUE(test.c1) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]@1 as test.c1, FIRST_VALUE(test.c2) ORDER BY [test.c1 ASC NULLS LAST, test.c3 ASC NULLS LAST]@2 as test.c2] |
| | SortPreservingMergeExec: [c1@0 ASC NULLS LAST] |
| | SortExec: expr=[c1@0 ASC NULLS LAST] |
| | AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1], aggr=[FIRST_VALUE(test.c1), FIRST_VALUE(test.c2)] |
| | CoalesceBatchesExec: target_batch_size=8192 |
| | RepartitionExec: partitioning=Hash([c1@0], 10), input_partitions=10 |
| | AggregateExec: mode=Partial, gby=[c1@0 as c1], aggr=[FIRST_VALUE(test.c1), FIRST_VALUE(test.c2)], ordering_mode=Sorted |
| | SortExec: expr=[c1@0 ASC NULLS LAST,c3@2 ASC NULLS LAST] |
| | RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1 |
| | CsvExec: file_groups={1 group: [[Users/markogrujic/Splitgraph/arrow-datafusion/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1, c2, c3], has_header=true |
| | |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
```
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1390389939
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,6 +238,16 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation expressions if they have changed
+ // TODO: This should be handled in `build_recover_project_plan` once qualified aliases
Review Comment:
I think this comment is out of date
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1379324782
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,8 +238,17 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation epxressions if they have changed
+ // TODO: This should have really been identified above and handled in the `else` branch
+ let aggr_exprs = new_aggr_expr
+ .iter()
+ .zip(aggr_expr.iter())
+ .map(|(new_expr, old_expr)| {
+ new_expr.clone().alias_if_changed(old_expr.display_name()?)
+ })
+ .collect::<Result<Vec<Expr>>>()?;
// Since group_epxr changes, schema changes also. Use try_new method.
- Aggregate::try_new(Arc::new(new_input), new_group_expr, new_aggr_expr)
+ Aggregate::try_new(Arc::new(new_input), new_group_expr, aggr_exprs)
.map(LogicalPlan::Aggregate)
Review Comment:
Ok, I think I understand what is going on:
- in general the root aggregation expressions need not be rewritten/aliased above since they can only appear once (i.e. they're not common), so this is ok
- however, some of their child expressions may get rewritten; an example is the query `select distinct on (c1 + 1) c2 from test order by c1 + 1, c3`, where the initial aggregation of the sole projected field gets rewritten from `FIRST_VALUE(test.column2) ORDER BY [test.column1 + Int64(1) ASC NULLS LAST, test.column3 ASC NULLS LAST]` to `FIRST_VALUE(test.column2) ORDER BY [test.column1 + Int64(1)Int64(1)test.column1 AS test.column1 + Int64(1) ASC NULLS LAST, test.column3 ASC NULLS LAST]` (i.e. `c1 + 1` is a common expression of it and the `ON` expression so it gets rewritten)
- it seems this was hidden before thanks to using `Aggregate::try_new_with_schema` with the original schema, however with #7919 this is no longer the case, so the newly created schema is mismatched with schema expected by the parent node
I think this explains the behavior I've encountered here, and that this behavior is not anomalous. The only question is whether there is a better way to align the new aggregation schema with the old one besides aliasing as above.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1381591595
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
@@ -0,0 +1,129 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+statement ok
+CREATE EXTERNAL TABLE aggregate_test_100 (
+ c1 VARCHAR NOT NULL,
+ c2 TINYINT NOT NULL,
+ c3 SMALLINT NOT NULL,
+ c4 SMALLINT,
+ c5 INT,
+ c6 BIGINT NOT NULL,
+ c7 SMALLINT NOT NULL,
+ c8 INT NOT NULL,
+ c9 BIGINT UNSIGNED NOT NULL,
+ c10 VARCHAR NOT NULL,
+ c11 FLOAT NOT NULL,
+ c12 DOUBLE NOT NULL,
+ c13 VARCHAR NOT NULL
+)
+STORED AS CSV
+WITH HEADER ROW
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
+
+# Basic example: distinct on the first column project the second one, and
+# order by the third
+# TODO: Note that the value for 'a' is actually degenerate, and while Postgres
+# returns the actual first match (4), DataFusion returns the last one, presumably
+# because `first_value` aggregation function doesn't have a dedicated `GroupsAccumulator`
Review Comment:
Oh I see what this is, it's another case of unstable sort, i.e. the sorted rows that have equal ordering expressions evaluations do not preserve the original order of those rows (but they're stable in the sense that output is always the same):
```sql
❯ select c1, c2, c3 from test order by c1, c3;
+----+----+------+
| c1 | c2 | c3 |
+----+----+------+
| a | 5 | -101 |
| a | 4 | -101 |
| a | 1 | -85 |
| a | 3 | -72 |
| a | 1 | -56 |
...
```
I've removed the misleading comment now.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1377481562
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2132,9 +2181,96 @@ pub struct Limit {
/// Removes duplicate rows from the input
#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct Distinct {
+pub enum Distinct {
+ /// Plain `DISTINCT` referencing all selection expressions
+ All(Arc<LogicalPlan>),
+ /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
+ On(DistinctOn),
+}
+
+/// Removes duplicate rows from the input
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct DistinctOn {
+ /// The `DISTINCT ON` clause expression list
+ pub on_expr: Vec<Expr>,
+ /// The selected projection expression list
+ pub select_expr: Vec<Expr>,
+ /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause
+ pub sort_expr: Option<Vec<Expr>>,
/// The logical plan that is being DISTINCT'd
pub input: Arc<LogicalPlan>,
+ /// The schema description of the DISTINCT ON output
+ pub schema: DFSchemaRef,
+}
+
+impl DistinctOn {
+ /// Create a new `DistintOn` struct.
+ pub fn try_new(
+ on_expr: Vec<Expr>,
+ select_expr: Vec<Expr>,
+ sort_expr: Option<Vec<Expr>>,
+ input: Arc<LogicalPlan>,
+ ) -> Result<Self> {
+ let on_expr = normalize_cols(on_expr, input.as_ref())?;
+
+ // Create fields with any qualifier stuffed in the name itself
Review Comment:
This is sort of a hack, but the reason I was forced to do it is so that the plan schema remains unchanged after `replace_distinct_aggregate`. In particular, in that optimizer I alias the final projected expressions with the selection expression display name, but that always results in a schema without qualifiers (with any original qualifier crammed into the field name).
I feel like this case should be handled by `Projection::try_new_with_schema` over at `replace_distinct_aggregate`, but it seems that after #7919 the custom schema doesn't propagate through the optimization chain.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1379180032
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2132,9 +2181,96 @@ pub struct Limit {
/// Removes duplicate rows from the input
#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct Distinct {
+pub enum Distinct {
+ /// Plain `DISTINCT` referencing all selection expressions
+ All(Arc<LogicalPlan>),
+ /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
+ On(DistinctOn),
+}
+
+/// Removes duplicate rows from the input
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct DistinctOn {
+ /// The `DISTINCT ON` clause expression list
+ pub on_expr: Vec<Expr>,
+ /// The selected projection expression list
+ pub select_expr: Vec<Expr>,
+ /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause
+ pub sort_expr: Option<Vec<Expr>>,
/// The logical plan that is being DISTINCT'd
pub input: Arc<LogicalPlan>,
+ /// The schema description of the DISTINCT ON output
+ pub schema: DFSchemaRef,
+}
+
+impl DistinctOn {
+ /// Create a new `DistintOn` struct.
+ pub fn try_new(
+ on_expr: Vec<Expr>,
+ select_expr: Vec<Expr>,
+ sort_expr: Option<Vec<Expr>>,
+ input: Arc<LogicalPlan>,
+ ) -> Result<Self> {
+ let on_expr = normalize_cols(on_expr, input.as_ref())?;
+
+ // Create fields with any qualifier stuffed in the name itself
Review Comment:
> This is sort of a hack, but the reason I was forced to do it is so that the plan schema remains unchanged after replace_distinct_aggregate.
Note https://github.com/apache/arrow-datafusion/pull/7997 may be related
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1379324782
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,8 +238,17 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation epxressions if they have changed
+ // TODO: This should have really been identified above and handled in the `else` branch
+ let aggr_exprs = new_aggr_expr
+ .iter()
+ .zip(aggr_expr.iter())
+ .map(|(new_expr, old_expr)| {
+ new_expr.clone().alias_if_changed(old_expr.display_name()?)
+ })
+ .collect::<Result<Vec<Expr>>>()?;
// Since group_epxr changes, schema changes also. Use try_new method.
- Aggregate::try_new(Arc::new(new_input), new_group_expr, new_aggr_expr)
+ Aggregate::try_new(Arc::new(new_input), new_group_expr, aggr_exprs)
.map(LogicalPlan::Aggregate)
Review Comment:
Ok, I think I understand what is going on:
- in general the root aggregation expressions need not be rewritten/aliased above since they can only appear once (i.e. they're not common), so this is ok
- however, some of their child expressions may get rewritten; an example is the query `select distinct on (c1 + 1) c2 from test order by c1 + 1, c3`, where the initial aggregation of the sole projected field gets rewritten from `FIRST_VALUE(test.column2) ORDER BY [test.column1 + Int64(1) ASC NULLS LAST, test.column3 ASC NULLS LAST]` to `FIRST_VALUE(test.column2) ORDER BY [test.column1 + Int64(1)Int64(1)test.column1 AS test.column1 + Int64(1) ASC NULLS LAST, test.column3 ASC NULLS LAST]` (i.e. `c1 + 1` is a common expression of it and the `ON` expression so it gets rewritten)
I think this explains the behavior I've encountered here, and that this behavior is not anomalous. The only question is whether there is a better way to align the new aggregation schema with the old one besides aliasing as above.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "waynexia (via GitHub)" <gi...@apache.org>.
waynexia commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1379552746
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -163,7 +163,8 @@ impl LogicalPlan {
}) => projected_schema,
LogicalPlan::Projection(Projection { schema, .. }) => schema,
LogicalPlan::Filter(Filter { input, .. }) => input.schema(),
- LogicalPlan::Distinct(Distinct { input }) => input.schema(),
+ LogicalPlan::Distinct(Distinct::All(input)) => input.schema(),
+ LogicalPlan::Distinct(Distinct::On(DistinctOn { schema, .. })) => schema,
Review Comment:
Add `api-change` label for this diff
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1382416566
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -60,6 +80,62 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
)?);
Ok(Some(aggregate))
}
+ LogicalPlan::Distinct(Distinct::On(DistinctOn {
+ select_expr,
+ on_expr,
+ sort_expr,
+ input,
+ ..
+ })) => {
+ // Construct the aggregation expression to be used to fetch the selected expressions.
+ let aggr_expr = select_expr
+ .iter()
+ .map(|e| {
+ Expr::AggregateFunction(AggregateFunction::new(
+ AggregateFunctionFunc::FirstValue,
+ vec![e.clone()],
+ false,
+ None,
+ sort_expr.clone(),
+ ))
+ })
+ .collect::<Vec<Expr>>();
+
+ // Build the aggregation plan
+ let plan = LogicalPlanBuilder::from(input.as_ref().clone())
+ .aggregate(on_expr.clone(), aggr_expr.to_vec())?
+ .build()?;
+
+ let plan = if let Some(sort_expr) = sort_expr {
+ // While sort expressions were used in the `FIRST_VALUE` aggregation itself above,
+ // this on it's own isn't enough to guarantee the proper output order of the grouping
+ // (`ON`) expression, so we need to sort those as well.
+ LogicalPlanBuilder::from(plan)
+ .sort(sort_expr[..on_expr.len()].to_vec())?
Review Comment:
Yeah, my thinking was that the `first_value` aggregation actually doesn't do any real sorting and is more like top-1, i.e. it only works on the incoming stream and picks up one row per-grouping set (`ON` clause), updating it if it's a better match according to the sorting expressions:
https://github.com/apache/arrow-datafusion/blob/2af326a0bdbb60e7d7420089c7aa9ff22c7319ee/datafusion/physical-expr/src/aggregate/first_last.rs#L236-L244
The subsequent actual sort works on the grouped data, so it should have a much lower cardinality.
Whereas if we sorted prior to aggregation then, I think we would materialize all the data initially (whether in memory or by spilling) only to throw away most of it, but I could be wrong.
Granted I haven't done any perf testing between the two approaches for now, this is just from my reading of the code.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1382413418
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
@@ -0,0 +1,125 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+statement ok
+CREATE EXTERNAL TABLE aggregate_test_100 (
+ c1 VARCHAR NOT NULL,
+ c2 TINYINT NOT NULL,
+ c3 SMALLINT NOT NULL,
+ c4 SMALLINT,
+ c5 INT,
+ c6 BIGINT NOT NULL,
+ c7 SMALLINT NOT NULL,
+ c8 INT NOT NULL,
+ c9 BIGINT UNSIGNED NOT NULL,
+ c10 VARCHAR NOT NULL,
+ c11 FLOAT NOT NULL,
+ c12 DOUBLE NOT NULL,
+ c13 VARCHAR NOT NULL
+)
+STORED AS CSV
+WITH HEADER ROW
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
+
+# Basic example: distinct on the first column project the second one, and
+# order by the third
+query TI
+SELECT DISTINCT ON (c1) c1, c2 FROM aggregate_test_100 ORDER BY c1, c3;
+----
+a 5
+b 4
+c 2
+d 1
+e 3
+
+# Basic example + reverse order of the selected column
+query TI
+SELECT DISTINCT ON (c1) c1, c2 FROM aggregate_test_100 ORDER BY c1, c3 DESC;
+----
+a 1
+b 5
+c 4
+d 1
+e 1
+
+# Basic example + reverse order of the ON column
+query TI
+SELECT DISTINCT ON (c1) c1, c2 FROM aggregate_test_100 ORDER BY c1 DESC, c3;
+----
+e 3
+d 1
+c 2
+b 4
+a 4
+
+# Basic example + reverse order of both columns + limit
+query TI
+SELECT DISTINCT ON (c1) c1, c2 FROM aggregate_test_100 ORDER BY c1 DESC, c3 DESC LIMIT 3;
+----
+e 1
+d 1
+c 4
+
+# Basic example + omit ON column from selection
+query I
+SELECT DISTINCT ON (c1) c2 FROM aggregate_test_100 ORDER BY c1, c3;
Review Comment:
Good idea, adding that test as well!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1391212308
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,6 +238,16 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation expressions if they have changed
+ // TODO: This should be handled in `build_recover_project_plan` once qualified aliases
Review Comment:
> Thanks for your patience on this one! :)
It is **I** who should be thanking **you**.
Thanks again @gruuya
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1376005289
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
Review Comment:
All tests here were compared for accuracy against Postgres.
Note that, while the [docs](https://www.postgresql.org/docs/current/sql-select.html#SQL-DISTINCT) state that in the absence of the `ORDER BY` clause the output is unpredictable, in practice I've noticed it is stable on Postgres for the table below, whereas that isn't the case for DataFusion, hence why I omitted those tests for now.
##########
datafusion/optimizer/src/push_down_projection.rs:
##########
@@ -228,7 +228,7 @@ impl OptimizerRule for PushDownProjection {
// Gather all columns needed for expressions in this Aggregate
let mut new_aggr_expr = vec![];
for e in agg.aggr_expr.iter() {
- let column = Column::from_name(e.display_name()?);
+ let column = Column::from(e.display_name()?);
Review Comment:
This allows one to correctly re-construct a column whose qualifier has been crammed into the name by aliasing, which is what I'm doing in `replace_distinct_aggregate` in this PR. Not sure whether this makes sense in general though.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1376010499
##########
datafusion/optimizer/src/push_down_projection.rs:
##########
@@ -228,7 +228,7 @@ impl OptimizerRule for PushDownProjection {
// Gather all columns needed for expressions in this Aggregate
let mut new_aggr_expr = vec![];
for e in agg.aggr_expr.iter() {
- let column = Column::from_name(e.display_name()?);
+ let column = Column::from(e.display_name()?);
Review Comment:
This allows one to correctly re-construct a column whose qualifier has been crammed into the name by aliasing with `.display_name()`, which is what I'm doing in `replace_distinct_aggregate` in this PR. Not sure whether this makes sense in general though.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1377468064
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
Review Comment:
All tests here were compared for accuracy against Postgres.
Note that, while the [docs](https://www.postgresql.org/docs/current/sql-select.html#SQL-DISTINCT) state that in the absence of the ORDER BY clause the output is unpredictable, in practice I've noticed it is stable on Postgres for the table below, whereas that isn't the case for DataFusion, hence why I omitted those tests for now.
Any ideas on what else can be tested here are welcome (cc @universalmind303).
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2132,9 +2181,96 @@ pub struct Limit {
/// Removes duplicate rows from the input
#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct Distinct {
+pub enum Distinct {
+ /// Plain `DISTINCT` referencing all selection expressions
+ All(Arc<LogicalPlan>),
+ /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
+ On(DistinctOn),
+}
+
+/// Removes duplicate rows from the input
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct DistinctOn {
+ /// The `DISTINCT ON` clause expression list
+ pub on_expr: Vec<Expr>,
+ /// The selected projection expression list
+ pub select_expr: Vec<Expr>,
+ /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause
+ pub sort_expr: Option<Vec<Expr>>,
/// The logical plan that is being DISTINCT'd
pub input: Arc<LogicalPlan>,
+ /// The schema description of the DISTINCT ON output
+ pub schema: DFSchemaRef,
+}
+
+impl DistinctOn {
+ /// Create a new `DistintOn` struct.
+ pub fn try_new(
+ on_expr: Vec<Expr>,
+ select_expr: Vec<Expr>,
+ sort_expr: Option<Vec<Expr>>,
+ input: Arc<LogicalPlan>,
+ ) -> Result<Self> {
+ let on_expr = normalize_cols(on_expr, input.as_ref())?;
+
+ // Create fields with any qualifier stuffed in the name itself
Review Comment:
This is sort of a hack, but the reason I was forced to do it is so that the plan schema remains unchanged after `replace_distinct_aggregate`. In particular, in that optimizer I alias the final projected expressions with the selection expression display name, but that always results in a schema without qualifiers (with any original qualifier crammed into the field name).
I feel like this case should be handled by `Projection::try_new_with_schema`, but it seems that after #7919 the custom schema doesn't propagate through the optimization chain.
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,8 +238,17 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation epxressions if they have changed
+ // TODO: This should have really been identified above and handled in the `else` branch
+ let aggr_exprs = new_aggr_expr
+ .iter()
+ .zip(aggr_expr.iter())
+ .map(|(new_expr, old_expr)| {
+ new_expr.clone().alias_if_changed(old_expr.display_name()?)
+ })
+ .collect::<Result<Vec<Expr>>>()?;
// Since group_epxr changes, schema changes also. Use try_new method.
- Aggregate::try_new(Arc::new(new_input), new_group_expr, new_aggr_expr)
+ Aggregate::try_new(Arc::new(new_input), new_group_expr, aggr_exprs)
.map(LogicalPlan::Aggregate)
Review Comment:
I don't yet understand why this wasn't identified/aliased automatically, but I found this part to be necessary in order to get `DISTINCT ON` with complex expressions to work.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1382374720
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -60,6 +80,62 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
)?);
Ok(Some(aggregate))
}
+ LogicalPlan::Distinct(Distinct::On(DistinctOn {
+ select_expr,
+ on_expr,
+ sort_expr,
+ input,
+ ..
+ })) => {
+ // Construct the aggregation expression to be used to fetch the selected expressions.
+ let aggr_expr = select_expr
+ .iter()
+ .map(|e| {
+ Expr::AggregateFunction(AggregateFunction::new(
+ AggregateFunctionFunc::FirstValue,
+ vec![e.clone()],
+ false,
+ None,
+ sort_expr.clone(),
+ ))
+ })
+ .collect::<Vec<Expr>>();
+
+ // Build the aggregation plan
+ let plan = LogicalPlanBuilder::from(input.as_ref().clone())
+ .aggregate(on_expr.clone(), aggr_expr.to_vec())?
+ .build()?;
+
+ let plan = if let Some(sort_expr) = sort_expr {
+ // While sort expressions were used in the `FIRST_VALUE` aggregation itself above,
+ // this on it's own isn't enough to guarantee the proper output order of the grouping
+ // (`ON`) expression, so we need to sort those as well.
+ LogicalPlanBuilder::from(plan)
+ .sort(sort_expr[..on_expr.len()].to_vec())?
Review Comment:
Would it make sense to sort the input prior to aggregating it? It seems like sorting to do the aggregate and then sorting again is not as efficient as being able to just sort once and reuse the output.
In theory the optimizer should handle this rewrite, but it might be better to just start with the sort under the aggregate
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2132,9 +2181,100 @@ pub struct Limit {
/// Removes duplicate rows from the input
#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct Distinct {
+pub enum Distinct {
+ /// Plain `DISTINCT` referencing all selection expressions
+ All(Arc<LogicalPlan>),
+ /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
+ On(DistinctOn),
+}
+
+/// Removes duplicate rows from the input
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct DistinctOn {
+ /// The `DISTINCT ON` clause expression list
+ pub on_expr: Vec<Expr>,
+ /// The selected projection expression list
+ pub select_expr: Vec<Expr>,
+ /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause
Review Comment:
I think your explanation makes sense. Maybe we can take some of this rationale (e.g. that `sort_exprs` are not a superset as they are wrapped by Expr::Sort) and put them into the doc comments. We can do this as a follow on PR
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
@@ -0,0 +1,125 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+statement ok
+CREATE EXTERNAL TABLE aggregate_test_100 (
+ c1 VARCHAR NOT NULL,
+ c2 TINYINT NOT NULL,
+ c3 SMALLINT NOT NULL,
+ c4 SMALLINT,
+ c5 INT,
+ c6 BIGINT NOT NULL,
+ c7 SMALLINT NOT NULL,
+ c8 INT NOT NULL,
+ c9 BIGINT UNSIGNED NOT NULL,
+ c10 VARCHAR NOT NULL,
+ c11 FLOAT NOT NULL,
+ c12 DOUBLE NOT NULL,
+ c13 VARCHAR NOT NULL
+)
+STORED AS CSV
+WITH HEADER ROW
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
+
+# Basic example: distinct on the first column project the second one, and
+# order by the third
+query TI
+SELECT DISTINCT ON (c1) c1, c2 FROM aggregate_test_100 ORDER BY c1, c3;
+----
+a 5
+b 4
+c 2
+d 1
+e 3
+
+# Basic example + reverse order of the selected column
+query TI
+SELECT DISTINCT ON (c1) c1, c2 FROM aggregate_test_100 ORDER BY c1, c3 DESC;
+----
+a 1
+b 5
+c 4
+d 1
+e 1
+
+# Basic example + reverse order of the ON column
+query TI
+SELECT DISTINCT ON (c1) c1, c2 FROM aggregate_test_100 ORDER BY c1 DESC, c3;
+----
+e 3
+d 1
+c 2
+b 4
+a 4
+
+# Basic example + reverse order of both columns + limit
+query TI
+SELECT DISTINCT ON (c1) c1, c2 FROM aggregate_test_100 ORDER BY c1 DESC, c3 DESC LIMIT 3;
+----
+e 1
+d 1
+c 4
+
+# Basic example + omit ON column from selection
+query I
+SELECT DISTINCT ON (c1) c2 FROM aggregate_test_100 ORDER BY c1, c3;
Review Comment:
Can you maybe also include an explain plan here so we can validate the plan?
Something like
```
EXPLAIN SELECT DISTINCT ON (c1) c3, c2 FROM aggregate_test_100 ORDER BY c1, c3;
```
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -32,6 +36,22 @@ use datafusion_expr::{Aggregate, Distinct, LogicalPlan};
/// ```text
/// SELECT a, b FROM tab GROUP BY a, b
/// ```
+///
+/// On the other hand, for a `DISTINCT ON` query the replacement is
+/// a bit more involved and effectively converts
+/// ```text
+/// SELECT DISTINCT ON (a) b FROM tab ORDER BY a DESC, c
+/// ```
+///
+/// into
+/// ```text
+/// SELECT b FROM (
+/// SELECT a, FIRST_VALUE(b ORDER BY a DESC, c) AS b
Review Comment:
Sorry -- thank you for the clarification
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "waynexia (via GitHub)" <gi...@apache.org>.
waynexia commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1379556954
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
Review Comment:
These cases are excellent 👍 . It might be better if we could include some negative cases. Here are two I got from PG:
```postgresql
SELECT DISTINCT ON () c1, c2 FROM aggregate_test_100 ORDER BY c1, c3;
ERROR: syntax error at or near ")"
LINE 1: SELECT DISTINCT ON () c1, c2 FROM aggregate_test_100 ORDER B...
SELECT DISTINCT ON (c2) c1, c2 FROM aggregate_test_100 ORDER BY c1, c3;
ERROR: SELECT DISTINCT ON expressions must match initial ORDER BY expressions
LINE 1: SELECT DISTINCT ON (c2) c1, c2 FROM aggregate_test_100 ORDER...
```
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
Review Comment:
These cases are excellent 👍 . It might be better if we could include some negative cases. Here are two I got from PG:
```postgresql
SELECT DISTINCT ON () c1, c2 FROM aggregate_test_100 ORDER BY c1, c3;
ERROR: syntax error at or near ")"
LINE 1: SELECT DISTINCT ON () c1, c2 FROM aggregate_test_100 ORDER B...
SELECT DISTINCT ON (c2) c1, c2 FROM aggregate_test_100 ORDER BY c1, c3;
ERROR: SELECT DISTINCT ON expressions must match initial ORDER BY expressions
LINE 1: SELECT DISTINCT ON (c2) c1, c2 FROM aggregate_test_100 ORDER...
```
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1381215153
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
@@ -0,0 +1,129 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+statement ok
+CREATE EXTERNAL TABLE aggregate_test_100 (
+ c1 VARCHAR NOT NULL,
+ c2 TINYINT NOT NULL,
+ c3 SMALLINT NOT NULL,
+ c4 SMALLINT,
+ c5 INT,
+ c6 BIGINT NOT NULL,
+ c7 SMALLINT NOT NULL,
+ c8 INT NOT NULL,
+ c9 BIGINT UNSIGNED NOT NULL,
+ c10 VARCHAR NOT NULL,
+ c11 FLOAT NOT NULL,
+ c12 DOUBLE NOT NULL,
+ c13 VARCHAR NOT NULL
+)
+STORED AS CSV
+WITH HEADER ROW
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
+
+# Basic example: distinct on the first column project the second one, and
+# order by the third
+# TODO: Note that the value for 'a' is actually degenerate, and while Postgres
+# returns the actual first match (4), DataFusion returns the last one, presumably
+# because `first_value` aggregation function doesn't have a dedicated `GroupsAccumulator`
Review Comment:
Will investigate in more depth and report back. (It's indicative that the [equivalent query](https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1381195039) without using `DISTINCT ON` returns the same results.)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1380703847
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2132,9 +2181,100 @@ pub struct Limit {
/// Removes duplicate rows from the input
#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct Distinct {
+pub enum Distinct {
+ /// Plain `DISTINCT` referencing all selection expressions
+ All(Arc<LogicalPlan>),
+ /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
+ On(DistinctOn),
+}
+
+/// Removes duplicate rows from the input
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct DistinctOn {
+ /// The `DISTINCT ON` clause expression list
+ pub on_expr: Vec<Expr>,
+ /// The selected projection expression list
+ pub select_expr: Vec<Expr>,
+ /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause
Review Comment:
Since the same list of exprs is included in both the `select_list` and `sort_exprs` I wonder if it would be less error prone to have references rather than keep three parallel lists:
```
/// The selected projection expression list
pub select_expr: Vec<Expr>,
/// The sort expressions
pub sort_expr: Vec<Expr>,
/// the number of prefix columns from `sort_expr` that form the `ON` clause used for deduplicating
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -32,6 +36,22 @@ use datafusion_expr::{Aggregate, Distinct, LogicalPlan};
/// ```text
/// SELECT a, b FROM tab GROUP BY a, b
/// ```
+///
+/// On the other hand, for a `DISTINCT ON` query the replacement is
+/// a bit more involved and effectively converts
+/// ```text
+/// SELECT DISTINCT ON (a) b FROM tab ORDER BY a DESC, c
+/// ```
+///
+/// into
+/// ```text
+/// SELECT b FROM (
+/// SELECT a, FIRST_VALUE(b ORDER BY a DESC, c) AS b
Review Comment:
```suggestion
/// SELECT a, FIRST_VALUE(PARTITION BY b ORDER BY a DESC, c) AS b
```
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -32,6 +36,22 @@ use datafusion_expr::{Aggregate, Distinct, LogicalPlan};
/// ```text
/// SELECT a, b FROM tab GROUP BY a, b
/// ```
+///
+/// On the other hand, for a `DISTINCT ON` query the replacement is
Review Comment:
Thank you -- this is very helpful to understand what is going on
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
@@ -0,0 +1,129 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+statement ok
+CREATE EXTERNAL TABLE aggregate_test_100 (
+ c1 VARCHAR NOT NULL,
+ c2 TINYINT NOT NULL,
+ c3 SMALLINT NOT NULL,
+ c4 SMALLINT,
+ c5 INT,
+ c6 BIGINT NOT NULL,
+ c7 SMALLINT NOT NULL,
+ c8 INT NOT NULL,
+ c9 BIGINT UNSIGNED NOT NULL,
+ c10 VARCHAR NOT NULL,
+ c11 FLOAT NOT NULL,
+ c12 DOUBLE NOT NULL,
+ c13 VARCHAR NOT NULL
+)
+STORED AS CSV
+WITH HEADER ROW
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
+
+# Basic example: distinct on the first column project the second one, and
+# order by the third
+# TODO: Note that the value for 'a' is actually degenerate, and while Postgres
+# returns the actual first match (4), DataFusion returns the last one, presumably
+# because `first_value` aggregation function doesn't have a dedicated `GroupsAccumulator`
Review Comment:
this doesn't make sense to me -- we should get the correct values, regardless of the presence of a groups accumulator
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -458,16 +467,19 @@ fn build_common_expr_project_plan(
/// the "intermediate" projection plan built in [build_common_expr_project_plan].
///
/// This is for those plans who don't keep its own output schema like `Filter` or `Sort`.
-fn build_recover_project_plan(schema: &DFSchema, input: LogicalPlan) -> LogicalPlan {
+fn build_recover_project_plan(
+ schema: &DFSchema,
+ input: LogicalPlan,
+) -> Result<LogicalPlan> {
let col_exprs = schema
.fields()
.iter()
.map(|field| Expr::Column(field.qualified_column()))
.collect();
- LogicalPlan::Projection(
- Projection::try_new(col_exprs, Arc::new(input))
- .expect("Cannot build projection plan from an invalid schema"),
- )
+ Ok(LogicalPlan::Projection(Projection::try_new(
Review Comment:
👍
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -458,16 +467,19 @@ fn build_common_expr_project_plan(
/// the "intermediate" projection plan built in [build_common_expr_project_plan].
///
/// This is for those plans who don't keep its own output schema like `Filter` or `Sort`.
-fn build_recover_project_plan(schema: &DFSchema, input: LogicalPlan) -> LogicalPlan {
+fn build_recover_project_plan(
+ schema: &DFSchema,
+ input: LogicalPlan,
+) -> Result<LogicalPlan> {
let col_exprs = schema
.fields()
.iter()
.map(|field| Expr::Column(field.qualified_column()))
.collect();
- LogicalPlan::Projection(
- Projection::try_new(col_exprs, Arc::new(input))
- .expect("Cannot build projection plan from an invalid schema"),
- )
+ Ok(LogicalPlan::Projection(Projection::try_new(
Review Comment:
👍
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,8 +238,17 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation epxressions if they have changed
+ // TODO: This should have really been identified above and handled in the `else` branch
+ let aggr_exprs = new_aggr_expr
+ .iter()
+ .zip(aggr_expr.iter())
+ .map(|(new_expr, old_expr)| {
+ new_expr.clone().alias_if_changed(old_expr.display_name()?)
+ })
+ .collect::<Result<Vec<Expr>>>()?;
// Since group_epxr changes, schema changes also. Use try_new method.
- Aggregate::try_new(Arc::new(new_input), new_group_expr, new_aggr_expr)
+ Aggregate::try_new(Arc::new(new_input), new_group_expr, aggr_exprs)
.map(LogicalPlan::Aggregate)
Review Comment:
Maybe @waynexia has some thoughts
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1381211603
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2132,9 +2181,100 @@ pub struct Limit {
/// Removes duplicate rows from the input
#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct Distinct {
+pub enum Distinct {
+ /// Plain `DISTINCT` referencing all selection expressions
+ All(Arc<LogicalPlan>),
+ /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
+ On(DistinctOn),
+}
+
+/// Removes duplicate rows from the input
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct DistinctOn {
+ /// The `DISTINCT ON` clause expression list
+ pub on_expr: Vec<Expr>,
+ /// The selected projection expression list
+ pub select_expr: Vec<Expr>,
+ /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause
Review Comment:
Yeah, `on_expr` and `sort_expr` should have a (partial) overlap (`select_expr` and `sort_expr` are arbitrary and different in general).
Condensing `on_expr` and `sort_expr` into one is doable, though it seems to me it would introduce more complexity, and make the logic harder to grasp. For starters `sort_expr` is not an exact super-set of `on_expr`, but instead wraps them inside `Expr::Sort`s to capture `asc` and `nulls_first`, but it can also be omitted (unlike `ON` exprs). On the other hand, these two are parsed at different places: first the `on_expr` is parsed in `select_to_plan`, and then `sort_expr` is extracted in `order_by`.
So in order to keep them both in one vec we'd have to: a) add an additional field to `DistinctOn` to track the length of the `ON` expressions as you mention (e.g. `on_expr_count`), b) if there are no actual `sort_expr` provided keep the `ON` expressions in them, otherwise validate that the wrapped sorting expressions match the existing `ON` expressions and replace the vector with the sorting expressions, and then c) inside `replace_distinct_aggregate` deconstruct that. That would mean take `on_expr_count` first expressions from `sort_expr`, and unwrapping the underlying expressions from `Expr::Sort` in case sort expression length is > `sort_expr`. In case sort expression length is equal to `on_expr_count` it could mean no `ORDER BY` was provided (no need for unwrapping), or the `ORDER BY` expressions match the `ON` expressions modulo the aforementioned sorting specifiers (`asc` and `nulls_first`).
This strikes me as less clear all in all, though if you'd prefer it I'll make that change.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1381211603
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -2132,9 +2181,100 @@ pub struct Limit {
/// Removes duplicate rows from the input
#[derive(Clone, PartialEq, Eq, Hash)]
-pub struct Distinct {
+pub enum Distinct {
+ /// Plain `DISTINCT` referencing all selection expressions
+ All(Arc<LogicalPlan>),
+ /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
+ On(DistinctOn),
+}
+
+/// Removes duplicate rows from the input
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct DistinctOn {
+ /// The `DISTINCT ON` clause expression list
+ pub on_expr: Vec<Expr>,
+ /// The selected projection expression list
+ pub select_expr: Vec<Expr>,
+ /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause
Review Comment:
Yeah, `on_expr` and `sort_expr` should have a (partial) overlap (`select_expr` and `sort_expr` are arbitrary and different in general).
Condensing `on_expr` and `sort_expr` into one is doable, though it seems to me it would introduce more complexity, and make the logic harder to grasp. For starters `sort_expr` is not an exact super-set of `on_expr`, but instead wraps them inside `Expr::Sort`s to capture `asc` and `nulls_first`, but it can also be omitted (unlike `ON` exprs). On the other hand, these two are parsed at different places: first the `on_expr` is parsed in `select_to_plan`, and then `sort_expr` is extracted in `order_by`.
So in order to keep them both in one vec we'd have to: a) add an additional field to `DistinctOn` to track the length of the `ON` expressions as you mention (e.g. `on_expr_count`), b) if there are no actual `sort_expr` provided keep the `ON` expressions in them, otherwise validate that the wrapped sorting expressions match the existing `ON` expressions and replace the vector with the sorting expressions, and then c) inside `replace_distinct_aggregate` deconstruct that. That would mean take `on_expr_count` first expressions from `sort_expr`, and unwrapping the underlying expressions from `Expr::Sort` in case sort expression length is > `on_expr_count`. In case sort expression length is equal to `on_expr_count` it could mean no `ORDER BY` was provided (no need for unwrapping), or the `ORDER BY` expressions match the `ON` expressions modulo the aforementioned sorting specifiers (`asc` and `nulls_first`).
This strikes me as less clear all in all, though if you'd prefer it I'll make that change.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1376005289
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
Review Comment:
All tests here were compared for accuracy against Postgres.
Note that, while the [docs](https://www.postgresql.org/docs/current/sql-select.html#SQL-DISTINCT) state that in the absence of the `ORDER BY` clause the output is unpredictable, in practice I've noticed it is stable on Postgres for the table below, whereas that isn't the case for DataFusion, hence why I omitted those tests for now.
##########
datafusion/optimizer/src/push_down_projection.rs:
##########
@@ -228,7 +228,7 @@ impl OptimizerRule for PushDownProjection {
// Gather all columns needed for expressions in this Aggregate
let mut new_aggr_expr = vec![];
for e in agg.aggr_expr.iter() {
- let column = Column::from_name(e.display_name()?);
+ let column = Column::from(e.display_name()?);
Review Comment:
This allows one to correctly re-construct a column whose qualifier has been crammed into the name by aliasing with `.display_name()`, which is what I'm doing in `replace_distinct_aggregate` in this PR. Not sure whether this makes sense in general though.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#issuecomment-1793475679
@alamb thanks for bearing with me here! I agree, ideally #8008 gets some kind of a resolution prior to merging this.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#issuecomment-1791460185
I am a little worried about the TODO and the seemingly incorrect results,
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1381591595
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
@@ -0,0 +1,129 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+statement ok
+CREATE EXTERNAL TABLE aggregate_test_100 (
+ c1 VARCHAR NOT NULL,
+ c2 TINYINT NOT NULL,
+ c3 SMALLINT NOT NULL,
+ c4 SMALLINT,
+ c5 INT,
+ c6 BIGINT NOT NULL,
+ c7 SMALLINT NOT NULL,
+ c8 INT NOT NULL,
+ c9 BIGINT UNSIGNED NOT NULL,
+ c10 VARCHAR NOT NULL,
+ c11 FLOAT NOT NULL,
+ c12 DOUBLE NOT NULL,
+ c13 VARCHAR NOT NULL
+)
+STORED AS CSV
+WITH HEADER ROW
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
+
+# Basic example: distinct on the first column project the second one, and
+# order by the third
+# TODO: Note that the value for 'a' is actually degenerate, and while Postgres
+# returns the actual first match (4), DataFusion returns the last one, presumably
+# because `first_value` aggregation function doesn't have a dedicated `GroupsAccumulator`
Review Comment:
Oh I see what this is, it's another case of unstable sort, i.e. the sorted rows that have equal ordering expressions evaluations do not preserve the original order of those rows:
```sql
❯ select c1, c2, c3 from test order by c1, c3;
+----+----+------+
| c1 | c2 | c3 |
+----+----+------+
| a | 5 | -101 |
| a | 4 | -101 |
| a | 1 | -85 |
| a | 3 | -72 |
| a | 1 | -56 |
...
```
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
@@ -0,0 +1,129 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+statement ok
+CREATE EXTERNAL TABLE aggregate_test_100 (
+ c1 VARCHAR NOT NULL,
+ c2 TINYINT NOT NULL,
+ c3 SMALLINT NOT NULL,
+ c4 SMALLINT,
+ c5 INT,
+ c6 BIGINT NOT NULL,
+ c7 SMALLINT NOT NULL,
+ c8 INT NOT NULL,
+ c9 BIGINT UNSIGNED NOT NULL,
+ c10 VARCHAR NOT NULL,
+ c11 FLOAT NOT NULL,
+ c12 DOUBLE NOT NULL,
+ c13 VARCHAR NOT NULL
+)
+STORED AS CSV
+WITH HEADER ROW
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
+
+# Basic example: distinct on the first column project the second one, and
+# order by the third
+# TODO: Note that the value for 'a' is actually degenerate, and while Postgres
+# returns the actual first match (4), DataFusion returns the last one, presumably
+# because `first_value` aggregation function doesn't have a dedicated `GroupsAccumulator`
Review Comment:
Oh I see what this is, it's another case of unstable sort, i.e. the sorted rows that have equal ordering expressions evaluations do not preserve the original order of those rows:
```sql
❯ select c1, c2, c3 from test order by c1, c3;
+----+----+------+
| c1 | c2 | c3 |
+----+----+------+
| a | 5 | -101 |
| a | 4 | -101 |
| a | 1 | -85 |
| a | 3 | -72 |
| a | 1 | -56 |
...
```
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#issuecomment-1805327473
Ok, the last commit (8ebdc13cee54bdd10003b3d7549a3eb7fe157da7) resolves #8008, so I think this might actually be mergable now.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1389432940
##########
datafusion/proto/proto/datafusion.proto:
##########
@@ -480,7 +488,8 @@ message Not {
message AliasNode {
LogicalExprNode expr = 1;
- string alias = 2;
+ repeated OwnedTableReference relation = 2;
+ string alias = 3;
Review Comment:
Good point, committed suggestion.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "jonahgao (via GitHub)" <gi...@apache.org>.
jonahgao commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1389315926
##########
datafusion/proto/proto/datafusion.proto:
##########
@@ -480,7 +488,8 @@ message Not {
message AliasNode {
LogicalExprNode expr = 1;
- string alias = 2;
+ repeated OwnedTableReference relation = 2;
+ string alias = 3;
Review Comment:
```suggestion
string alias = 2;
repeated OwnedTableReference relation = 3;
```
Reusing old tag numbers may be dangerous.
References:
- [Don’t Re-use a Tag Number](https://protobuf.dev/programming-guides/dos-donts/#reuse-number)
- [Consequences of Reusing Field Numbers
](https://protobuf.dev/programming-guides/proto3/#consequences)
##########
datafusion/proto/proto/datafusion.proto:
##########
@@ -480,7 +488,8 @@ message Not {
message AliasNode {
LogicalExprNode expr = 1;
- string alias = 2;
+ repeated OwnedTableReference relation = 2;
+ string alias = 3;
Review Comment:
```suggestion
string alias = 2;
repeated OwnedTableReference relation = 3;
```
Reusing old tag numbers may be dangerous.
References:
- [Don’t Re-use a Tag Number](https://protobuf.dev/programming-guides/dos-donts/#reuse-number)
- [Consequences of Reusing Field Numbers
](https://protobuf.dev/programming-guides/proto3/#consequences)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#issuecomment-1806377857
Thank you @gruuya -- this is on my review queue
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1390468860
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,6 +238,16 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation expressions if they have changed
+ // TODO: This should be handled in `build_recover_project_plan` once qualified aliases
Review Comment:
Yup, removed and merged latest main.
Thanks for your patience on this one! :)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1382416566
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -60,6 +80,62 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
)?);
Ok(Some(aggregate))
}
+ LogicalPlan::Distinct(Distinct::On(DistinctOn {
+ select_expr,
+ on_expr,
+ sort_expr,
+ input,
+ ..
+ })) => {
+ // Construct the aggregation expression to be used to fetch the selected expressions.
+ let aggr_expr = select_expr
+ .iter()
+ .map(|e| {
+ Expr::AggregateFunction(AggregateFunction::new(
+ AggregateFunctionFunc::FirstValue,
+ vec![e.clone()],
+ false,
+ None,
+ sort_expr.clone(),
+ ))
+ })
+ .collect::<Vec<Expr>>();
+
+ // Build the aggregation plan
+ let plan = LogicalPlanBuilder::from(input.as_ref().clone())
+ .aggregate(on_expr.clone(), aggr_expr.to_vec())?
+ .build()?;
+
+ let plan = if let Some(sort_expr) = sort_expr {
+ // While sort expressions were used in the `FIRST_VALUE` aggregation itself above,
+ // this on it's own isn't enough to guarantee the proper output order of the grouping
+ // (`ON`) expression, so we need to sort those as well.
+ LogicalPlanBuilder::from(plan)
+ .sort(sort_expr[..on_expr.len()].to_vec())?
Review Comment:
Yeah, my thinking was that the `first_value` aggregation actually doesn't do any real sorting and is more like top-1, i.e. it only works on the incoming stream and picks up one row per-grouping set (`ON` clause) if it's a better match according to the sorting expressions:
https://github.com/apache/arrow-datafusion/blob/2af326a0bdbb60e7d7420089c7aa9ff22c7319ee/datafusion/physical-expr/src/aggregate/first_last.rs#L236-L244
The subsequent actual sort works on the grouped data, so it should have a much lower cardinality.
Whereas if we sorted prior to aggregation then, I think we would materialize all the data initially (whether in memory or by spilling) only to throw away most of it, but I could be wrong.
Granted I haven't done any perf testing between the two approaches for now, this is just from my reading of the code.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1382416566
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -60,6 +80,62 @@ impl OptimizerRule for ReplaceDistinctWithAggregate {
)?);
Ok(Some(aggregate))
}
+ LogicalPlan::Distinct(Distinct::On(DistinctOn {
+ select_expr,
+ on_expr,
+ sort_expr,
+ input,
+ ..
+ })) => {
+ // Construct the aggregation expression to be used to fetch the selected expressions.
+ let aggr_expr = select_expr
+ .iter()
+ .map(|e| {
+ Expr::AggregateFunction(AggregateFunction::new(
+ AggregateFunctionFunc::FirstValue,
+ vec![e.clone()],
+ false,
+ None,
+ sort_expr.clone(),
+ ))
+ })
+ .collect::<Vec<Expr>>();
+
+ // Build the aggregation plan
+ let plan = LogicalPlanBuilder::from(input.as_ref().clone())
+ .aggregate(on_expr.clone(), aggr_expr.to_vec())?
+ .build()?;
+
+ let plan = if let Some(sort_expr) = sort_expr {
+ // While sort expressions were used in the `FIRST_VALUE` aggregation itself above,
+ // this on it's own isn't enough to guarantee the proper output order of the grouping
+ // (`ON`) expression, so we need to sort those as well.
+ LogicalPlanBuilder::from(plan)
+ .sort(sort_expr[..on_expr.len()].to_vec())?
Review Comment:
Yeah, my thinking was that the `first_value` aggregation actually doesn't do any real sorting and is more like top-1, i.e. it only works on the incoming stream and picks up one row per-grouping set (`ON` clause) if it's a better match according to the sorting expressions:
https://github.com/apache/arrow-datafusion/blob/2af326a0bdbb60e7d7420089c7aa9ff22c7319ee/datafusion/physical-expr/src/aggregate/first_last.rs#L236-L244
The subsequent actual sort works on the grouped data, so it should have a much lower cardinality.
Whereas if we sorted prior to aggregation then we'd need to materialize all the data (whether in memory or by spilling), only to throw away most of it.
Granted I haven't done any perf testing between the two approaches for now, this is just from my reading of the code.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1381195039
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -32,6 +36,22 @@ use datafusion_expr::{Aggregate, Distinct, LogicalPlan};
/// ```text
/// SELECT a, b FROM tab GROUP BY a, b
/// ```
+///
+/// On the other hand, for a `DISTINCT ON` query the replacement is
+/// a bit more involved and effectively converts
+/// ```text
+/// SELECT DISTINCT ON (a) b FROM tab ORDER BY a DESC, c
+/// ```
+///
+/// into
+/// ```text
+/// SELECT b FROM (
+/// SELECT a, FIRST_VALUE(b ORDER BY a DESC, c) AS b
Review Comment:
Actually there is no partitioning involved in the transformed plans. In fact the query from docs returns the same results as the `DISTINCT ON` query:
```sql
❯ select c2 from (
select c1, first_value(c2 order by c1, c3) as c2
from test group by c1
)
order by c1;
+----+
| c2 |
+----+
| 5 |
| 4 |
| 2 |
| 1 |
| 3 |
+----+
5 rows in set. Query took 0.035 seconds.
❯ select distinct on (c1) c2 from test order by c1, c3;
+---------+
| test.c2 |
+---------+
| 5 |
| 4 |
| 2 |
| 1 |
| 3 |
+---------+
5 rows in set. Query took 0.034 seconds
```
##########
datafusion/optimizer/src/replace_distinct_aggregate.rs:
##########
@@ -32,6 +36,22 @@ use datafusion_expr::{Aggregate, Distinct, LogicalPlan};
/// ```text
/// SELECT a, b FROM tab GROUP BY a, b
/// ```
+///
+/// On the other hand, for a `DISTINCT ON` query the replacement is
+/// a bit more involved and effectively converts
+/// ```text
+/// SELECT DISTINCT ON (a) b FROM tab ORDER BY a DESC, c
+/// ```
+///
+/// into
+/// ```text
+/// SELECT b FROM (
+/// SELECT a, FIRST_VALUE(b ORDER BY a DESC, c) AS b
Review Comment:
Actually there is no partitioning involved in the transformed plans. In fact the query from docs returns the same results as the `DISTINCT ON` query:
```sql
❯ select c2 from (
select c1, first_value(c2 order by c1, c3) as c2
from test group by c1
)
order by c1;
+----+
| c2 |
+----+
| 5 |
| 4 |
| 2 |
| 1 |
| 3 |
+----+
5 rows in set. Query took 0.035 seconds.
❯ select distinct on (c1) c2 from test order by c1, c3;
+---------+
| test.c2 |
+---------+
| 5 |
| 4 |
| 2 |
| 1 |
| 3 |
+---------+
5 rows in set. Query took 0.034 seconds
```
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1379645442
##########
datafusion/sqllogictest/test_files/distinct_on.slt:
##########
Review Comment:
Thanks! Also good catch, the first example was throwing a panic; added handling and a test case for it.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement `DISTINCT ON` from Postgres [arrow-datafusion]
Posted by "gruuya (via GitHub)" <gi...@apache.org>.
gruuya commented on code in PR #7981:
URL: https://github.com/apache/arrow-datafusion/pull/7981#discussion_r1383818073
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -238,8 +238,17 @@ impl CommonSubexprEliminate {
let rewritten = pop_expr(&mut rewritten)?;
if affected_id.is_empty() {
+ // Alias aggregation epxressions if they have changed
+ // TODO: This should have really been identified above and handled in the `else` branch
+ let aggr_exprs = new_aggr_expr
+ .iter()
+ .zip(aggr_expr.iter())
+ .map(|(new_expr, old_expr)| {
+ new_expr.clone().alias_if_changed(old_expr.display_name()?)
+ })
+ .collect::<Result<Vec<Expr>>>()?;
// Since group_epxr changes, schema changes also. Use try_new method.
- Aggregate::try_new(Arc::new(new_input), new_group_expr, new_aggr_expr)
+ Aggregate::try_new(Arc::new(new_input), new_group_expr, aggr_exprs)
.map(LogicalPlan::Aggregate)
Review Comment:
Btw, I've now clarified this comment since it was misleading, i.e. the aliasing should really be done in `build_recover_project_plan` once qualified aliasing is enabled.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org