Repository: apache/flink
File: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeSemanticTestPrograms.java (Java)

Code before:
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.table.planner.plan.nodes.exec.stream;

import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

/**
 * {@link TableTestProgram} definitions for semantic testing {@link StreamExecChangelogNormalize}.
 */
public class ChangelogNormalizeSemanticTestPrograms {

    private static final String[] SINK_SCHEMA = {
        "a STRING", "b INT", "c STRING", "PRIMARY KEY(a) NOT ENFORCED"
    };
    private static final String[] SOURCE_SCHEMA = {
        "a STRING", "b INT", "c STRING", "PRIMARY KEY(a) NOT ENFORCED"
    };

    static final TableTestProgram UPSERT_SOURCE_WITH_FILTER =
            TableTestProgram.of("upsert-with-filter", "validates upsert with filter")
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .enforceUpsert()
                                    .addSchema(SOURCE_SCHEMA)
                                    .producedValues(
                                            Row.ofKind(RowKind.INSERT, "one", 1, "a"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 2, "bb"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "two", 0, "dd"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "two", 1, "dd"),
                                            Row.ofKind(RowKind.INSERT, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.DELETE, "three", 5, "ccc"))
                                    .build())
                    .setupTableSink(
                            SinkTestStep.newBuilder("sink_t")
                                    .enforceRetract()
                                    .addSchema(SINK_SCHEMA)
                                    .consumedValues(
                                            "+I[one, 1, a]",
                                            "-U[one, 1, a]",
                                            "+U[one, 2, bb]",
                                            "-D[one, 2, bb]",
                                            "+I[two, 0, dd]",

Review comment (suggestion):
```suggestion
            TableTestProgram.of("upsert-non-key-filter", "validates upsert with filter")
```

Code after:
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.table.planner.plan.nodes.exec.stream;

import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

/**
 * {@link TableTestProgram} definitions for semantic testing {@link StreamExecChangelogNormalize}.
 */
public class ChangelogNormalizeSemanticTestPrograms {

    private static final String[] SINK_SCHEMA = {
        "a STRING", "b INT", "c STRING", "PRIMARY KEY(a) NOT ENFORCED"
    };

    private static final String[] SOURCE_SCHEMA = {
        "a STRING", "b INT", "c STRING", "PRIMARY KEY(a) NOT ENFORCED"
    };

    private static final ChangelogMode KAFKA_WITH_PARTIAL_DELETES_CHANGELOG =
            ChangelogMode.newBuilder()
                    .addContainedKind(RowKind.UPDATE_AFTER)
                    .addContainedKind(RowKind.DELETE)
                    .keyOnlyDeletes(true)
                    .build();

    static final TableTestProgram UPSERT_SOURCE_WITH_NON_KEY_FILTER =
            TableTestProgram.of(
                            "upsert-with-non-key-filter", "validates upsert with non key filter")
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .addMode(ChangelogMode.upsert())
                                    .addSchema(SOURCE_SCHEMA)
                                    .producedValues(
                                            Row.ofKind(RowKind.INSERT, "one", 1, "a"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 2, "bb"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "two", 0, "dd"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "two", 1, "dd"),
                                            Row.ofKind(RowKind.INSERT, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.DELETE, "three", 5, "ccc"))
                                    .build())