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
   *
   *     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 static org.junit.Assert.assertArrayEquals;
  import static org.junit.Assert.assertEquals;
  
  import java.io.ByteArrayInputStream;
  import java.io.ByteArrayOutputStream;
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.nio.charset.StandardCharsets;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.stream.Stream;
  
  import org.apache.avro.generic.GenericData;
  import org.apache.avro.generic.GenericData.EnumSymbol;
  import org.apache.avro.generic.GenericData.Record;
  import org.apache.avro.generic.GenericDatumReader;
  import org.apache.avro.generic.GenericDatumWriter;
  import org.apache.avro.generic.GenericRecord;
  import org.apache.avro.io.DatumWriter;
  import org.apache.avro.io.Decoder;
  import org.apache.avro.io.DecoderFactory;
  import org.apache.avro.io.Encoder;
  import org.apache.avro.io.EncoderFactory;
  
  import org.junit.jupiter.api.Assertions;
  import org.junit.jupiter.params.ParameterizedTest;
  import org.junit.jupiter.params.provider.EnumSource;
  
  public class TestReadingWritingDataInEvolvedSchemas {
  
    private static final String RECORD_A = "RecordA";
    private static final String FIELD_A = "fieldA";
    private static final char LATIN_SMALL_LETTER_O_WITH_DIARESIS = '\u00F6';
  
    private static final Schema DOUBLE_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().doubleType().noDefault() //
        .endRecord();
    private static final Schema FLOAT_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().floatType().noDefault() //
        .endRecord();
    private static final Schema LONG_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().longType().noDefault() //
        .endRecord();
    private static final Schema INT_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().intType().noDefault() //
        .endRecord();
    private static final Schema UNION_INT_LONG_FLOAT_DOUBLE_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().unionOf().doubleType().and().floatType().and().longType().and().intType().endUnion()
        .noDefault() //
        .endRecord();
    private static final Schema STRING_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().stringType().noDefault() //
        .endRecord();
    private static final Schema BYTES_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().bytesType().noDefault() //
        .endRecord();
    private static final Schema UNION_STRING_BYTES_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().unionOf().stringType().and().bytesType().endUnion().noDefault() //
        .endRecord();
  
    private static final Schema ENUM_AB = SchemaBuilder.enumeration("Enum1").symbols("A", "B");
  
    private static final Schema ENUM_AB_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type(ENUM_AB).noDefault() //
        .endRecord();
  
    private static final Schema ENUM_ABC = SchemaBuilder.enumeration("Enum1").symbols("A", "B", "C");
    private static final Schema ENUM_ABC_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type(ENUM_ABC).noDefault() //
        .endRecord();
    private static final Schema UNION_INT_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().unionOf().intType().endUnion().noDefault() //
        .endRecord();
    private static final Schema UNION_LONG_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().unionOf().longType().endUnion().noDefault() //
        .endRecord();
    private static final Schema UNION_FLOAT_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().unionOf().floatType().endUnion().noDefault() //
        .endRecord();
    private static final Schema UNION_DOUBLE_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().unionOf().doubleType().endUnion().noDefault() //
        .endRecord();
    private static final Schema UNION_LONG_FLOAT_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().unionOf().floatType().and().longType().endUnion().noDefault() //
        .endRecord();
    private static final Schema UNION_FLOAT_DOUBLE_RECORD = SchemaBuilder.record(RECORD_A) //
        .fields() //
        .name(FIELD_A).type().unionOf().floatType().and().doubleType().endUnion().noDefault() //
        .endRecord();
  
    enum EncoderType {
      BINARY, JSON
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void doubleWrittenWithUnionSchemaIsConvertedToDoubleSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_INT_LONG_FLOAT_DOUBLE_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42.0);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(DOUBLE_RECORD, writer, encoded, encoderType);
      assertEquals(42.0, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void longWrittenWithUnionSchemaIsConvertedToUnionLongFloatSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_LONG_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42L);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(UNION_LONG_FLOAT_RECORD, writer, encoded, encoderType);
      assertEquals(42L, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void longWrittenWithUnionSchemaIsConvertedToDoubleSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_LONG_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42L);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(UNION_DOUBLE_RECORD, writer, encoded, encoderType);
      assertEquals(42.0, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void intWrittenWithUnionSchemaIsConvertedToDoubleSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_INT_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(UNION_DOUBLE_RECORD, writer, encoded, encoderType);
      assertEquals(42.0, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void intWrittenWithUnionSchemaIsReadableByFloatSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_INT_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(FLOAT_RECORD, writer, encoded, encoderType);
      assertEquals(42.0f, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void intWrittenWithUnionSchemaIsReadableByFloatUnionSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_INT_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(UNION_FLOAT_RECORD, writer, encoded, encoderType);
      assertEquals(42.0f, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void longWrittenWithUnionSchemaIsReadableByFloatSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_LONG_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42L);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(FLOAT_RECORD, writer, encoded, encoderType);
      assertEquals(42.0f, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void longWrittenWithUnionSchemaIsReadableByFloatUnionSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_LONG_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42L);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(UNION_FLOAT_RECORD, writer, encoded, encoderType);
      assertEquals(42.0f, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void longWrittenWithUnionSchemaIsConvertedToLongFloatUnionSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_LONG_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42L);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(UNION_LONG_FLOAT_RECORD, writer, encoded, encoderType);
      assertEquals(42L, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void longWrittenWithUnionSchemaIsConvertedToFloatDoubleUnionSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_LONG_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42L);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(UNION_FLOAT_DOUBLE_RECORD, writer, encoded, encoderType);
      assertEquals(42.0F, decoded.get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void doubleWrittenWithUnionSchemaIsNotConvertedToFloatSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_INT_LONG_FLOAT_DOUBLE_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42.0);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      AvroTypeException exception = Assertions.assertThrows(AvroTypeException.class,
          () -> decodeGenericBlob(FLOAT_RECORD, writer, encoded, encoderType));
      Assertions.assertEquals("Found double, expecting float", exception.getMessage());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void floatWrittenWithUnionSchemaIsNotConvertedToLongSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_INT_LONG_FLOAT_DOUBLE_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42.0f);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      AvroTypeException exception = Assertions.assertThrows(AvroTypeException.class,
          () -> decodeGenericBlob(LONG_RECORD, writer, encoded, encoderType));
      Assertions.assertEquals("Found float, expecting long", exception.getMessage());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void longWrittenWithUnionSchemaIsNotConvertedToIntSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_INT_LONG_FLOAT_DOUBLE_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42L);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      AvroTypeException exception = Assertions.assertThrows(AvroTypeException.class,
          () -> decodeGenericBlob(INT_RECORD, writer, encoded, encoderType));
      Assertions.assertEquals("Found long, expecting int", exception.getMessage());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void intWrittenWithUnionSchemaIsConvertedToAllNumberSchemas(EncoderType encoderType) throws Exception {
      Schema writer = UNION_INT_LONG_FLOAT_DOUBLE_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      assertEquals(42.0, decodeGenericBlob(DOUBLE_RECORD, writer, encoded, encoderType).get(FIELD_A));
      assertEquals(42.0f, decodeGenericBlob(FLOAT_RECORD, writer, encoded, encoderType).get(FIELD_A));
      assertEquals(42L, decodeGenericBlob(LONG_RECORD, writer, encoded, encoderType).get(FIELD_A));
      assertEquals(42, decodeGenericBlob(INT_RECORD, writer, encoded, encoderType).get(FIELD_A));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void asciiStringWrittenWithUnionSchemaIsConvertedToBytesSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_STRING_BYTES_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, "42");
      byte[] encoded = encodeGenericBlob(record, encoderType);
      ByteBuffer actual = (ByteBuffer) decodeGenericBlob(BYTES_RECORD, writer, encoded, encoderType).get(FIELD_A);
      assertArrayEquals("42".getBytes(StandardCharsets.UTF_8), actual.array());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void utf8StringWrittenWithUnionSchemaIsConvertedToBytesSchema(EncoderType encoderType) throws Exception {
      String goeran = String.format("G%sran", LATIN_SMALL_LETTER_O_WITH_DIARESIS);
      Schema writer = UNION_STRING_BYTES_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, goeran);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      ByteBuffer actual = (ByteBuffer) decodeGenericBlob(BYTES_RECORD, writer, encoded, encoderType).get(FIELD_A);
      assertArrayEquals(goeran.getBytes(StandardCharsets.UTF_8), actual.array());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void asciiBytesWrittenWithUnionSchemaIsConvertedToStringSchema(EncoderType encoderType) throws Exception {
      Schema writer = UNION_STRING_BYTES_RECORD;
      ByteBuffer buf = ByteBuffer.wrap("42".getBytes(StandardCharsets.UTF_8));
      Record record = defaultRecordWithSchema(writer, FIELD_A, buf);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      CharSequence read = (CharSequence) decodeGenericBlob(STRING_RECORD, writer, encoded, encoderType).get(FIELD_A);
      assertEquals("42", read.toString());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void utf8BytesWrittenWithUnionSchemaIsConvertedToStringSchema(EncoderType encoderType) throws Exception {
      String goeran = String.format("G%sran", LATIN_SMALL_LETTER_O_WITH_DIARESIS);
      Schema writer = UNION_STRING_BYTES_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, goeran);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      CharSequence read = (CharSequence) decodeGenericBlob(STRING_RECORD, writer, encoded, encoderType).get(FIELD_A);
      assertEquals(goeran, read.toString());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void enumRecordCanBeReadWithExtendedEnumSchema(EncoderType encoderType) throws Exception {
      Schema writer = ENUM_AB_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, new EnumSymbol(ENUM_AB, "A"));
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(ENUM_ABC_RECORD, writer, encoded, encoderType);
      assertEquals("A", decoded.get(FIELD_A).toString());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void enumRecordWithExtendedSchemaCanBeReadWithOriginalEnumSchemaIfOnlyOldValues(EncoderType encoderType)
        throws Exception {
      Schema writer = ENUM_ABC_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, new EnumSymbol(ENUM_ABC, "A"));
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(ENUM_AB_RECORD, writer, encoded, encoderType);
      assertEquals("A", decoded.get(FIELD_A).toString());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void enumRecordWithExtendedSchemaCanNotBeReadIfNewValuesAreUsed(EncoderType encoderType) throws Exception {
      Schema writer = ENUM_ABC_RECORD;
      Record record = defaultRecordWithSchema(writer, FIELD_A, new EnumSymbol(ENUM_ABC, "C"));
      byte[] encoded = encodeGenericBlob(record, encoderType);
  
      AvroTypeException exception = Assertions.assertThrows(AvroTypeException.class,
          () -> decodeGenericBlob(ENUM_AB_RECORD, writer, encoded, encoderType));
      Assertions.assertEquals("No match for C", exception.getMessage());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void recordWrittenWithExtendedSchemaCanBeReadWithOriginalSchemaButLossOfData(EncoderType encoderType)
        throws Exception {
      Schema writer = SchemaBuilder.record(RECORD_A) //
          .fields() //
          .name("newTopField").type().stringType().noDefault() //
          .name(FIELD_A).type().intType().noDefault() //
          .endRecord();
      Record record = defaultRecordWithSchema(writer, FIELD_A, 42);
      record.put("newTopField", "not decoded");
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(INT_RECORD, writer, encoded, encoderType);
      assertEquals(42, decoded.get(FIELD_A));
      try {
        decoded.get("newTopField");
        Assertions.fail("get should throw a exception");
      } catch (AvroRuntimeException ex) {
        Assertions.assertEquals("Not a valid schema field: newTopField", ex.getMessage());
      }
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void readerWithoutDefaultValueThrowsException(EncoderType encoderType) throws Exception {
      Schema reader = SchemaBuilder.record(RECORD_A) //
          .fields() //
          .name("newField").type().intType().noDefault() //
          .name(FIELD_A).type().intType().noDefault() //
          .endRecord();
      Record record = defaultRecordWithSchema(INT_RECORD, FIELD_A, 42);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      AvroTypeException exception = Assertions.assertThrows(AvroTypeException.class,
          () -> decodeGenericBlob(reader, INT_RECORD, encoded, encoderType));
      Assertions.assertTrue(exception.getMessage().contains("missing required field newField"), exception.getMessage());
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void readerWithDefaultValueIsApplied(EncoderType encoderType) throws Exception {
      Schema reader = SchemaBuilder.record(RECORD_A) //
          .fields() //
          .name("newFieldWithDefault").type().intType().intDefault(314) //
          .name(FIELD_A).type().intType().noDefault() //
          .endRecord();
      Record record = defaultRecordWithSchema(INT_RECORD, FIELD_A, 42);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      Record decoded = decodeGenericBlob(reader, INT_RECORD, encoded, encoderType);
      assertEquals(42, decoded.get(FIELD_A));
      assertEquals(314, decoded.get("newFieldWithDefault"));
    }
  
    @ParameterizedTest
    @EnumSource(EncoderType.class)
    void aliasesInSchema(EncoderType encoderType) throws Exception {
      Schema writer = new Schema.Parser()
          .parse("{\"namespace\": \"example.avro\", \"type\": \"record\", \"name\": \"User\", \"fields\": ["
              + "{\"name\": \"name\", \"type\": \"int\"}\n" + "]}\n");
      Schema reader = new Schema.Parser()
          .parse("{\"namespace\": \"example.avro\", \"type\": \"record\", \"name\": \"User\", \"fields\": ["
              + "{\"name\": \"fname\", \"type\": \"int\", \"aliases\" : [ \"name\" ]}\n" + "]}\n");
  
      GenericData.Record record = defaultRecordWithSchema(writer, "name", 1);
      byte[] encoded = encodeGenericBlob(record, encoderType);
      GenericData.Record decoded = decodeGenericBlob(reader, reader, encoded, encoderType);
  
      assertEquals(1, decoded.get("fname"));
    }
  
    private <T> Record defaultRecordWithSchema(Schema schema, String key, T value) {
      Record data = new GenericData.Record(schema);
      data.put(key, value);
      return data;
    }
  
    private byte[] encodeGenericBlob(GenericRecord data, EncoderType encoderType) throws IOException {
      DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(data.getSchema());
      ByteArrayOutputStream outStream = new ByteArrayOutputStream();
      Encoder encoder = encoderType == EncoderType.BINARY ? EncoderFactory.get().binaryEncoder(outStream, null)
          : EncoderFactory.get().jsonEncoder(data.getSchema(), outStream);
      writer.write(data, encoder);
      encoder.flush();
      outStream.close();
      return outStream.toByteArray();
    }
  
    private Record decodeGenericBlob(Schema expectedSchema, Schema schemaOfBlob, byte[] blob, EncoderType encoderType)
        throws IOException {
      if (blob == null) {
        return null;
      }
      GenericData data = new GenericData();
      data.setFastReaderEnabled(true);
      GenericDatumReader<Record> reader = new GenericDatumReader<>(null, null, data);
      reader.setExpected(expectedSchema);
      reader.setSchema(schemaOfBlob);
      Decoder decoder = encoderType == EncoderType.BINARY ? DecoderFactory.get().binaryDecoder(blob, null)
          : DecoderFactory.get().jsonDecoder(schemaOfBlob, new ByteArrayInputStream(blob));
      return reader.read(null, decoder);
    }
  }
  