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

Code before:
                            "CREATE VIEW UsersNullSafe AS SELECT * FROM (VALUES "
                                    + "('1','Gus'),"
                                    + "(CAST(NULL AS STRING), 'NullUser')"
                                    + ") AS T(user_id, name)")
                    .setupSql(
                            "CREATE VIEW OrdersNullSafe AS SELECT * FROM (VALUES "
                                    + "('1','order1'),"
                                    + "(CAST(NULL AS STRING), 'nullOrder')"
                                    + ") AS T(user_id, order_id)")
                    .setupTableSink(
                            SinkTestStep.newBuilder("sink")
                                    .addSchema("user_id STRING", "name STRING", "order_id STRING")
                                    .consumedValues(
                                            "+I[1, Gus, order1]", "+I[null, NullUser, nullOrder]")
                                    .testMaterializedData()
                                    .build())
                    .runSql(
                            "INSERT INTO sink "
                                    + "SELECT u.user_id, u.name, o.order_id "
                                    + "FROM UsersNullSafe u "
                                    + "INNER JOIN OrdersNullSafe o ON u.user_id IS NOT DISTINCT FROM o.user_id")
                    .build();

    public static final TableTestProgram MULTI_JOIN_CONDITIONS_MATERIALIZATION_NEXMARK_AGGREGATION =
            TableTestProgram.of(
                            "nexmark-auction-bid-aggregation",
                            "Nexmark auction and bid aggregation with time-based filtering")
                    .setupTableSource(
                            SourceTestStep.newBuilder("auctions")
                                    .addSchema(
                                            "id BIGINT PRIMARY KEY NOT ENFORCED",
                                            "category STRING",
                                            "auctionTimestamp STRING",
                                            "expiresTimestamp STRING",
                                            "auctionDateTime AS TO_TIMESTAMP(auctionTimestamp)",
                                            "expires AS TO_TIMESTAMP(expiresTimestamp)",
                                            "WATERMARK FOR auctionDateTime AS auctionDateTime - INTERVAL '1' SECOND")
                                    .addOption("changelog-mode", "I")
                                    .producedValues(
                                            Row.ofKind(
                                                    RowKind.INSERT,
                                                    1L,
                                                    "Electronics",
                                                    "2024-01-01 12:00:00",
                                                    "2024-01-01 12:30:00"),
                                            Row.ofKind(
                                                    RowKind.INSERT,
                                                    3L,
                                                    "Electronics",
                                                    "2024-01-01 12:10:00",
                                                    "2024-01-01 12:40:00"),

Review comment (suggestion):
nit: I see you have named the constant Nexmark, I think it would be better to include the test description/goal in the constant name , similar to MULTI_JOIN_NULL_SAFE_JOIN_WITH_NULL_KEYS.  

Code after:
                            "CREATE VIEW UsersNullSafe AS SELECT * FROM (VALUES "
                                    + "('1','Gus'),"
                                    + "(CAST(NULL AS STRING), 'NullUser')"
                                    + ") AS T(user_id, name)")
                    .setupSql(
                            "CREATE VIEW OrdersNullSafe AS SELECT * FROM (VALUES "
                                    + "('1','order1'),"
                                    + "(CAST(NULL AS STRING), 'nullOrder')"
                                    + ") AS T(user_id, order_id)")
                    .setupTableSink(
                            SinkTestStep.newBuilder("sink")
                                    .addSchema("user_id STRING", "name STRING", "order_id STRING")
                                    .consumedValues(
                                            "+I[1, Gus, order1]", "+I[null, NullUser, nullOrder]")
                                    .testMaterializedData()
                                    .build())
                    .runSql(
                            "INSERT INTO sink "
                                    + "SELECT u.user_id, u.name, o.order_id "
                                    + "FROM UsersNullSafe u "
                                    + "INNER JOIN OrdersNullSafe o ON u.user_id IS NOT DISTINCT FROM o.user_id")
                    .build();

    public static final TableTestProgram
            MULTI_JOIN_WITH_TIME_ATTRIBUTES_IN_CONDITIONS_MATERIALIZATION =
                    TableTestProgram.of(
                                    "three-way-join-with-time-attributes-in-join-conditions",
                                    "A query from the nexmark benchmark: "
                                            + "auction and bid aggregation with time-based filtering")
                            .setupTableSource(
                                    SourceTestStep.newBuilder("auctions")
                                            .addSchema(
                                                    "id BIGINT PRIMARY KEY NOT ENFORCED",
                                                    "category STRING",
                                                    "auctionTimestamp STRING",
                                                    "expiresTimestamp STRING",
                                                    "auctionDateTime AS TO_TIMESTAMP(auctionTimestamp)",
                                                    "expires AS TO_TIMESTAMP(expiresTimestamp)",
                                                    "WATERMARK FOR auctionDateTime AS auctionDateTime - INTERVAL '1' SECOND")
                                            .addOption("changelog-mode", "I")
                                            .producedValues(
                                                    Row.ofKind(
                                                            RowKind.INSERT,
                                                            1L,
                                                            "Electronics",
                                                            "2024-01-01 12:00:00",
                                                            "2024-01-01 12:30:00"),
                                                    Row.ofKind(
                                                            RowKind.INSERT,
                                                            3L,
                                                            "Electronics",