You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "xuyangzhong (via GitHub)" <gi...@apache.org> on 2024/12/02 09:23:16 UTC
[PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
xuyangzhong opened a new pull request, #25717:
URL: https://github.com/apache/flink/pull/25717
## What is the purpose of the change
Introduce FastTop1Function in Rank with Async State API.
## Brief change log
- *Add async fast top1 function and fast top1 helper*
- *Add ITs and HarnessTests*
## Verifying this change
Existent tests and new added tests can verify this change.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? A independent pr will be introduced later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "lincoln-lil (via GitHub)" <gi...@apache.org>.
lincoln-lil merged PR #25717:
URL: https://github.com/apache/flink/pull/25717
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on PR #25717:
URL: https://github.com/apache/flink/pull/25717#issuecomment-2567257847
Hi, @davidradl thanks for your review! I have updated it, and can you take a look again when you have time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1903721138
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/asyncprocessing/AsyncStateKeyedProcessOperator.java:
##########
@@ -0,0 +1,196 @@
+/*
+ * 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.streaming.api.operators.asyncprocessing;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateUdfStreamOperator;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.SimpleTimerService;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.TimerService;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.TimestampedCollector;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link StreamOperator} for executing {@link KeyedProcessFunction KeyedProcessFunctions}.
+ *
+ * <p>This class is nearly identical with {@link KeyedProcessOperator}, but extending from {@link
+ * AbstractAsyncStateUdfStreamOperator} to integrate with asynchronous state access. Another
+ * difference is this class is internal.
+ */
+@Internal
+public class AsyncStateKeyedProcessOperator<K, IN, OUT>
+ extends AbstractAsyncStateUdfStreamOperator<OUT, KeyedProcessFunction<K, IN, OUT>>
+ implements OneInputStreamOperator<IN, OUT>, Triggerable<K, VoidNamespace> {
+
+ private static final long serialVersionUID = 1L;
+
+ private transient TimestampedCollector<OUT> collector;
+
+ private transient ContextImpl context;
+
+ private transient OnTimerContextImpl onTimerContext;
+
+ public AsyncStateKeyedProcessOperator(KeyedProcessFunction<K, IN, OUT> function) {
+ super(function);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ collector = new TimestampedCollector<>(output);
Review Comment:
The timestamp is set before the actual processing, which may involves several async procedure. In the meantime, other record may reset the timestamp, resulting in previous record may produce downstream event with wrong timestamp.
I have implement a class `TimestampedCollectorWithDeclaredVariable` to keep the timestamp alongwith record context. The timestamp is independent between records. Please use `AsyncKeyedProcessOperator` instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "lincoln-lil (via GitHub)" <gi...@apache.org>.
lincoln-lil commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1910000403
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/AbstractTopNFunction.java:
##########
@@ -44,7 +41,12 @@
import java.util.Comparator;
import java.util.function.Function;
-/** Base class for TopN Function. */
+/**
+ * Base class for TopN Function.
+ *
+ * <p>TODO: move util methods and metrics to AbstractTopNHelper after using AbstractTopNHelper in
Review Comment:
Is there any Jira tracking this？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1869709030
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/asyncprocessing/AbstractAsyncSyncStateTopNFunction.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.rank.asyncprocessing;
+
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.AbstractTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/**
+ * Base class for TopN Function with async state api.
+ *
+ * <p>Currently, only constant rank end is supported in async state rank.
Review Comment:
NIT: It would be helpful to say why this is the case, in the comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1909730165
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/async/AbstractAsyncStateTopNFunction.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.rank.async;
+
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.AbstractTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/**
+ * Base class for TopN Function with async state api.
+ *
+ * <p>TODO FLINK-36831 support variable rank end in async state rank later.
+ */
+public abstract class AbstractAsyncStateTopNFunction extends AbstractTopNFunction {
+
+ public AbstractAsyncStateTopNFunction(
+ StateTtlConfig ttlConfig,
+ InternalTypeInfo<RowData> inputRowType,
+ GeneratedRecordComparator generatedSortKeyComparator,
+ RowDataKeySelector sortKeySelector,
+ RankType rankType,
+ RankRange rankRange,
+ boolean generateUpdateBefore,
+ boolean outputRankNumber) {
+ super(
+ ttlConfig,
+ inputRowType,
+ generatedSortKeyComparator,
+ sortKeySelector,
+ rankType,
+ rankRange,
+ generateUpdateBefore,
+ outputRankNumber);
+ if (!isConstantRankEnd) {
+ throw new UnsupportedOperationException(
+ "Variable rank end is not supported in rank with async state api.");
Review Comment:
Nit:how about giving a hint of supported version in the error message?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/AbstractTopNFunction.java:
##########
@@ -326,4 +294,69 @@ private RowData createOutputRow(RowData inputRow, long rank, RowKind rowKind) {
public void setKeyContext(KeyContext keyContext) {
this.keyContext = keyContext;
}
+
+ /** An abstract helper to do the logic Top-n used for all top-n functions. */
+ public abstract static class AbstractTopNHelper {
+
+ protected final AbstractTopNFunction topNFunction;
+
+ protected final StateTtlConfig ttlConfig;
+
+ protected final KeySelector<RowData, RowData> sortKeySelector;
+
+ protected final Comparator<RowData> sortKeyComparator;
+
+ protected final boolean outputRankNumber;
+
+ protected final KeyContext keyContext;
+
+ // metrics
Review Comment:
Nit: could this comment be clarified further by elaborating on the meaning of this metric?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/FastTop1Function.java:
##########
@@ -112,66 +89,58 @@ public void open(OpenContext openContext) throws Exception {
}
dataState = getRuntimeContext().getState(valueStateDescriptor);
+ helper = new SyncStateFastTop1Helper();
+
// metrics
Review Comment:
Ditto
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/async/AsyncStateFastTop1Function.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.rank.async;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.state.StateFutureUtils;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.runtime.state.v2.ValueStateDescriptor;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.runtime.operators.asyncprocessing.AsyncStateProcessingOperator;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.AppendOnlyTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.FastTop1Function;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+import org.apache.flink.table.runtime.operators.rank.UpdatableTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.utils.FastTop1Helper;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+/**
+ * A more concise implementation for {@link AppendOnlyTopNFunction} and {@link
+ * UpdatableTopNFunction} when only Top-1 is desired. This function can handle updating stream
+ * because the RankProcessStrategy is inferred as UpdateFastStrategy, i.e., 1) the upsert key of
+ * input steam contains partition key; 2) the sort field is updated monotonely under the upsert key.
+ *
+ * <p>Different with {@link FastTop1Function}, this function is used with async state api.
+ */
+public class AsyncStateFastTop1Function extends AbstractAsyncStateTopNFunction
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private final TypeSerializer<RowData> inputRowSer;
+ private final long cacheSize;
+
+ // a value state stores the latest record
+ private transient ValueState<RowData> dataState;
+
+ private transient AsyncStateFastTop1Helper helper;
+
+ public AsyncStateFastTop1Function(
+ StateTtlConfig ttlConfig,
+ InternalTypeInfo<RowData> inputRowType,
+ GeneratedRecordComparator generatedSortKeyComparator,
+ RowDataKeySelector sortKeySelector,
+ RankType rankType,
+ RankRange rankRange,
+ boolean generateUpdateBefore,
+ boolean outputRankNumber,
+ long cacheSize) {
+ super(
+ ttlConfig,
+ inputRowType,
+ generatedSortKeyComparator,
+ sortKeySelector,
+ rankType,
+ rankRange,
+ generateUpdateBefore,
+ outputRankNumber);
+
+ this.inputRowSer = inputRowType.createSerializer(new SerializerConfigImpl());
+ this.cacheSize = cacheSize;
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ super.open(openContext);
+
+ ValueStateDescriptor<RowData> valueStateDescriptor =
+ new ValueStateDescriptor<>("Top1-Rank-State", inputRowType);
+ if (ttlConfig.isEnabled()) {
+ valueStateDescriptor.enableTimeToLive(ttlConfig);
+ }
+ dataState =
+ ((StreamingRuntimeContext) getRuntimeContext()).getValueState(valueStateDescriptor);
+
+ helper = new AsyncStateFastTop1Helper();
+
+ // metrics
Review Comment:
Nit: the comment here appears to be redundant because the method name is already clear and self-explanatory.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/FastTop1Function.java:
##########
@@ -112,66 +89,58 @@ public void open(OpenContext openContext) throws Exception {
}
dataState = getRuntimeContext().getState(valueStateDescriptor);
+ helper = new SyncStateFastTop1Helper();
+
// metrics
- registerMetric(kvCache.size() * getDefaultTopNSize());
+ helper.registerMetric();
}
@Override
public void processElement(RowData input, Context ctx, Collector<RowData> out)
throws Exception {
- requestCount += 1;
+ helper.accRequestCount();
+
// load state under current key if necessary
RowData currentKey = (RowData) keyContext.getCurrentKey();
- RowData prevRow = kvCache.getIfPresent(currentKey);
+ RowData prevRow = helper.getPrevRowFromCache(currentKey);
if (prevRow == null) {
prevRow = dataState.value();
} else {
- hitCount += 1;
+ helper.accHitCount();
}
// first row under current key.
if (prevRow == null) {
- kvCache.put(currentKey, inputRowSer.copy(input));
- if (outputRankNumber) {
- collectInsert(out, input, 1);
- } else {
- collectInsert(out, input);
- }
- return;
- }
-
- RowData curSortKey = sortKeySelector.getKey(input);
- RowData oldSortKey = sortKeySelector.getKey(prevRow);
- int compare = sortKeyComparator.compare(curSortKey, oldSortKey);
- // current sort key is higher than old sort key
- if (compare < 0) {
- kvCache.put(currentKey, inputRowSer.copy(input));
- // Note: partition key is unique key if only top-1 is desired,
- // thus emitting UB and UA here
- if (outputRankNumber) {
- collectUpdateBefore(out, prevRow, 1);
- collectUpdateAfter(out, input, 1);
- } else {
- collectUpdateBefore(out, prevRow);
- collectUpdateAfter(out, input);
- }
+ helper.processAsFirstRow(input, currentKey, out);
+ } else {
+ helper.processWithPrevRow(input, currentKey, prevRow, out);
}
}
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
- for (Map.Entry<RowData, RowData> entry : kvCache.asMap().entrySet()) {
- keyContext.setCurrentKey(entry.getKey());
- flushBufferToState(entry.getValue());
- }
+ helper.flushAllCacheToState();
}
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
// nothing to do
}
- private void flushBufferToState(RowData rowData) throws Exception {
- dataState.update(rowData);
+ private class SyncStateFastTop1Helper extends FastTop1Helper {
+
+ public SyncStateFastTop1Helper() {
+ super(
+ FastTop1Function.this,
+ inputRowSer,
+ cacheSize,
+ FastTop1Function.this.getDefaultTopNSize());
+ }
+
+ @Override
+ public void flushBufferToState(RowData currentKey, RowData value) throws Exception {
+ keyContext.setCurrentKey(currentKey);
+ FastTop1Function.this.dataState.update(value);
Review Comment:
This implementation breaks the encapsulation of the class by directly accessing the private field.. Would it be possible to add a new method in FastTop1Function to handle this logic instead?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/AbstractSyncStateTopNFunction.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.rank;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/** Base class for TopN Function with sync state api. */
+public abstract class AbstractSyncStateTopNFunction extends AbstractTopNFunction {
+
+ private ValueState<Long> rankEndState;
+
+ public AbstractSyncStateTopNFunction(
+ StateTtlConfig ttlConfig,
+ InternalTypeInfo<RowData> inputRowType,
+ GeneratedRecordComparator generatedSortKeyComparator,
+ RowDataKeySelector sortKeySelector,
+ RankType rankType,
+ RankRange rankRange,
+ boolean generateUpdateBefore,
+ boolean outputRankNumber) {
+ super(
+ ttlConfig,
+ inputRowType,
+ generatedSortKeyComparator,
+ sortKeySelector,
+ rankType,
+ rankRange,
+ generateUpdateBefore,
+ outputRankNumber);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ super.open(openContext);
+
+ if (!isConstantRankEnd) {
+ ValueStateDescriptor<Long> rankStateDesc =
+ new ValueStateDescriptor<>("rankEnd", Types.LONG);
+ if (ttlConfig.isEnabled()) {
+ rankStateDesc.enableTimeToLive(ttlConfig);
+ }
+ rankEndState = getRuntimeContext().getState(rankStateDesc);
+ }
+ }
+
+ /**
+ * Initialize rank end.
+ *
+ * @param row input record
+ * @return rank end
+ * @throws Exception
+ */
+ protected long initRankEnd(RowData row) throws Exception {
Review Comment:
The return value of this function is never used.
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/RankHarnessTest.scala:
##########
@@ -451,4 +467,128 @@ class RankHarnessTest(mode: StateBackendMode) extends HarnessTestBase(mode) {
assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
testHarness.close()
}
+
+ def prepareTop1Tester(query: String, operatorNameIdentifier: String)
+ : (KeyedOneInputStreamOperatorTestHarness[RowData, RowData, RowData], RowDataHarnessAssertor) = {
+ val sourceDDL =
+ s"""
+ |CREATE TEMPORARY TABLE T(
+ | a STRING PRIMARY KEY NOT ENFORCED,
+ | b BIGINT
+ |) WITH (
+ | 'connector' = 'values',
+ | 'changelog-mode' = 'I'
+ |)
+ |""".stripMargin
+ tEnv.executeSql(sourceDDL)
+
+ val t1 = tEnv.sqlQuery(query)
+
+ val testHarness =
+ createHarnessTester(t1.toRetractStream[Row], operatorNameIdentifier)
+ val assertor = new RowDataHarnessAssertor(
+ Array(
+ DataTypes.STRING().getLogicalType,
+ DataTypes.BIGINT().getLogicalType,
+ DataTypes.BIGINT().getLogicalType))
+
+ (testHarness, assertor)
+ }
+
+ @TestTemplate
+ def testAppendFastTop1(): Unit = {
+ tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(1))
+ val query =
+ """
+ |SELECT a, b, rn
+ |FROM
+ |(
+ | SELECT a, b,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY b DESC) AS rn
+ | FROM T
+ |) t1
+ |WHERE rn <= 1
+ """.stripMargin
+ val (testHarness, assertor) =
+ prepareTop1Tester(query, "Rank(strategy=[AppendFastStrategy")
+
+ if (enableAsyncState) {
+ assertThat(isAsyncStateOperator(testHarness)).isTrue
+ } else {
+ assertThat(isAsyncStateOperator(testHarness)).isFalse
+ }
+
+ testHarness.open()
+
+ testHarness.processElement(binaryRecord(INSERT, "a", 2L: JLong))
+ testHarness.processElement(binaryRecord(INSERT, "a", 1L: JLong))
+ testHarness.processElement(binaryRecord(INSERT, "a", 3L: JLong))
+
+ val result = dropWatermarks(testHarness.getOutput.toArray)
+
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+ expectedOutput.add(binaryRecord(INSERT, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+
+ assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
+
+ testHarness.close()
+ }
+
+ @TestTemplate
+ def testUpdateFastTop1(): Unit = {
+ tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(1))
+ val query =
+ """
+ |SELECT a, b, rn
+ |FROM
+ |(
+ | SELECT a, b,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY b DESC) AS rn
+ | FROM (
+ | select a, count(*) as b from T group by a
+ | ) t1
+ |) t2
+ |WHERE rn <= 1
+ """.stripMargin
+ val (testHarness, assertor) =
+ prepareTop1Tester(query, "Rank(strategy=[UpdateFastStrategy")
+
+ if (enableAsyncState) {
+ assertThat(isAsyncStateOperator(testHarness)).isTrue
+ } else {
+ assertThat(isAsyncStateOperator(testHarness)).isFalse
+ }
+
+ testHarness.open()
+
+ testHarness.processElement(binaryRecord(INSERT, "a", 2L: JLong))
+ testHarness.processElement(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+ testHarness.processElement(binaryRecord(UPDATE_AFTER, "a", 4L: JLong))
+
+ val result = dropWatermarks(testHarness.getOutput.toArray)
+
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+ expectedOutput.add(binaryRecord(INSERT, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 3L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 4L: JLong))
+
Review Comment:
This test has left me a bit confused. According to the SQL query, we are calculating the maximum number of rows for each distinct value of a. How does number which is greater than 1 is calculated in the expected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1912428298
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/FastTop1Function.java:
##########
@@ -112,66 +89,58 @@ public void open(OpenContext openContext) throws Exception {
}
dataState = getRuntimeContext().getState(valueStateDescriptor);
+ helper = new SyncStateFastTop1Helper();
+
// metrics
- registerMetric(kvCache.size() * getDefaultTopNSize());
+ helper.registerMetric();
}
@Override
public void processElement(RowData input, Context ctx, Collector<RowData> out)
throws Exception {
- requestCount += 1;
+ helper.accRequestCount();
+
// load state under current key if necessary
RowData currentKey = (RowData) keyContext.getCurrentKey();
- RowData prevRow = kvCache.getIfPresent(currentKey);
+ RowData prevRow = helper.getPrevRowFromCache(currentKey);
if (prevRow == null) {
prevRow = dataState.value();
} else {
- hitCount += 1;
+ helper.accHitCount();
}
// first row under current key.
if (prevRow == null) {
- kvCache.put(currentKey, inputRowSer.copy(input));
- if (outputRankNumber) {
- collectInsert(out, input, 1);
- } else {
- collectInsert(out, input);
- }
- return;
- }
-
- RowData curSortKey = sortKeySelector.getKey(input);
- RowData oldSortKey = sortKeySelector.getKey(prevRow);
- int compare = sortKeyComparator.compare(curSortKey, oldSortKey);
- // current sort key is higher than old sort key
- if (compare < 0) {
- kvCache.put(currentKey, inputRowSer.copy(input));
- // Note: partition key is unique key if only top-1 is desired,
- // thus emitting UB and UA here
- if (outputRankNumber) {
- collectUpdateBefore(out, prevRow, 1);
- collectUpdateAfter(out, input, 1);
- } else {
- collectUpdateBefore(out, prevRow);
- collectUpdateAfter(out, input);
- }
+ helper.processAsFirstRow(input, currentKey, out);
+ } else {
+ helper.processWithPrevRow(input, currentKey, prevRow, out);
}
}
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
- for (Map.Entry<RowData, RowData> entry : kvCache.asMap().entrySet()) {
- keyContext.setCurrentKey(entry.getKey());
- flushBufferToState(entry.getValue());
- }
+ helper.flushAllCacheToState();
}
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
// nothing to do
}
- private void flushBufferToState(RowData rowData) throws Exception {
- dataState.update(rowData);
+ private class SyncStateFastTop1Helper extends FastTop1Helper {
+
+ public SyncStateFastTop1Helper() {
+ super(
+ FastTop1Function.this,
+ inputRowSer,
+ cacheSize,
+ FastTop1Function.this.getDefaultTopNSize());
+ }
+
+ @Override
+ public void flushBufferToState(RowData currentKey, RowData value) throws Exception {
+ keyContext.setCurrentKey(currentKey);
+ FastTop1Function.this.dataState.update(value);
Review Comment:
Fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1909886420
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/RankHarnessTest.scala:
##########
@@ -451,4 +467,128 @@ class RankHarnessTest(mode: StateBackendMode) extends HarnessTestBase(mode) {
assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
testHarness.close()
}
+
+ def prepareTop1Tester(query: String, operatorNameIdentifier: String)
+ : (KeyedOneInputStreamOperatorTestHarness[RowData, RowData, RowData], RowDataHarnessAssertor) = {
+ val sourceDDL =
+ s"""
+ |CREATE TEMPORARY TABLE T(
+ | a STRING PRIMARY KEY NOT ENFORCED,
+ | b BIGINT
+ |) WITH (
+ | 'connector' = 'values',
+ | 'changelog-mode' = 'I'
+ |)
+ |""".stripMargin
+ tEnv.executeSql(sourceDDL)
+
+ val t1 = tEnv.sqlQuery(query)
+
+ val testHarness =
+ createHarnessTester(t1.toRetractStream[Row], operatorNameIdentifier)
+ val assertor = new RowDataHarnessAssertor(
+ Array(
+ DataTypes.STRING().getLogicalType,
+ DataTypes.BIGINT().getLogicalType,
+ DataTypes.BIGINT().getLogicalType))
+
+ (testHarness, assertor)
+ }
+
+ @TestTemplate
+ def testAppendFastTop1(): Unit = {
+ tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(1))
+ val query =
+ """
+ |SELECT a, b, rn
+ |FROM
+ |(
+ | SELECT a, b,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY b DESC) AS rn
+ | FROM T
+ |) t1
+ |WHERE rn <= 1
+ """.stripMargin
+ val (testHarness, assertor) =
+ prepareTop1Tester(query, "Rank(strategy=[AppendFastStrategy")
+
+ if (enableAsyncState) {
+ assertThat(isAsyncStateOperator(testHarness)).isTrue
+ } else {
+ assertThat(isAsyncStateOperator(testHarness)).isFalse
+ }
+
+ testHarness.open()
+
+ testHarness.processElement(binaryRecord(INSERT, "a", 2L: JLong))
+ testHarness.processElement(binaryRecord(INSERT, "a", 1L: JLong))
+ testHarness.processElement(binaryRecord(INSERT, "a", 3L: JLong))
+
+ val result = dropWatermarks(testHarness.getOutput.toArray)
+
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+ expectedOutput.add(binaryRecord(INSERT, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+
+ assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
+
+ testHarness.close()
+ }
+
+ @TestTemplate
+ def testUpdateFastTop1(): Unit = {
+ tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(1))
+ val query =
+ """
+ |SELECT a, b, rn
+ |FROM
+ |(
+ | SELECT a, b,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY b DESC) AS rn
+ | FROM (
+ | select a, count(*) as b from T group by a
+ | ) t1
+ |) t2
+ |WHERE rn <= 1
+ """.stripMargin
+ val (testHarness, assertor) =
+ prepareTop1Tester(query, "Rank(strategy=[UpdateFastStrategy")
+
+ if (enableAsyncState) {
+ assertThat(isAsyncStateOperator(testHarness)).isTrue
+ } else {
+ assertThat(isAsyncStateOperator(testHarness)).isFalse
+ }
+
+ testHarness.open()
+
+ testHarness.processElement(binaryRecord(INSERT, "a", 2L: JLong))
+ testHarness.processElement(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+ testHarness.processElement(binaryRecord(UPDATE_AFTER, "a", 4L: JLong))
+
+ val result = dropWatermarks(testHarness.getOutput.toArray)
+
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+ expectedOutput.add(binaryRecord(INSERT, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 3L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 4L: JLong))
+
Review Comment:
It's a changelog output and the final result is `"a", 4L: JLong`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1909891358
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/AbstractSyncStateTopNFunction.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.rank;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/** Base class for TopN Function with sync state api. */
+public abstract class AbstractSyncStateTopNFunction extends AbstractTopNFunction {
+
+ private ValueState<Long> rankEndState;
+
+ public AbstractSyncStateTopNFunction(
+ StateTtlConfig ttlConfig,
+ InternalTypeInfo<RowData> inputRowType,
+ GeneratedRecordComparator generatedSortKeyComparator,
+ RowDataKeySelector sortKeySelector,
+ RankType rankType,
+ RankRange rankRange,
+ boolean generateUpdateBefore,
+ boolean outputRankNumber) {
+ super(
+ ttlConfig,
+ inputRowType,
+ generatedSortKeyComparator,
+ sortKeySelector,
+ rankType,
+ rankRange,
+ generateUpdateBefore,
+ outputRankNumber);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ super.open(openContext);
+
+ if (!isConstantRankEnd) {
+ ValueStateDescriptor<Long> rankStateDesc =
+ new ValueStateDescriptor<>("rankEnd", Types.LONG);
+ if (ttlConfig.isEnabled()) {
+ rankStateDesc.enableTimeToLive(ttlConfig);
+ }
+ rankEndState = getRuntimeContext().getState(rankStateDesc);
+ }
+ }
+
+ /**
+ * Initialize rank end.
+ *
+ * @param row input record
+ * @return rank end
+ * @throws Exception
+ */
+ protected long initRankEnd(RowData row) throws Exception {
Review Comment:
Let's keep the current situation... It has always been like this, and I don't want to introduce any unnecessary changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1909899286
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/FastTop1Function.java:
##########
@@ -112,66 +89,58 @@ public void open(OpenContext openContext) throws Exception {
}
dataState = getRuntimeContext().getState(valueStateDescriptor);
+ helper = new SyncStateFastTop1Helper();
+
// metrics
- registerMetric(kvCache.size() * getDefaultTopNSize());
+ helper.registerMetric();
}
@Override
public void processElement(RowData input, Context ctx, Collector<RowData> out)
throws Exception {
- requestCount += 1;
+ helper.accRequestCount();
+
// load state under current key if necessary
RowData currentKey = (RowData) keyContext.getCurrentKey();
- RowData prevRow = kvCache.getIfPresent(currentKey);
+ RowData prevRow = helper.getPrevRowFromCache(currentKey);
if (prevRow == null) {
prevRow = dataState.value();
} else {
- hitCount += 1;
+ helper.accHitCount();
}
// first row under current key.
if (prevRow == null) {
- kvCache.put(currentKey, inputRowSer.copy(input));
- if (outputRankNumber) {
- collectInsert(out, input, 1);
- } else {
- collectInsert(out, input);
- }
- return;
- }
-
- RowData curSortKey = sortKeySelector.getKey(input);
- RowData oldSortKey = sortKeySelector.getKey(prevRow);
- int compare = sortKeyComparator.compare(curSortKey, oldSortKey);
- // current sort key is higher than old sort key
- if (compare < 0) {
- kvCache.put(currentKey, inputRowSer.copy(input));
- // Note: partition key is unique key if only top-1 is desired,
- // thus emitting UB and UA here
- if (outputRankNumber) {
- collectUpdateBefore(out, prevRow, 1);
- collectUpdateAfter(out, input, 1);
- } else {
- collectUpdateBefore(out, prevRow);
- collectUpdateAfter(out, input);
- }
+ helper.processAsFirstRow(input, currentKey, out);
+ } else {
+ helper.processWithPrevRow(input, currentKey, prevRow, out);
}
}
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
- for (Map.Entry<RowData, RowData> entry : kvCache.asMap().entrySet()) {
- keyContext.setCurrentKey(entry.getKey());
- flushBufferToState(entry.getValue());
- }
+ helper.flushAllCacheToState();
}
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
// nothing to do
}
- private void flushBufferToState(RowData rowData) throws Exception {
- dataState.update(rowData);
+ private class SyncStateFastTop1Helper extends FastTop1Helper {
+
+ public SyncStateFastTop1Helper() {
+ super(
+ FastTop1Function.this,
+ inputRowSer,
+ cacheSize,
+ FastTop1Function.this.getDefaultTopNSize());
+ }
+
+ @Override
+ public void flushBufferToState(RowData currentKey, RowData value) throws Exception {
+ keyContext.setCurrentKey(currentKey);
+ FastTop1Function.this.dataState.update(value);
Review Comment:
I don't think there's much of an issue because all access to this private class is managed within the FastTop1Function itself, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1898405938
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/asyncprocessing/AbstractAsyncSyncStateTopNFunction.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.rank.asyncprocessing;
+
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.AbstractTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/**
+ * Base class for TopN Function with async state api.
+ *
+ * <p>Currently, only constant rank end is supported in async state rank.
Review Comment:
Oh, it's just that FastTop1Rank currently doesn't require the variable rank end. It will be introduced in this PR later(https://github.com/apache/flink/pull/25723). Let me revise this section and add a TODO note.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1900519913
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecRank.java:
##########
@@ -250,17 +257,31 @@ protected Transformation<RowData> translateToPlanInternal(
generateUpdateBefore,
outputRankNumber);
} else if (RankUtil.isTop1(rankRange)) {
- processFunction =
- new FastTop1Function(
- ttlConfig,
- inputRowTypeInfo,
- sortKeyComparator,
- sortKeySelector,
- rankType,
- rankRange,
- generateUpdateBefore,
- outputRankNumber,
- cacheSize);
+ if (enableAsyncState) {
Review Comment:
I was trying to use a `RankOperatorBuilder` in table-runtime module like `WindowAggOperatorBuilder` before. However, some classes in table-planner need to be moved to table-runtime to adapt this change(such as SortSpec, RankProcessStrategy and etc).
I want to do this independent improvement in a new jira to avoid introducing many irrelevant changes in this jira. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1869737400
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/FastTop1Function.java:
##########
@@ -30,38 +30,28 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.utils.FastTop1Helper;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.util.Collector;
-import org.apache.flink.shaded.guava32.com.google.common.cache.Cache;
-import org.apache.flink.shaded.guava32.com.google.common.cache.CacheBuilder;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
/**
* A more concise implementation for {@link AppendOnlyTopNFunction} and {@link
* UpdatableTopNFunction} when only Top-1 is desired. This function can handle updating stream
* because the RankProcessStrategy is inferred as UpdateFastStrategy, i.e., 1) the upsert key of
* input steam contains partition key; 2) the sort field is updated monotonely under the upsert key.
Review Comment:
NIT: steam -> stream.
NIT: I am not sure what you are saying by the monotonely sentence - I wonder if you could rephrase please.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1898410221
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/FastTop1Function.java:
##########
@@ -30,38 +30,28 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.utils.FastTop1Helper;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.util.Collector;
-import org.apache.flink.shaded.guava32.com.google.common.cache.Cache;
-import org.apache.flink.shaded.guava32.com.google.common.cache.CacheBuilder;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
/**
* A more concise implementation for {@link AppendOnlyTopNFunction} and {@link
* UpdatableTopNFunction} when only Top-1 is desired. This function can handle updating stream
* because the RankProcessStrategy is inferred as UpdateFastStrategy, i.e., 1) the upsert key of
* input steam contains partition key; 2) the sort field is updated monotonely under the upsert key.
Review Comment:
In fact, I haven't changed those comments, but I think I can update them.
Regarding `monotonely`, you can find more info in `FlinkRelMdModifiedMonotonicity`. In simple terms, it refers to whether a particular column is monotonically increasing or monotonically decreasing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1909886994
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/async/AbstractAsyncStateTopNFunction.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.rank.async;
+
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.AbstractTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/**
+ * Base class for TopN Function with async state api.
+ *
+ * <p>TODO FLINK-36831 support variable rank end in async state rank later.
+ */
+public abstract class AbstractAsyncStateTopNFunction extends AbstractTopNFunction {
+
+ public AbstractAsyncStateTopNFunction(
+ StateTtlConfig ttlConfig,
+ InternalTypeInfo<RowData> inputRowType,
+ GeneratedRecordComparator generatedSortKeyComparator,
+ RowDataKeySelector sortKeySelector,
+ RankType rankType,
+ RankRange rankRange,
+ boolean generateUpdateBefore,
+ boolean outputRankNumber) {
+ super(
+ ttlConfig,
+ inputRowType,
+ generatedSortKeyComparator,
+ sortKeySelector,
+ rankType,
+ rankRange,
+ generateUpdateBefore,
+ outputRankNumber);
+ if (!isConstantRankEnd) {
+ throw new UnsupportedOperationException(
+ "Variable rank end is not supported in rank with async state api.");
Review Comment:
You can see the comment in this class: `<p>TODO FLINK-36831 support variable rank end in async state rank later.`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1912428355
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/AbstractSyncStateTopNFunction.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.rank;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/** Base class for TopN Function with sync state api. */
+public abstract class AbstractSyncStateTopNFunction extends AbstractTopNFunction {
+
+ private ValueState<Long> rankEndState;
+
+ public AbstractSyncStateTopNFunction(
+ StateTtlConfig ttlConfig,
+ InternalTypeInfo<RowData> inputRowType,
+ GeneratedRecordComparator generatedSortKeyComparator,
+ RowDataKeySelector sortKeySelector,
+ RankType rankType,
+ RankRange rankRange,
+ boolean generateUpdateBefore,
+ boolean outputRankNumber) {
+ super(
+ ttlConfig,
+ inputRowType,
+ generatedSortKeyComparator,
+ sortKeySelector,
+ rankType,
+ rankRange,
+ generateUpdateBefore,
+ outputRankNumber);
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ super.open(openContext);
+
+ if (!isConstantRankEnd) {
+ ValueStateDescriptor<Long> rankStateDesc =
+ new ValueStateDescriptor<>("rankEnd", Types.LONG);
+ if (ttlConfig.isEnabled()) {
+ rankStateDesc.enableTimeToLive(ttlConfig);
+ }
+ rankEndState = getRuntimeContext().getState(rankStateDesc);
+ }
+ }
+
+ /**
+ * Initialize rank end.
+ *
+ * @param row input record
+ * @return rank end
+ * @throws Exception
+ */
+ protected long initRankEnd(RowData row) throws Exception {
Review Comment:
Fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on PR #25717:
URL: https://github.com/apache/flink/pull/25717#issuecomment-2585695117
LGTM!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xishuaidelin (via GitHub)" <gi...@apache.org>.
xishuaidelin commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1907207187
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/asyncprocessing/AsyncStateFastTop1Function.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.rank.asyncprocessing;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.state.StateFutureUtils;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.runtime.state.v2.ValueStateDescriptor;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.runtime.operators.asyncprocessing.AsyncStateProcessingOperator;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.AppendOnlyTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.FastTop1Function;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+import org.apache.flink.table.runtime.operators.rank.UpdatableTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.utils.FastTop1Helper;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Collector;
+
+/**
+ * A more concise implementation for {@link AppendOnlyTopNFunction} and {@link
+ * UpdatableTopNFunction} when only Top-1 is desired. This function can handle updating stream
+ * because the RankProcessStrategy is inferred as UpdateFastStrategy, i.e., 1) the upsert key of
+ * input steam contains partition key; 2) the sort field is updated monotonely under the upsert key.
+ *
+ * <p>Different with {@link FastTop1Function}, this function is used with async state api.
+ */
+public class AsyncStateFastTop1Function extends AbstractAsyncSyncStateTopNFunction
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private final TypeSerializer<RowData> inputRowSer;
+ private final long cacheSize;
+
+ // a map state stores list of records
+ private transient ValueState<RowData> dataState;
Review Comment:
This comment is not accurate.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/asyncprocessing/AbstractAsyncSyncStateTopNFunction.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.rank.asyncprocessing;
+
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.AbstractTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/**
+ * Base class for TopN Function with async state api.
+ *
+ * <p>TODO FLINK-36831 support variable rank end in async state rank later.
+ */
+public abstract class AbstractAsyncSyncStateTopNFunction extends AbstractTopNFunction {
Review Comment:
rename to AbstractAsyncStateTopNFunction?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1909891982
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/RankHarnessTest.scala:
##########
@@ -451,4 +467,128 @@ class RankHarnessTest(mode: StateBackendMode) extends HarnessTestBase(mode) {
assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
testHarness.close()
}
+
+ def prepareTop1Tester(query: String, operatorNameIdentifier: String)
+ : (KeyedOneInputStreamOperatorTestHarness[RowData, RowData, RowData], RowDataHarnessAssertor) = {
+ val sourceDDL =
+ s"""
+ |CREATE TEMPORARY TABLE T(
+ | a STRING PRIMARY KEY NOT ENFORCED,
+ | b BIGINT
+ |) WITH (
+ | 'connector' = 'values',
+ | 'changelog-mode' = 'I'
+ |)
+ |""".stripMargin
+ tEnv.executeSql(sourceDDL)
+
+ val t1 = tEnv.sqlQuery(query)
+
+ val testHarness =
+ createHarnessTester(t1.toRetractStream[Row], operatorNameIdentifier)
+ val assertor = new RowDataHarnessAssertor(
+ Array(
+ DataTypes.STRING().getLogicalType,
+ DataTypes.BIGINT().getLogicalType,
+ DataTypes.BIGINT().getLogicalType))
+
+ (testHarness, assertor)
+ }
+
+ @TestTemplate
+ def testAppendFastTop1(): Unit = {
+ tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(1))
+ val query =
+ """
+ |SELECT a, b, rn
+ |FROM
+ |(
+ | SELECT a, b,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY b DESC) AS rn
+ | FROM T
+ |) t1
+ |WHERE rn <= 1
+ """.stripMargin
+ val (testHarness, assertor) =
+ prepareTop1Tester(query, "Rank(strategy=[AppendFastStrategy")
+
+ if (enableAsyncState) {
+ assertThat(isAsyncStateOperator(testHarness)).isTrue
+ } else {
+ assertThat(isAsyncStateOperator(testHarness)).isFalse
+ }
+
+ testHarness.open()
+
+ testHarness.processElement(binaryRecord(INSERT, "a", 2L: JLong))
+ testHarness.processElement(binaryRecord(INSERT, "a", 1L: JLong))
+ testHarness.processElement(binaryRecord(INSERT, "a", 3L: JLong))
+
+ val result = dropWatermarks(testHarness.getOutput.toArray)
+
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+ expectedOutput.add(binaryRecord(INSERT, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+
+ assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
+
+ testHarness.close()
+ }
+
+ @TestTemplate
+ def testUpdateFastTop1(): Unit = {
+ tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(1))
+ val query =
+ """
+ |SELECT a, b, rn
+ |FROM
+ |(
+ | SELECT a, b,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY b DESC) AS rn
+ | FROM (
+ | select a, count(*) as b from T group by a
+ | ) t1
+ |) t2
+ |WHERE rn <= 1
+ """.stripMargin
+ val (testHarness, assertor) =
+ prepareTop1Tester(query, "Rank(strategy=[UpdateFastStrategy")
+
+ if (enableAsyncState) {
+ assertThat(isAsyncStateOperator(testHarness)).isTrue
+ } else {
+ assertThat(isAsyncStateOperator(testHarness)).isFalse
+ }
+
+ testHarness.open()
+
+ testHarness.processElement(binaryRecord(INSERT, "a", 2L: JLong))
+ testHarness.processElement(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+ testHarness.processElement(binaryRecord(UPDATE_AFTER, "a", 4L: JLong))
+
+ val result = dropWatermarks(testHarness.getOutput.toArray)
+
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+ expectedOutput.add(binaryRecord(INSERT, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 3L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", 3L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", 4L: JLong))
+
Review Comment:
All input data is sent directly to Rank, not to Agg.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on PR #25717:
URL: https://github.com/apache/flink/pull/25717#issuecomment-2581904311
@xishuaidelin Updated! Could you please take a look again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1909886994
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/async/AbstractAsyncStateTopNFunction.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.rank.async;
+
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.operators.rank.AbstractTopNFunction;
+import org.apache.flink.table.runtime.operators.rank.RankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+/**
+ * Base class for TopN Function with async state api.
+ *
+ * <p>TODO FLINK-36831 support variable rank end in async state rank later.
+ */
+public abstract class AbstractAsyncStateTopNFunction extends AbstractTopNFunction {
+
+ public AbstractAsyncStateTopNFunction(
+ StateTtlConfig ttlConfig,
+ InternalTypeInfo<RowData> inputRowType,
+ GeneratedRecordComparator generatedSortKeyComparator,
+ RowDataKeySelector sortKeySelector,
+ RankType rankType,
+ RankRange rankRange,
+ boolean generateUpdateBefore,
+ boolean outputRankNumber) {
+ super(
+ ttlConfig,
+ inputRowType,
+ generatedSortKeyComparator,
+ sortKeySelector,
+ rankType,
+ rankRange,
+ generateUpdateBefore,
+ outputRankNumber);
+ if (!isConstantRankEnd) {
+ throw new UnsupportedOperationException(
+ "Variable rank end is not supported in rank with async state api.");
Review Comment:
You can see the comment in this class: `<p>TODO FLINK-36831 support variable rank end in async state rank later.` And it will be implemented in next pr https://github.com/apache/flink/pull/25723.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25717:
URL: https://github.com/apache/flink/pull/25717#issuecomment-2510998764
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "85c1a74cd71f64052f85722ef4d59e947e03cc26",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "85c1a74cd71f64052f85722ef4d59e947e03cc26",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 85c1a74cd71f64052f85722ef4d59e947e03cc26 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36825][table] Introduce FastTop1Function in Rank with Async State API [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25717:
URL: https://github.com/apache/flink/pull/25717#discussion_r1869702588
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecRank.java:
##########
@@ -250,17 +257,31 @@ protected Transformation<RowData> translateToPlanInternal(
generateUpdateBefore,
outputRankNumber);
} else if (RankUtil.isTop1(rankRange)) {
- processFunction =
- new FastTop1Function(
- ttlConfig,
- inputRowTypeInfo,
- sortKeyComparator,
- sortKeySelector,
- rankType,
- rankRange,
- generateUpdateBefore,
- outputRankNumber,
- cacheSize);
+ if (enableAsyncState) {
Review Comment:
NIT: I see the logic
```
if (RankUtil.isTop1(rankRange)) {
if (enableAsyncState) {
else
```
Repeated later in the file and in the tests. Can we remove the duplicate code, maybe as a static method in the Helper or maybe a provider / factory 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org