[FLINK-27618][sql] Flink supports CUME_DIST function #19727

luoyuxia · 2022-05-16T01:43:58Z

What is the purpose of the change

To make Flink support CUME_DIST function.

Brief change log

Introduce CUME_DIST in FlinkSqlOperatorTable.java
Introduce a new interface SizeBasedWindowFunction.java, which enables the implementation to get windowsize.
Add a new method setWindowSize in ...le-runtime/src/main/java/org/apache/flink/table/runtime/generated/AggsHandleFunction.java which enable to generate setWindowSize function in code gen phase.
In code gen phase for BatchExecOverAggregate.java, if the aggInfoList contains any function that implement SizeBasedWindowFunction, the method setWindowSize will really set window's size.
In different over window, call method setWindowSize to set the window size.
Introduce CumeDistAggFunction.java which implements SizeBasedWindowFunction for CUME_DIST.

Verifying this change

Test in OverAggregateITCase.scala

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs

flinkbot · 2022-05-16T01:45:39Z

CI report:

2d09a24 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

luoyuxia · 2022-07-04T03:48:10Z

@flinkbot run azure

luoyuxia · 2022-07-08T02:00:13Z

@flinkbot run azure

wuchong

Thank @luoyuxia for the contribution. I have left some comments.

wuchong · 2022-07-13T11:40:02Z

.../src/main/java/org/apache/flink/table/planner/expressions/DeclarativeExpressionResolver.java

        if (expression instanceof UnresolvedReferenceExpression) {
            UnresolvedReferenceExpression expr = (UnresolvedReferenceExpression) expression;
            String name = expr.getName();
+            if (function instanceof SizeBasedWindowFunction


wuchong · 2022-07-20T03:41:56Z

...main/java/org/apache/flink/table/planner/functions/aggfunctions/SizeBasedWindowFunction.java

+public interface SizeBasedWindowFunction {
+
+    /** The field for the window size. */
+    UnresolvedReferenceExpression windowSize();


A window function should have only a context-provided window size expression. So we don't need to let users define the expression. The interface can pre-define one. In addition, the expression can be a resolved one, to make it easy to use when code generation (don't need toWindowSizeExpr). For example:

default ResolvedExpression windowSizeAttribute() { return localRef("window_size", DataTypes.INT()); }

Then devs dont' need to implement this method and can directly be used just like method operand(i).

Really a good suggestion!

...anner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/AggCodeGenHelper.scala

...ink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggCodeGen.scala

...ntime/src/main/java/org/apache/flink/table/runtime/generated/AggsFunctionWithWindowSize.java

wuchong · 2022-07-20T13:13:57Z

...e-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/DistinctAggCodeGen.scala

+
+  override def setWindowSize(generator: ExprCodeGenerator): String = {
+    throw new TableException(
+      "Distinct shouldn't set window size, this is a bug, please file a issue.")


Is it possible that COUNT(DISTINCT) OVER ... may hit this?

No, currently, the dictinct + over is not supported, see the code.
Also, in this pr, it'll skip setWindowSize when it's for DistinctAggCodeGen.

wuchong · 2022-07-20T13:23:10Z

...er/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala

  }
+
+  @Test
+  def testCumeDist(): Unit = {


Please add additional tests to cover values of order-key containing duplicates.

luoyuxia · 2022-07-22T01:40:26Z

@wuchong Thanks for review. I have updated this pr.

wuchong · 2022-07-22T07:30:01Z

...ner/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala

    initialAggregateInformation(aggInfoList)

    // generates all methods body first to add necessary reuse code to context
+    val setWindowSizeCode = if (isWindowSizeNeeded) genSetWindowSize() else ""


Please follow the way of genRetract() to generate throw exception when window size is not needed.

Based on the current implementation, we will always call method setWindowSize, so we can't generate the code to throw exception . But I add some comments for this method genSetWindowSize.

wuchong · 2022-07-22T07:31:49Z

...table-runtime/src/main/java/org/apache/flink/table/runtime/generated/AggsHandleFunction.java

+     * Set window size for the aggregate function. Some aggregate functions may requires the size of
+     * current window to do calculation.
+     */


Add more explanation about the "window size", it may confuse devs the relationship with tumble/hop window size.

wuchong

LGTM

flinkbot added the component=TableSQL/API label May 16, 2022

luoyuxia force-pushed the FLINK-27617 branch 5 times, most recently from 05d76dd to 44e9bc4 Compare June 22, 2022 02:13

luoyuxia force-pushed the FLINK-27617 branch 3 times, most recently from 43e7fba to 6e3706f Compare June 29, 2022 07:05

luoyuxia changed the title ~~[FLINK-27617][sql] Flink supports CUME_DIST function~~ [FLINK-27618][sql] Flink supports CUME_DIST function Jul 4, 2022

flinkbot added component=<none> and removed component=TableSQL/API labels Jul 4, 2022

luoyuxia force-pushed the FLINK-27617 branch from 6e3706f to 433de6b Compare July 7, 2022 04:11

wuchong reviewed Jul 20, 2022

View reviewed changes

luoyuxia force-pushed the FLINK-27617 branch 5 times, most recently from e206652 to e94b0d2 Compare July 22, 2022 01:38

wuchong reviewed Jul 22, 2022

View reviewed changes

luoyuxia and others added 2 commits July 22, 2022 19:37

[FLINK-27617][sql] Flink supports CUME_DIST function

471dd65

fix comments

2d09a24

luoyuxia force-pushed the FLINK-27617 branch from e94b0d2 to 2d09a24 Compare July 22, 2022 12:31

wuchong approved these changes Jul 22, 2022

View reviewed changes

wuchong merged commit 8017a41 into apache:master Jul 23, 2022

huangxiaofeng10047 pushed a commit to huangxiaofeng10047/flink that referenced this pull request Nov 3, 2022

[FLINK-27618][sql] Flink SQL supports CUME_DIST function (apache#19727)

babfc78

[FLINK-27618][sql] Flink supports CUME_DIST function #19727

[FLINK-27618][sql] Flink supports CUME_DIST function #19727

Uh oh!

Conversation

luoyuxia commented May 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented May 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

luoyuxia commented Jul 4, 2022

Uh oh!

luoyuxia commented Jul 8, 2022

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

luoyuxia Jul 22, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

luoyuxia commented Jul 22, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

luoyuxia commented May 16, 2022 •

edited

Loading

flinkbot commented May 16, 2022 •

edited

Loading

luoyuxia Jul 22, 2022 •

edited

Loading