You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/09/05 10:00:53 UTC
[GitHub] [flink] leonardBang opened a new pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
leonardBang opened a new pull request #13331:
URL: https://github.com/apache/flink/pull/13331
## What is the purpose of the change
* This pull request import both `first row` and `last row` deduplicate operator on row time. This PR also support deduplicate in `mini-batch` mode.
## Brief change log
- import `RowTimeDeduplicateKeepFirstRowFunction` to support deduplicate with first row
- import `RowTimeDeduplicateKeepLastRowFunction` to support deduplicate with last row
- import `RowTimeMiniBatchDeduplicateKeepFirstRowFunction` to support deduplicate with first row in min-batch mode
- import `RowTimeMiniBatchDeduplicateKeepLastRowFunction` to support deduplicate with last row in min-batch mode
## Verifying this change
* Add plan test in `DeduplicateTest.scala`
* Add operator harness test in:
`RowTimeDeduplicateKeepFirstRowFunctionTest.java`
`RowTimeDeduplicateKeepLastRowFunctionTest.java`
`RowTimeMiniBatchDeduplicateKeepFirstRowFunctionTest.java`
`RowTimeMiniBatchDeduplicateKeepLastRowFunctionTest.java`
* Add Deduplicate ITCase in `DeduplicateITCase.scala`
* Add ITCase for Temporal Join with a view which contains a Deduplicate query in `TemporalJoinITCase.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, Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5e0c14bcf30a4640f90210be692e922a11d3dd1c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856)
* b4a2d274bd70b8006c267be02ee48ee5b9def3bb UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
}, {
"hash" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8671",
"triggerID" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"triggerType" : "PUSH"
}, {
"hash" : "4e977036b9484acda997c8dd76709601bb70a418",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8677",
"triggerID" : "4e977036b9484acda997c8dd76709601bb70a418",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8703",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987",
"triggerID" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"triggerType" : "PUSH"
}, {
"hash" : "1755f4c73e181e591f39adaddd10413f75d6989e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1755f4c73e181e591f39adaddd10413f75d6989e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b3252a0d7e4481aebb849b2777ae544209cff1e8 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987)
* 1755f4c73e181e591f39adaddd10413f75d6989e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leonardBang commented on a change in pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
leonardBang commented on a change in pull request #13331:
URL: https://github.com/apache/flink/pull/13331#discussion_r517239991
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -147,6 +150,148 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void deduplicateOnRowTime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData preRow = state.value();
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ return;
+ }
+
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ null
+ );
+ state.update(currentRow);
+ }
+
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param bufferedRows latest row received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void miniBatchDeduplicateOnRowTime(
+ ValueState<RowData> state,
+ List<RowData> bufferedRows,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+ if (bufferedRows == null) {
+ return;
+ }
+
+ RowData preRow = state.value();
+ for (RowData currentRow : bufferedRows) {
+ checkInsertOnly(currentRow);
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ continue;
+ }
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ serializer
+ );
+ preRow = currentRow;
+ }
+ state.update(preRow);
+ }
+
+ private static void collectRetractResult(
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ RowData preRow,
+ RowData currentRow,
+ Collector<RowData> out,
+ TypeSerializer<RowData> serializer) {
+ if (generateUpdateBefore || generateInsert) {
+ if (preRow == null) {
+ // the first row, send INSERT message
+ currentRow.setRowKind(RowKind.INSERT);
+ out.collect(currentRow);
+ } else {
+ if (generateUpdateBefore) {
+ RowData copyRow;
+ // when miniBatch enabled, do a copy here, the serializer is not null
+ if (serializer != null) {
+ copyRow = serializer.copy(preRow);
Review comment:
the state update logic is a little different, I skip this one
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
}, {
"hash" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8671",
"triggerID" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"triggerType" : "PUSH"
}, {
"hash" : "4e977036b9484acda997c8dd76709601bb70a418",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8677",
"triggerID" : "4e977036b9484acda997c8dd76709601bb70a418",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8703",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987",
"triggerID" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b3252a0d7e4481aebb849b2777ae544209cff1e8 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
wuchong commented on a change in pull request #13331:
URL: https://github.com/apache/flink/pull/13331#discussion_r515793467
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -147,6 +150,148 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void deduplicateOnRowTime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData preRow = state.value();
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ return;
+ }
+
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ null
+ );
+ state.update(currentRow);
+ }
+
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param bufferedRows latest row received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void miniBatchDeduplicateOnRowTime(
Review comment:
It is only used in `RowTimeMiniBatchDeduplicateFunction`. We don't need to put it in the util class.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -147,6 +150,148 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void deduplicateOnRowTime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData preRow = state.value();
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ return;
+ }
+
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ null
+ );
+ state.update(currentRow);
+ }
+
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param bufferedRows latest row received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void miniBatchDeduplicateOnRowTime(
+ ValueState<RowData> state,
+ List<RowData> bufferedRows,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+ if (bufferedRows == null) {
Review comment:
I think `bufferedRows` would never be null, but maybe empty.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/MiniBatchDeduplicateFunctionBase.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.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.typeutils.TypeSerializer;
+import org.apache.flink.table.runtime.context.ExecutionContext;
+import org.apache.flink.table.runtime.operators.bundle.MapBundleFunction;
+
+import static org.apache.flink.table.runtime.util.StateTtlConfigUtil.createTtlConfig;
+
+/**
+ * Base class for miniBatch deduplicate function.
+ * @param <T> The type of the value in the state.
+ * @param <K> The type of the key in the bundle map.
+ * @param <V> The type of the value in the bundle map.
+ * @param <IN> Type of the input elements.
+ * @param <OUT> Type of the returned elements.
+ */
+abstract class MiniBatchDeduplicateFunctionBase<T, K, V, IN, OUT> extends MapBundleFunction<K, V, IN, OUT> {
+
+ protected final TypeInformation<T> typeInfo;
+ protected final TypeSerializer<OUT> serializer;
+ protected final long minRetentionTime;
+ // state stores previous message under the key.
+ protected ValueState<T> state;
+
+ public MiniBatchDeduplicateFunctionBase(
+ TypeInformation<T> typeInfo,
+ TypeSerializer<OUT> serializer,
Review comment:
It would be better to remove this `serializer`, as it is never used in the base class. It confuses what's this used for, and it's error-prone, becuase it's nullable.
##########
File path: flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/RowTimeDeduplicateFunctionTest.java
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.data.RowData;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
+
+/**
+ * Tests for {@link RowTimeDeduplicateFunction}.
+ */
+@RunWith(Parameterized.class)
+public class RowTimeDeduplicateFunctionTest extends RowTimeDeduplicateFunctionTestBase {
+
+ private final boolean generateUpdateBefore;
+ private final boolean generateInsert;
+
+ public RowTimeDeduplicateFunctionTest(boolean generateUpdateBefore, boolean generateInsert) {
+ this.generateUpdateBefore = generateUpdateBefore;
+ this.generateInsert = generateInsert;
+ }
+
+ @Test
+ public void testRowTimeDeduplicateKeepFirstRow() throws Exception {
+ final boolean keepLastRow = false;
+ RowTimeDeduplicateFunction func = new RowTimeDeduplicateFunction(
+ inputRowType,
+ minTtlTime.toMilliseconds(),
+ rowTimeIndex,
+ generateUpdateBefore,
+ generateInsert,
+ keepLastRow);
+ KeyedProcessOperator<RowData, RowData, RowData> operator = new KeyedProcessOperator<>(func);
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);
+
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 13, 99L));
+ testHarness.processElement(insertRecord("key1", 13, 99L));
+ testHarness.processElement(insertRecord("key1", 12, 100L));
+ testHarness.processElement(insertRecord("key2", 11, 101L));
+
+ // test 1: keep first row with row time
+ testHarness.processWatermark(new Watermark(102));
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 1), testHarness.getOutput());
+
+ // do a snapshot, close and restore again
+ OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
+ testHarness.close();
+
+ testHarness = createTestHarness(operator);
+ testHarness.setup();
+ testHarness.initializeState(snapshot);
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 12, 300L));
+ testHarness.processElement(insertRecord("key2", 11, 301L));
+ testHarness.processElement(insertRecord("key3", 5, 299L));
+
+ // test 2: load snapshot state
+ testHarness.processWatermark(new Watermark(302));
+
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 2), testHarness.getOutput());
+
+ // test 3: expire the state
+ testHarness.setStateTtlProcessingTime(minTtlTime.toMilliseconds() + 1);
+ testHarness.processElement(insertRecord("key1", 12, 400L));
+ testHarness.processElement(insertRecord("key2", 11, 401L));
+ testHarness.processWatermark(402);
+
+ // ("key1", 13, 99L) and ("key2", 11, 101L) had retired, thus output ("key1", 12, 200L),("key2", 11, 201L)
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 3), testHarness.getOutput());
+ testHarness.close();
+ }
+
+ public void testRowTimeDeduplicateKeepLastRow() throws Exception {
Review comment:
Miss `@Test`?
##########
File path: flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/RowTimeDeduplicateFunctionTest.java
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.data.RowData;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
+
+/**
+ * Tests for {@link RowTimeDeduplicateFunction}.
+ */
+@RunWith(Parameterized.class)
+public class RowTimeDeduplicateFunctionTest extends RowTimeDeduplicateFunctionTestBase {
+
+ private final boolean generateUpdateBefore;
+ private final boolean generateInsert;
+
+ public RowTimeDeduplicateFunctionTest(boolean generateUpdateBefore, boolean generateInsert) {
+ this.generateUpdateBefore = generateUpdateBefore;
+ this.generateInsert = generateInsert;
+ }
+
+ @Test
+ public void testRowTimeDeduplicateKeepFirstRow() throws Exception {
+ final boolean keepLastRow = false;
+ RowTimeDeduplicateFunction func = new RowTimeDeduplicateFunction(
+ inputRowType,
+ minTtlTime.toMilliseconds(),
+ rowTimeIndex,
+ generateUpdateBefore,
+ generateInsert,
+ keepLastRow);
+ KeyedProcessOperator<RowData, RowData, RowData> operator = new KeyedProcessOperator<>(func);
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);
+
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 13, 99L));
+ testHarness.processElement(insertRecord("key1", 13, 99L));
+ testHarness.processElement(insertRecord("key1", 12, 100L));
+ testHarness.processElement(insertRecord("key2", 11, 101L));
+
+ // test 1: keep first row with row time
+ testHarness.processWatermark(new Watermark(102));
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 1), testHarness.getOutput());
+
+ // do a snapshot, close and restore again
+ OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
+ testHarness.close();
+
+ testHarness = createTestHarness(operator);
+ testHarness.setup();
+ testHarness.initializeState(snapshot);
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 12, 300L));
+ testHarness.processElement(insertRecord("key2", 11, 301L));
+ testHarness.processElement(insertRecord("key3", 5, 299L));
+
+ // test 2: load snapshot state
+ testHarness.processWatermark(new Watermark(302));
+
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 2), testHarness.getOutput());
+
+ // test 3: expire the state
+ testHarness.setStateTtlProcessingTime(minTtlTime.toMilliseconds() + 1);
+ testHarness.processElement(insertRecord("key1", 12, 400L));
+ testHarness.processElement(insertRecord("key2", 11, 401L));
+ testHarness.processWatermark(402);
+
+ // ("key1", 13, 99L) and ("key2", 11, 101L) had retired, thus output ("key1", 12, 200L),("key2", 11, 201L)
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 3), testHarness.getOutput());
+ testHarness.close();
+ }
+
+ public void testRowTimeDeduplicateKeepLastRow() throws Exception {
+ final boolean keepLastRow = true;
+ RowTimeDeduplicateFunction func = new RowTimeDeduplicateFunction(
+ inputRowType,
+ minTtlTime.toMilliseconds(),
+ rowTimeIndex,
+ generateUpdateBefore,
+ generateInsert,
+ true);
+ KeyedProcessOperator<RowData, RowData, RowData> operator = new KeyedProcessOperator<>(func);
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 13, 99L));
+ testHarness.processElement(insertRecord("key1", 12, 100L));
+ testHarness.processElement(insertRecord("key2", 11, 101L));
+
+ // test 1: keep last row with row time
+ testHarness.processWatermark(new Watermark(102));
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 1), testHarness.getOutput());
+
+ // do a snapshot, close and restore again
+ OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
+ testHarness.close();
+
+ testHarness = createTestHarness(operator);
+ testHarness.setup();
+ testHarness.initializeState(snapshot);
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 12, 300L));
+ testHarness.processElement(insertRecord("key2", 11, 301L));
+ testHarness.processElement(insertRecord("key3", 5, 299L));
+
+ // test 2: load snapshot state
+ testHarness.processWatermark(new Watermark(302));
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 2), testHarness.getOutput());
+
+ // test 3: expire the state
+ testHarness.setStateTtlProcessingTime(minTtlTime.toMilliseconds() + 1);
+ testHarness.processElement(insertRecord("key1", 12, 400L));
+ testHarness.processElement(insertRecord("key2", 11, 401L));
+ testHarness.processWatermark(402);
+
+ // all state has expired, so the record ("key1", 12, 400L), ("key2", 12, 401L) will be INSERT message
+ assertor.assertOutputEqualsSorted("output wrong.", getExpectOutput(keepLastRow, 3), testHarness.getOutput());
Review comment:
It's super hard to understand what the expect result is.
I think if the result are different, then we shouldn't parameterize them into one test. We can follow the way of `TemporalRowTimeJoinOperatorTest` that resue the same test data in a common private method, but delcare differenet expected output in different test methods.
Besides, I think we can't share the expected output for both minibatch and non-minibatch mode?
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -147,6 +150,148 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void deduplicateOnRowTime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData preRow = state.value();
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ return;
+ }
+
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ null
+ );
+ state.update(currentRow);
+ }
+
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param bufferedRows latest row received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void miniBatchDeduplicateOnRowTime(
+ ValueState<RowData> state,
+ List<RowData> bufferedRows,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+ if (bufferedRows == null) {
+ return;
+ }
+
+ RowData preRow = state.value();
+ for (RowData currentRow : bufferedRows) {
+ checkInsertOnly(currentRow);
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ continue;
+ }
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ serializer
+ );
+ preRow = currentRow;
+ }
+ state.update(preRow);
+ }
+
+ private static void collectRetractResult(
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ RowData preRow,
+ RowData currentRow,
+ Collector<RowData> out,
+ TypeSerializer<RowData> serializer) {
+ if (generateUpdateBefore || generateInsert) {
+ if (preRow == null) {
+ // the first row, send INSERT message
+ currentRow.setRowKind(RowKind.INSERT);
+ out.collect(currentRow);
+ } else {
+ if (generateUpdateBefore) {
+ RowData copyRow;
+ // when miniBatch enabled, do a copy here, the serializer is not null
+ if (serializer != null) {
+ copyRow = serializer.copy(preRow);
Review comment:
If this copying can be removed, the implementations of `collectRetractResult` and `processLastRowOnProcTime` are totally the same. I feel that `collectRetractResult` is not suitable, because it not always generates retractions.
What about combine them in the the following new method?
```java
static void updateDeduplicateResult(
RowData currentRow, // new row result of deduplicate under current key
ValueState<RowData> state, // the state stores the previous collected row under current key
boolean generateUpdateBefore,
boolean generateInsert,
Collector<RowData> out) throws Exception {
```
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/MiniBatchDeduplicateFunctionBase.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.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.typeutils.TypeSerializer;
+import org.apache.flink.table.runtime.context.ExecutionContext;
+import org.apache.flink.table.runtime.operators.bundle.MapBundleFunction;
+
+import static org.apache.flink.table.runtime.util.StateTtlConfigUtil.createTtlConfig;
+
+/**
+ * Base class for miniBatch deduplicate function.
+ * @param <T> The type of the value in the state.
+ * @param <K> The type of the key in the bundle map.
+ * @param <V> The type of the value in the bundle map.
+ * @param <IN> Type of the input elements.
+ * @param <OUT> Type of the returned elements.
+ */
+abstract class MiniBatchDeduplicateFunctionBase<T, K, V, IN, OUT> extends MapBundleFunction<K, V, IN, OUT> {
Review comment:
Add `serialVersionUID`.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/MiniBatchDeduplicateFunctionBase.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.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.typeutils.TypeSerializer;
+import org.apache.flink.table.runtime.context.ExecutionContext;
+import org.apache.flink.table.runtime.operators.bundle.MapBundleFunction;
+
+import static org.apache.flink.table.runtime.util.StateTtlConfigUtil.createTtlConfig;
+
+/**
+ * Base class for miniBatch deduplicate function.
+ * @param <T> The type of the value in the state.
+ * @param <K> The type of the key in the bundle map.
+ * @param <V> The type of the value in the bundle map.
+ * @param <IN> Type of the input elements.
+ * @param <OUT> Type of the returned elements.
+ */
+abstract class MiniBatchDeduplicateFunctionBase<T, K, V, IN, OUT> extends MapBundleFunction<K, V, IN, OUT> {
+
+ protected final TypeInformation<T> typeInfo;
+ protected final TypeSerializer<OUT> serializer;
+ protected final long minRetentionTime;
+ // state stores previous message under the key.
+ protected ValueState<T> state;
+
+ public MiniBatchDeduplicateFunctionBase(
+ TypeInformation<T> typeInfo,
Review comment:
`typeInfo` -> `stateType` to be more specifically.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -147,6 +150,148 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void deduplicateOnRowTime(
Review comment:
ditto.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -147,6 +150,148 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void deduplicateOnRowTime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData preRow = state.value();
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ return;
+ }
+
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ null
+ );
+ state.update(currentRow);
+ }
+
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param bufferedRows latest row received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void miniBatchDeduplicateOnRowTime(
+ ValueState<RowData> state,
+ List<RowData> bufferedRows,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+ if (bufferedRows == null) {
+ return;
+ }
+
+ RowData preRow = state.value();
+ for (RowData currentRow : bufferedRows) {
+ checkInsertOnly(currentRow);
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ continue;
+ }
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ serializer
+ );
+ preRow = currentRow;
+ }
+ state.update(preRow);
+ }
+
+ private static void collectRetractResult(
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ RowData preRow,
+ RowData currentRow,
+ Collector<RowData> out,
+ TypeSerializer<RowData> serializer) {
+ if (generateUpdateBefore || generateInsert) {
+ if (preRow == null) {
+ // the first row, send INSERT message
+ currentRow.setRowKind(RowKind.INSERT);
+ out.collect(currentRow);
+ } else {
+ if (generateUpdateBefore) {
+ RowData copyRow;
+ // when miniBatch enabled, do a copy here, the serializer is not null
+ if (serializer != null) {
+ copyRow = serializer.copy(preRow);
Review comment:
Why we have to do a deep copy here?
If it is for not affecting the original `preRow`, we can set back `RowKind` after collecting `preRow` (non-minibatch mode also has this problem).
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -147,6 +150,148 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void deduplicateOnRowTime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData preRow = state.value();
+ if (!orderFunctionProvider.matches(preRow, currentRow)) {
+ return;
+ }
+
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ preRow,
+ currentRow,
+ out,
+ null
+ );
+ state.update(currentRow);
+ }
+
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last
+ * or first row, retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param bufferedRows latest row received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ * @param orderFunctionProvider provider that provides an order function to judge first or last
+ */
+ public static void miniBatchDeduplicateOnRowTime(
+ ValueState<RowData> state,
+ List<RowData> bufferedRows,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ DeduplicateOrderFunctionProvider orderFunctionProvider) throws Exception {
Review comment:
If this method can be moved in `RowTimeMiniBatchDeduplicateFunction`, I think we don't need the special `DeduplicateOrderFunctionProvider` parameter then, because we have `keepLastRow` flag.
##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeduplicateITCase.scala
##########
@@ -137,4 +152,93 @@ class DeduplicateITCase(miniBatch: MiniBatchMode, mode: StateBackendMode)
assertEquals(expected.sorted, sink.getRetractResults.sorted)
}
+ @Test
+ def testFirstRowOnRowtime(): Unit = {
+ val t = env.fromCollection(rowtimeTestData)
+ .assignTimestampsAndWatermarks(new RowtimeExtractor)
+ .toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime())
+ tEnv.registerTable("T", t)
+ createSinkTable("rowtime_sink")
+
+ val sql =
+ """
+ |INSERT INTO rowtime_sink
+ | SELECT a, b, c, rowtime
+ | FROM (
+ | SELECT *,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY rowtime) as rowNum
+ | FROM T
+ | )
+ | WHERE rowNum = 1
+ """.stripMargin
+
+ tEnv.executeSql(sql).await()
+ val rawResult = TestValuesTableFactory.getRawResults("rowtime_sink")
+
+ val 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)")
+ assertEquals(expected.sorted, rawResult.sorted)
Review comment:
If we want to verify the changelog result, I would suggest to use single parallelism and do not sort the result. Otherwise, we didn't actually test the changelog order. For example, it's hard to understand why `-U` is after `+U` in the above test.
We can also just verify the merged result and can use the default parallelism.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 61470e6514fd5cb80250ed641e9e7e293c97fa07 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279)
* baed656a328a88791c638d87ac3b15091bc43408 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
}, {
"hash" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8671",
"triggerID" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"triggerType" : "PUSH"
}, {
"hash" : "4e977036b9484acda997c8dd76709601bb70a418",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8677",
"triggerID" : "4e977036b9484acda997c8dd76709601bb70a418",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8703",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987",
"triggerID" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0fcea2d9a006b405865e558b2bdb07311b87c2a6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8703) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693)
* b3252a0d7e4481aebb849b2777ae544209cff1e8 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f37fa30b387d0c443139107f7b2a8dbdc6a8364d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6a79badae175128326ed4e77708b48488009731c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261)
* 6988d189034d8ad024e5db27962cf589273a5534 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
wuchong commented on a change in pull request #13331:
URL: https://github.com/apache/flink/pull/13331#discussion_r517854922
##########
File path: flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/RowTimeMiniBatchDeduplicateFunctionTest.java
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.time.Time;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.bundle.KeyedMapBundleOperator;
+import org.apache.flink.table.runtime.operators.bundle.trigger.CountBundleTrigger;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.runtime.util.BinaryRowDataKeySelector;
+import org.apache.flink.table.runtime.util.GenericRowRecordSortComparator;
+import org.apache.flink.table.runtime.util.RowDataHarnessAssertor;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.types.RowKind;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.record;
+
+/**
+ * Harness tests for {@link RowTimeDeduplicateFunction} and {@link RowTimeMiniBatchDeduplicateFunction}.
+ */
+@RunWith(Parameterized.class)
+public class RowTimeMiniBatchDeduplicateFunctionTest {
Review comment:
nit: would be better to call `RowTimeDeduplicateFunctionTest`, `minibatch` is the parameter.
##########
File path: flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/RowTimeMiniBatchDeduplicateFunctionTest.java
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.time.Time;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.bundle.KeyedMapBundleOperator;
+import org.apache.flink.table.runtime.operators.bundle.trigger.CountBundleTrigger;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.runtime.util.BinaryRowDataKeySelector;
+import org.apache.flink.table.runtime.util.GenericRowRecordSortComparator;
+import org.apache.flink.table.runtime.util.RowDataHarnessAssertor;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.types.RowKind;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.record;
+
+/**
+ * Harness tests for {@link RowTimeDeduplicateFunction} and {@link RowTimeMiniBatchDeduplicateFunction}.
+ */
+@RunWith(Parameterized.class)
+public class RowTimeMiniBatchDeduplicateFunctionTest {
+
+ private final long miniBatchSize = 4L;
+ private Time minTtlTime = Time.milliseconds(10);
+ private InternalTypeInfo inputRowType = InternalTypeInfo.ofFields(
+ new VarCharType(VarCharType.MAX_LENGTH),
+ new IntType(),
+ new BigIntType());
+ private TypeSerializer<RowData> serializer = inputRowType.toSerializer();
+ private int rowTimeIndex = 2;
+ private int rowKeyIndex = 0;
+ private BinaryRowDataKeySelector rowKeySelector = new BinaryRowDataKeySelector(
+ new int[]{rowKeyIndex},
+ inputRowType.toRowFieldTypes());
+ private RowDataHarnessAssertor assertor = new RowDataHarnessAssertor(
+ inputRowType.toRowFieldTypes(),
+ new GenericRowRecordSortComparator(rowKeyIndex, inputRowType.toRowFieldTypes()[rowKeyIndex]));
+
+ private final boolean miniBatchEnable;
+
+ public RowTimeMiniBatchDeduplicateFunctionTest(boolean miniBacthEnable) {
+ this.miniBatchEnable = miniBacthEnable;
+ }
+
+ @Test
+ public void testRowTimeDeduplicateKeepFirstRow() throws Exception {
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record(RowKind.INSERT, "key1", 13, 99L));
+ expectedOutput.add(record(RowKind.INSERT, "key2", 11, 101L));
+ expectedOutput.add(new Watermark(102));
+ expectedOutput.add(record(RowKind.INSERT, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.INSERT, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.INSERT, "key1", 12, 400L));
+ expectedOutput.add(record(RowKind.INSERT, "key2", 11, 401L));
+ expectedOutput.add(new Watermark(402));
+
+ // generateUpdateBefore: true, generateInsert: true
+ testRowTimeDeduplicateKeepFirstRow(true, true, expectedOutput);
+
+ // generateUpdateBefore: true, generateInsert: false
+ testRowTimeDeduplicateKeepFirstRow(true, false, expectedOutput);
+
+ // generateUpdateBefore: false, generateInsert: true
+ testRowTimeDeduplicateKeepFirstRow(false, true, expectedOutput);
+
+ // generateUpdateBefore: false, generateInsert: false
+ expectedOutput.clear();
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 13, 99L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 101L));
+ expectedOutput.add(new Watermark(102));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 400L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 401L));
+ expectedOutput.add(new Watermark(402));
+ testRowTimeDeduplicateKeepFirstRow(false, false, expectedOutput);
+
+ }
+
+ @Test
+ public void testRowTimeDeduplicateKeepLastRow() throws Exception {
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record(RowKind.INSERT, "key1", 13, 99L));
+ expectedOutput.add(record(RowKind.UPDATE_BEFORE, "key1", 13, 99L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 100L));
+ expectedOutput.add(record(RowKind.INSERT, "key2", 11, 101L));
+ expectedOutput.add(new Watermark(102));
+ expectedOutput.add(record(RowKind.UPDATE_BEFORE, "key1", 12, 100L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 300L));
+ expectedOutput.add(record(RowKind.UPDATE_BEFORE, "key2", 11, 101L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 301L));
+ expectedOutput.add(record(RowKind.INSERT, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.UPDATE_BEFORE, "key1", 12, 100L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 300L));
+ expectedOutput.add(record(RowKind.UPDATE_BEFORE, "key2", 11, 101L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 301L));
+ expectedOutput.add(record(RowKind.INSERT, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.INSERT, "key1", 12, 400L));
+ expectedOutput.add(record(RowKind.INSERT, "key2", 11, 401L));
+ expectedOutput.add(new Watermark(402));
+
+ // generateUpdateBefore: true, generateInsert: true
+ testRowTimeDeduplicateKeepLastRow(true, true, expectedOutput);
+
+ // generateUpdateBefore: true, generateInsert: false
+ testRowTimeDeduplicateKeepLastRow(true, false, expectedOutput);
+
+ // generateUpdateBefore: false, generateInsert: true
+ expectedOutput.clear();
+ expectedOutput.add(record(RowKind.INSERT, "key1", 13, 99L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 100L));
+ expectedOutput.add(record(RowKind.INSERT, "key2", 11, 101L));
+ expectedOutput.add(new Watermark(102));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 300L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 301L));
+ expectedOutput.add(record(RowKind.INSERT, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 300L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 301L));
+ expectedOutput.add(record(RowKind.INSERT, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.INSERT, "key1", 12, 400L));
+ expectedOutput.add(record(RowKind.INSERT, "key2", 11, 401L));
+ expectedOutput.add(new Watermark(402));
+ testRowTimeDeduplicateKeepLastRow(false, true, expectedOutput);
+
+ // generateUpdateBefore: false, generateInsert: false
+ expectedOutput.clear();
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 13, 99L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 100L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 101L));
+ expectedOutput.add(new Watermark(102));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 300L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 301L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 300L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 301L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key3", 5, 299L));
+ expectedOutput.add(new Watermark(302));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key1", 12, 400L));
+ expectedOutput.add(record(RowKind.UPDATE_AFTER, "key2", 11, 401L));
+ expectedOutput.add(new Watermark(402));
+ testRowTimeDeduplicateKeepLastRow(false, false, expectedOutput);
+ }
+
+ private void testRowTimeDeduplicateKeepFirstRow(
+ boolean generateUpdateBefore,
+ boolean generateInsert,
+ List<Object> expectedOutput) throws Exception {
+ final boolean keepLastRow = false;
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness;
+ KeyedMapBundleOperator<RowData, RowData, RowData, RowData> keyedMapBundleOperator = null;
+ KeyedProcessOperator keyedProcessOperator = null;
+ if (miniBatchEnable) {
+ RowTimeMiniBatchDeduplicateFunction func = new RowTimeMiniBatchDeduplicateFunction(
+ inputRowType,
+ serializer,
+ minTtlTime.toMilliseconds(),
+ rowTimeIndex,
+ generateUpdateBefore,
+ generateInsert,
+ keepLastRow);
+ CountBundleTrigger trigger = new CountBundleTrigger<RowData>(miniBatchSize);
+ keyedMapBundleOperator = new KeyedMapBundleOperator(func, trigger);
+ testHarness = createTestHarness(keyedMapBundleOperator);
+ } else {
+ RowTimeDeduplicateFunction func = new RowTimeDeduplicateFunction(
+ inputRowType,
+ minTtlTime.toMilliseconds(),
+ rowTimeIndex,
+ generateUpdateBefore,
+ generateInsert,
+ keepLastRow);
+ keyedProcessOperator = new KeyedProcessOperator<>(func);
+ testHarness = createTestHarness(keyedProcessOperator);
+ }
+
+ List<Object> actualOutput = new ArrayList<>();
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 13, 99L));
+ testHarness.processElement(insertRecord("key1", 13, 99L));
+ testHarness.processElement(insertRecord("key1", 12, 100L));
+ testHarness.processElement(insertRecord("key2", 11, 101L));
+
+ // test 1: keep first row with row time
+ testHarness.processWatermark(new Watermark(102));
+ actualOutput.addAll(testHarness.getOutput());
+
+ // do a snapshot, close and restore again
+ OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
+ testHarness.close();
+
+ if (miniBatchEnable) {
+ testHarness = createTestHarness(keyedMapBundleOperator);
+ } else {
+ testHarness = createTestHarness(keyedProcessOperator);
+ }
+
+ testHarness.setup();
+ testHarness.initializeState(snapshot);
+ testHarness.open();
+
+ testHarness.processElement(insertRecord("key1", 12, 300L));
+ testHarness.processElement(insertRecord("key2", 11, 301L));
+ testHarness.processElement(insertRecord("key3", 5, 299L));
+
+ // test 2: load snapshot state
+ testHarness.processWatermark(new Watermark(302));
+ actualOutput.addAll(testHarness.getOutput());
Review comment:
This adds duplicate result into the 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b4a2d274bd70b8006c267be02ee48ee5b9def3bb Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861)
* 42f18261f80da589a648414d6bd81c20730e0cc0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6988d189034d8ad024e5db27962cf589273a5534 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1fa2f9690c7d83d4a93720a53c1344620b8849e6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231)
* f37fa30b387d0c443139107f7b2a8dbdc6a8364d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 61470e6514fd5cb80250ed641e9e7e293c97fa07 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 42f18261f80da589a648414d6bd81c20730e0cc0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6a79badae175128326ed4e77708b48488009731c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
}, {
"hash" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8671",
"triggerID" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"triggerType" : "PUSH"
}, {
"hash" : "4e977036b9484acda997c8dd76709601bb70a418",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8677",
"triggerID" : "4e977036b9484acda997c8dd76709601bb70a418",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8703",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987",
"triggerID" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"triggerType" : "PUSH"
}, {
"hash" : "1755f4c73e181e591f39adaddd10413f75d6989e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9059",
"triggerID" : "1755f4c73e181e591f39adaddd10413f75d6989e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b3252a0d7e4481aebb849b2777ae544209cff1e8 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987)
* 1755f4c73e181e591f39adaddd10413f75d6989e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9059)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f37fa30b387d0c443139107f7b2a8dbdc6a8364d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247)
* 6a79badae175128326ed4e77708b48488009731c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1fa2f9690c7d83d4a93720a53c1344620b8849e6 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* baed656a328a88791c638d87ac3b15091bc43408 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b4a2d274bd70b8006c267be02ee48ee5b9def3bb Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1fa2f9690c7d83d4a93720a53c1344620b8849e6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
}, {
"hash" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f3b2192b6cc216c691e833e40e52dfc64807e0a0 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642)
* d4463d6be66ef3a125e23c4a63c4efad8359454d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687583105
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 1fa2f9690c7d83d4a93720a53c1344620b8849e6 (Sat Sep 05 10:02:33 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 61470e6514fd5cb80250ed641e9e7e293c97fa07 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279)
* baed656a328a88791c638d87ac3b15091bc43408 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6988d189034d8ad024e5db27962cf589273a5534 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268)
* 61470e6514fd5cb80250ed641e9e7e293c97fa07 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wuchong merged pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
wuchong merged pull request #13331:
URL: https://github.com/apache/flink/pull/13331
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f3b2192b6cc216c691e833e40e52dfc64807e0a0 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* baed656a328a88791c638d87ac3b15091bc43408 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372)
* 5e0c14bcf30a4640f90210be692e922a11d3dd1c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] danny0405 commented on a change in pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #13331:
URL: https://github.com/apache/flink/pull/13331#discussion_r514915286
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -147,6 +149,226 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last row,
+ * retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param out underlying collector
+ * @param rowtimeIndex index of row time field
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ */
+ static void processLastRowOnRowtime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ Collector<RowData> out,
+ int rowtimeIndex,
+ boolean generateUpdateBefore,
+ boolean generateInsert) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData prevRow = state.value();
+ if (!isLastRow(prevRow, currentRow, rowtimeIndex)) {
+ return;
+ }
+ state.update(currentRow);
+
+ // store all needed data to state
+ collectRetractResult(
+ generateUpdateBefore,
+ generateInsert,
+ prevRow,
+ currentRow,
+ out,
+ null
+ );
+ }
+
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last row,
+ * retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param bufferedRows latest rows received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param rowtimeIndex index of row time field
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ */
+ static void processMiniBatchLastRowOnRowtime(
+ ValueState<RowData> state,
+ List<RowData> bufferedRows,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ int rowtimeIndex,
+ boolean generateUpdateBefore,
+ boolean generateInsert) throws Exception {
+
+ if (bufferedRows == null) {
+ return;
+ }
+
+ RowData preRow = state.value();
+ for (RowData currentRow : bufferedRows) {
+ checkInsertOnly(currentRow);
+ if (!isLastRow(preRow, currentRow, rowtimeIndex)) {
Review comment:
The code is almost same with `processMiniBatchFirstRowOnRowtime`, we can abstract the common code out here.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/RowTimeMiniBatchDeduplicateKeepFirstRowFunction.java
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.typeutils.TypeSerializer;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.table.runtime.operators.deduplicate.DeduplicateFunctionHelper.processMiniBatchFirstRowOnRowTime;
+
+/**
+ * This function is used to get the first row for every key partition in miniBatch mode.
+ */
+public class RowTimeMiniBatchDeduplicateKeepFirstRowFunction
+ extends MiniBatchDeduplicateFunctionBase<RowData, RowData, List<RowData>, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int rowtimeIndex;
+ private final boolean generateUpdateBefore;
+ private final boolean generateInsert;
+
+ public RowTimeMiniBatchDeduplicateKeepFirstRowFunction(
+ InternalTypeInfo<RowData> typeInfo,
+ TypeSerializer<RowData> serializer,
+ long minRetentionTime,
+ int rowtimeIndex,
+ boolean generateUpdateBefore,
+ boolean generateInsert) {
+ super(typeInfo, serializer, minRetentionTime);
+ this.rowtimeIndex = rowtimeIndex;
+ this.generateUpdateBefore = generateUpdateBefore;
+ this.generateInsert = generateInsert;
+ }
+
+ @Override
+ public List<RowData> addInput(@Nullable List<RowData> value, RowData input) throws Exception {
+ if (value == null) {
+ value = new ArrayList<>();
+ }
+ value.add(serializer.copy(input));
+ return value;
+ }
+
+ @Override
+ public void finishBundle(Map<RowData, List<RowData>> buffer, Collector<RowData> out) throws Exception {
+ for (Map.Entry<RowData, List<RowData>> entry : buffer.entrySet()) {
+ RowData currentKey = entry.getKey();
+ List<RowData> bufferedRows = entry.getValue();
Review comment:
Both `addInput` and `finishBundle` can be moved to the base class.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 008e4ae2c2e10b712d84a4f324634a71fe5f84ac Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 115ad9e4e8c7258d5831848a7144bfbb46a5b6f9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478)
* f3b2192b6cc216c691e833e40e52dfc64807e0a0 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1fa2f9690c7d83d4a93720a53c1344620b8849e6 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 42f18261f80da589a648414d6bd81c20730e0cc0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208)
* 008e4ae2c2e10b712d84a4f324634a71fe5f84ac Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6988d189034d8ad024e5db27962cf589273a5534 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268)
* 61470e6514fd5cb80250ed641e9e7e293c97fa07 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5e0c14bcf30a4640f90210be692e922a11d3dd1c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] danny0405 commented on a change in pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
danny0405 commented on a change in pull request #13331:
URL: https://github.com/apache/flink/pull/13331#discussion_r508472527
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecDeduplicate.scala
##########
@@ -102,44 +110,109 @@ class StreamExecDeduplicate(
.asInstanceOf[Transformation[RowData]]
val rowTypeInfo = inputTransform.getOutputType.asInstanceOf[InternalTypeInfo[RowData]]
+ val inputFieldTypes = rowTypeInfo.toRowFieldTypes
+ val keyFieldTypes = new Array[LogicalType](uniqueKeys.length)
+ for (i <- 0 until uniqueKeys.length) {
+ keyFieldTypes(i) = inputFieldTypes(uniqueKeys(i))
+ }
+
Review comment:
You can use `uniqueKeys .map(idx => inputFieldTypes(idx))` directly.
##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecDeduplicate.scala
##########
@@ -102,44 +110,109 @@ class StreamExecDeduplicate(
.asInstanceOf[Transformation[RowData]]
val rowTypeInfo = inputTransform.getOutputType.asInstanceOf[InternalTypeInfo[RowData]]
+ val inputFieldTypes = rowTypeInfo.toRowFieldTypes
+ val keyFieldTypes = new Array[LogicalType](uniqueKeys.length)
+ for (i <- 0 until uniqueKeys.length) {
+ keyFieldTypes(i) = inputFieldTypes(uniqueKeys(i))
+ }
+
val generateUpdateBefore = ChangelogPlanUtils.generateUpdateBefore(this)
val tableConfig = planner.getTableConfig
val generateInsert = tableConfig.getConfiguration
.getBoolean(TABLE_EXEC_INSERT_AND_UPDATE_AFTER_SENSITIVE)
val isMiniBatchEnabled = tableConfig.getConfiguration.getBoolean(
ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED)
val minRetentionTime = tableConfig.getMinIdleStateRetentionTime
- val operator = if (isMiniBatchEnabled) {
- val exeConfig = planner.getExecEnv.getConfig
- val rowSerializer = rowTypeInfo.createSerializer(exeConfig)
+
+ val rowtimeField = input.getRowType.getFieldList
+ .filter(f => FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
+ val rowtimeIndex = if (isRowtime) {
+ Preconditions.checkArgument(rowtimeField.nonEmpty)
+ rowtimeField.get(0).getIndex
+ } else {
+ -1
+ }
+
+ val miniBatchsize = if (isMiniBatchEnabled) {
+ val size = tableConfig.getConfiguration.getLong(
+ ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE)
+ Preconditions.checkArgument(size > 0)
+ size
+ } else {
+ -1L
+ }
+ val exeConfig = planner.getExecEnv.getConfig
+ val rowSerializer = rowTypeInfo.createSerializer(exeConfig)
+
+ val operator = if (isRowtime) {
+ if(isMiniBatchEnabled) {
+ val processFunction = if (keepLastRow) {
+ new RowTimeMiniBatchDeduplicateKeepLastRowFunction(
+ rowTypeInfo,
Review comment:
Can we have 2 sub-class here ? One is for `proctime` another is for `rowtime`, in each class, we can have a method to return the row function `getRowFunctction(isMiniBatchEnabled, keepLastRow)`.
There are too many if else branches here and it is hard to maintain.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -93,6 +95,239 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last row,
+ * retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param rowtimeIndex index of row time field
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ */
+ static void processLastRowOnRowtime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ int rowtimeIndex,
+ boolean generateUpdateBefore,
+ boolean generateInsert) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData prevRow = state.value();
+ if (!isLastRow(prevRow, currentRow, rowtimeIndex)) {
+ return;
+ }
+ state.update(currentRow);
+
+ // store all needed data to state
+ if (generateUpdateBefore || generateInsert) {
+ if (prevRow == null) {
+ // the first row, send INSERT message
+ currentRow.setRowKind(RowKind.INSERT);
+ out.collect(currentRow);
+ } else {
+ if (generateUpdateBefore) {
+ RowData copyRow = serializer.copy(prevRow);
+ copyRow.setRowKind(RowKind.UPDATE_BEFORE);
Review comment:
Do not need copy.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/DeduplicateFunctionHelper.java
##########
@@ -93,6 +95,239 @@ static void processFirstRow(
out.collect(currentRow);
}
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last row,
+ * retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param currentRow latest row received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param rowtimeIndex index of row time field
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ */
+ static void processLastRowOnRowtime(
+ ValueState<RowData> state,
+ RowData currentRow,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ int rowtimeIndex,
+ boolean generateUpdateBefore,
+ boolean generateInsert) throws Exception {
+
+ checkInsertOnly(currentRow);
+ RowData prevRow = state.value();
+ if (!isLastRow(prevRow, currentRow, rowtimeIndex)) {
+ return;
+ }
+ state.update(currentRow);
+
+ // store all needed data to state
+ if (generateUpdateBefore || generateInsert) {
+ if (prevRow == null) {
+ // the first row, send INSERT message
+ currentRow.setRowKind(RowKind.INSERT);
+ out.collect(currentRow);
+ } else {
+ if (generateUpdateBefore) {
+ RowData copyRow = serializer.copy(prevRow);
+ copyRow.setRowKind(RowKind.UPDATE_BEFORE);
+ out.collect(copyRow);
+ }
+ currentRow.setRowKind(RowKind.UPDATE_AFTER);
+ out.collect(currentRow);
+ }
+ } else {
+ currentRow.setRowKind(RowKind.UPDATE_AFTER);
+ out.collect(currentRow);
+ }
+ }
+
+ /**
+ * Processes element to deduplicate on keys with row time semantic, sends current element if it is last row,
+ * retracts previous element if needed.
+ *
+ * @param state state of function
+ * @param bufferedRows latest rows received by deduplicate function
+ * @param serializer serializer to serialize the data
+ * @param out underlying collector
+ * @param rowtimeIndex index of row time field
+ * @param generateUpdateBefore flag to generate UPDATE_BEFORE message or not
+ * @param generateInsert flag to gennerate INSERT message or not
+ */
+ static void processLastRowOnRowtime(
+ ValueState<RowData> state,
+ List<RowData> bufferedRows,
+ TypeSerializer<RowData> serializer,
+ Collector<RowData> out,
+ int rowtimeIndex,
+ boolean generateUpdateBefore,
+ boolean generateInsert) throws Exception {
+
+ if (bufferedRows == null) {
+ return;
+ }
+
+ RowData prevRow = state.value();
+ for (RowData currentRow: bufferedRows) {
+ checkInsertOnly(currentRow);
+ if (!isLastRow(prevRow, currentRow, rowtimeIndex)) {
+ continue;
+ }
+ // store all needed data to state
+ if (generateUpdateBefore || generateInsert) {
+ if (prevRow == null) {
+ // the first row, send INSERT message
+ currentRow.setRowKind(RowKind.INSERT);
+ out.collect(currentRow);
+ } else {
+ if (generateUpdateBefore) {
+ RowData copyRow = serializer.copy(prevRow);
+ copyRow.setRowKind(RowKind.UPDATE_BEFORE);
+ out.collect(copyRow);
+ }
Review comment:
Can reuse the code in
```java
static void processLastRowOnRowtime(
ValueState<RowData> state,
RowData currentRow,
TypeSerializer<RowData> serializer,
Collector<RowData> out,
int rowtimeIndex,
boolean generateUpdateBefore,
boolean generateInsert)
```
##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
##########
@@ -635,7 +635,25 @@ class FlinkRelMdHandlerTestBase {
// select a, b, c, proctime
// ROW_NUMBER() over (partition by b, c order by proctime desc) rn from TemporalTable3
// ) t where rn <= 1
- protected lazy val (streamDeduplicateFirstRow, streamDeduplicateLastRow) = {
+ protected lazy val (streamProcTimeDeduplicateFirstRow, streamProcTimeDeduplicateLastRow) = {
+ buildFirstRowAndLastRowDeduplicateNode(false)
+ }
+
+ // equivalent SQL is
+ // select a, b, c from (
+ // select a, b, c, rowtime
+ // ROW_NUMBER() over (partition by b order by rowtime) rn from TemporalTable3
+ // ) t where rn <= 1
+ //
+ // select a, b, c from (
+ // select a, b, c, proctime
+ // ROW_NUMBER() over (partition by b, c order by rowtime desc) rn from TemporalTable3
Review comment:
`proctime` => `rowtime` ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1fa2f9690c7d83d4a93720a53c1344620b8849e6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231)
* f37fa30b387d0c443139107f7b2a8dbdc6a8364d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 42f18261f80da589a648414d6bd81c20730e0cc0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208)
* 008e4ae2c2e10b712d84a4f324634a71fe5f84ac UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5e0c14bcf30a4640f90210be692e922a11d3dd1c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856)
* b4a2d274bd70b8006c267be02ee48ee5b9def3bb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b4a2d274bd70b8006c267be02ee48ee5b9def3bb Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861)
* 42f18261f80da589a648414d6bd81c20730e0cc0 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 008e4ae2c2e10b712d84a4f324634a71fe5f84ac Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235)
* 115ad9e4e8c7258d5831848a7144bfbb46a5b6f9 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6a79badae175128326ed4e77708b48488009731c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261)
* 6988d189034d8ad024e5db27962cf589273a5534 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 115ad9e4e8c7258d5831848a7144bfbb46a5b6f9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 115ad9e4e8c7258d5831848a7144bfbb46a5b6f9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478)
* f3b2192b6cc216c691e833e40e52dfc64807e0a0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
}, {
"hash" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8671",
"triggerID" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"triggerType" : "PUSH"
}, {
"hash" : "4e977036b9484acda997c8dd76709601bb70a418",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8677",
"triggerID" : "4e977036b9484acda997c8dd76709601bb70a418",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8703",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0fcea2d9a006b405865e558b2bdb07311b87c2a6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8703) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693)
* b3252a0d7e4481aebb849b2777ae544209cff1e8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13331: [FLINK-19079][table-runtime] Import rowtime deduplicate operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13331:
URL: https://github.com/apache/flink/pull/13331#issuecomment-687585649
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6231",
"triggerID" : "1fa2f9690c7d83d4a93720a53c1344620b8849e6",
"triggerType" : "PUSH"
}, {
"hash" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6247",
"triggerID" : "f37fa30b387d0c443139107f7b2a8dbdc6a8364d",
"triggerType" : "PUSH"
}, {
"hash" : "6a79badae175128326ed4e77708b48488009731c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6261",
"triggerID" : "6a79badae175128326ed4e77708b48488009731c",
"triggerType" : "PUSH"
}, {
"hash" : "6988d189034d8ad024e5db27962cf589273a5534",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6268",
"triggerID" : "6988d189034d8ad024e5db27962cf589273a5534",
"triggerType" : "PUSH"
}, {
"hash" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6279",
"triggerID" : "61470e6514fd5cb80250ed641e9e7e293c97fa07",
"triggerType" : "PUSH"
}, {
"hash" : "baed656a328a88791c638d87ac3b15091bc43408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6372",
"triggerID" : "baed656a328a88791c638d87ac3b15091bc43408",
"triggerType" : "PUSH"
}, {
"hash" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7856",
"triggerID" : "5e0c14bcf30a4640f90210be692e922a11d3dd1c",
"triggerType" : "PUSH"
}, {
"hash" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7861",
"triggerID" : "b4a2d274bd70b8006c267be02ee48ee5b9def3bb",
"triggerType" : "PUSH"
}, {
"hash" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8208",
"triggerID" : "42f18261f80da589a648414d6bd81c20730e0cc0",
"triggerType" : "PUSH"
}, {
"hash" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8235",
"triggerID" : "008e4ae2c2e10b712d84a4f324634a71fe5f84ac",
"triggerType" : "PUSH"
}, {
"hash" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8478",
"triggerID" : "115ad9e4e8c7258d5831848a7144bfbb46a5b6f9",
"triggerType" : "PUSH"
}, {
"hash" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8642",
"triggerID" : "f3b2192b6cc216c691e833e40e52dfc64807e0a0",
"triggerType" : "PUSH"
}, {
"hash" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8671",
"triggerID" : "d4463d6be66ef3a125e23c4a63c4efad8359454d",
"triggerType" : "PUSH"
}, {
"hash" : "4e977036b9484acda997c8dd76709601bb70a418",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8677",
"triggerID" : "4e977036b9484acda997c8dd76709601bb70a418",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"triggerType" : "PUSH"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8703",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "0fcea2d9a006b405865e558b2bdb07311b87c2a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8693",
"triggerID" : "719892276",
"triggerType" : "MANUAL"
}, {
"hash" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8987",
"triggerID" : "b3252a0d7e4481aebb849b2777ae544209cff1e8",
"triggerType" : "PUSH"
}, {
"hash" : "1755f4c73e181e591f39adaddd10413f75d6989e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9059",
"triggerID" : "1755f4c73e181e591f39adaddd10413f75d6989e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1755f4c73e181e591f39adaddd10413f75d6989e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9059)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org