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 mergeExcludingWatermarksDuplicate() {
        Schema sourceSchema =
                Schema.newBuilder()
                        .column("one", DataTypes.INT())
                        .column("timestamp", DataTypes.TIMESTAMP())
                        .watermark("timestamp", "timestamp - INTERVAL '5' SECOND")
                        .build();

        List<SqlWatermark> derivedWatermarkSpecs =
                Collections.singletonList(
                        new SqlWatermark(
                                SqlParserPos.ZERO,
                                identifier("timestamp"),
                                boundedStrategy("timestamp", "10")));

        Map<FeatureOption, MergingStrategy> mergingStrategies = getDefaultMergingStrategies();
        mergingStrategies.put(FeatureOption.WATERMARKS, MergingStrategy.EXCLUDING);

        Schema mergedSchema =
                util.mergeTables(
                        mergingStrategies,
                        sourceSchema,
                        Collections.emptyList(),
                        derivedWatermarkSpecs,
                        null);

        Schema expectedSchema =
                Schema.newBuilder()
                        .column("one", DataTypes.INT())
                        .column("timestamp", DataTypes.TIMESTAMP())
                        .watermark("timestamp", "`timestamp` - INTERVAL '10' SECOND")
                        .build();

        assertThat(mergedSchema).isEqualTo(expectedSchema);
    }