/*
 * 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
 *
 *     https://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.avro;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import org.apache.avro.Schema.Field;
import org.apache.avro.Schema.Type;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Evaluate the compatibility between a reader schema and a writer schema. A
reader and a writer schema are declared compatible if all datum instances of
the writer schema can be successfully decoded using the specified reader
schema.
/**
 * Evaluate the compatibility between a reader schema and a writer schema. A
 * reader and a writer schema are declared compatible if all datum instances of
 * the writer schema can be successfully decoded using the specified reader
 * schema.
 */
public class SchemaCompatibility {
  private static final Logger LOG = LoggerFactory.getLogger(SchemaCompatibility.class);

  
Utility class cannot be instantiated. /** Utility class cannot be instantiated. */
  private SchemaCompatibility() {
  }

  
Message to annotate reader/writer schema pairs that are compatible. /** Message to annotate reader/writer schema pairs that are compatible. */
  public static final String READER_WRITER_COMPATIBLE_MESSAGE = "Reader schema can always successfully decode data written using the writer schema.";

  
Validates that the provided reader schema can be used to decode avro data
written with the provided writer schema.
Params:
reader – schema to check.
writer – schema to check.
Returns:
a result object identifying any compatibility errors./**
   * Validates that the provided reader schema can be used to decode avro data
   * written with the provided writer schema.
   *
   * @param reader schema to check.
   * @param writer schema to check.
   * @return a result object identifying any compatibility errors.
   */
  public static SchemaPairCompatibility checkReaderWriterCompatibility(final Schema reader, final Schema writer) {
    final SchemaCompatibilityResult compatibility = new ReaderWriterCompatibilityChecker().getCompatibility(reader,
        writer);

    final String message;
    switch (compatibility.getCompatibility()) {
    case INCOMPATIBLE: {
      message = String.format(
          "Data encoded using writer schema:%n%s%n" + "will or may fail to decode using reader schema:%n%s%n",
          writer.toString(true), reader.toString(true));
      break;
    }
    case COMPATIBLE: {
      message = READER_WRITER_COMPATIBLE_MESSAGE;
      break;
    }
    default:
      throw new AvroRuntimeException("Unknown compatibility: " + compatibility);
    }

    return new SchemaPairCompatibility(compatibility, reader, writer, message);
  }

