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:
[email protected]
