You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "pnowojski (via GitHub)" <gi...@apache.org> on 2025/01/22 09:29:07 UTC
[PR] [FLINK-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
pnowojski opened a new pull request, #26051:
URL: https://github.com/apache/flink/pull/26051
## What is the purpose of the change
This PR:
- implements the `RowTimeDeduplicateKeepFirstRowFunction` using watermarks to avoid retracting previous results
- changes planner to take into account that both proc time and rowtime deduplicate in keep first row variants are append only
Thanks to that, planner can avoid costly operators like `SinkUpsertMaterializer` downstream the whole query can be append-only - removing a need for retracting/upserting results from the output table.
Currently there is no variant for the async state backend.
## Verifying this change
This adds new tests and is also covered by existing ITCases.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (**yes** / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski merged PR #26051:
URL: https://github.com/apache/flink/pull/26051
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1946598188
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java:
##########
@@ -339,25 +362,39 @@ OneInputStreamOperator<RowData, RowData> createDeduplicateOperator() {
}
} else {
if (isAsyncStateEnabled()) {
- AsyncStateRowTimeDeduplicateFunction processFunction =
- new AsyncStateRowTimeDeduplicateFunction(
- rowTypeInfo,
- stateRetentionTime,
- rowtimeIndex,
- generateUpdateBefore,
- generateInsert(),
- keepLastRow);
- return new AsyncKeyedProcessOperator<>(processFunction);
+ if (!keepLastRow && outputInsertOnly) {
+ checkState(canBeInsertOnly(config, keepLastRow));
Review Comment:
Generating plans and actually using those plans can happen quite far apart. Yes it doesn't make much sense to `checkState` this if we assume both generation and creating job graph happens with the same Flink version, but that's often not the case. With those `checkState` I wanted to add a bit of extra safety if in the future someones changes the code. Similarly how for example the dedupe operator itself makes assertions that the input is insert only.
Also they don't hurt.
But if you have stronger feelings, about dropping them I would be fine 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1951073995
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java:
##########
@@ -339,25 +362,39 @@ OneInputStreamOperator<RowData, RowData> createDeduplicateOperator() {
}
} else {
if (isAsyncStateEnabled()) {
- AsyncStateRowTimeDeduplicateFunction processFunction =
- new AsyncStateRowTimeDeduplicateFunction(
- rowTypeInfo,
- stateRetentionTime,
- rowtimeIndex,
- generateUpdateBefore,
- generateInsert(),
- keepLastRow);
- return new AsyncKeyedProcessOperator<>(processFunction);
+ if (!keepLastRow && outputInsertOnly) {
+ checkState(canBeInsertOnly(config, keepLastRow));
Review Comment:
Ahh, sorry I misunderstood you.
You mean why don't I add the reverse check state in the else branch below? The problem with that is that old compiled plans would fail that check state. I mean plans that were created before this PR, but are executed with this PR. `RankUtil.outputInsertOnlyInDeduplicate(config, keepLastRow)` could return `true` for them, while `outputInsertOnly` is false (absent value/null interpreted as false)
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1946730432
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.scala:
##########
@@ -139,14 +139,17 @@ class DeduplicateTest extends TableTestBase {
@Test
def testSimpleFirstRowOnRowtime(): Unit = {
+ // indirectly check output insert only via used SUM or SUM_RETRACT aggregation function
Review Comment:
The output type of the actual resource is then slightly different, whole explain instead of 3 individual resources for sql, ast and optimized exec plan, but I guess that doesn't matter. (please check the fixup to the expected output)
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2607039595
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "63f9ed19290ca306170c67664f67750ce8d6e8e3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "63f9ed19290ca306170c67664f67750ce8d6e8e3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 63f9ed19290ca306170c67664f67750ce8d6e8e3 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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2655806513
@flinkbot run azure
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2637506032
@Zakelly / @xuyangzhong I'm struggling with some failures in tests. `DeduplicateITCase#testFirstRowOnRowtime` fails for mini batch in a very weird way.
```
[ERROR] Run 3: DeduplicateITCase.testFirstRowOnRowtime:232
expected: List(
+I(1,1,Hi,1970-01-01T00:00:00.001),
+I(2,3,I am fine.,1970-01-01T00:00:00.003),
+I(3,5,Comment#2,1970-01-01T00:00:00.005),
+I(4,4,Comment#3,1970-01-01T00:00:00.004),
+U(3,4,Comment#2,1970-01-01T00:00:00.004),
-U(3,5,Comment#2,1970-01-01T00:00:00.005)) # <<<<<<<<<<< this is missing
but was: ArrayBuffer(
+I(1,1,Hi,1970-01-01T00:00:00.001),
+I(2,3,I am fine.,1970-01-01T00:00:00.003),
+I(3,5,Comment#2,1970-01-01T00:00:00.005),
+I(4,4,Comment#3,1970-01-01T00:00:00.004),
+U(3,4,Comment#2,1970-01-01T00:00:00.004))
```
The issue is that in the test with mini batch enabled, `StreamExecDeduplicate` ends up with `StreamExecDeduplicate#generateUpdateBefore` set to `false`.
I've tracked the problem down to some inconsistency in the configuration. In `FlinkChangelogModeInferenceProgram` I'm doing a check if the deduplication can be executed as insert only (line 238 https://github.com/apache/flink/pull/26051/files#diff-ddf95eb3949ab889e8e3bccdacb9f57553aac06657574fd6316ca17dd48321a1R238 ). One of the conditions is that mini batch must be disabled.
But the problem is that the value of `table.exec.mini-batch.enabled` is overwritten and inconsistent in the `TableConfig`:
note `configuration` and `rootConfiguration` having different values.
I've tracked it down to this code: https://github.com/apache/flink/blob/b1544e4e513d2b75b350c20dbb1c17a8232c22fd/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala#L106
introduced here: https://github.com/apache/flink/pull/23470
Because of that, during planning we encounter different configuration value for the `mini-batch.enabled` vs during execution. So the planner code decides to use `insert-only` code path, which doesn't requires `generateUpdateBefore` but but during runtime it's still using mini batch versions of the operators.
First of all it seems to me like this code in `StreamCommonSubGraphBasedOptimizer` is bugged, isn't it?
Secondly, that raised a question for me. Is it expected that a compiled plan created with mini-batch DISABLED, can it be later executed with mini-batch ENABLED? Or the other way?
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2608722972
@pnowojski Glad to see you here. @xuyangzhong is right, you have that guarantee. The same-key records and timers will happen in order of arrival. That is, when watermark advance, the timer will fire after all arrived records for that key finishes. And any records for that key arrive after the watermark advance will start processing after the timer finishes.
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2655376288
BTW, the ci seems failing during compiling
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1951860691
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java:
##########
@@ -339,25 +362,39 @@ OneInputStreamOperator<RowData, RowData> createDeduplicateOperator() {
}
} else {
if (isAsyncStateEnabled()) {
- AsyncStateRowTimeDeduplicateFunction processFunction =
- new AsyncStateRowTimeDeduplicateFunction(
- rowTypeInfo,
- stateRetentionTime,
- rowtimeIndex,
- generateUpdateBefore,
- generateInsert(),
- keepLastRow);
- return new AsyncKeyedProcessOperator<>(processFunction);
+ if (!keepLastRow && outputInsertOnly) {
+ checkState(canBeInsertOnly(config, keepLastRow));
Review Comment:
Maybe I didn't make it clear, I just want to add one more check in the `if` branch like:
```
if (isMiniBatchEnabled()) {
checkState(!RankUtil.outputInsertOnlyInDeduplicate(config, keepLastRow));
checkState(!outputInsertOnly);
...
} else {
...
}
```
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2693914171
Thanks for the reviews. Everything is finally green. Sorry for the delay, but I had to deal with some other things in the mean time.
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1944682832
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java:
##########
@@ -339,25 +362,39 @@ OneInputStreamOperator<RowData, RowData> createDeduplicateOperator() {
}
} else {
if (isAsyncStateEnabled()) {
- AsyncStateRowTimeDeduplicateFunction processFunction =
- new AsyncStateRowTimeDeduplicateFunction(
- rowTypeInfo,
- stateRetentionTime,
- rowtimeIndex,
- generateUpdateBefore,
- generateInsert(),
- keepLastRow);
- return new AsyncKeyedProcessOperator<>(processFunction);
+ if (!keepLastRow && outputInsertOnly) {
+ checkState(canBeInsertOnly(config, keepLastRow));
Review Comment:
nit: can we need to double check the field `outputInsertOnly`?
The same goes for the following.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalRank.scala:
##########
@@ -109,28 +109,34 @@ class StreamPhysicalRank(
.item("select", getRowType.getFieldNames.mkString(", "))
}
- private def getDeduplicateDescription(isRowtime: Boolean, isLastRow: Boolean): String = {
+ private def getDeduplicateDescription(
+ isRowtime: Boolean,
+ isLastRow: Boolean,
+ insertOnly: Boolean): String = {
val fieldNames = getRowType.getFieldNames
val orderString = if (isRowtime) "ROWTIME" else "PROCTIME"
val keep = if (isLastRow) "LastRow" else "FirstRow"
- s"Deduplicate(keep=[$keep], key=[${partitionKey.toArray.map(fieldNames.get).mkString(", ")}], order=[$orderString])"
+ s"Deduplicate(keep=[$keep], key=[${partitionKey.toArray.map(fieldNames.get).mkString(", ")}], order=[$orderString], outputInsertOnly=[$insertOnly])"
}
override def translateToExecNode(): ExecNode[_] = {
val generateUpdateBefore = ChangelogPlanUtils.generateUpdateBefore(this)
if (RankUtil.canConvertToDeduplicate(this)) {
val keepLastRow = RankUtil.keepLastDeduplicateRow(orderKey)
+ val tableConfig = unwrapTableConfig(this)
+ val outputInsertOnly = StreamExecDeduplicate.canBeInsertOnly(tableConfig, keepLastRow)
Review Comment:
nit: use `val outputInsertOnly = ChangelogPlanUtils.isInsertOnly(this)`?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -220,6 +221,38 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
val providedTrait = ModifyKindSetTrait.INSERT_ONLY
createNewNode(rel, children, providedTrait, requiredTrait, requester)
+ case rank: StreamPhysicalRank if RankUtil.isDeduplication(rank) =>
+ val children = visitChildren(rel, ModifyKindSetTrait.ALL_CHANGES)
+ val tableConfig = unwrapTableConfig(rank)
+
+ // if the rank is deduplication and can be executed as insert-only, forward that information
+ val insertOnly = children
Review Comment:
nit: using `val insertOnly = children.forall(ChangelogPlanUtils.isInsertOnly)` to resolve idea warning
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeduplicateTest.scala:
##########
@@ -139,14 +139,17 @@ class DeduplicateTest extends TableTestBase {
@Test
def testSimpleFirstRowOnRowtime(): Unit = {
+ // indirectly check output insert only via used SUM or SUM_RETRACT aggregation function
Review Comment:
Tips, use `util.verifyExplain(sql, ExplainDetail.CHANGELOG_MODE)` can also print the changelog mode in physical nodes. For example:
```
@Test
def test(): Unit = {
val sql =
"""
| SELECT a, b, c
| FROM (
| SELECT *,
| ROW_NUMBER() OVER (PARTITION BY a ORDER BY rowtime ASC) as rank_num
| FROM MyTable)
| WHERE rank_num <= 1
""".stripMargin
util.verifyExplain(sql, ExplainDetail.CHANGELOG_MODE)
}
```
Before this pr:
```
== Abstract Syntax Tree ==
LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[<=($5, 1)])
+- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], rank_num=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $4 NULLS FIRST)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
== Optimized Physical Plan ==
Calc(select=[a, b, c], changelogMode=[I,UA,D])
+- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=1], partitionBy=[a], orderBy=[ROWTIME rowtime ASC], select=[a, b, c, rowtime], changelogMode=[I,UA,D])
+- Exchange(distribution=[hash[a]], changelogMode=[I])
+- Calc(select=[a, b, c, rowtime], changelogMode=[I])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime], changelogMode=[I])
== Optimized Execution Plan ==
Calc(select=[a, b, c])
+- Deduplicate(keep=[FirstRow], key=[a], order=[ROWTIME], outputInsertOnly=[false])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, b, c, rowtime])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime])
```
After applying this pr:
```
== Abstract Syntax Tree ==
LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[<=($5, 1)])
+- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], rank_num=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $4 NULLS FIRST)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
== Optimized Physical Plan ==
Calc(select=[a, b, c], changelogMode=[I])
+- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=1], partitionBy=[a], orderBy=[ROWTIME rowtime ASC], select=[a, b, c, rowtime], changelogMode=[I])
+- Exchange(distribution=[hash[a]], changelogMode=[I])
+- Calc(select=[a, b, c, rowtime], changelogMode=[I])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime], changelogMode=[I])
== Optimized Execution Plan ==
Calc(select=[a, b, c])
+- Deduplicate(keep=[FirstRow], key=[a], order=[ROWTIME], outputInsertOnly=[true])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, b, c, rowtime])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime])
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -220,6 +221,38 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
val providedTrait = ModifyKindSetTrait.INSERT_ONLY
createNewNode(rel, children, providedTrait, requiredTrait, requester)
+ case rank: StreamPhysicalRank if RankUtil.isDeduplication(rank) =>
+ val children = visitChildren(rel, ModifyKindSetTrait.ALL_CHANGES)
+ val tableConfig = unwrapTableConfig(rank)
+
+ // if the rank is deduplication and can be executed as insert-only, forward that information
+ val insertOnly = children
+ .filterNot(
+ rel => {
+ rel.getTraitSet.contains(ModifyKindSetTrait.INSERT_ONLY)
+ })
+ .isEmpty
+
+ val providedTrait = {
+ if (
+ insertOnly && StreamExecDeduplicate.canBeInsertOnly(
Review Comment:
It's a bit strange to use the exec node method in a place that only handles physical nodes. How about moving this method to `RankUtil` and naming it something like `RankUtil#outputInsertOnlyInDeduplicate`?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/RowTimeDeduplicateKeepFirstRowFunction.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.table.runtime.operators.deduplicate;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import static org.apache.flink.table.runtime.operators.deduplicate.utils.DeduplicateFunctionHelper.checkInsertOnly;
+import static org.apache.flink.table.runtime.operators.deduplicate.utils.DeduplicateFunctionHelper.shouldKeepCurrentRow;
+import static org.apache.flink.table.runtime.util.StateConfigUtil.createTtlConfig;
+
+/**
+ * This function is used to deduplicate on keys and keeps only first row on row time. It produces
+ * append only stream thanks to emitting results only via firing the timers.
+ */
+public class RowTimeDeduplicateKeepFirstRowFunction
+ extends KeyedProcessFunction<RowData, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ // the TypeInformation of the values in the state.
+ private final TypeInformation<RowData> typeInfo;
+ private final long stateRetentionTime;
+ private final int rowtimeIndex;
+
+ // state stores previous message under the key.
+ protected ValueState<RowData> waitingToEmitOnTimerState;
+ protected ValueState<Boolean> alreadyEmittedState;
+
+ public RowTimeDeduplicateKeepFirstRowFunction(
+ InternalTypeInfo<RowData> typeInfo, long minRetentionTime, int rowtimeIndex) {
+ this.typeInfo = typeInfo;
+ this.stateRetentionTime = minRetentionTime;
+ this.rowtimeIndex = rowtimeIndex;
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ super.open(openContext);
+
+ // We don't enable TTL on the timer's state, because we rely on the state cleaning up on
+ // watermark. Also otherwise TTL clean up before firing the watermark would cause a data
+ // loss.
+ ValueStateDescriptor<RowData> timerStateDesc =
+ new ValueStateDescriptor<>("waiting-to-emit-on-timer", typeInfo);
+ waitingToEmitOnTimerState = getRuntimeContext().getState(timerStateDesc);
+
+ ValueStateDescriptor<Boolean> stateDesc =
+ new ValueStateDescriptor<>("already-emitted-state-boolean", Types.BOOLEAN);
+ StateTtlConfig ttlConfig = createTtlConfig(stateRetentionTime);
+ if (ttlConfig.isEnabled()) {
+ stateDesc.enableTimeToLive(ttlConfig);
+ }
+ alreadyEmittedState = getRuntimeContext().getState(stateDesc);
+ }
+
+ @Override
+ public void processElement(RowData input, Context ctx, Collector<RowData> out)
+ throws Exception {
+ checkInsertOnly(input);
+ Boolean allreadyEmitted = alreadyEmittedState.value();
+ if (allreadyEmitted != null && allreadyEmitted) {
+ // result has already been emitted, we can not retract/emit anything different.
+ return;
+ }
+ long rowtime = input.getLong(rowtimeIndex);
+ if (rowtime < ctx.timerService().currentWatermark()) {
Review Comment:
What about introducing a metric to log late num used for debugging just like other window operators?
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1950698484
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalRank.scala:
##########
@@ -109,28 +109,34 @@ class StreamPhysicalRank(
.item("select", getRowType.getFieldNames.mkString(", "))
}
- private def getDeduplicateDescription(isRowtime: Boolean, isLastRow: Boolean): String = {
+ private def getDeduplicateDescription(
+ isRowtime: Boolean,
+ isLastRow: Boolean,
+ insertOnly: Boolean): String = {
val fieldNames = getRowType.getFieldNames
val orderString = if (isRowtime) "ROWTIME" else "PROCTIME"
val keep = if (isLastRow) "LastRow" else "FirstRow"
- s"Deduplicate(keep=[$keep], key=[${partitionKey.toArray.map(fieldNames.get).mkString(", ")}], order=[$orderString])"
+ s"Deduplicate(keep=[$keep], key=[${partitionKey.toArray.map(fieldNames.get).mkString(", ")}], order=[$orderString], outputInsertOnly=[$insertOnly])"
}
override def translateToExecNode(): ExecNode[_] = {
val generateUpdateBefore = ChangelogPlanUtils.generateUpdateBefore(this)
if (RankUtil.canConvertToDeduplicate(this)) {
val keepLastRow = RankUtil.keepLastDeduplicateRow(orderKey)
+ val tableConfig = unwrapTableConfig(this)
+ val outputInsertOnly = StreamExecDeduplicate.canBeInsertOnly(tableConfig, keepLastRow)
Review Comment:
Yes you're right!
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1952125243
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java:
##########
@@ -339,25 +362,39 @@ OneInputStreamOperator<RowData, RowData> createDeduplicateOperator() {
}
} else {
if (isAsyncStateEnabled()) {
- AsyncStateRowTimeDeduplicateFunction processFunction =
- new AsyncStateRowTimeDeduplicateFunction(
- rowTypeInfo,
- stateRetentionTime,
- rowtimeIndex,
- generateUpdateBefore,
- generateInsert(),
- keepLastRow);
- return new AsyncKeyedProcessOperator<>(processFunction);
+ if (!keepLastRow && outputInsertOnly) {
+ checkState(canBeInsertOnly(config, keepLastRow));
Review Comment:
Ahh, ok. Done
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2643486689
Thanks for the review and comments @xuyangzhong . Can you take a look at my responoses/fixes? Fixes are in separate fixup commits. I've also added a small fix for https://issues.apache.org/jira/browse/FLINK-37280 as you suggested (if I understood you correctly).
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2651197059
> check the changelog mode is not I when enabling mini-batch or keeping last row
I think they are covered by the existing cases. For example `org.apache.flink.table.planner.plan.stream.sql.DeduplicateTest#testMiniBatchInferFirstRowOnRowtime` for mini batch or `org.apache.flink.table.planner.plan.stream.sql.DeduplicateTest#testSimpleLastRowOnProctime` for last row in plans and we also have ITCases for those. However indeed the `DeduplicateTest` planing tests are relaying on the `COUNT_RETRACT`. I will change them to use `util.verifyExplain(sql, ExplainDetail.CHANGELOG_MODE)`.
> I'm sorry for this late reply, please ping me again next time if this pr is ready :)
no problem!
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2659343708
@flinkbot run azure
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2639758425
@pnowojski I believe you’re correct.
1. I think the current implementation of StreamCommonSubGraphBasedOptimizer has a flaw in how it handles the removal of mini-batch configs. The rank/dedup node is generated during optimization via specific patterns involving project and filter nodes. Upon further reflection, it might be more effective to remove unnecessary mini-batch assigner nodes only after the exec graph is fully generated.
2. Regarding the second point: if different key configs are used during compiling plan and execute plan, it can indeed lead to issues. This applies not only to configs related to mini-batch, but also to any changes in configs marked in `ExecNodeMetadata#consumedOptions`. It would be advisable for users to recompile the plan in such cases.
To avoid blocking this PR, my suggestion is to make a slight improvement to `FlinkRexUtil#shouldSkipMiniBatch` to identify projects that include the `rank_number` function, allowing to temporarily bypass the issue. We can then create a separate jira to address this problem more thoroughly at a later time.
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2606730242
@lincoln-lil @xuyangzhong can you take a look at this? You were recently involved in a couple of changes around here: FLINK-36837 and FLINK-34702.
I could try to implement the async append-only row-time version, but I would probably need a bit of guidance from your side.
For example, the function that I introduced `RowTimeDeduplicateKeepFirstRowFunction` doesn't output the data in one step and it also relies on firing timers. Are there any guarantees to the order of firing timers interleaved with pending async state access for the given key? In other words, if I implement async version of the `processElement`, is it guaranteed that it will finish before any timer for that key can be fired?
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1946606727
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalRank.scala:
##########
@@ -109,28 +109,34 @@ class StreamPhysicalRank(
.item("select", getRowType.getFieldNames.mkString(", "))
}
- private def getDeduplicateDescription(isRowtime: Boolean, isLastRow: Boolean): String = {
+ private def getDeduplicateDescription(
+ isRowtime: Boolean,
+ isLastRow: Boolean,
+ insertOnly: Boolean): String = {
val fieldNames = getRowType.getFieldNames
val orderString = if (isRowtime) "ROWTIME" else "PROCTIME"
val keep = if (isLastRow) "LastRow" else "FirstRow"
- s"Deduplicate(keep=[$keep], key=[${partitionKey.toArray.map(fieldNames.get).mkString(", ")}], order=[$orderString])"
+ s"Deduplicate(keep=[$keep], key=[${partitionKey.toArray.map(fieldNames.get).mkString(", ")}], order=[$orderString], outputInsertOnly=[$insertOnly])"
}
override def translateToExecNode(): ExecNode[_] = {
val generateUpdateBefore = ChangelogPlanUtils.generateUpdateBefore(this)
if (RankUtil.canConvertToDeduplicate(this)) {
val keepLastRow = RankUtil.keepLastDeduplicateRow(orderKey)
+ val tableConfig = unwrapTableConfig(this)
+ val outputInsertOnly = StreamExecDeduplicate.canBeInsertOnly(tableConfig, keepLastRow)
Review Comment:
Hmm, I guess you are right. But can you confirm if I understand the code correctly?
`ChangelogPlanUtils.isInsertOnly(this)` checks for the trait that is actually set in `FlinkChangelogModeInferenceProgram`. `FlinkChangelogModeInferenceProgram` works on physical plan, and therefore those traits are being set before `StreamPhysicalRank#translateToExecNode` is being call? So there is no need to recalculate from scratch (and defacto duplicate the logic of checking) if this rank can output inserts only, we can just use what is already stored/persisted in the trait?
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1950710265
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java:
##########
@@ -339,25 +362,39 @@ OneInputStreamOperator<RowData, RowData> createDeduplicateOperator() {
}
} else {
if (isAsyncStateEnabled()) {
- AsyncStateRowTimeDeduplicateFunction processFunction =
- new AsyncStateRowTimeDeduplicateFunction(
- rowTypeInfo,
- stateRetentionTime,
- rowtimeIndex,
- generateUpdateBefore,
- generateInsert(),
- keepLastRow);
- return new AsyncKeyedProcessOperator<>(processFunction);
+ if (!keepLastRow && outputInsertOnly) {
+ checkState(canBeInsertOnly(config, keepLastRow));
Review Comment:
Emmm... I meant to add one more assertion to do what you say, as shown below to ensure consistency between visiting FlinkChangelogModeInferenceProgram during optimizing and executing this plan. WDYT?
```
checkState(!RankUtil.outputInsertOnlyInDeduplicate(config, keepLastRow));
checkState(!outputInsertOnly)
```
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #26051:
URL: https://github.com/apache/flink/pull/26051#discussion_r1952327615
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java:
##########
@@ -392,6 +431,7 @@ protected ProcTimeDeduplicateOperatorTranslator(
@Override
OneInputStreamOperator<RowData, RowData> createDeduplicateOperator() {
if (isMiniBatchEnabled()) {
+ checkState(!RankUtil.outputInsertOnlyInDeduplicate(config, keepLastRow));
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
Re: [PR] [FLINK-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2608696358
@pnowojski IIUC, the async state operator currently processes watermarks **only** after all prior async state requests have been completed. That means when the event time timer is triggered, all preceding async state access operations have already been processed. Please correct me if there is any mistake @Zakelly .
--
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-37005][table] Make StreamExecDeduplicate ouput insert only where possible [flink]
Posted by "pnowojski (via GitHub)" <gi...@apache.org>.
pnowojski commented on PR #26051:
URL: https://github.com/apache/flink/pull/26051#issuecomment-2612061387
Thanks for your answers @Zakelly and @xuyangzhong!
One more thing. At the moment I have only sync implementation of the append-only deduplicate operator. I guess there is no harm if this is merged as is, so that the synchronous version will be used despite `isAsyncStateEnabled()` being enabled, right? The code will remain correct? And I can provide async version in a separate PR/ticket? (relevant code is in the `StreamExecDeduplicate`).
--
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