  // -----------------------------------------------------------------------------------------------

  
Tests the equality of two Avro named schemas.

Matching includes reader name aliases.

Params:
reader – Named reader schema.
writer – Named writer schema.
Returns:
whether the names of the named schemas match or not./**
   * Tests the equality of two Avro named schemas.
   *
   * <p>
   * Matching includes reader name aliases.
   * </p>
   *
   * @param reader Named reader schema.
   * @param writer Named writer schema.
   * @return whether the names of the named schemas match or not.
   */
  public static boolean schemaNameEquals(final Schema reader, final Schema writer) {
    if (objectsEqual(reader.getName(), writer.getName())) {
      return true;
    }
    // Apply reader aliases:
    return reader.getAliases().contains(writer.getFullName());
  }

  
Identifies the writer field that corresponds to the specified reader field.

Matching includes reader name aliases.

Params:
writerSchema – Schema of the record where to look for the writer field.
readerField –  Reader field to identify the corresponding writer field
                    of.
Returns:
the writer field, if any does correspond, or None./**
   * Identifies the writer field that corresponds to the specified reader field.
   *
   * <p>
   * Matching includes reader name aliases.
   * </p>
   *
   * @param writerSchema Schema of the record where to look for the writer field.
   * @param readerField  Reader field to identify the corresponding writer field
   *                     of.
   * @return the writer field, if any does correspond, or None.
   */
  public static Field lookupWriterField(final Schema writerSchema, final Field readerField) {
    assert (writerSchema.getType() == Type.RECORD);
    final List<Field> writerFields = new ArrayList<>();
    final Field direct = writerSchema.getField(readerField.name());
    if (direct != null) {
      writerFields.add(direct);
    }
    for (final String readerFieldAliasName : readerField.aliases()) {
      final Field writerField = writerSchema.getField(readerFieldAliasName);
      if (writerField != null) {
        writerFields.add(writerField);
      }
    }
    switch (writerFields.size()) {
    case 0:
      return null;
    case 1:
      return writerFields.get(0);
    default: {
      throw new AvroRuntimeException(String.format(
          "Reader record field %s matches multiple fields in writer record schema %s", readerField, writerSchema));
    }
    }
  }

  
Reader/writer schema pair that can be used as a key in a hash map.
This reader/writer pair differentiates Schema objects based on their system
hash code.
/**
   * Reader/writer schema pair that can be used as a key in a hash map.
   *
   * This reader/writer pair differentiates Schema objects based on their system
   * hash code.
   */
  private static final class ReaderWriter {
    private final Schema mReader;
    private final Schema mWriter;

    
Initializes a new reader/writer pair.
Params:
reader – Reader schema.
writer – Writer schema./**
     * Initializes a new reader/writer pair.
     *
     * @param reader Reader schema.
     * @param writer Writer schema.
     */
    public ReaderWriter(final Schema reader, final Schema writer) {
      mReader = reader;
      mWriter = writer;
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public int hashCode() {
      return System.identityHashCode(mReader) ^ System.identityHashCode(mWriter);
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public boolean equals(Object obj) {
      if (!(obj instanceof ReaderWriter)) {
        return false;
      }
      final ReaderWriter that = (ReaderWriter) obj;
      // Use pointer comparison here:
      return (this.mReader == that.mReader) && (this.mWriter == that.mWriter);
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public String toString() {
      return String.format("ReaderWriter{reader:%s, writer:%s}", mReader, mWriter);
    }
  }

  
Determines the compatibility of a reader/writer schema pair.

Provides memoization to handle recursive schemas.

/**
   * Determines the compatibility of a reader/writer schema pair.
   *
   * <p>
   * Provides memoization to handle recursive schemas.
   * </p>
   */
  private static final class ReaderWriterCompatibilityChecker {
    private static final String ROOT_REFERENCE_TOKEN = "";
    private final Map<ReaderWriter, SchemaCompatibilityResult> mMemoizeMap = new HashMap<>();

    
Reports the compatibility of a reader/writer schema pair.

Memoizes the compatibility results.

Params:
reader – Reader schema to test.
writer – Writer schema to test.
Returns:
the compatibility of the reader/writer schema pair./**
     * Reports the compatibility of a reader/writer schema pair.
     *
     * <p>
     * Memoizes the compatibility results.
     * </p>
     *
     * @param reader Reader schema to test.
     * @param writer Writer schema to test.
     * @return the compatibility of the reader/writer schema pair.
     */
    public SchemaCompatibilityResult getCompatibility(final Schema reader, final Schema writer) {
      Deque<String> location = new ArrayDeque<>();
      return getCompatibility(ROOT_REFERENCE_TOKEN, reader, writer, location);
    }

    
Reports the compatibility of a reader/writer schema pair.

Memoizes the compatibility results.

Params:
referenceToken – The equivalent JSON pointer reference token
                      representation of the schema node being visited.
reader –         Reader schema to test.
writer –         Writer schema to test.
location –       Stack with which to track the location within the
                      schema.
Returns:
the compatibility of the reader/writer schema pair./**
     * Reports the compatibility of a reader/writer schema pair.
     * <p>
     * Memoizes the compatibility results.
     * </p>
     *
     * @param referenceToken The equivalent JSON pointer reference token
     *                       representation of the schema node being visited.
     * @param reader         Reader schema to test.
     * @param writer         Writer schema to test.
     * @param location       Stack with which to track the location within the
     *                       schema.
     * @return the compatibility of the reader/writer schema pair.
     */
    private SchemaCompatibilityResult getCompatibility(String referenceToken, final Schema reader, final Schema writer,
        final Deque<String> location) {
      location.addFirst(referenceToken);
      LOG.debug("Checking compatibility of reader {} with writer {}", reader, writer);
      final ReaderWriter pair = new ReaderWriter(reader, writer);
      SchemaCompatibilityResult result = mMemoizeMap.get(pair);
      if (result != null) {
        if (result.getCompatibility() == SchemaCompatibilityType.RECURSION_IN_PROGRESS) {
          // Break the recursion here.
          // schemas are compatible unless proven incompatible:
          result = SchemaCompatibilityResult.compatible();
        }
      } else {
        // Mark this reader/writer pair as "in progress":
        mMemoizeMap.put(pair, SchemaCompatibilityResult.recursionInProgress());
        result = calculateCompatibility(reader, writer, location);
        mMemoizeMap.put(pair, result);
      }
      location.removeFirst();
      return result;
    }

    
Calculates the compatibility of a reader/writer schema pair.
 Relies on external memoization performed by getCompatibility(Schema, Schema). 
Params:
reader –   Reader schema to test.
writer –   Writer schema to test.
location – Stack with which to track the location within the schema.
Returns:
the compatibility of the reader/writer schema pair./**
     * Calculates the compatibility of a reader/writer schema pair.
     *
     * <p>
     * Relies on external memoization performed by
     * {@link #getCompatibility(Schema, Schema)}.
     * </p>
     *
     * @param reader   Reader schema to test.
     * @param writer   Writer schema to test.
     * @param location Stack with which to track the location within the schema.
     * @return the compatibility of the reader/writer schema pair.
     */
    private SchemaCompatibilityResult calculateCompatibility(final Schema reader, final Schema writer,
        final Deque<String> location) {
      assert (reader != null);
      assert (writer != null);
      SchemaCompatibilityResult result = SchemaCompatibilityResult.compatible();

      if (reader.getType() == writer.getType()) {
        switch (reader.getType()) {
        case NULL:
        case BOOLEAN:
        case INT:
        case LONG:
        case FLOAT:
        case DOUBLE:
        case BYTES:
        case STRING: {
          return result;
        }
        case ARRAY: {
          return result
              .mergedWith(getCompatibility("items", reader.getElementType(), writer.getElementType(), location));
        }
        case MAP: {
          return result.mergedWith(getCompatibility("values", reader.getValueType(), writer.getValueType(), location));
        }
        case FIXED: {
          result = result.mergedWith(checkSchemaNames(reader, writer, location));
          return result.mergedWith(checkFixedSize(reader, writer, location));
        }
        case ENUM: {
          result = result.mergedWith(checkSchemaNames(reader, writer, location));
          return result.mergedWith(checkReaderEnumContainsAllWriterEnumSymbols(reader, writer, location));
        }
        case RECORD: {
          result = result.mergedWith(checkSchemaNames(reader, writer, location));
          return result.mergedWith(checkReaderWriterRecordFields(reader, writer, location));
        }
        case UNION: {
          // Check that each individual branch of the writer union can be decoded:
          int i = 0;
          for (final Schema writerBranch : writer.getTypes()) {
            location.addFirst(Integer.toString(i));
            SchemaCompatibilityResult compatibility = getCompatibility(reader, writerBranch);
            if (compatibility.getCompatibility() == SchemaCompatibilityType.INCOMPATIBLE) {
              String message = String.format("reader union lacking writer type: %s", writerBranch.getType());
              result = result.mergedWith(SchemaCompatibilityResult.incompatible(
                  SchemaIncompatibilityType.MISSING_UNION_BRANCH, reader, writer, message, asList(location)));
            }
            location.removeFirst();
            i++;
          }
          // Each schema in the writer union can be decoded with the reader:
          return result;
        }

        default: {
          throw new AvroRuntimeException("Unknown schema type: " + reader.getType());
        }
        }

      } else {
        // Reader and writer have different schema types:

        // Reader compatible with all branches of a writer union is compatible
        if (writer.getType() == Schema.Type.UNION) {
          for (Schema s : writer.getTypes()) {
            result = result.mergedWith(getCompatibility(reader, s));
          }
          return result;
        }

        switch (reader.getType()) {
        case NULL:
          return result.mergedWith(typeMismatch(reader, writer, location));
        case BOOLEAN:
          return result.mergedWith(typeMismatch(reader, writer, location));
        case INT:
          return result.mergedWith(typeMismatch(reader, writer, location));
        case LONG: {
          return (writer.getType() == Type.INT) ? result : result.mergedWith(typeMismatch(reader, writer, location));
        }
        case FLOAT: {
          return ((writer.getType() == Type.INT) || (writer.getType() == Type.LONG)) ? result
              : result.mergedWith(typeMismatch(reader, writer, location));

        }
        case DOUBLE: {
          return ((writer.getType() == Type.INT) || (writer.getType() == Type.LONG) || (writer.getType() == Type.FLOAT))
              ? result
              : result.mergedWith(typeMismatch(reader, writer, location));
        }
        case BYTES: {
          return (writer.getType() == Type.STRING) ? result : result.mergedWith(typeMismatch(reader, writer, location));
        }
        case STRING: {
          return (writer.getType() == Type.BYTES) ? result : result.mergedWith(typeMismatch(reader, writer, location));
        }

        case ARRAY:
          return result.mergedWith(typeMismatch(reader, writer, location));
        case MAP:
          return result.mergedWith(typeMismatch(reader, writer, location));
        case FIXED:
          return result.mergedWith(typeMismatch(reader, writer, location));
        case ENUM:
          return result.mergedWith(typeMismatch(reader, writer, location));
        case RECORD:
          return result.mergedWith(typeMismatch(reader, writer, location));
        case UNION: {
          for (final Schema readerBranch : reader.getTypes()) {
            SchemaCompatibilityResult compatibility = getCompatibility(readerBranch, writer);
            if (compatibility.getCompatibility() == SchemaCompatibilityType.COMPATIBLE) {
              return result;
            }
          }
          // No branch in the reader union has been found compatible with the writer
          // schema:
          String message = String.format("reader union lacking writer type: %s", writer.getType());
          return result.mergedWith(SchemaCompatibilityResult
              .incompatible(SchemaIncompatibilityType.MISSING_UNION_BRANCH, reader, writer, message, asList(location)));
        }

        default: {
          throw new AvroRuntimeException("Unknown schema type: " + reader.getType());
        }
        }
      }
    }

    private SchemaCompatibilityResult checkReaderWriterRecordFields(final Schema reader, final Schema writer,
        final Deque<String> location) {
      SchemaCompatibilityResult result = SchemaCompatibilityResult.compatible();
      location.addFirst("fields");
      // Check that each field in the reader record can be populated from the writer
      // record:
      for (final Field readerField : reader.getFields()) {
        location.addFirst(Integer.toString(readerField.pos()));
        final Field writerField = lookupWriterField(writer, readerField);
        if (writerField == null) {
          // Reader field does not correspond to any field in the writer record schema, so
          // the
          // reader field must have a default value.
          if (!readerField.hasDefaultValue()) {
            // reader field has no default value. Check for the enum default value
            if (readerField.schema().getType() == Type.ENUM && readerField.schema().getEnumDefault() != null) {
              result = result
                  .mergedWith(getCompatibility("type", readerField.schema(), writerField.schema(), location));
            } else {
              result = result.mergedWith(
                  SchemaCompatibilityResult.incompatible(SchemaIncompatibilityType.READER_FIELD_MISSING_DEFAULT_VALUE,
                      reader, writer, readerField.name(), asList(location)));
            }
          }
        } else {
          result = result.mergedWith(getCompatibility("type", readerField.schema(), writerField.schema(), location));
        }
        // POP field index
        location.removeFirst();
      }
      // All fields in the reader record can be populated from the writer record:
      // POP "fields" literal
      location.removeFirst();
      return result;
    }

    private SchemaCompatibilityResult checkReaderEnumContainsAllWriterEnumSymbols(final Schema reader,
        final Schema writer, final Deque<String> location) {
      SchemaCompatibilityResult result = SchemaCompatibilityResult.compatible();
      location.addFirst("symbols");
      final Set<String> symbols = new TreeSet<>(writer.getEnumSymbols());
      symbols.removeAll(reader.getEnumSymbols());
      if (!symbols.isEmpty()) {
        if (reader.getEnumDefault() != null && reader.getEnumSymbols().contains(reader.getEnumDefault())) {
          symbols.clear();
          result = SchemaCompatibilityResult.compatible();
        } else {
          result = SchemaCompatibilityResult.incompatible(SchemaIncompatibilityType.MISSING_ENUM_SYMBOLS, reader,
              writer, symbols.toString(), asList(location));
        }
      }
      // POP "symbols" literal
      location.removeFirst();
      return result;
    }

    private SchemaCompatibilityResult checkFixedSize(final Schema reader, final Schema writer,
        final Deque<String> location) {
      SchemaCompatibilityResult result = SchemaCompatibilityResult.compatible();
      location.addFirst("size");
      int actual = reader.getFixedSize();
      int expected = writer.getFixedSize();
      if (actual != expected) {
        String message = String.format("expected: %d, found: %d", expected, actual);
        result = SchemaCompatibilityResult.incompatible(SchemaIncompatibilityType.FIXED_SIZE_MISMATCH, reader, writer,
            message, asList(location));
      }
      // POP "size" literal
      location.removeFirst();
      return result;
    }

    private SchemaCompatibilityResult checkSchemaNames(final Schema reader, final Schema writer,
        final Deque<String> location) {
      SchemaCompatibilityResult result = SchemaCompatibilityResult.compatible();
      location.addFirst("name");
      if (!schemaNameEquals(reader, writer)) {
        String message = String.format("expected: %s", writer.getFullName());
        result = SchemaCompatibilityResult.incompatible(SchemaIncompatibilityType.NAME_MISMATCH, reader, writer,
            message, asList(location));
      }
      // POP "name" literal
      location.removeFirst();
      return result;
    }

    private SchemaCompatibilityResult typeMismatch(final Schema reader, final Schema writer,
        final Deque<String> location) {
      String message = String.format("reader type: %s not compatible with writer type: %s", reader.getType(),
          writer.getType());
      return SchemaCompatibilityResult.incompatible(SchemaIncompatibilityType.TYPE_MISMATCH, reader, writer, message,
          asList(location));
    }
  }

  
Identifies the type of a schema compatibility result.
/**
   * Identifies the type of a schema compatibility result.
   */
  public enum SchemaCompatibilityType {
    COMPATIBLE, INCOMPATIBLE,

    
Used internally to tag a reader/writer schema pair and prevent recursion. /** Used internally to tag a reader/writer schema pair and prevent recursion. */
    RECURSION_IN_PROGRESS;
  }

  public enum SchemaIncompatibilityType {
    NAME_MISMATCH, FIXED_SIZE_MISMATCH, MISSING_ENUM_SYMBOLS, READER_FIELD_MISSING_DEFAULT_VALUE, TYPE_MISMATCH,
    MISSING_UNION_BRANCH;
  }

  
Immutable class representing details about a particular schema pair
compatibility check.
/**
   * Immutable class representing details about a particular schema pair
   * compatibility check.
   */
  public static final class SchemaCompatibilityResult {

    
Merges the current SchemaCompatibilityResult with the supplied result into a new instance, combining the list of Incompatibility Incompatibilities and regressing to the SchemaCompatibilityType#INCOMPATIBLE INCOMPATIBLE state if any incompatibilities are encountered. 
Params:
toMerge – The SchemaCompatibilityResult to merge with the current instance.
Returns:
A SchemaCompatibilityResult that combines the state of the current and supplied instances./**
     * Merges the current {@code SchemaCompatibilityResult} with the supplied result
     * into a new instance, combining the list of
     * {@code Incompatibility Incompatibilities} and regressing to the
     * {@code SchemaCompatibilityType#INCOMPATIBLE INCOMPATIBLE} state if any
     * incompatibilities are encountered.
     *
     * @param toMerge The {@code SchemaCompatibilityResult} to merge with the
     *                current instance.
     * @return A {@code SchemaCompatibilityResult} that combines the state of the
     *         current and supplied instances.
     */
    public SchemaCompatibilityResult mergedWith(SchemaCompatibilityResult toMerge) {
      List<Incompatibility> mergedIncompatibilities = new ArrayList<>(mIncompatibilities);
      mergedIncompatibilities.addAll(toMerge.getIncompatibilities());
      SchemaCompatibilityType compatibilityType = mCompatibilityType == SchemaCompatibilityType.COMPATIBLE
          ? toMerge.mCompatibilityType
          : SchemaCompatibilityType.INCOMPATIBLE;
      return new SchemaCompatibilityResult(compatibilityType, mergedIncompatibilities);
    }

    private final SchemaCompatibilityType mCompatibilityType;
    // the below fields are only valid if INCOMPATIBLE
    private final List<Incompatibility> mIncompatibilities;
    // cached objects for stateless details
    private static final SchemaCompatibilityResult COMPATIBLE = new SchemaCompatibilityResult(
        SchemaCompatibilityType.COMPATIBLE, Collections.emptyList());
    private static final SchemaCompatibilityResult RECURSION_IN_PROGRESS = new SchemaCompatibilityResult(
        SchemaCompatibilityType.RECURSION_IN_PROGRESS, Collections.emptyList());

    private SchemaCompatibilityResult(SchemaCompatibilityType compatibilityType,
        List<Incompatibility> incompatibilities) {
      this.mCompatibilityType = compatibilityType;
      this.mIncompatibilities = incompatibilities;
    }

    
Returns a details object representing a compatible schema pair.
Returns:
a SchemaCompatibilityDetails object with COMPATIBLE
        SchemaCompatibilityType, and no other state./**
     * Returns a details object representing a compatible schema pair.
     *
     * @return a SchemaCompatibilityDetails object with COMPATIBLE
     *         SchemaCompatibilityType, and no other state.
     */
    public static SchemaCompatibilityResult compatible() {
      return COMPATIBLE;
    }

    
Returns a details object representing a state indicating that recursion is in
progress.
Returns:
a SchemaCompatibilityDetails object with RECURSION_IN_PROGRESS
        SchemaCompatibilityType, and no other state./**
     * Returns a details object representing a state indicating that recursion is in
     * progress.
     *
     * @return a SchemaCompatibilityDetails object with RECURSION_IN_PROGRESS
     *         SchemaCompatibilityType, and no other state.
     */
    public static SchemaCompatibilityResult recursionInProgress() {
      return RECURSION_IN_PROGRESS;
    }

    
Returns a details object representing an incompatible schema pair, including
error details.
Returns:
a SchemaCompatibilityDetails object with INCOMPATIBLE
        SchemaCompatibilityType, and state representing the violating part./**
     * Returns a details object representing an incompatible schema pair, including
     * error details.
     *
     * @return a SchemaCompatibilityDetails object with INCOMPATIBLE
     *         SchemaCompatibilityType, and state representing the violating part.
     */
    public static SchemaCompatibilityResult incompatible(SchemaIncompatibilityType incompatibilityType,
        Schema readerFragment, Schema writerFragment, String message, List<String> location) {
      Incompatibility incompatibility = new Incompatibility(incompatibilityType, readerFragment, writerFragment,
          message, location);
      return new SchemaCompatibilityResult(SchemaCompatibilityType.INCOMPATIBLE,
          Collections.singletonList(incompatibility));
    }

    
Returns the SchemaCompatibilityType, always non-null.
Returns:
a SchemaCompatibilityType instance, always non-null/**
     * Returns the SchemaCompatibilityType, always non-null.
     *
     * @return a SchemaCompatibilityType instance, always non-null
     */
    public SchemaCompatibilityType getCompatibility() {
      return mCompatibilityType;
    }

    
If the compatibility is INCOMPATIBLE, returns 
Incompatibilities found, otherwise an empty list. 
Returns:
a list of Incompatibilities, may be empty, never null./**
     * If the compatibility is INCOMPATIBLE, returns {@link Incompatibility
     * Incompatibilities} found, otherwise an empty list.
     *
     * @return a list of {@link Incompatibility Incompatibilities}, may be empty,
     *         never null.
     */
    public List<Incompatibility> getIncompatibilities() {
      return mIncompatibilities;
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public int hashCode() {
      final int prime = 31;
      int result = 1;
      result = prime * result + ((mCompatibilityType == null) ? 0 : mCompatibilityType.hashCode());
      result = prime * result + ((mIncompatibilities == null) ? 0 : mIncompatibilities.hashCode());
      return result;
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public boolean equals(Object obj) {
      if (this == obj)
        return true;
      if (obj == null)
        return false;
      if (getClass() != obj.getClass())
        return false;
      SchemaCompatibilityResult other = (SchemaCompatibilityResult) obj;
      if (mIncompatibilities == null) {
        if (other.mIncompatibilities != null)
          return false;
      } else if (!mIncompatibilities.equals(other.mIncompatibilities))
        return false;
      return mCompatibilityType == other.mCompatibilityType;
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public String toString() {
      return String.format("SchemaCompatibilityResult{compatibility:%s, incompatibilities:%s}", mCompatibilityType,
          mIncompatibilities);
    }
  }
  // -----------------------------------------------------------------------------------------------

  public static final class Incompatibility {
    private final SchemaIncompatibilityType mType;
    private final Schema mReaderFragment;
    private final Schema mWriterFragment;
    private final String mMessage;
    private final List<String> mLocation;

    Incompatibility(SchemaIncompatibilityType type, Schema readerFragment, Schema writerFragment, String message,
        List<String> location) {
      super();
      this.mType = type;
      this.mReaderFragment = readerFragment;
      this.mWriterFragment = writerFragment;
      this.mMessage = message;
      this.mLocation = location;
    }

    
Returns the SchemaIncompatibilityType.
Returns:
a SchemaIncompatibilityType instance./**
     * Returns the SchemaIncompatibilityType.
     *
     * @return a SchemaIncompatibilityType instance.
     */
    public SchemaIncompatibilityType getType() {
      return mType;
    }

    
Returns the fragment of the reader schema that failed compatibility check.
Returns:
a Schema instance (fragment of the reader schema)./**
     * Returns the fragment of the reader schema that failed compatibility check.
     *
     * @return a Schema instance (fragment of the reader schema).
     */
    public Schema getReaderFragment() {
      return mReaderFragment;
    }

    
Returns the fragment of the writer schema that failed compatibility check.
Returns:
a Schema instance (fragment of the writer schema)./**
     * Returns the fragment of the writer schema that failed compatibility check.
     *
     * @return a Schema instance (fragment of the writer schema).
     */
    public Schema getWriterFragment() {
      return mWriterFragment;
    }

    
Returns a human-readable message with more details about what failed. Syntax
depends on the SchemaIncompatibilityType.
See Also:
getType()
Returns:
a String with details about the incompatibility./**
     * Returns a human-readable message with more details about what failed. Syntax
     * depends on the SchemaIncompatibilityType.
     *
     * @see #getType()
     * @return a String with details about the incompatibility.
     */
    public String getMessage() {
      return mMessage;
    }

    
Returns a
JSON
Pointer describing the node location within the schema's JSON document
tree where the incompatibility was encountered.
Returns:
JSON Pointer encoded as a string./**
     * Returns a
     * <a href="https://tools.ietf.org/html/draft-ietf-appsawg-json-pointer-08">JSON
     * Pointer</a> describing the node location within the schema's JSON document
     * tree where the incompatibility was encountered.
     *
     * @return JSON Pointer encoded as a string.
     */
    public String getLocation() {
      StringBuilder s = new StringBuilder("/");
      boolean first = true;
      // ignore root element
      for (String coordinate : mLocation.subList(1, mLocation.size())) {
        if (first) {
          first = false;
        } else {
          s.append('/');
        }
        // Apply JSON pointer escaping.
        s.append(coordinate.replace("~", "~0").replace("/", "~1"));
      }
      return s.toString();
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public int hashCode() {
      final int prime = 31;
      int result = 1;
      result = prime * result + ((mType == null) ? 0 : mType.hashCode());
      result = prime * result + ((mReaderFragment == null) ? 0 : mReaderFragment.hashCode());
      result = prime * result + ((mWriterFragment == null) ? 0 : mWriterFragment.hashCode());
      result = prime * result + ((mMessage == null) ? 0 : mMessage.hashCode());
      result = prime * result + ((mLocation == null) ? 0 : mLocation.hashCode());
      return result;
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public boolean equals(Object obj) {
      if (this == obj) {
        return true;
      }
      if (obj == null) {
        return false;
      }
      if (getClass() != obj.getClass()) {
        return false;
      }
      Incompatibility other = (Incompatibility) obj;
      if (mType != other.mType) {
        return false;
      }
      if (mReaderFragment == null) {
        if (other.mReaderFragment != null) {
          return false;
        }
      } else if (!mReaderFragment.equals(other.mReaderFragment)) {
        return false;
      }
      if (mWriterFragment == null) {
        if (other.mWriterFragment != null) {
          return false;
        }
      } else if (!mWriterFragment.equals(other.mWriterFragment)) {
        return false;
      }
      if (mMessage == null) {
        if (other.mMessage != null) {
          return false;
        }
      } else if (!mMessage.equals(other.mMessage)) {
        return false;
      }
      if (mLocation == null) {
        return other.mLocation == null;
      } else
        return mLocation.equals(other.mLocation);
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public String toString() {
      return String.format("Incompatibility{type:%s, location:%s, message:%s, reader:%s, writer:%s}", mType,
          getLocation(), mMessage, mReaderFragment, mWriterFragment);
    }
  }
  // -----------------------------------------------------------------------------------------------

  
Provides information about the compatibility of a single reader and writer
schema pair.
Note: This class represents a one-way relationship from the reader to the
writer schema.
/**
   * Provides information about the compatibility of a single reader and writer
   * schema pair.
   *
   * Note: This class represents a one-way relationship from the reader to the
   * writer schema.
   */
  public static final class SchemaPairCompatibility {
    
The details of this result. /** The details of this result. */
    private final SchemaCompatibilityResult mResult;

    
Validated reader schema. /** Validated reader schema. */
    private final Schema mReader;

    
Validated writer schema. /** Validated writer schema. */
    private final Schema mWriter;

    
Human readable description of this result. /** Human readable description of this result. */
    private final String mDescription;

    
Constructs a new instance.
Params:
result –      The result of the compatibility check.
reader –      schema that was validated.
writer –      schema that was validated.
description – of this compatibility result./**
     * Constructs a new instance.
     *
     * @param result      The result of the compatibility check.
     * @param reader      schema that was validated.
     * @param writer      schema that was validated.
     * @param description of this compatibility result.
     */
    public SchemaPairCompatibility(SchemaCompatibilityResult result, Schema reader, Schema writer, String description) {
      mResult = result;
      mReader = reader;
      mWriter = writer;
      mDescription = description;
    }

    
Gets the type of this result.
Returns:
the type of this result./**
     * Gets the type of this result.
     *
     * @return the type of this result.
     */
    public SchemaCompatibilityType getType() {
      return mResult.getCompatibility();
    }

    
Gets more details about the compatibility, in particular if getType() is
INCOMPATIBLE.
Returns:
the details of this compatibility check./**
     * Gets more details about the compatibility, in particular if getType() is
     * INCOMPATIBLE.
     *
     * @return the details of this compatibility check.
     */
    public SchemaCompatibilityResult getResult() {
      return mResult;
    }

    
Gets the reader schema that was validated.
Returns:
reader schema that was validated./**
     * Gets the reader schema that was validated.
     *
     * @return reader schema that was validated.
     */
    public Schema getReader() {
      return mReader;
    }

    
Gets the writer schema that was validated.
Returns:
writer schema that was validated./**
     * Gets the writer schema that was validated.
     *
     * @return writer schema that was validated.
     */
    public Schema getWriter() {
      return mWriter;
    }

    
Gets a human readable description of this validation result.
Returns:
a human readable description of this validation result./**
     * Gets a human readable description of this validation result.
     *
     * @return a human readable description of this validation result.
     */
    public String getDescription() {
      return mDescription;
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public String toString() {
      return String.format("SchemaPairCompatibility{result:%s, readerSchema:%s, writerSchema:%s, description:%s}",
          mResult, mReader, mWriter, mDescription);
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public boolean equals(Object other) {
      if ((other instanceof SchemaPairCompatibility)) {
        final SchemaPairCompatibility result = (SchemaPairCompatibility) other;
        return objectsEqual(result.mResult, mResult) && objectsEqual(result.mReader, mReader)
            && objectsEqual(result.mWriter, mWriter) && objectsEqual(result.mDescription, mDescription);
      } else {
        return false;
      }
    }

    
{@inheritDoc} /** {@inheritDoc} */
    @Override
    public int hashCode() {
      return Arrays.hashCode(new Object[] { mResult, mReader, mWriter, mDescription });
    }
  }

  
Borrowed from Guava's Objects.equal(a, b) /** Borrowed from Guava's Objects.equal(a, b) */
  private static boolean objectsEqual(Object obj1, Object obj2) {
    return Objects.equals(obj1, obj2);
  }

  private static List<String> asList(Deque<String> deque) {
    List<String> list = new ArrayList<>(deque);
    Collections.reverse(list);
    return Collections.unmodifiableList(list);
  }
}