[FLINK-35184] Hash collision inside MiniBatchStreamingJoin operator - ASF JIRA

Attach files

Attach Screenshot

Voters

Watch issue

Watchers

Create sub-task

Link

Clone

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Bug
Status: Resolved
Priority: Major
Resolution: Fixed
Affects Version/s: 1.19.0
Fix Version/s: 1.20.0
Component/s: Table SQL / Runtime
Labels:
- pull-request-available

Description

The hash collision is possible for InputSideHasNoUniqueKeyBundle. To reproduce it just launch the following test within StreamingMiniBatchJoinOperatorTest:

@Tag("miniBatchSize=6")
@Test
public void testInnerJoinWithNoUniqueKeyHashCollision(TestInfo testInfo) throws Exception {

    leftTypeInfo =
            InternalTypeInfo.of(
                    RowType.of(
                            new LogicalType[] {new IntType(), new BigIntType()},
                            new String[] {"id1", "val1"}));

    rightTypeInfo =
            InternalTypeInfo.of(
                    RowType.of(
                            new LogicalType[] {new IntType(), new BigIntType()},
                            new String[] {"id2", "val2"}));

    leftKeySelector =
            HandwrittenSelectorUtil.getRowDataSelector(
                    new int[] {0},
                    leftTypeInfo.toRowType().getChildren().toArray(new LogicalType[0]));
    rightKeySelector =
            HandwrittenSelectorUtil.getRowDataSelector(
                    new int[] {0},
                    rightTypeInfo.toRowType().getChildren().toArray(new LogicalType[0]));

    joinKeyTypeInfo = InternalTypeInfo.of(new IntType());

    super.beforeEach(testInfo);

    testHarness.setStateTtlProcessingTime(1);
    testHarness.processElement2(insertRecord(1, 1L));
    testHarness.processElement1(insertRecord(1, 4294967296L));
    testHarness.processElement2(insertRecord(1, 4294967296L));
    testHarness.processElement2(deleteRecord(1, 1L));

    testHarness.close();

    assertor.shouldEmit(testHarness, rowOfKind(RowKind.INSERT, 1, 4294967296L, 1, 4294967296L));
}