Skip to content

Commit

Permalink
[hotfix][table] Avoid sending duplicate delete record for left join w…
Browse files Browse the repository at this point in the history 
…hen state expired

This closes apache#24164
  • Loading branch information
@lincoln-lil
lincoln-lil authored Jan 22, 2024
1 parent c3c8362 commit b79d3f0
Show file tree
Hide file tree
Showing 2 changed files with 80 additions and 12 deletions.
9 changes: 3 additions & 6 deletions ...ain/java/org/apache/flink/table/runtime/operators/join/lookup/KeyedLookupJoinWrapper.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -150,15 +150,15 @@ public void processElement(
RowData rightRow = uniqueState.value();
// should distinguish null from empty(join condition unsatisfied)
if (null == rightRow) {
stateStaledErrorHandle(in, out);
stateStaledErrorHandle();
} else if (!emptyRow.equals(rightRow)) {
collectDeleteRow(in, rightRow, out);
collected = true;
}
} else {
List<RowData> rightRows = state.value();
if (null == rightRows) {
stateStaledErrorHandle(in, out);
stateStaledErrorHandle();
} else {
for (RowData row : rightRows) {
if (!emptyRow.equals(row)) {
Expand Down Expand Up @@ -238,12 +238,9 @@ public void onCollect(RowData record) {
}
}

private void stateStaledErrorHandle(RowData in, Collector out) {
private void stateStaledErrorHandle() {
if (lenient) {
LOG.warn(STATE_CLEARED_WARN_MSG);
if (lookupJoinRunner.isLeftOuterJoin) {
lookupJoinRunner.padNullForLeftJoin(in, out);
}
} else {
throw new RuntimeException(STATE_CLEARED_WARN_MSG);
}
Expand Down
83 changes: 77 additions & 6 deletions ...c/test/java/org/apache/flink/table/runtime/operators/join/KeyedLookupJoinHarnessTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,8 +74,6 @@ public class KeyedLookupJoinHarnessTest {
DataTypes.STRING().getLogicalType()
});

StateTtlConfig ttlConfig = StateConfigUtil.createTtlConfig(10_000_000);

@Test
public void testTemporalInnerJoin() throws Exception {
OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
Expand Down Expand Up @@ -352,12 +350,85 @@ public void testTemporalLeftJoinWithFilterLookupKeyContainsPk() throws Exception
testHarness.close();
}

// ---------------------------------------------------------------------------------
@Test
public void testTemporalLeftJoinWithTtlLookupKeyContainsPk() throws Exception {
OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
createHarness(JoinType.LEFT_JOIN, FilterOnTable.WITH_FILTER, true, 1_000);

testHarness.open();
// set TtlTimeProvider with 1
testHarness.setStateTtlProcessingTime(1);
testHarness.processElement(insertRecord(1, "a"));
testHarness.processElement(insertRecord(2, "b"));
testHarness.processElement(insertRecord(3, "c"));

// set TtlTimeProvider with 1001 to trigger expired state cleanup
testHarness.setStateTtlProcessingTime(1002);
// should output a delete message (pad null) since it's left join
testHarness.processElement(deleteRecord(2, "b"));

testHarness.processElement(insertRecord(2, "b2"));
testHarness.processElement(updateBeforeRecord(3, "c"));
testHarness.processElement(updateAfterRecord(3, "c2"));

List<Object> expectedOutput = new ArrayList<>();
expectedOutput.add(insertRecord(1, "a", 1, "Julian"));
expectedOutput.add(insertRecord(2, "b", null, null));
expectedOutput.add(insertRecord(3, "c", null, null));
expectedOutput.add(deleteRecord(2, "b", null, null));
expectedOutput.add(insertRecord(2, "b2", 2, "default-2"));
expectedOutput.add(deleteRecord(3, "c", null, null));
expectedOutput.add(insertRecord(3, "c2", 6, "Jark-2"));

assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.close();
}

@Test
public void testTemporalInnerJoinWithTtlLookupKeyContainsPk() throws Exception {
OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
createHarness(JoinType.INNER_JOIN, FilterOnTable.WITH_FILTER, true, 1_000);

testHarness.open();
// set TtlTimeProvider with 1
testHarness.setStateTtlProcessingTime(1);
testHarness.processElement(insertRecord(1, "a"));
testHarness.processElement(insertRecord(2, "b"));
testHarness.processElement(insertRecord(3, "c"));

@SuppressWarnings("unchecked")
// set TtlTimeProvider with 1001 to trigger expired state cleanup
testHarness.setStateTtlProcessingTime(1002);
// should not output a delete message (pad null) since it's inner join
testHarness.processElement(deleteRecord(2, "b"));

testHarness.processElement(insertRecord(2, "b2"));
testHarness.processElement(updateBeforeRecord(3, "c"));
testHarness.processElement(updateAfterRecord(3, "c2"));

List<Object> expectedOutput = new ArrayList<>();
expectedOutput.add(insertRecord(1, "a", 1, "Julian"));
expectedOutput.add(insertRecord(2, "b2", 2, "default-2"));
expectedOutput.add(insertRecord(3, "c2", 6, "Jark-2"));

assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.close();
}

// ---------------------------------------------------------------------------------
private KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData> createHarness(
JoinType joinType, FilterOnTable filterOnTable, boolean lookupKeyContainsPrimaryKey)
throws Exception {
return createHarness(joinType, filterOnTable, lookupKeyContainsPrimaryKey, -1);
}

private KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData> createHarness(
JoinType joinType,
FilterOnTable filterOnTable,
boolean lookupKeyContainsPrimaryKey,
long stateTtl)
throws Exception {
StateTtlConfig ttlConfig =
StateConfigUtil.createTtlConfig(stateTtl < 1 ? 1_000_000 : stateTtl);
boolean isLeftJoin = joinType == JoinType.LEFT_JOIN;
LookupJoinRunner joinRunner;
TestingEvolvingOutputFetcherFunction fetcher;
Expand All @@ -372,7 +443,7 @@ private KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData> create
new GeneratedFunctionWrapper<>(fetcher),
new GeneratedCollectorWrapper<>(
new LookupJoinHarnessTest.TestingFetcherCollector()),
new GeneratedFunctionWrapper(
new GeneratedFunctionWrapper<>(
new LookupJoinHarnessTest.TestingPreFilterCondition()),
isLeftJoin,
2);
Expand All @@ -384,7 +455,7 @@ private KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData> create
new LookupJoinHarnessTest.CalculateOnTemporalTable()),
new GeneratedCollectorWrapper<>(
new LookupJoinHarnessTest.TestingFetcherCollector()),
new GeneratedFunctionWrapper(
new GeneratedFunctionWrapper<>(
new LookupJoinHarnessTest.TestingPreFilterCondition()),
isLeftJoin,
2);
Expand Down

0 comments on commit b79d3f0

Please sign in to comment.