Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(frontend): support single phase approx percentile in batch #18083

Merged
merged 5 commits into from
Aug 19, 2024
+157 −14
Merged

feat(frontend): support single phase approx percentile in batch #18083

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
4f09732
only support single phase approx_percentile batch agg
kwannoel Aug 19, 2024
42c35b5
add e2e tests
kwannoel Aug 19, 2024
35c7e10
add delete tests
kwannoel Aug 19, 2024
b4f290c
Merge branch 'main' into kwannoel/approx-percentile-batch
kwannoel Aug 19, 2024
7153cf7
Merge branch 'main' into kwannoel/approx-percentile-batch
kwannoel Aug 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions e2e_test/streaming/aggregate/two_phase_approx_percentile_merge_stateless_agg.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,26 @@ select * from m1;
----
-963.1209598593477 0.5501000000000007 0.00009999833511933609 3965.1209598593477

query I
select
approx_percentile(0.01, 0.01) within group (order by p_col) as p01,
round(sum(p_col)::numeric, 2) as s,
approx_percentile(0.5, 0.01) within group (order by p_col) as p50,
count(*)::double + approx_percentile(0.99, 0.01) within group (order by p_col) as p99
from t;
----
-963.1209598593477 0.55 0.00009999833511933609 3965.1209598593477

query I
select
approx_percentile(0.01, 0.01) within group (order by p_col) as p01,
round(sum(p_col)::numeric, 2) as s,
approx_percentile(0.5, 0.01) within group (order by p_col) as p50,
count(*)::double + approx_percentile(0.99, 0.01) within group (order by p_col) as p99
from t group by grp_col;
----
-963.1209598593477 0.55 0.00009999833511933609 3965.1209598593477

