public static TableSchema.Builder builderWithGivenSchema(TableSchema oriSchema) {
        TableSchema.Builder builder = builderWithGivenColumns(oriSchema.getTableColumns());
        // Copy watermark specification.
        for (WatermarkSpec wms : oriSchema.getWatermarkSpecs()) {
            builder.watermark(
                    wms.getRowtimeAttribute(),
                    wms.getWatermarkExpr(),
                    wms.getWatermarkExprOutputType());
        }
        // Copy primary key constraint.
        oriSchema
                .getPrimaryKey()
                .map(
                        pk ->
                                builder.primaryKey(
                                        pk.getName(), pk.getColumns().toArray(new String[0])));
        return builder;
    }
@Test
    void testBuilderWithGivenSchema() {
        TableSchema oriSchema =
                TableSchema.builder()
                        .field("a", DataTypes.INT().notNull())
                        .field("b", DataTypes.STRING())
                        .field("c", DataTypes.INT(), "a + 1")
                        .field("t", DataTypes.TIMESTAMP(3))
                        .primaryKey("ct1", new String[] {"a"})
                        .watermark("t", "t", DataTypes.TIMESTAMP(3))
                        .build();
        TableSchema newSchema = TableSchemaUtils.builderWithGivenSchema(oriSchema).build();
        assertThat(newSchema).isEqualTo(oriSchema);
    }