public Schema mergeTables(
            Map<FeatureOption, MergingStrategy> mergingStrategies,
            Schema sourceSchema,
            List<SqlNode> derivedColumns,
            List<SqlWatermark> derivedWatermarkSpecs,
            SqlTableConstraint derivedPrimaryKey) {

        SchemaBuilder schemaBuilder =
                new SchemaBuilder(
                        mergingStrategies,
                        sourceSchema,
                        (FlinkTypeFactory) validator.getTypeFactory(),
                        dataTypeFactory,
                        validator,
                        escapeExpression);
        schemaBuilder.appendDerivedColumns(mergingStrategies, derivedColumns);
        schemaBuilder.appendDerivedWatermarks(mergingStrategies, derivedWatermarkSpecs);
        schemaBuilder.appendDerivedPrimaryKey(derivedPrimaryKey);

        return schemaBuilder.build();
    }
@Test
    void mergeWatermarks() {
        Schema sourceSchema =
                Schema.newBuilder()
                        .column("one", DataTypes.INT())
                        .columnByExpression("two", "one +1")
                        .column("timestamp", DataTypes.TIMESTAMP())
                        .watermark("timestamp", "timestamp - INTERVAL '5' SECOND")
                        .build();

        List<SqlNode> derivedColumns =
                Arrays.asList(
                        regularColumn("three", DataTypes.INT()),
                        computedColumn("four", plus("one", "3")));

        Schema mergedSchema =
                util.mergeTables(
                        getDefaultMergingStrategies(),
                        sourceSchema,
                        derivedColumns,
                        Collections.emptyList(),
                        null);

        Schema expectedSchema =
                Schema.newBuilder()
                        .column("one", DataTypes.INT())
                        .columnByExpression("two", "one +1")
                        .column("timestamp", DataTypes.TIMESTAMP())
                        .watermark("timestamp", "timestamp - INTERVAL '5' SECOND")
                        .column("three", DataTypes.INT())
                        .columnByExpression("four", "`one` + 3")
                        .build();

        assertThat(mergedSchema).isEqualTo(expectedSchema);
    }