query I
select
percentile_cont(0.01) within group (order by p_col) as p01,
Expand Down
16 changes: 16 additions & 0 deletions src/frontend/planner_test/tests/testdata/input/agg.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1013,20 +1013,23 @@
SELECT approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1) from t;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
- name: test simple approx_percentile with other simple aggs
sql: |
CREATE TABLE t (v1 int);
SELECT approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1), sum(v1) from t;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
- name: test simple approx_percentile with other simple aggs (sum, count)
sql: |
CREATE TABLE t (v1 int);
SELECT sum(v1) as s1, approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1), sum(v1) as s2, count(v1) from t;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
- name: test simple approx_percentile with duplicate approx_percentile
sql: |
Expand All @@ -1041,32 +1044,45 @@
SELECT approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1) as x, approx_percentile(0.5, 0.01) WITHIN GROUP (order by v2) as y from t;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
- name: test simple approx_percentile with different approx_percentile interleaved with stateless simple aggs
sql: |
CREATE TABLE t (v1 int, v2 int);
SELECT sum(v1) as s1, approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1) as x, count(*), sum(v2) + approx_percentile(0.9, 0.01) WITHIN GROUP (order by v2) as y from t;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
- name: test simple approx_percentile with duplicated approx_percentile interleaved with stateless simple aggs
sql: |
CREATE TABLE t (v1 int, v2 int);
SELECT sum(v1) as s1, approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1) as x, count(*), sum(v2) + approx_percentile(0.5, 0.01) WITHIN GROUP (order by v2) as y from t;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
- name: test simple approx_percentile with descending order
sql: |
CREATE TABLE t (v1 int, v2 int);
SELECT sum(v1) as s1, approx_percentile(0.2, 0.01) WITHIN GROUP (order by v1 desc) from t;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
- name: test simple approx_percentile with different approx_percentile interleaved with stateless + stateful simple aggs
sql: |
CREATE TABLE t (v1 int, v2 int);
SELECT sum(v1) as s1, approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1) as x, count(*), max(v2) as m2, approx_percentile(0.5, 0.01) WITHIN GROUP (order by v2) as y from t;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
- name: test hash approx_percentile
sql: |
CREATE TABLE t (v1 int, v2 int);
SELECT approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1) from t group by v2;
expected_outputs:
- logical_plan
- batch_plan
- stream_plan
66 changes: 66 additions & 0 deletions src/frontend/planner_test/tests/testdata/output/agg.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1888,6 +1888,11 @@
└─LogicalAgg { aggs: [approx_percentile($expr1)] }
└─LogicalProject { exprs: [t.v1::Float64 as $expr1] }
└─LogicalScan { table: t, columns: [t.v1, t._row_id] }
batch_plan: |-
BatchSimpleAgg { aggs: [approx_percentile($expr1)] }
└─BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [t.v1::Float64 as $expr1] }
└─BatchScan { table: t, columns: [t.v1], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [approx_percentile], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
└─StreamGlobalApproxPercentile { quantile: 0.5:Float64, relative_error: 0.01:Float64 }
Expand All @@ -1904,6 +1909,11 @@
└─LogicalAgg { aggs: [approx_percentile($expr1), sum(t.v1)] }
└─LogicalProject { exprs: [t.v1::Float64 as $expr1, t.v1] }
└─LogicalScan { table: t, columns: [t.v1, t._row_id] }
batch_plan: |-
BatchSimpleAgg { aggs: [approx_percentile($expr1), sum(t.v1)] }
└─BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [t.v1::Float64 as $expr1, t.v1] }
└─BatchScan { table: t, columns: [t.v1], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [approx_percentile, sum], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
└─StreamRowMerge { output: [approx_percentile:Float64, sum(sum(t.v1)):Int64] }
Expand All @@ -1928,6 +1938,12 @@
└─LogicalAgg { aggs: [sum(t.v1), approx_percentile($expr1), count(t.v1)] }
└─LogicalProject { exprs: [t.v1, t.v1::Float64 as $expr1] }
└─LogicalScan { table: t, columns: [t.v1, t._row_id] }
batch_plan: |-
BatchProject { exprs: [sum(t.v1), approx_percentile($expr1), sum(t.v1), count(t.v1)] }
└─BatchSimpleAgg { aggs: [sum(t.v1), approx_percentile($expr1), count(t.v1)] }
└─BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [t.v1, t.v1::Float64 as $expr1] }
└─BatchScan { table: t, columns: [t.v1], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [s1, approx_percentile, s2, count], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
└─StreamProject { exprs: [sum(sum(t.v1)), approx_percentile, sum(sum(t.v1)), sum0(count(t.v1))] }
Expand Down Expand Up @@ -1970,6 +1986,11 @@
└─LogicalAgg { aggs: [approx_percentile($expr1), approx_percentile($expr2)] }
└─LogicalProject { exprs: [t.v1::Float64 as $expr1, t.v2::Float64 as $expr2] }
└─LogicalScan { table: t, columns: [t.v1, t.v2, t._row_id] }
batch_plan: |-
BatchSimpleAgg { aggs: [approx_percentile($expr1), approx_percentile($expr2)] }
└─BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [t.v1::Float64 as $expr1, t.v2::Float64 as $expr2] }
└─BatchScan { table: t, columns: [t.v1, t.v2], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [x, y], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
└─StreamRowMerge { output: [approx_percentile:Float64, approx_percentile:Float64] }
Expand All @@ -1994,6 +2015,12 @@
└─LogicalAgg { aggs: [sum(t.v1), approx_percentile($expr1), count, sum(t.v2), approx_percentile($expr2)] }
└─LogicalProject { exprs: [t.v1, t.v1::Float64 as $expr1, t.v2, t.v2::Float64 as $expr2] }
└─LogicalScan { table: t, columns: [t.v1, t.v2, t._row_id] }
batch_plan: |-
BatchProject { exprs: [sum(t.v1), approx_percentile($expr1), count, (sum(t.v2)::Float64 + approx_percentile($expr2)) as $expr3] }
└─BatchSimpleAgg { aggs: [sum(t.v1), approx_percentile($expr1), count, sum(t.v2), approx_percentile($expr2)] }
└─BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [t.v1, t.v1::Float64 as $expr1, t.v2, t.v2::Float64 as $expr2] }
└─BatchScan { table: t, columns: [t.v1, t.v2], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [s1, x, count, y], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
└─StreamProject { exprs: [sum(sum(t.v1)), approx_percentile, sum0(count), (sum(sum(t.v2))::Float64 + approx_percentile) as $expr3] }
Expand Down Expand Up @@ -2026,6 +2053,12 @@
└─LogicalAgg { aggs: [sum(t.v1), approx_percentile($expr1), count, sum(t.v2), approx_percentile($expr2)] }
└─LogicalProject { exprs: [t.v1, t.v1::Float64 as $expr1, t.v2, t.v2::Float64 as $expr2] }
└─LogicalScan { table: t, columns: [t.v1, t.v2, t._row_id] }
batch_plan: |-
BatchProject { exprs: [sum(t.v1), approx_percentile($expr1), count, (sum(t.v2)::Float64 + approx_percentile($expr2)) as $expr3] }
└─BatchSimpleAgg { aggs: [sum(t.v1), approx_percentile($expr1), count, sum(t.v2), approx_percentile($expr2)] }
└─BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [t.v1, t.v1::Float64 as $expr1, t.v2, t.v2::Float64 as $expr2] }
└─BatchScan { table: t, columns: [t.v1, t.v2], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [s1, x, count, y], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
└─StreamProject { exprs: [sum(sum(t.v1)), approx_percentile, sum0(count), (sum(sum(t.v2))::Float64 + approx_percentile) as $expr3] }
Expand Down Expand Up @@ -2058,6 +2091,11 @@
└─LogicalAgg { aggs: [sum(t.v1), approx_percentile($expr1)] }
└─LogicalProject { exprs: [t.v1, t.v1::Float64 as $expr1] }
└─LogicalScan { table: t, columns: [t.v1, t.v2, t._row_id] }
batch_plan: |-
BatchSimpleAgg { aggs: [sum(t.v1), approx_percentile($expr1)] }
└─BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [t.v1, t.v1::Float64 as $expr1] }
└─BatchScan { table: t, columns: [t.v1], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [s1, approx_percentile], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
└─StreamRowMerge { output: [sum(sum(t.v1)):Int64, approx_percentile:Float64] }
Expand All @@ -2082,6 +2120,11 @@
└─LogicalAgg { aggs: [sum(t.v1), approx_percentile($expr1), count, max(t.v2), approx_percentile($expr2)] }
└─LogicalProject { exprs: [t.v1, t.v1::Float64 as $expr1, t.v2, t.v2::Float64 as $expr2] }
└─LogicalScan { table: t, columns: [t.v1, t.v2, t._row_id] }
batch_plan: |-
BatchSimpleAgg { aggs: [sum(t.v1), approx_percentile($expr1), count, max(t.v2), approx_percentile($expr2)] }
└─BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [t.v1, t.v1::Float64 as $expr1, t.v2, t.v2::Float64 as $expr2] }
└─BatchScan { table: t, columns: [t.v1, t.v2], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [s1, x, count, m2, y], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
└─StreamRowMerge { output: [sum(sum(t.v1)):Int64, approx_percentile:Float64, sum0(count):Int64, max(max(t.v2)):Int32, approx_percentile:Float64] }
Expand All @@ -2103,3 +2146,26 @@
└─StreamHashAgg { group_key: [$expr5], aggs: [sum(t.v1), count, max(t.v2)] }
└─StreamProject { exprs: [t.v1, t.v1::Float64 as $expr3, t.v2, t.v2::Float64 as $expr4, t._row_id, Vnode(t._row_id) as $expr5] }
└─StreamTableScan { table: t, columns: [t.v1, t.v2, t._row_id], stream_scan_type: ArrangementBackfill, stream_key: [t._row_id], pk: [_row_id], dist: UpstreamHashShard(t._row_id) }
- name: test hash approx_percentile
sql: |
CREATE TABLE t (v1 int, v2 int);
SELECT approx_percentile(0.5, 0.01) WITHIN GROUP (order by v1) from t group by v2;
logical_plan: |-
LogicalProject { exprs: [approx_percentile($expr1)] }
└─LogicalAgg { group_key: [t.v2], aggs: [approx_percentile($expr1)] }
└─LogicalProject { exprs: [t.v2, t.v1::Float64 as $expr1] }
└─LogicalScan { table: t, columns: [t.v1, t.v2, t._row_id] }
batch_plan: |-
BatchExchange { order: [], dist: Single }
└─BatchProject { exprs: [approx_percentile($expr1)] }
└─BatchHashAgg { group_key: [t.v2], aggs: [approx_percentile($expr1)] }
└─BatchExchange { order: [], dist: HashShard(t.v2) }
└─BatchProject { exprs: [t.v2, t.v1::Float64 as $expr1] }
└─BatchScan { table: t, columns: [t.v1, t.v2], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [approx_percentile, t.v2(hidden)], stream_key: [t.v2], pk_columns: [t.v2], pk_conflict: NoCheck }
└─StreamProject { exprs: [approx_percentile($expr1), t.v2] }
└─StreamHashAgg { group_key: [t.v2], aggs: [approx_percentile($expr1), count] }
└─StreamExchange { dist: HashShard(t.v2) }
└─StreamProject { exprs: [t.v2, t.v1::Float64 as $expr1, t._row_id] }
└─StreamTableScan { table: t, columns: [t.v1, t.v2, t._row_id], stream_scan_type: ArrangementBackfill, stream_key: [t._row_id], pk: [_row_id], dist: UpstreamHashShard(t._row_id) }
11 changes: 10 additions & 1 deletion src/frontend/src/optimizer/plan_node/batch_simple_agg.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.

use risingwave_expr::aggregate::{AggKind, PbAggKind};
use risingwave_pb::batch_plan::plan_node::NodeBody;
use risingwave_pb::batch_plan::SortAggNode;

Expand Down Expand Up @@ -51,7 +52,15 @@ impl BatchSimpleAgg {
}

pub(crate) fn can_two_phase_agg(&self) -> bool {
self.core.can_two_phase_agg() && self.two_phase_agg_enabled()
self.core.can_two_phase_agg()
&& self
.core
// Ban two phase approx percentile.
.agg_calls
.iter()
.map(|agg_call| &agg_call.agg_kind)
.all(|agg_kind| !matches!(agg_kind, AggKind::Builtin(PbAggKind::ApproxPercentile)))
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is specific to batch, so we ban it here, instead of inside the core.can_two_phase_agg.

&& self.two_phase_agg_enabled()
}
}

Expand Down
Loading