feat(stream): add two-phase stateless simple approx percentile

[kwannoel]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

#17531

We add stateless simple approx percentile executors (local and global). This is the barebones version, without any caching, and which only works when its used alone without other aggregations.

Subsequent PRs will take care of:

• Cache
• Keyed Merge (incl. changes to simple agg executor, to flush once per epoch).
• Performance test
• Fuzz test (test relative error vs percentile_cont vs shuffle approx percentile)

LocalApproxPercentile is stateless, and just construct the buckets and corresponding counts. For negative numbers we need to get the absolute value, and log that instead. Then we pass the sign in a separate column. The schema will look like:

sign	bucket_id	count
-1	1	100
0	doesn't matter	10
1	1	10

GlobalApproxPercentile stores the bucket_ids, prefixed with a sign. On barrier, we will iterate over all buckets, and output the approximated percentile.

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

This is an experimental feature. The interface may change in the future. Only Streaming Approx percentile is supported at the moment.

Approx Percentile is an aggregation with the following interface:

approx_percentile(DOUBLE percentile [, DOUBLE relative_error]) within group (order by percentile_column)

• percentile refers to the percentile to approximate. For example, 50th, 90th will be 0.5, 0.9 respectively.
• relative_error refers to how far the approximated percentile value can stray from the actual percentile value. If unspecified, it will default to 0.01 (1%).
• percentile_column refers to the column we will maintain the approx percentile for. It must be of a numeric type.

[github-actions]

license-eye has totally checked 5253 files.

Valid	Invalid	Ignored	Fixed
2253	2	2998	0

Click to see the invalid file list

• src/stream/src/executor/approx_percentile/global.rs
• src/stream/src/executor/keyed_merge.rs

[github-actions]

license-eye has totally checked 5257 files.

Valid	Invalid	Ignored	Fixed
2256	2	2999	0

Click to see the invalid file list

• src/stream/src/executor/approx_percentile/global.rs
• src/stream/src/executor/keyed_merge.rs

[github-actions]

license-eye has totally checked 5259 files.

Valid	Invalid	Ignored	Fixed
2259	1	2999	0

Click to see the invalid file list

• src/stream/src/executor/keyed_merge.rs

[github-actions]

license-eye has totally checked 5259 files.

Valid	Invalid	Ignored	Fixed
2259	1	2999	0

Click to see the invalid file list

• src/stream/src/executor/keyed_merge.rs

[kwannoel]

TBH I'm not totally sure about this design decision.
The alternative is to split into 3 tables:

• zeros
• neg
• pos

Then we don't have to use reverse iteration.

The current design is to group everyting in a single table, and add a sign column.
Because everything is ascending, this means that when iterating negative values, the order will be:

- 1
- 10000
- 10000000

When we want this instead:

- 10000000
- 10000
- 1

So we will need to use a separate reverse iterator for the negative values. This approach is simpler, but may be less efficient.

[chenzl25]

Can we just use an empty prefix to scan the state table? The order should start from neg to pos.

[kwannoel]

No it doesn't work. Because we aren't just iterating on negative values.

Consider the following is in the state table (with the count column elided)

sign	bucket_id
-1	-3
-1	-1
-1	1
-1	2
0	0
1	1
1	2

Our iteration order will be:

(-1, -3), (-1,-1), (-1, 1), (-1, 2), (0,0), (+1, 1), (+1, 2)

But we actually want

(-1, 2), (-1,1), (-1, -1), (-1, -3), (0,0), (+1, 1), (+1, 2)

Because as an example:

-(base^2) < -(base ^1)

[chenzl25]

I see. Actually, we can do a math trick here, for example, encoding bucket_id as - bucket_id for sign with -1. Then we will get the order we want.

[kwannoel]

I see. Actually, we can do a math trick here, for example, encoding bucket_id as - bucket_id for sign with -1. Then we will get the order we want.

That's a good idea for optimization, but I intend to keep this suggestion as a future optimization, rather than change it in this PR. Currently this is just a few lines in one location:

        for keyed_row in bucket_state_table
            .rev_iter_with_prefix(&[Datum::None; 0], &neg_bounds, PrefetchOptions::default())
            .await?

Adding -1 sign, means at all areas where we encode / decode, we will need to add the logic to reverse it. And probably have to add comments there to explain why as well. Here we can just add some comments in a single spot.

Once we add caching, I think we won't do as much state table iter anymore, and so it will mitigate any performance hit from the reverse iteration of negative values.

If we add an abstraction over state_table and cache in the caching PR, we can consider adding this optimization there.

[kwannoel]

Bump, PTAL, the tests have passed.

[chenzl25]

It seems we don't have any shuffle between the global approx percentile and the local approx percentile?

[kwannoel]

Didn't quite get it. We do have shuffle between local -> global approx percentile. May take a look at the output inside agg.yaml. There we have:

      ├─StreamGlobalApproxPercentile { quantile: 0.8:Float64, relative_error: 0.01:Float64 }
      │ └─StreamExchange { dist: Single }
      │   └─StreamLocalApproxPercentile { percentile_col: $expr1, quantile: 0.8:Float64, relative_error: 0.01:Float64 }

[chenzl25]

Oh, I see it. My main branch is out of date.

[chenzl25]

BTW, I think StreamMerge inputs could be changed from binary inputs to multi inputs like StreamUnion. Maybe the in next PR.

[chenzl25]

LGTM!