class TableSchemaUtils {

	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;
	}

}