Au-Miner commented on code in PR #26649: URL: https://github.com/apache/flink/pull/26649#discussion_r2151872423
########## flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeltaJoinITCase.scala: ########## @@ -0,0 +1,520 @@ +/* + * 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.stream.sql + +import org.apache.flink.streaming.api.CheckpointingMode +import org.apache.flink.table.api.Schema +import org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl +import org.apache.flink.table.api.config.OptimizerConfigOptions +import org.apache.flink.table.api.config.OptimizerConfigOptions.DeltaJoinStrategy +import org.apache.flink.table.catalog.{CatalogTable, ObjectPath, ResolvedCatalogTable} +import org.apache.flink.table.planner.factories.TestValuesRuntimeFunctions.AsyncTestValueLookupFunction +import org.apache.flink.table.planner.factories.TestValuesTableFactory +import org.apache.flink.table.planner.factories.TestValuesTableFactory.changelogRow +import org.apache.flink.table.planner.runtime.utils.{FailingCollectionSource, StreamingTestBase} +import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension +import org.apache.flink.types.Row + +import org.assertj.core.api.Assertions.{assertThat, assertThatThrownBy} +import org.junit.jupiter.api.{BeforeEach, Test} +import org.junit.jupiter.api.extension.ExtendWith + +import javax.annotation.Nullable + +import java.time.LocalDateTime + +import scala.collection.JavaConversions._ + +@ExtendWith(Array(classOf[ParameterizedTestExtension])) +class DeltaJoinITCase() extends StreamingTestBase { + + @BeforeEach + override def before(): Unit = { + super.before() + + // avoid data disorder + env.setParallelism(1) + + tEnv.getConfig.set( + OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY, + DeltaJoinStrategy.FORCE) + + AsyncTestValueLookupFunction.invokeCount.set(0) + } + + @Test + def testJoinKeyEqualsPKWithSamePK(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Double.box(2.0), Int.box(1), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2023, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(11.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+U", Int.box(1), Double.box(22.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List("+I[2.0, 1, 2022-02-02T02:02:02, 1, 22.0, 2022-02-02T02:02:22]") + testUpsertResult(List("a0"), List("b0"), data1, data2, "a0 = b0", expected, 6) + } + + @Test + def testJoinKeyEqualsPKWithDifferentPK(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+I", Double.box(2.0), Int.box(2), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2023, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(11.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+I", Int.box(2), Double.box(22.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List( + "+I[1.0, 1, 2021-01-01T01:01:01, 1, 11.0, 2021-01-01T01:01:11]", + "+I[2.0, 2, 2022-02-02T02:02:02, 2, 22.0, 2022-02-02T02:02:22]" + ) + testUpsertResult(List("a0"), List("b0"), data1, data2, "a0 = b0", expected, 6) + } + + @Test + def testJoinKeyContainsPKWithSamePK(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Double.box(2.0), Int.box(1), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2023, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(2.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+U", Int.box(1), Double.box(2.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List("+I[2.0, 1, 2022-02-02T02:02:02, 1, 2.0, 2022-02-02T02:02:22]") + + testUpsertResult(List("a0"), List("b0"), data1, data2, "a0 = b0 and a1 = b1", expected, 6) + } + + @Test + def testJoinKeyContainsPKWithDifferentPK(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+I", Double.box(2.0), Int.box(2), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2033, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(1.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+I", Int.box(2), Double.box(2.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List( + "+I[1.0, 1, 2021-01-01T01:01:01, 1, 1.0, 2021-01-01T01:01:11]", + "+I[2.0, 2, 2022-02-02T02:02:02, 2, 2.0, 2022-02-02T02:02:22]" + ) + + testUpsertResult(List("a0"), List("b0"), data1, data2, "a0 = b0 and a1 = b1", expected, 6) + } + + @Test + def testJoinKeyEqualsIndexWithDifferentPK(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Double.box(2.0), Int.box(2), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2023, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(1.0), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Int.box(2), Double.box(2.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List( + "+I[1.0, 1, 2021-01-01T01:01:01, 1, 1.0, 2021-01-01T01:01:01]", + "+I[2.0, 2, 2022-02-02T02:02:02, 2, 2.0, 2022-02-02T02:02:22]" + ) + testUpsertResult(List("a1"), List("b1"), data1, data2, "a1 = b1", expected, 6) + } + + @Test + def testJoinKeyEqualsIndexWithSamePK(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Double.box(1.0), Int.box(1), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2023, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(1.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+U", Int.box(1), Double.box(1.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List("+I[1.0, 1, 2022-02-02T02:02:02, 1, 1.0, 2022-02-02T02:02:22]") + testUpsertResult(List("a1"), List("b1"), data1, data2, "a1 = b1", expected, 6) + } + + @Test + def testJoinKeyContainsIndexWithDifferentPK(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Double.box(2.0), Int.box(2), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2023, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(1.0), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Int.box(2), Double.box(2.0), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List( + "+I[2.0, 2, 2022-02-02T02:02:02, 2, 2.0, 2022-02-02T02:02:02]", + "+I[1.0, 1, 2021-01-01T01:01:01, 1, 1.0, 2021-01-01T01:01:01]" + ) + testUpsertResult(List("a1"), List("b1"), data1, data2, "a1 = b1 and a2 = b2", expected, 6) + } + + @Test + def testJoinKeyNotContainsIndexWithDifferentPK(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Double.box(1.0), Int.box(1), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2023, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(1.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+U", Int.box(1), Double.box(1.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List("+I[1.0, 1, 2022-02-02T02:02:02, 1, 1.0, 2022-02-02T02:02:22]") + + assertThatThrownBy( + () => + testUpsertResult(List("a0", "a1"), List("b0", "b1"), data1, data2, "a1 = b1", expected, 6)) + .hasMessageContaining("The current sql doesn't support to do delta join optimization.") + } + + @Test + def testNotEquiFilterAfterJoin(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+I", Double.box(2.0), Int.box(2), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2033, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(1.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+I", Int.box(2), Double.box(2.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List("+I[2.0, 2, 2022-02-02T02:02:02, 2, 2.0, 2022-02-02T02:02:22]") + + assertThatThrownBy( + () => + testUpsertResult( + List("a0"), + List("b0"), + data1, + data2, + "a0 = b0 and a1 = b1 and a2 > TO_TIMESTAMP('2021-01-01 01:01:11')", + expected, + 6)) + .hasMessageContaining("The current sql doesn't support to do delta join optimization.") + + assertThatThrownBy( + () => testUpsertResult(List(), List(), data1, data2, "a0 = b0 and b1 > 1.0", expected, 12)) + .hasMessageContaining("The current sql doesn't support to do delta join optimization.") + } + + @Test + def testEquiFilterAfterJoin(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+I", Double.box(2.0), Int.box(2), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2033, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(1.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+I", Int.box(2), Double.box(2.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List("+I[1.0, 1, 2021-01-01T01:01:01, 1, 1.0, 2021-01-01T01:01:11]") + + assertThatThrownBy( + () => + testUpsertResult( + List("a0"), + List("b0"), + data1, + data2, + "a0 = b0 and a1 = b1 and a2 = TO_TIMESTAMP('2021-01-01 01:01:01')", + expected, + 6)) + .hasMessageContaining("The current sql doesn't support to do delta join optimization.") + + assertThatThrownBy( + () => testUpsertResult(List(), List(), data1, data2, "a0 = b0 and b1 = 1.0", expected, 12)) + .hasMessageContaining("The current sql doesn't support to do delta join optimization.") + } + + @Test + def testFailOverAndRestore(): Unit = { + // enable checkpoint, we are using failing source to force have a complete checkpoint + // and cover restore path + env.enableCheckpointing(100, CheckpointingMode.EXACTLY_ONCE) + FailingCollectionSource.reset() + + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Double.box(2.0), Int.box(1), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2023, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(2.0), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Int.box(1), Double.box(2.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + // TestValuesRuntimeFunctions#KeyedUpsertingSinkFunction will change the RowKind from + // "+U" to "+I" + val expected = List("+I[2.0, 1, 2022-02-02T02:02:02, 1, 2.0, 2022-02-02T02:02:22]") + + testUpsertResult( + List("a0"), + List("b0"), + data1, + data2, + "a0 = b0 and a1 = b1", + expected, + null, + testFailingSource = true) + } + + @Test + def testProjectionAfterJoin(): Unit = { + val data1 = List( + changelogRow("+I", Double.box(1.0), Int.box(1), LocalDateTime.of(2021, 1, 1, 1, 1, 1)), + changelogRow("+U", Double.box(2.0), Int.box(1), LocalDateTime.of(2022, 2, 2, 2, 2, 2)), + // mismatch + changelogRow("+I", Double.box(3.0), Int.box(3), LocalDateTime.of(2033, 3, 3, 3, 3, 3)) + ) + + val data2 = List( + changelogRow("+I", Int.box(1), Double.box(1.0), LocalDateTime.of(2021, 1, 1, 1, 1, 11)), + changelogRow("+U", Int.box(1), Double.box(2.0), LocalDateTime.of(2022, 2, 2, 2, 2, 22)), + // mismatch + changelogRow("+I", Int.box(99), Double.box(99.0), LocalDateTime.of(2099, 2, 2, 2, 2, 2)) + ) + + prepareTable(List("a0"), List("b0"), data1, data2) + tEnv.executeSql(s""" + |create table projectedSink( + | l0 int, + | r0 int, + | l1 double, + | l2 timestamp(3), + | primary key(l0, r0) not enforced + |) with ( + | 'connector' = 'values', + | 'bounded' = 'false', + | 'sink-insert-only' = 'false' + |) + |""".stripMargin) + + assertThatThrownBy( + () => + tEnv + .executeSql( + s"insert into projectedSink select testLeft.a0, testRight.b0, testLeft.a1, testLeft.a2 " + + s"from testLeft join testRight on a0 = b0") + .await()) + .hasMessageContaining("The current sql doesn't support to do delta join optimization.") Review Comment: `testProjectFieldsAfterJoin` has added. -- 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]
