content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.connect.data;
  
  import org.apache.kafka.connect.errors.DataException;
  import org.apache.kafka.connect.errors.SchemaBuilderException;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.LinkedHashMap;
  import java.util.List;
  import java.util.Map;
  
  /**
   * <p>
   *     SchemaBuilder provides a fluent API for constructing {@link Schema} objects. It allows you to set each of the
   *     properties for the schema and each call returns the SchemaBuilder so the calls can be chained. When nested types
   *     are required, use one of the predefined schemas from {@link Schema} or use a second SchemaBuilder inline.
   * </p>
   * <p>
   *     Here is an example of building a struct schema:
   *     <pre>
   *     Schema dateSchema = SchemaBuilder.struct()
   *         .name("com.example.CalendarDate").version(2).doc("A calendar date including month, day, and year.")
   *         .field("month", Schema.STRING_SCHEMA)
   *         .field("day", Schema.INT8_SCHEMA)
   *         .field("year", Schema.INT16_SCHEMA)
   *         .build();
   *     </pre>
   * </p>
   * <p>
   *     Here is an example of using a second SchemaBuilder to construct complex, nested types:
   *     <pre>
   *     Schema userListSchema = SchemaBuilder.array(
   *         SchemaBuilder.struct().name("com.example.User").field("username", Schema.STRING_SCHEMA).field("id", Schema.INT64_SCHEMA).build()
   *     ).build();
   *     </pre>
   * </p>
   */
  public class SchemaBuilder implements Schema {
      private static final String TYPE_FIELD = "type";
      private static final String OPTIONAL_FIELD = "optional";
      private static final String DEFAULT_FIELD = "default";
      private static final String NAME_FIELD = "name";
      private static final String VERSION_FIELD = "version";
      private static final String DOC_FIELD = "doc";
  
  
      private final Type type;
      private Boolean optional = null;
      private Object defaultValue = null;
  
      private Map<String, Field> fields = null;
      private Schema keySchema = null;
      private Schema valueSchema = null;
  
      private String name;
      private Integer version;
      // Optional human readable documentation describing this schema.
      private String doc;
      // Additional parameters for logical types.
      private Map<String, String> parameters;
  
      public SchemaBuilder(Type type) {
          if (null == type)
              throw new SchemaBuilderException("type cannot be null");
          this.type = type;
          if (type == Type.STRUCT) {
              fields = new LinkedHashMap<>();
          }
      }
  
      // Common/metadata fields
  
      @Override
      public boolean isOptional() {
          return optional != null && optional;
      }
  
      /**
       * Set this schema as optional.
       * @return the SchemaBuilder
       */
      public SchemaBuilder optional() {
          checkCanSet(OPTIONAL_FIELD, optional, true);
          optional = true;
          return this;
      }
  
      /**
       * Set this schema as required. This is the default, but this method can be used to make this choice explicit.
       * @return the SchemaBuilder
       */
      public SchemaBuilder required() {
          checkCanSet(OPTIONAL_FIELD, optional, false);
          optional = false;
          return this;
      }
  
      @Override
      public Object defaultValue() {
          return defaultValue;
      }
  
      /**
       * Set the default value for this schema. The value is validated against the schema type, throwing a
       * {@link SchemaBuilderException} if it does not match.
       * @param value the default value
       * @return the SchemaBuilder
       */
      public SchemaBuilder defaultValue(Object value) {
          checkCanSet(DEFAULT_FIELD, defaultValue, value);
          checkNotNull(TYPE_FIELD, type, DEFAULT_FIELD);
          try {
              ConnectSchema.validateValue(this, value);
          } catch (DataException e) {
              throw new SchemaBuilderException("Invalid default value", e);
          }
          defaultValue = value;
          return this;
      }
  
      @Override
      public String name() {
          return name;
      }
  
      /**
       * Set the name of this schema.
       * @param name the schema name
       * @return the SchemaBuilder
       */
      public SchemaBuilder name(String name) {
          checkCanSet(NAME_FIELD, this.name, name);
          this.name = name;
          return this;
      }
  
      @Override
      public Integer version() {
          return version;
      }
  
      /**
       * Set the version of this schema. Schema versions are integers which, if provided, must indicate which schema is
       * newer and which is older by their ordering.
       * @param version the schema version
       * @return the SchemaBuilder
       */
      public SchemaBuilder version(Integer version) {
          checkCanSet(VERSION_FIELD, this.version, version);
          this.version = version;
          return this;
      }
  
      @Override
      public String doc() {
          return doc;
      }
  
      /**
       * Set the documentation for this schema.
       * @param doc the documentation
       * @return the SchemaBuilder
       */
      public SchemaBuilder doc(String doc) {
          checkCanSet(DOC_FIELD, this.doc, doc);
          this.doc = doc;
          return this;
      }
  
      @Override
      public Map<String, String> parameters() {
          return parameters == null ? null : Collections.unmodifiableMap(parameters);
      }
  
      /**
       * Set a schema parameter.
       * @param propertyName name of the schema property to define
       * @param propertyValue value of the schema property to define, as a String
       * @return the SchemaBuilder
       */
      public SchemaBuilder parameter(String propertyName, String propertyValue) {
          // Preserve order of insertion with a LinkedHashMap. This isn't strictly necessary, but is nice if logical types
          // can print their properties in a consistent order.
          if (parameters == null)
              parameters = new LinkedHashMap<>();
          parameters.put(propertyName, propertyValue);
          return this;
      }
  
      /**
       * Set schema parameters. This operation is additive; it does not remove existing parameters that do not appear in
       * the set of properties pass to this method.
       * @param props Map of properties to set
       * @return the SchemaBuilder
       */
      public SchemaBuilder parameters(Map<String, String> props) {
          // Avoid creating an empty set of properties so we never have an empty map
          if (props.isEmpty())
              return this;
          if (parameters == null)
              parameters = new LinkedHashMap<>();
          parameters.putAll(props);
          return this;
      }
  
      @Override
      public Type type() {
          return type;
      }
  
      /**
       * Create a SchemaBuilder for the specified type.
       * <p>
       * Usually it will be simpler to use one of the variants like {@link #string()} or {@link #struct()}, but this form
       * can be useful when generating schemas dynamically.
       *
       * @param type the schema type
       * @return a new SchemaBuilder
       */
      public static SchemaBuilder type(Type type) {
          return new SchemaBuilder(type);
      }
  
      // Primitive types
  
      /**
       * @return a new {@link Schema.Type#INT8} SchemaBuilder
       */
      public static SchemaBuilder int8() {
          return new SchemaBuilder(Type.INT8);
      }
  
      /**
       * @return a new {@link Schema.Type#INT16} SchemaBuilder
       */
      public static SchemaBuilder int16() {
          return new SchemaBuilder(Type.INT16);
      }
  
      /**
       * @return a new {@link Schema.Type#INT32} SchemaBuilder
       */
      public static SchemaBuilder int32() {
          return new SchemaBuilder(Type.INT32);
      }
  
      /**
       * @return a new {@link Schema.Type#INT64} SchemaBuilder
       */
      public static SchemaBuilder int64() {
          return new SchemaBuilder(Type.INT64);
      }
  
      /**
       * @return a new {@link Schema.Type#FLOAT32} SchemaBuilder
       */
      public static SchemaBuilder float32() {
          return new SchemaBuilder(Type.FLOAT32);
      }
  
      /**
       * @return a new {@link Schema.Type#FLOAT64} SchemaBuilder
       */
      public static SchemaBuilder float64() {
          return new SchemaBuilder(Type.FLOAT64);
      }
  
      /**
       * @return a new {@link Schema.Type#BOOLEAN} SchemaBuilder
       */
      public static SchemaBuilder bool() {
          return new SchemaBuilder(Type.BOOLEAN);
      }
  
      /**
       * @return a new {@link Schema.Type#STRING} SchemaBuilder
       */
      public static SchemaBuilder string() {
          return new SchemaBuilder(Type.STRING);
      }
  
      /**
       * @return a new {@link Schema.Type#BYTES} SchemaBuilder
       */
      public static SchemaBuilder bytes() {
          return new SchemaBuilder(Type.BYTES);
      }
  
  
      // Structs
  
      /**
       * @return a new {@link Schema.Type#STRUCT} SchemaBuilder
       */
      public static SchemaBuilder struct() {
          return new SchemaBuilder(Type.STRUCT);
      }
  
      /**
       * Add a field to this {@link Schema.Type#STRUCT} schema. Throws a {@link SchemaBuilderException} if this is not a struct schema.
       * @param fieldName the name of the field to add
       * @param fieldSchema the Schema for the field's value
       * @return the SchemaBuilder
       */
      public SchemaBuilder field(String fieldName, Schema fieldSchema) {
          if (type != Type.STRUCT)
              throw new SchemaBuilderException("Cannot create fields on type " + type);
          if (null == fieldName || fieldName.isEmpty())
              throw new SchemaBuilderException("fieldName cannot be null.");
          if (null == fieldSchema)
              throw new SchemaBuilderException("fieldSchema for field " + fieldName + " cannot be null.");
          int fieldIndex = fields.size();
          if (fields.containsKey(fieldName))
              throw new SchemaBuilderException("Cannot create field because of field name duplication " + fieldName);
          fields.put(fieldName, new Field(fieldName, fieldIndex, fieldSchema));
          return this;
      }
  
      /**
       * Get the list of fields for this Schema. Throws a {@link DataException} if this schema is not a {@link Schema.Type#STRUCT}.
       * @return the list of fields for this Schema
       */
      @Override
      public List<Field> fields() {
          if (type != Type.STRUCT)
              throw new DataException("Cannot list fields on non-struct type");
          return new ArrayList<>(fields.values());
      }
  
      @Override
      public Field field(String fieldName) {
          if (type != Type.STRUCT)
              throw new DataException("Cannot look up fields on non-struct type");
          return fields.get(fieldName);
      }
  
  
  
      // Maps & Arrays
  
      /**
       * @param valueSchema the schema for elements of the array
       * @return a new {@link Schema.Type#ARRAY} SchemaBuilder
       */
      public static SchemaBuilder array(Schema valueSchema) {
          if (null == valueSchema)
              throw new SchemaBuilderException("valueSchema cannot be null.");
          SchemaBuilder builder = new SchemaBuilder(Type.ARRAY);
          builder.valueSchema = valueSchema;
          return builder;
      }
  
      /**
       * @param keySchema the schema for keys in the map
       * @param valueSchema the schema for values in the map
       * @return a new {@link Schema.Type#MAP} SchemaBuilder
       */
      public static SchemaBuilder map(Schema keySchema, Schema valueSchema) {
          if (null == keySchema)
              throw new SchemaBuilderException("keySchema cannot be null.");
          if (null == valueSchema)
              throw new SchemaBuilderException("valueSchema cannot be null.");
          SchemaBuilder builder = new SchemaBuilder(Type.MAP);
          builder.keySchema = keySchema;
          builder.valueSchema = valueSchema;
          return builder;
      }
  
      static SchemaBuilder arrayOfNull() {
          return new SchemaBuilder(Type.ARRAY);
      }
  
      static SchemaBuilder mapOfNull() {
          return new SchemaBuilder(Type.MAP);
      }
  
      static SchemaBuilder mapWithNullKeys(Schema valueSchema) {
          SchemaBuilder result = new SchemaBuilder(Type.MAP);
          result.valueSchema = valueSchema;
          return result;
      }
  
      static SchemaBuilder mapWithNullValues(Schema keySchema) {
          SchemaBuilder result = new SchemaBuilder(Type.MAP);
          result.keySchema = keySchema;
          return result;
      }
  
      @Override
      public Schema keySchema() {
          return keySchema;
      }
  
      @Override
      public Schema valueSchema() {
          return valueSchema;
      }
  
  
      /**
       * Build the Schema using the current settings
       * @return the {@link Schema}
       */
      public Schema build() {
          return new ConnectSchema(type, isOptional(), defaultValue, name, version, doc,
                  parameters == null ? null : Collections.unmodifiableMap(parameters),
                  fields == null ? null : List.copyOf(fields.values()), keySchema, valueSchema);
      }
  
      /**
       * Return a concrete instance of the {@link Schema} specified by this builder
       * @return the {@link Schema}
       */
      @Override
      public Schema schema() {
          return build();
      }
  
      private static void checkCanSet(String fieldName, Object fieldVal, Object val) {
          if (fieldVal != null && fieldVal != val)
              throw new SchemaBuilderException("Invalid SchemaBuilder call: " + fieldName + " has already been set.");
      }
  
      private static void checkNotNull(String fieldName, Object val, String fieldToSet) {
          if (val == null)
              throw new SchemaBuilderException("Invalid SchemaBuilder call: " + fieldName + " must be specified to set " + fieldToSet);
      }
  }
  