You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2021/10/28 07:53:06 UTC
[flink] branch master updated: [FLINK-24654][table] Fix NPE on
RetractableTopNFunction when some records were cleared by state ttl
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 2ae9c4d [FLINK-24654][table] Fix NPE on RetractableTopNFunction when some records were cleared by state ttl
2ae9c4d is described below
commit 2ae9c4db0a00ab08c85f0813e473c8f1fce141ea
Author: lincoln lee <li...@gmail.com>
AuthorDate: Thu Oct 28 15:52:48 2021 +0800
[FLINK-24654][table] Fix NPE on RetractableTopNFunction when some records were cleared by state ttl
This closes #17570
---
.../planner/runtime/harness/RankHarnessTest.scala | 223 +++++++++++++++++++++
.../operators/rank/RetractableTopNFunction.java | 70 +++----
2 files changed, 258 insertions(+), 35 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/RankHarnessTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/RankHarnessTest.scala
new file mode 100644
index 0000000..a80cf0a
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/harness/RankHarnessTest.scala
@@ -0,0 +1,223 @@
+/*
+ * 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.planner.runtime.harness
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api._
+import org.apache.flink.table.api.bridge.scala._
+import org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl
+import org.apache.flink.table.planner.runtime.utils.JavaUserDefinedTableFunctions
+import org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase.StateBackendMode
+import org.apache.flink.table.runtime.util.RowDataHarnessAssertor
+import org.apache.flink.table.runtime.util.StreamRecordUtils.binaryRecord
+import org.apache.flink.types.Row
+import org.apache.flink.types.RowKind._
+
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.{Before, Test}
+
+import java.lang.{Long => JLong}
+import java.time.Duration
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import scala.collection.mutable
+
+@RunWith(classOf[Parameterized])
+class RankHarnessTest(mode: StateBackendMode) extends HarnessTestBase(mode) {
+
+ @Before
+ override def before(): Unit = {
+ super.before()
+ val setting = EnvironmentSettings.newInstance().inStreamingMode().build()
+ val config = new TestTableConfig
+ this.tEnv = StreamTableEnvironmentImpl.create(env, setting, config)
+ }
+
+ @Test
+ def testRetractRankWithRowNumber(): Unit = {
+ val data = new mutable.MutableList[(String, String, Long)]
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.createTemporaryView("T", t)
+ tEnv.createTemporarySystemFunction(
+ "STRING_SPLIT", new JavaUserDefinedTableFunctions.StringSplit)
+ tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(1))
+
+ val sql =
+ """
+ |SELECT a, b, c, id, rn1
+ |FROM (
+ | SELECT a, b, c, t3.id id,
+ | ROW_NUMBER() OVER (PARTITION BY a, t3.id ORDER BY c DESC) AS rn1
+ | FROM (
+ | SELECT a, b, c, rn
+ | FROM
+ | (
+ | -- append rank
+ | SELECT a, b, c,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY c DESC) AS rn
+ | FROM T
+ | ) t1
+ | WHERE rn = 1
+ | ) t2, LATERAL TABLE(STRING_SPLIT(b, '#')) AS t3(id)
+ |) WHERE rn1 <= 2
+ """.stripMargin
+
+ val t1 = tEnv.sqlQuery(sql)
+
+ val testHarness = createHarnessTester(t1.toRetractStream[Row], "Rank(strategy=[RetractStrategy")
+ val assertor = new RowDataHarnessAssertor(
+ Array(
+ DataTypes.STRING().getLogicalType,
+ DataTypes.STRING().getLogicalType,
+ DataTypes.BIGINT().getLogicalType,
+ DataTypes.STRING().getLogicalType,
+ DataTypes.BIGINT().getLogicalType))
+
+ testHarness.open()
+
+ // set TtlTimeProvider with 1
+ testHarness.setStateTtlProcessingTime(1)
+
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 1L: JLong, "1"))
+
+ testHarness.setStateTtlProcessingTime(1000)
+ testHarness.processElement(binaryRecord(DELETE, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(DELETE, "a", "1", 2L: JLong, "1"))
+
+ // set TtlTimeProvider with 1001 to trigger expired state cleanup
+ testHarness.setStateTtlProcessingTime(1001)
+ testHarness.processElement(binaryRecord(DELETE, "a", "1", 2L: JLong, "1"))
+
+ // currently there should only exists one record in state, test adding two new record
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+
+ val result = dropWatermarks(testHarness.getOutput.toArray)
+
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+
+ expectedOutput.add(binaryRecord(INSERT, "a", "1", 2L: JLong, "1", 1L: JLong))
+ expectedOutput.add(binaryRecord(INSERT, "a", "1", 2L: JLong, "1", 2L: JLong))
+
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", "1", 2L: JLong, "1", 1L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", "1", 2L: JLong, "1", 1L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", "1", 2L: JLong, "1", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", "1", 2L: JLong, "1", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", "1", 2L: JLong, "1", 1L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", "1", 2L: JLong, "1", 1L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", "1", 2L: JLong, "1", 2L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", "1", 2L: JLong, "1", 2L: JLong))
+
+ expectedOutput.add(binaryRecord(UPDATE_BEFORE, "a", "1", 2L: JLong, "1", 1L: JLong))
+ expectedOutput.add(binaryRecord(UPDATE_AFTER, "a", "1", 2L: JLong, "1", 1L: JLong))
+
+ expectedOutput.add(binaryRecord(DELETE, "a", "1", 2L: JLong, "1", 2L: JLong))
+
+ // if not expired, this is expected result
+ // expectedOutput.add(binaryRecord(INSERT, "a", "1", 1L: JLong, "1", 2L: JLong))
+
+ // only output one result
+ expectedOutput.add(binaryRecord(INSERT, "a", "1", 2L: JLong, "1", 2L: JLong))
+
+ assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
+ testHarness.close()
+ }
+
+ @Test
+ def testRetractRankWithoutRowNumber(): Unit = {
+ val data = new mutable.MutableList[(String, String, Long)]
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.createTemporaryView("T", t)
+ tEnv.createTemporarySystemFunction(
+ "STRING_SPLIT", new JavaUserDefinedTableFunctions.StringSplit)
+ tEnv.getConfig.setIdleStateRetention(Duration.ofSeconds(1))
+
+ val sql =
+ """
+ |SELECT a, b, c, id
+ |FROM (
+ | SELECT a, b, c, t3.id id,
+ | ROW_NUMBER() OVER (PARTITION BY a, t3.id ORDER BY c DESC) AS rn1
+ | FROM (
+ | SELECT a, b, c, rn
+ | FROM
+ | (
+ | -- append rank
+ | SELECT a, b, c,
+ | ROW_NUMBER() OVER (PARTITION BY a ORDER BY c DESC) AS rn
+ | FROM T
+ | ) t1
+ | WHERE rn = 1
+ | ) t2, LATERAL TABLE(STRING_SPLIT(b, '#')) AS t3(id)
+ |) WHERE rn1 = 1
+ """.stripMargin
+
+ val t1 = tEnv.sqlQuery(sql)
+
+ val testHarness = createHarnessTester(t1.toRetractStream[Row], "Rank(strategy=[RetractStrategy")
+ val assertor = new RowDataHarnessAssertor(
+ Array(
+ DataTypes.STRING().getLogicalType,
+ DataTypes.STRING().getLogicalType,
+ DataTypes.BIGINT().getLogicalType,
+ DataTypes.STRING().getLogicalType))
+
+ testHarness.open()
+
+ // set TtlTimeProvider with 1
+ testHarness.setStateTtlProcessingTime(1)
+
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 1L: JLong, "1"))
+
+ testHarness.setStateTtlProcessingTime(1000)
+ testHarness.processElement(binaryRecord(DELETE, "a", "1", 2L: JLong, "1"))
+
+ // set TtlTimeProvider with 1001 to trigger expired state cleanup
+ testHarness.setStateTtlProcessingTime(1001)
+ testHarness.processElement(binaryRecord(DELETE, "a", "1", 2L: JLong, "1"))
+
+ // currently there should not exists any record in state, test adding two new record
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ testHarness.processElement(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+
+ val result = dropWatermarks(testHarness.getOutput.toArray)
+
+ val expectedOutput = new ConcurrentLinkedQueue[Object]()
+
+ expectedOutput.add(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ expectedOutput.add(binaryRecord(DELETE, "a", "1", 2L: JLong, "1"))
+ expectedOutput.add(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+ expectedOutput.add(binaryRecord(DELETE, "a", "1", 2L: JLong, "1"))
+
+ // if not expired, this is expected result
+ // expectedOutput.add(binaryRecord(INSERT, "a", "1", 1L: JLong, "1"))
+
+ expectedOutput.add(binaryRecord(INSERT, "a", "1", 2L: JLong, "1"))
+
+ assertor.assertOutputEqualsSorted("result mismatch", expectedOutput, result)
+ testHarness.close()
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/RetractableTopNFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/RetractableTopNFunction.java
index da6a852..6694a12 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/RetractableTopNFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank/RetractableTopNFunction.java
@@ -224,6 +224,18 @@ public class RetractableTopNFunction extends AbstractTopNFunction {
// ------------- ROW_NUMBER-------------------------------
+ private void processStateStaled(Iterator<Map.Entry<RowData, Long>> sortedMapIterator)
+ throws RuntimeException {
+ // Skip the data if it's state is cleared because of state ttl.
+ if (lenient) {
+ // Sync with dataState
+ sortedMapIterator.remove();
+ LOG.warn(STATE_CLEARED_WARN_MSG);
+ } else {
+ throw new RuntimeException(STATE_CLEARED_WARN_MSG);
+ }
+ }
+
private void emitRecordsWithRowNumber(
SortedMap<RowData, Long> sortedMap,
RowData sortKey,
@@ -244,12 +256,7 @@ public class RetractableTopNFunction extends AbstractTopNFunction {
} else if (findsSortKey) {
List<RowData> inputs = dataState.get(key);
if (inputs == null) {
- // Skip the data if it's state is cleared because of state ttl.
- if (lenient) {
- LOG.warn(STATE_CLEARED_WARN_MSG);
- } else {
- throw new RuntimeException(STATE_CLEARED_WARN_MSG);
- }
+ processStateStaled(iterator);
} else {
int i = 0;
while (i < inputs.size() && isInRankEnd(currentRank)) {
@@ -294,12 +301,7 @@ public class RetractableTopNFunction extends AbstractTopNFunction {
} else if (findsSortKey) {
List<RowData> inputs = dataState.get(key);
if (inputs == null) {
- // Skip the data if it's state is cleared because of state ttl.
- if (lenient) {
- LOG.warn(STATE_CLEARED_WARN_MSG);
- } else {
- throw new RuntimeException(STATE_CLEARED_WARN_MSG);
- }
+ processStateStaled(iterator);
} else {
long count = entry.getValue();
// gets the rank of last record with same sortKey
@@ -346,12 +348,7 @@ public class RetractableTopNFunction extends AbstractTopNFunction {
if (!findsSortKey && key.equals(sortKey)) {
List<RowData> inputs = dataState.get(key);
if (inputs == null) {
- // Skip the data if it's state is cleared because of state ttl.
- if (lenient) {
- LOG.warn(STATE_CLEARED_WARN_MSG);
- } else {
- throw new RuntimeException(STATE_CLEARED_WARN_MSG);
- }
+ processStateStaled(iterator);
} else {
Iterator<RowData> inputIter = inputs.iterator();
while (inputIter.hasNext() && isInRankEnd(currentRank)) {
@@ -375,14 +372,18 @@ public class RetractableTopNFunction extends AbstractTopNFunction {
}
} else if (findsSortKey) {
List<RowData> inputs = dataState.get(key);
- int i = 0;
- while (i < inputs.size() && isInRankEnd(currentRank)) {
- RowData currentRow = inputs.get(i);
- collectUpdateBefore(out, prevRow, currentRank);
- collectUpdateAfter(out, currentRow, currentRank);
- prevRow = currentRow;
- currentRank += 1;
- i++;
+ if (inputs == null) {
+ processStateStaled(iterator);
+ } else {
+ int i = 0;
+ while (i < inputs.size() && isInRankEnd(currentRank)) {
+ RowData currentRow = inputs.get(i);
+ collectUpdateBefore(out, prevRow, currentRank);
+ collectUpdateAfter(out, currentRow, currentRank);
+ prevRow = currentRow;
+ currentRank += 1;
+ i++;
+ }
}
} else {
currentRank += entry.getValue();
@@ -417,12 +418,7 @@ public class RetractableTopNFunction extends AbstractTopNFunction {
if (!findsSortKey && key.equals(sortKey)) {
List<RowData> inputs = dataState.get(key);
if (inputs == null) {
- // Skip the data if it's state is cleared because of state ttl.
- if (lenient) {
- LOG.warn(STATE_CLEARED_WARN_MSG);
- } else {
- throw new RuntimeException(STATE_CLEARED_WARN_MSG);
- }
+ processStateStaled(iterator);
} else {
Iterator<RowData> inputIter = inputs.iterator();
while (inputIter.hasNext() && isInRankEnd(nextRank)) {
@@ -455,9 +451,13 @@ public class RetractableTopNFunction extends AbstractTopNFunction {
// sends the record if there is a record recently upgrades to Top-N
int index = Long.valueOf(rankEnd - nextRank).intValue();
List<RowData> inputs = dataState.get(key);
- RowData toAdd = inputs.get(index);
- collectInsert(out, toAdd);
- break;
+ if (inputs == null) {
+ processStateStaled(iterator);
+ } else {
+ RowData toAdd = inputs.get(index);
+ collectInsert(out, toAdd);
+ break;
+ }
}
} else {
nextRank += entry.getValue();