You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "LadyForest (via GitHub)" <gi...@apache.org> on 2024/01/09 15:08:24 UTC
[PR] [FLINK-31788][table] TableAggregateFunction supports emitUpdateWithRetract [flink]
LadyForest opened a new pull request, #24051:
URL: https://github.com/apache/flink/pull/24051
## What is the purpose of the change
This pull request supports implementing`emitUpdateWithRetract` for user-defined `TableAggregateFunction`, which is a missing feature since blink planner became the default planner.
## Brief change log
- **Planning phase** Check whether the `emitUpdateWithRetract` method is implemented. If so, the `TableAggsHandleFunction#emitValue` method should invoke `TableAggregateFunction#emitUpdateWithRetract` instead of `TableAggregateFunction#emitValue`. Meanwhile, change the collector type to `RetractableCollector` and implement `RetractableCollector#retract`.
- **Runtime phase** Do not invoke `emitValue` to retract previously sent unchanged data.
## Verifying this change
This change added tests and can be verified as follows:
- `TableAggregateITCase#testFlagAggregateWithOrWithoutIncrementalUpdate`
## 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): yes
- 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? no because it's a feature already mentioned in doc
- If yes, how is the feature documented? not applicable
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31788][table] TableAggregateFunction supports emitUpdateWithRetract [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24051:
URL: https://github.com/apache/flink/pull/24051#issuecomment-1883240330
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a923785e4b94a628893b60cff0748b78fa0b6f7b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a923785e4b94a628893b60cff0748b78fa0b6f7b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a923785e4b94a628893b60cff0748b78fa0b6f7b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31788][table] TableAggregateFunction supports emitUpdateWithRetract [flink]
Posted by "LadyForest (via GitHub)" <gi...@apache.org>.
LadyForest merged PR #24051:
URL: https://github.com/apache/flink/pull/24051
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31788][table] TableAggregateFunction supports emitUpdateWithRetract [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #24051:
URL: https://github.com/apache/flink/pull/24051#discussion_r1448431727
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala:
##########
@@ -83,7 +87,8 @@ class ImperativeAggCodeGen(
hasNamespace: Boolean,
mergedAccOnHeap: Boolean,
mergedAccExternalType: DataType,
- inputFieldCopy: Boolean)
+ inputFieldCopy: Boolean,
+ val isIncrementalUpdateNeeded: Boolean)
Review Comment:
nit: remove `val` here because the default attribute in construct is `val`.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala:
##########
@@ -488,10 +493,14 @@ class ImperativeAggCodeGen(
}
if (needEmitValue) {
+ val (emitMethod, collectorClass) =
+ if (isIncrementalUpdateNeeded)
+ (UserDefinedFunctionHelper.TABLE_AGGREGATE_EMIT_RETRACT, classOf[RetractableCollector[_]])
Review Comment:
nit: add braces for `If` block to resolve the warning
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala:
##########
@@ -447,6 +473,23 @@ class AggsHandlerCodeGenerator(
val recordInputName = newName("recordInput")
val recordToRowDataCode = genRecordToRowData(aggExternalType, recordInputName)
+ // for emitUpdateWithRetract, the collector needs to implement RetractableCollector
+ // and override retract method
+ val (collectorClassName, collectorRetractCode) =
+ if (isIncrementalUpdateNeeded)
Review Comment:
ditto
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedTableFunctions.java:
##########
@@ -166,4 +169,91 @@ public void eval(String str) {
str.length() + 5));
}
}
+
+ /** Mutable accumulator of structured type for the table aggregate function. */
Review Comment:
nit: add class JavaUserDefinedTableAggFunctions.java to restore these function (+1), or just using UserDefinedTableAggFunctions.scala.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31788][table] TableAggregateFunction supports emitUpdateWithRetract [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #24051:
URL: https://github.com/apache/flink/pull/24051#discussion_r1448429999
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala:
##########
@@ -488,10 +493,14 @@ class ImperativeAggCodeGen(
}
if (needEmitValue) {
+ val (emitMethod, collectorClass) =
+ if (isIncrementalUpdateNeeded)
+ (UserDefinedFunctionHelper.TABLE_AGGREGATE_EMIT_RETRACT, classOf[RetractableCollector[_]])
Review Comment:
nit: add braces for `If` block to resolve the warning
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31788][table] TableAggregateFunction supports emitUpdateWithRetract [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #24051:
URL: https://github.com/apache/flink/pull/24051#discussion_r1448449006
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala:
##########
@@ -447,6 +473,23 @@ class AggsHandlerCodeGenerator(
val recordInputName = newName("recordInput")
val recordToRowDataCode = genRecordToRowData(aggExternalType, recordInputName)
+ // for emitUpdateWithRetract, the collector needs to implement RetractableCollector
+ // and override retract method
+ val (collectorClassName, collectorRetractCode) =
+ if (isIncrementalUpdateNeeded)
Review Comment:
ditto
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org