[ https://issues.apache.org/jira/browse/FLINK-35184?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17839540#comment-17839540 ]
Roman Boyko commented on FLINK-35184: ------------------------------------- Most probably the easiest way to fix it would be to replace the Map<hashCode, List<record>> to Map<record, List<rowKind>> inside InputSideHasNoUniqueKeyBundle. In such attempt we should store every record as map key with constant rowKind=+I, while rowkinds would be stored as values. And before providing the record to processing we should set the proper rowKind back to it. > Hash collision inside MiniBatchStreamingJoin operator > ----------------------------------------------------- > > Key: FLINK-35184 > URL: https://issues.apache.org/jira/browse/FLINK-35184 > Project: Flink > Issue Type: Bug > Components: Table SQL / Runtime > Affects Versions: 1.19.0 > Reporter: Roman Boyko > Priority: Major > > The hash collision is possible for InputSideHasNoUniqueKeyBundle. To > reproduce it just launch the following test within > StreamingMiniBatchJoinOperatorTest: > > {code:java} > @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)); > } {code} > > -- This message was sent by Atlassian Jira (v8.20.10#820010)