gustavodemorais commented on code in PR #29583: URL: https://github.com/apache/flink/pull/29583#discussion_r4496292252
########## flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/multijoin/StreamingMultiJoinUpsertOperatorTest.java: ########## @@ -0,0 +1,280 @@ +/* + * 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.join.stream.multijoin; + +import org.apache.flink.api.common.functions.AbstractRichFunction; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.generated.GeneratedJoinCondition; +import org.apache.flink.table.runtime.generated.JoinCondition; +import org.apache.flink.table.runtime.keyselector.RowDataKeySelector; +import org.apache.flink.table.runtime.operators.join.FlinkJoinType; +import org.apache.flink.table.runtime.operators.join.stream.keyselector.AttributeBasedJoinKeyExtractor.ConditionAttributeRef; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.runtime.util.RuntimeChangelogMode; +import org.apache.flink.table.types.logical.CharType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.utils.HandwrittenSelectorUtil; +import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; + +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests that the multi-way join handles inputs that update a record without UPDATE_BEFORE and + * delete it with only its unique key. The joined rows and null padding of the old version of a + * record are retracted. + * + * <p>SQL: SELECT * FROM Users u JOIN Orders o ON u.user_id = o.user_id AND u.details <> o.details + * LEFT JOIN Payments p ON o.user_id = p.user_id AND o.details <> p.details LEFT JOIN Shipments s ON + * p.user_id = s.user_id AND p.details <> s.details + * + * <p>Schema: Users(user_id PRIMARY KEY, name, details), Orders, Payments and Shipments(user_id, id, + * details, PRIMARY KEY (user_id, id)). All inputs are upsert inputs, whose keys contain their join + * key user_id. + */ +@ExtendWith(ParameterizedTestExtension.class) +class StreamingMultiJoinUpsertOperatorTest extends StreamingMultiJoinOperatorTestBase { + + private static final Object[] NO_ROW = {null, null, null}; + private static final Object[] USER = {"1", "Gus", "A"}; + private static final Object[] ORDER = {"1", "o1", "B"}; + private static final Object[] PAYMENT = {"1", "p1", "X"}; + + public StreamingMultiJoinUpsertOperatorTest(StateBackendMode stateBackendMode) { + super( + stateBackendMode, + 4, + List.of( + FlinkJoinType.INNER, + FlinkJoinType.INNER, + FlinkJoinType.LEFT, + FlinkJoinType.LEFT), + Arrays.asList( + null, + createAndCondition(equalTo(0, 0), notEqualTo(2, 2)), + createAndCondition(equalTo(3, 0), notEqualTo(5, 2)), + createAndCondition(equalTo(6, 0), notEqualTo(8, 2))), + Map.of( + 1, + List.of(new ConditionAttributeRef(0, 0, 1, 0)), + 2, + List.of(new ConditionAttributeRef(1, 0, 2, 0)), + 3, + List.of(new ConditionAttributeRef(2, 0, 3, 0))), + false); + } + + @TestTemplate + void testUpdateThatLosesMatchRetractsJoinedRow() throws Exception { + insertUserAndOrder(); + + updateAfterOrder("1", "o1", "A"); + emits(DELETE, row(USER, ORDER, NO_ROW, NO_ROW)); + + // only the new version of the order is stored + updateAfterUser("1", "Gus", "C"); + emits(UPDATE_AFTER, row(r("1", "Gus", "C"), r("1", "o1", "A"), NO_ROW, NO_ROW)); + } + + @TestTemplate + void testInsertThatReplacesStoredRecordRetractsOldVersion() throws Exception { + insertUserAndOrder(); + + insertOrder("1", "o1", "A"); + emits(DELETE, row(USER, ORDER, NO_ROW, NO_ROW)); + } + + @TestTemplate + void testUpdateOfFirstInputThatLosesMatchRetractsJoinedRow() throws Exception { + insertUserAndOrder(); + + updateAfterUser("1", "Gus", "B"); + emits(DELETE, row(USER, ORDER, NO_ROW, NO_ROW)); + } + + @TestTemplate + void testUpdateWithSameConditionFieldsUpdatesJoinedRow() throws Exception { + insertUserAndOrder(); + + updateAfterUser("1", "Bob", "A"); + emits(UPDATE_AFTER, row(r("1", "Bob", "A"), ORDER, NO_ROW, NO_ROW)); + } + + @TestTemplate + void testKeyOnlyDeleteRetractsStoredRecord() throws Exception { + insertUserAndOrder(); + + deleteOrder("1", "o1", null); + emits(DELETE, row(USER, ORDER, NO_ROW, NO_ROW)); + assertThat(testHarness.numKeyedStateEntries()).isOne(); Review Comment: This doesn't compile: `KeyedMultiInputStreamOperatorTestHarness` has no `numKeyedStateEntries()`. Could we assert it through the output instead, e.g. a later update of the user no longer joins `o1`? ########## flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java: ########## @@ -406,10 +432,93 @@ public void processElement(int inputId, StreamRecord<RowData> element) throws Ex return; } + if (isUpsert(input)) { + final RowData replacedRecord = replacedRecord(input, inputId); + if (replacedRecord != null) { + replaceRecord(replacedRecord, input, inputId); + return; + } + } else { + input = retractedRecord(input, inputId); + } performMultiJoin(input, inputId); addRecordToState(input, inputId); } + /** + * Returns the stored record that the given record replaces, or null if there is none or if the + * join conditions match both versions the same way. An upsert input updates a record by sending + * only the new version, without retracting the old one. + */ + private @Nullable RowData replacedRecord(RowData record, int inputId) throws Exception { + if (!isUpsertInput[inputId] + || !inputSpecs.get(inputId).hasUniqueKey() + || conditionFields[inputId].length == 0) { + return null; + } + final RowData storedRecord = + stateHandlers + .get(inputId) + .getRecord(keyExtractor.getJoinKey(record, inputId), record); + if (storedRecord == null || conditionFieldsEqual(storedRecord, record, inputId)) { + // the new version produces the same joined rows, which update the old ones + return null; + } + return storedRecord; + } + + /** + * Returns the full record to retract. A DELETE may contain only the unique key, e.g. from an + * upsert source, so the stored record with the same unique key is returned if there is one. + */ + private RowData retractedRecord(RowData retraction, int inputId) throws Exception { + if (retraction.getRowKind() != RowKind.DELETE + || !hasKeyOnlyDeletes[inputId] + || !inputSpecs.get(inputId).hasUniqueKey() + || conditionFields[inputId].length == 0) { + return retraction; + } + final RowData storedRecord = + stateHandlers + .get(inputId) + .getRecord(keyExtractor.getJoinKey(retraction, inputId), retraction); + if (storedRecord == null) { + return retraction; + } + // the stored record is removed from state with this retraction + storedRecord.setRowKind(RowKind.DELETE); + return storedRecord; + } + + /** + * Replaces a stored record with its new version: the joined rows and null padding of the old + * version are retracted, then the new version is joined and stored. + */ + private void replaceRecord(RowData replacedRecord, RowData record, int inputId) + throws Exception { + // set the RowKind back, because the record may be the one in state + replacedRecord.setRowKind(RowKind.DELETE); + performMultiJoin(replacedRecord, inputId); + replacedRecord.setRowKind(RowKind.INSERT); + // the old version must not count as a match of the other inputs for the new version + stateHandlers + .get(inputId) + .retractRecord(keyExtractor.getJoinKey(replacedRecord, inputId), replacedRecord); + performMultiJoin(record, inputId); + addRecordToState(record, inputId); + } + + private boolean conditionFieldsEqual(RowData storedRecord, RowData record, int inputId) { + for (int field : conditionFields[inputId]) { + final RowData.FieldGetter getter = inputFieldGetters[inputId][field]; + if (!Objects.equals( + getter.getFieldOrNull(storedRecord), getter.getFieldOrNull(record))) { Review Comment: This runs on every upsert update: `getFieldOrNull` boxes primitives, and `Objects.equals` compares `byte[]` by reference, so BINARY/VARBINARY fields always count as changed and take the full replace path. Could we generate a `RecordEqualiser` over the condition fields in the planner, like ChangelogNormalize and SinkUpsertMaterializer do? ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMultiJoin.java: ########## @@ -382,36 +394,51 @@ private String formatInputUniqueKeysWithFieldNames() { /** * This is mainly used in `FlinkChangelogModeInferenceProgram.SatisfyUpdateKindTraitVisitor`. If - * the unique key of input is a superset of the common join key, then we can ignore - * UPDATE_BEFORE. Otherwise, we can't ignore UPDATE_BEFORE. + * an upsert key of the input contains its common join key and its join key, then we can ignore + * UPDATE_BEFORE and handle deletes by key. Otherwise, we can't. * - * <p>For example, if the input schema is [id, name, cnt] with the unique key (id) and the + * <p>For example, if the input schema is [id, name, cnt] with the upsert key (id) and the * common join key is (id, name) across joins, then an insert and update on the id: * * <p>+I(1001, Tim, 10) -U(1001, Tim, 10) +U(1001, Timo, 11) * * <p>If the UPDATE_BEFORE is ignored, the `+I(1001, Tim, 10)` record in join will never be - * retracted. Therefore, if we want to ignore UPDATE_BEFORE, the unique key must contain join - * key. + * retracted. Therefore, if we want to ignore UPDATE_BEFORE, the upsert key must contain the + * join key, so that all versions of a record are stored under the same key. * - * <p>This is similar to {@link StreamPhysicalJoin#inputUniqueKeyContainsJoinKey(int)} but here - * we use the common join key, since the multi join operator partitions on the common join key. + * <p>This is similar to {@link StreamPhysicalJoin#inputUniqueKeyContainsJoinKey(int)}, but the + * multi join operator partitions on the common join key and stores each input under its own + * join key, so the upsert key must contain both. */ - public boolean inputUniqueKeyContainsCommonJoinKey(int inputId) { - final RelNode input = getInputs().get(inputId); - final Set<ImmutableBitSet> inputUniqueKeysSet = getUniqueKeys(input); - if (inputUniqueKeysSet == null || inputUniqueKeysSet.isEmpty()) { - return false; - } + public boolean inputUpsertKeyContainsJoinKey(int inputId) { + return !getUpsertKeysContainingJoinKey(inputId).isEmpty(); + } + /** + * Returns the upsert keys of the input that contain its common join key and its join key. An + * input without UPDATE_BEFORE replaces the record with the same upsert key, which a unique key + * that isn't an upsert key doesn't identify. + */ + private List<int[]> getUpsertKeysContainingJoinKey(int inputId) { final int[] commonJoinKeyIndices = keyExtractor.getCommonJoinKeyIndices(inputId); if (commonJoinKeyIndices.length == 0) { - return false; + return Collections.emptyList(); } - - final ImmutableBitSet commonJoinKeys = ImmutableBitSet.of(commonJoinKeyIndices); - return inputUniqueKeysSet.stream() - .anyMatch(uniqueKey -> uniqueKey.contains(commonJoinKeys)); + final RelNode input = inputs.get(inputId); + final FlinkRelMetadataQuery fmq = + FlinkRelMetadataQuery.reuseOrCreate(input.getCluster().getMetadataQuery()); + final Set<ImmutableBitSet> upsertKeys = + fmq.getUpsertKeysInKeyGroupRange(input, commonJoinKeyIndices); + if (upsertKeys == null) { + return Collections.emptyList(); Review Comment: Input 0 isn't keyed by its join key (`getJoinKey` returns null for it), so a change of its join columns keeps its state key. Could we skip the join-key part of the rule for input 0? That avoids a ChangelogNormalize for e.g. `UsersPK u JOIN UserRegionsPK r ON u.user_id = r.user_id AND u.name = r.region`. Fine as a follow-up too. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
