{
+
+ R apply(T input) throws IOException;
+}
diff --git a/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/InMemoryEncoder.java b/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/InMemoryEncoder.java
new file mode 100644
index 000000000..b74d902cb
--- /dev/null
+++ b/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/InMemoryEncoder.java
@@ -0,0 +1,175 @@
+package com.linkedin.avro.fastserde.logical.types;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import org.apache.avro.Schema;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.Encoder;
+import org.apache.avro.util.Utf8;
+
+import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
+import com.linkedin.avroutil1.compatibility.SchemaNormalization;
+
+/**
+ * Helper class which wraps {@link BinaryEncoder} and its corresponding {@code OutputStream}
+ * ({@link ByteArrayOutputStream}) so that we can simplify some repeatable code just to:
+ *
+ * InMemoryEncoder encoder = new InMemoryEncoder(data.getSchema());
+ * fastSerializer.serialize(data, encoder);
+ * return encoder.toByteArray();
+ *
+ * Additionally, it prepends 10-bytes header to the output byte-array so that it can be easily compared with
+ * built-in serialization method: {@code someSpecificRecord.toByteBuffer().toByteArray()}.
+ */
+public class InMemoryEncoder extends Encoder {
+
+ private final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+ private final BinaryEncoder binaryEncoder = AvroCompatibilityHelper.newBinaryEncoder(baos);
+
+ public InMemoryEncoder(Schema schema) {
+ final int v1HeaderLength = 10;
+ byte[] v1Header = ByteBuffer.wrap(new byte[v1HeaderLength])
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .put(new byte[]{(byte) 0xC3, (byte) 0x01}) // BinaryMessageEncoder.V1_HEADER
+ .putLong(SchemaNormalization.parsingFingerprint64(schema))
+ .array();
+
+ baos.write(v1Header, 0, v1Header.length);
+ }
+
+ public byte[] toByteArray() {
+ try {
+ binaryEncoder.flush();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ return baos.toByteArray();
+ }
+
+ // generated delegate methods
+
+ @Override
+ public void writeNull() throws IOException {
+ binaryEncoder.writeNull();
+ }
+
+ @Override
+ public void writeString(Utf8 utf8) throws IOException {
+ binaryEncoder.writeString(utf8);
+ }
+
+ @Override
+ public void writeString(String string) throws IOException {
+ binaryEncoder.writeString(string);
+ }
+
+ @Override
+ public void writeBytes(ByteBuffer bytes) throws IOException {
+ binaryEncoder.writeBytes(bytes);
+ }
+
+ @Override
+ public void writeBytes(byte[] bytes, int start, int len) throws IOException {
+ binaryEncoder.writeBytes(bytes, start, len);
+ }
+
+ @Override
+ public void writeEnum(int e) throws IOException {
+ binaryEncoder.writeEnum(e);
+ }
+
+ @Override
+ public void writeArrayStart() throws IOException {
+ binaryEncoder.writeArrayStart();
+ }
+
+ @Override
+ public void setItemCount(long itemCount) throws IOException {
+ binaryEncoder.setItemCount(itemCount);
+ }
+
+ @Override
+ public void startItem() throws IOException {
+ binaryEncoder.startItem();
+ }
+
+ @Override
+ public void writeArrayEnd() throws IOException {
+ binaryEncoder.writeArrayEnd();
+ }
+
+ @Override
+ public void writeMapStart() throws IOException {
+ binaryEncoder.writeMapStart();
+ }
+
+ @Override
+ public void writeMapEnd() throws IOException {
+ binaryEncoder.writeMapEnd();
+ }
+
+ @Override
+ public void writeIndex(int unionIndex) throws IOException {
+ binaryEncoder.writeIndex(unionIndex);
+ }
+
+ @Override
+ public void writeBoolean(boolean b) throws IOException {
+ binaryEncoder.writeBoolean(b);
+ }
+
+ @Override
+ public void writeInt(int n) throws IOException {
+ binaryEncoder.writeInt(n);
+ }
+
+ @Override
+ public void writeLong(long n) throws IOException {
+ binaryEncoder.writeLong(n);
+ }
+
+ @Override
+ public void writeFloat(float f) throws IOException {
+ binaryEncoder.writeFloat(f);
+ }
+
+ @Override
+ public void writeDouble(double d) throws IOException {
+ binaryEncoder.writeDouble(d);
+ }
+
+ @Override
+ public void writeString(CharSequence charSequence) throws IOException {
+ binaryEncoder.writeString(charSequence);
+ }
+
+ @Override
+ public void writeBytes(byte[] bytes) throws IOException {
+ binaryEncoder.writeBytes(bytes);
+ }
+
+ @Override
+ public void writeFixed(byte[] bytes, int start, int len) throws IOException {
+ binaryEncoder.writeFixed(bytes, start, len);
+ }
+
+ @Override
+ public void writeFixed(byte[] bytes) throws IOException {
+ binaryEncoder.writeFixed(bytes);
+ }
+
+ @Override
+ public void writeFixed(ByteBuffer bytes) throws IOException {
+ binaryEncoder.writeFixed(bytes);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ binaryEncoder.flush();
+ }
+}
diff --git a/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/LogicalTypesFastSerdeTest.java b/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/LogicalTypesFastSerdeTest.java
new file mode 100644
index 000000000..d72597096
--- /dev/null
+++ b/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/LogicalTypesFastSerdeTest.java
@@ -0,0 +1,239 @@
+package com.linkedin.avro.fastserde.logical.types;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.apache.avro.util.Utf8;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+import org.testng.collections.Lists;
+
+import com.linkedin.avro.fastserde.Utils;
+import com.linkedin.avro.fastserde.generated.avro.FastSerdeLogicalTypesTest1;
+import com.linkedin.avro.fastserde.generated.avro.FastSerdeLogicalTypesWithDefaults;
+import com.linkedin.avro.fastserde.generated.avro.LocalTimestampRecord;
+import com.linkedin.avro.fastserde.generated.avro.LocalTimestampRecordWithDefaults;
+import com.linkedin.avroutil1.compatibility.AvroVersion;
+
+public class LogicalTypesFastSerdeTest extends LogicalTypesTestBase {
+
+ @DataProvider
+ public static Object[][] logicalTypesTestCases() {
+ LocalDate now = LocalDate.now();
+ LocalDate localDate = LocalDate.of(2023, 8, 11);
+
+ Map mapOfDates = mapOf(
+ new String[]{"today", "yesterday", "tomorrow"},
+ new LocalDate[]{now, now.minusDays(1), now.plusDays(1)});
+
+ Map mapOfTimestamps = mapOf(
+ new String[]{"today", "yesterday", "tomorrow"},
+ new Instant[]{toInstant(now), toInstant(now.minusDays(1)), toInstant(now.plusDays(1))});
+
+ Map mapOfDatesAndTimestamps = mapOf(
+ new String[]{"today", "yesterday", "tomorrow"},
+ new Object[]{toInstant(now), now.minusDays(1), now.plusDays(1)});
+
+ Object[] mapOfUnionsOfDateAndTimestampMillisOptions = {mapOfDates, mapOfTimestamps, mapOfDatesAndTimestamps};
+ Object[] nullableUnionOfDateAndLocalTimestampOptions = {null, now.minusDays(12), localDate.atStartOfDay()};
+ Object[] unionOfDateAndLocalTimestampOptions = {now.minusDays(12), localDate.atStartOfDay()};
+ Object[] unionOfArrayAndMapOptions = {
+ Lists.newArrayList(LocalTime.now(), LocalTime.now().plusMinutes(1)), mapOfDates};
+ Object[] nullableArrayOfDatesOptions = {
+ null, Lists.newArrayList(localDate, localDate.plusDays(11), localDate.plusDays(22))};
+ Object[] decimalOrDateOptions = {new BigDecimal("3.14"), LocalDate.of(2023, 3, 14)};
+
+ List allOptions = new ArrayList<>();
+
+ for (Object mapOfUnionsOfDateAndTimestampMillis : mapOfUnionsOfDateAndTimestampMillisOptions) {
+ for (Object nullableUnionOfDateAndLocalTimestamp : nullableUnionOfDateAndLocalTimestampOptions) {
+ for (Object unionOfDateAndLocalTimestamp : unionOfDateAndLocalTimestampOptions) {
+ for (Object unionOfArrayAndMap : unionOfArrayAndMapOptions) {
+ for (Object nullableArrayOfDates : nullableArrayOfDatesOptions) {
+ for (Object decimalOrDate : decimalOrDateOptions) {
+ allOptions.add(new Object[]{mapOfUnionsOfDateAndTimestampMillis,
+ nullableUnionOfDateAndLocalTimestamp, unionOfDateAndLocalTimestamp,
+ unionOfArrayAndMap, nullableArrayOfDates, decimalOrDate});
+ }
+ }
+ }
+ }
+ }
+ }
+
+ return allOptions.toArray(new Object[0][]);
+ }
+
+ @Test(groups = "serializationTest", dataProvider = "logicalTypesTestCases")
+ public void shouldWriteAndReadLogicalTypesSuccessfully(Map mapOfUnionsOfDateAndTimestampMillis,
+ Object nullableUnionOfDateAndLocalTimestamp, Object unionOfDateAndLocalTimestamp,
+ Object unionOfArrayAndMap, List nullableArrayOfDates, Object decimalOrDate) throws IOException {
+ // given
+ LocalDate localDate = LocalDate.of(2023, 8, 11);
+ Instant instant = localDate.atStartOfDay().toInstant(ZoneOffset.UTC);
+ LocalTimestampRecord localTimestampRecord = createLocalTimestampRecord(nullableUnionOfDateAndLocalTimestamp, unionOfDateAndLocalTimestamp);
+
+ FastSerdeLogicalTypesTest1.Builder builder = FastSerdeLogicalTypesTest1.newBuilder()
+ .setMapOfUnionsOfDateAndTimestampMillis(mapOfUnionsOfDateAndTimestampMillis)
+ .setUnionOfArrayAndMap(unionOfArrayAndMap)
+ .setTimestampMillisMap(createTimestampMillisMap())
+ .setNullableArrayOfDates(nullableArrayOfDates)
+ .setArrayOfDates(Lists.newArrayList(localDate, localDate.plusDays(1), localDate.plusDays(2)))
+ .setUnionOfDecimalOrDate(decimalOrDate)
+ .setTimestampMillisField(instant)
+ .setTimestampMicrosField(instant)
+ .setTimeMillisField(LocalTime.of(14, 17, 45, 12345))
+ .setTimeMicrosField(LocalTime.of(14, 17, 45, 12345))
+ .setDateField(localDate)
+ .setNestedLocalTimestampMillis(localTimestampRecord);
+ injectUuidField(builder);
+ FastSerdeLogicalTypesTest1 inputData = builder.build();
+
+ // all serializers produce the same array of bytes
+ byte[] bytesWithHeader = verifySerializers(inputData, FastSerdeLogicalTypesTest1::toByteBuffer);
+
+ // all deserializers create (logically) the same data (in generic or specific representation)
+ verifyDeserializers(bytesWithHeader, FastSerdeLogicalTypesTest1::fromByteBuffer);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(groups = "serializationTest")
+ public void shouldCreateLogicalTypesFromDefaults() {
+ // given
+ LocalTime localTime0 = LocalTime.ofNanoOfDay(0L);
+ LocalDate localDate0 = LocalDate.ofYearDay(1970, 1);
+ LocalDateTime localDateTime0 = LocalDateTime.of(localDate0, localTime0);
+
+ fixConversionsIfAvro19(new FastSerdeLogicalTypesWithDefaults().getSpecificData());
+
+ // when
+ FastSerdeLogicalTypesWithDefaults data = FastSerdeLogicalTypesWithDefaults.newBuilder()
+ .setNestedLocalTimestampMillisBuilder(LocalTimestampRecordWithDefaults.newBuilder())
+ .build();
+
+ // and
+ List arrayOfLocalTimes = (List) data.getUnionOfArrayAndMap();
+ Map mapOfLocalDates = data.getMapOfUnionsOfDateAndTimestampMillis();
+ Map timestampMillisMap = data.getTimestampMillisMap();
+ List arrayOfDates = data.getArrayOfDates();
+ BigDecimal decimal = (BigDecimal) data.getUnionOfDecimalOrDate();
+ Object uuidField = data.getUuidField();
+ Instant timestampMillisField = data.getTimestampMillisField();
+ Instant timestampMicrosField = data.getTimestampMicrosField();
+ LocalTime timeMillisField = data.getTimeMillisField();
+ LocalTime timeMicrosField = data.getTimeMicrosField();
+ LocalDate dateField = data.getDateField();
+ Object nestedTimestamp = data.getNestedLocalTimestampMillis().getNestedTimestamp();
+ LocalDate nestedDate = (LocalDate) data.getNestedLocalTimestampMillis().getUnionOfDateAndLocalTimestamp();
+
+ // then
+ Assert.assertEquals(arrayOfLocalTimes, Lists.newArrayList(
+ localTime0.plus(654321, ChronoUnit.MILLIS),
+ localTime0.plus(7415896, ChronoUnit.MILLIS)));
+
+ Assert.assertEquals(mapOfLocalDates, mapOf(
+ new Utf8[] {new Utf8("someDay"), new Utf8("anotherDay")},
+ new LocalDate[] {localDate0.plusDays(12345), localDate0.plusDays(23456)}));
+
+ Assert.assertEquals(timestampMillisMap, mapOf(
+ new Utf8[] {new Utf8("timestampMillis1"), new Utf8("timestampMillis2")},
+ new Instant[] {Instant.ofEpochMilli(123456789012L), Instant.ofEpochMilli(112233445566L)}));
+
+ Assert.assertNull(data.getNullableArrayOfDates());
+ Assert.assertEquals(arrayOfDates, Lists.newArrayList(localDate0.plusDays(7777), localDate0.plusDays(8888)));
+ Assert.assertEquals(decimal.scale(), 2);
+ Assert.assertEquals(decimal.precision(), 5);
+ Assert.assertEquals(decimal, new BigDecimal(new BigInteger("13".getBytes(StandardCharsets.UTF_8)), 2));
+ Assert.assertEquals(uuidField.toString(), "b4ddd079-a024-4cc3-ac6c-a14f174c9922");
+ Assert.assertEquals(timestampMillisField, Instant.ofEpochMilli(120120120120L));
+ Assert.assertEquals(timestampMicrosField, Instant.ofEpochMilli(0L).plusNanos(123451234512345L * 1000L));
+ Assert.assertEquals(timeMillisField, localTime0.plus(15L, ChronoUnit.MILLIS));
+ Assert.assertEquals(timeMicrosField, localTime0.plus(16L, ChronoUnit.MICROS));
+ Assert.assertEquals(dateField, localDate0.plusDays(223344));
+ if (Utils.getRuntimeAvroVersion() == AvroVersion.AVRO_1_9) {
+ Assert.assertEquals(nestedTimestamp, 99L);
+ } else {
+ Assert.assertEquals(nestedTimestamp, localDateTime0.plus(99, ChronoUnit.MILLIS));
+ }
+ Assert.assertEquals(nestedDate, localDate0.plusDays(45678));
+ }
+
+ private Map createTimestampMillisMap() {
+ return mapOf(new String[]{"one", "two", "three"}, new Instant[]{
+ toInstant(LocalDate.of(2023, 8, 18)),
+ toInstant(LocalDate.of(2023, 8, 19)),
+ toInstant(LocalDate.of(2023, 8, 20))});
+ }
+
+ private LocalTimestampRecord createLocalTimestampRecord(
+ Object nullableUnionOfDateAndLocalTimestamp, Object unionOfDateAndLocalTimestamp) {
+ Instant nestedTimestamp = toInstant(LocalDate.of(2023, 8, 21));
+ LocalTimestampRecord.Builder builder = LocalTimestampRecord.newBuilder();
+
+ try {
+ if (Utils.getRuntimeAvroVersion().laterThan(AvroVersion.AVRO_1_9)) {
+ builder.getClass().getMethod("setNestedTimestamp", LocalDateTime.class)
+ .invoke(builder, LocalDateTime.ofInstant(nestedTimestamp, ZoneId.systemDefault()));
+ builder.getClass().getMethod("setNullableNestedTimestamp", LocalDateTime.class)
+ .invoke(builder, LocalDateTime.ofInstant(nestedTimestamp.plusSeconds(10), ZoneId.systemDefault()));
+ } else {
+ nullableUnionOfDateAndLocalTimestamp = Optional.ofNullable(toInstant(nullableUnionOfDateAndLocalTimestamp))
+ .map(Instant::toEpochMilli)
+ .orElse(null);
+ unionOfDateAndLocalTimestamp = toInstant(unionOfDateAndLocalTimestamp).toEpochMilli();
+
+ builder.getClass().getMethod("setNestedTimestamp", Long.TYPE)
+ .invoke(builder, nestedTimestamp.toEpochMilli());
+ builder.getClass().getMethod("setNullableNestedTimestamp", Long.class)
+ .invoke(builder, nestedTimestamp.toEpochMilli() + 10L);
+ }
+
+ builder.setNullableUnionOfDateAndLocalTimestamp(nullableUnionOfDateAndLocalTimestamp);
+ builder.setUnionOfDateAndLocalTimestamp(unionOfDateAndLocalTimestamp);
+ } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+
+ return builder.build();
+ }
+
+ private static Map mapOf(K[] keys, V[] values) {
+ Map map = new LinkedHashMap<>();
+ for (int i = 0; i < keys.length; i++) {
+ map.put(keys[i], values[i]);
+ }
+
+ return map;
+ }
+
+ private void injectUuidField(FastSerdeLogicalTypesTest1.Builder builder) {
+ try {
+ if (Utils.getRuntimeAvroVersion().laterThan(AvroVersion.AVRO_1_10)) {
+ builder.getClass().getMethod("setUuidField", UUID.class)
+ .invoke(builder, UUID.randomUUID());
+ } else {
+ builder.getClass().getMethod("setUuidField", CharSequence.class)
+ .invoke(builder, UUID.randomUUID().toString());
+ }
+ } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/LogicalTypesTestBase.java b/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/LogicalTypesTestBase.java
new file mode 100644
index 000000000..2e156ae48
--- /dev/null
+++ b/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/LogicalTypesTestBase.java
@@ -0,0 +1,251 @@
+package com.linkedin.avro.fastserde.logical.types;
+
+import static com.linkedin.avro.fastserde.FastSerdeTestsSupport.getCodeGenDirectory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+import org.apache.avro.Conversion;
+import org.apache.avro.Conversions;
+import org.apache.avro.Schema;
+import org.apache.avro.data.TimeConversions;
+import org.apache.avro.generic.ColdGenericDatumReader;
+import org.apache.avro.generic.ColdSpecificDatumReader;
+import org.apache.avro.generic.GenericData;
+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.specific.SpecificData;
+import org.apache.avro.specific.SpecificDatumReader;
+import org.apache.avro.specific.SpecificDatumWriter;
+import org.apache.avro.specific.SpecificRecordBase;
+import org.testng.Assert;
+import org.testng.collections.Lists;
+
+import com.linkedin.avro.fastserde.FastDeserializer;
+import com.linkedin.avro.fastserde.FastGenericDeserializerGenerator;
+import com.linkedin.avro.fastserde.FastGenericSerializerGenerator;
+import com.linkedin.avro.fastserde.FastSerdeCache;
+import com.linkedin.avro.fastserde.FastSerializer;
+import com.linkedin.avro.fastserde.FastSpecificDeserializerGenerator;
+import com.linkedin.avro.fastserde.FastSpecificSerializerGenerator;
+import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelperCommon;
+import com.linkedin.avroutil1.compatibility.AvroVersion;
+
+public abstract class LogicalTypesTestBase {
+
+ private final File classesDir;
+ private final ClassLoader classLoader;
+
+ LogicalTypesTestBase() {
+ try {
+ classesDir = getCodeGenDirectory();
+ classLoader = URLClassLoader.newInstance(new URL[]{classesDir.toURI().toURL()}, getClass().getClassLoader());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected T copyConversions(SpecificData fromSpecificData, T toModelData) {
+ Optional.ofNullable(fromSpecificData.getConversions())
+ .orElse(Collections.emptyList())
+ .forEach(toModelData::addLogicalTypeConversion);
+
+ fixConversionsIfAvro19(toModelData);
+
+ return toModelData;
+ }
+
+ protected void fixConversionsIfAvro19(T modelData) {
+ if (AvroCompatibilityHelperCommon.getRuntimeAvroVersion() == AvroVersion.AVRO_1_9) {
+ // 1. These conversions should be injected by default (avro-compiler issue).
+ // 2. Missing DecimalConversion causes conflict with stringable feature (BigDecimal is considered as string, not bytes)
+ // details: GenericData.resolveUnion() + SpecificData.getSchemaName().
+ // 3. Time-conversions are needed for GenericDatumWriter/Reader which sees logicalType in the schema
+ // but is unable to find corresponding converter.
+ // 4. LocalTimestampMillisConversion not available in 1.9.
+
+ List> missingConversions = Lists.newArrayList(
+ new Conversions.DecimalConversion(),
+ new TimeConversions.TimeMicrosConversion(),
+ new TimeConversions.TimestampMicrosConversion());
+
+ for (Conversion conversion : missingConversions) {
+ modelData.addLogicalTypeConversion(conversion);
+ }
+ }
+ }
+
+ protected byte[] serialize(FastSerializer fastSerializer, T data) throws IOException {
+ InMemoryEncoder encoder = new InMemoryEncoder(data.getSchema());
+ fastSerializer.serialize(data, encoder);
+ return encoder.toByteArray();
+ }
+
+ protected byte[] serialize(DatumWriter datumWriter, T data) throws IOException {
+ InMemoryEncoder encoder = new InMemoryEncoder(data.getSchema());
+ datumWriter.write(data, encoder);
+
+ return encoder.toByteArray();
+ }
+
+ protected static Instant toInstant(Object maybeDate) {
+ if (maybeDate == null) {
+ return null;
+ } else if (maybeDate instanceof LocalDate) {
+ return ((LocalDate) maybeDate).atStartOfDay(ZoneId.systemDefault()).toInstant();
+ } else if (maybeDate instanceof LocalDateTime) {
+ return ((LocalDateTime) maybeDate).toInstant(ZoneOffset.UTC);
+ } else {
+ throw new UnsupportedOperationException(maybeDate + " is not supported (yet)");
+ }
+ }
+
+ /**
+ * Runs various serializers and ensures each one returned the same result.
+ * @return result from default serialization, normally it's just {@code data.toByteBuffer().array()}
+ */
+ protected byte[] verifySerializers(T data,
+ FunctionThrowingIOException toByteBuffer) throws IOException {
+ // given
+ Schema schema = data.getSchema();
+ GenericData genericData = copyConversions(data.getSpecificData(), new GenericData());
+ SpecificData specificData = copyConversions(data.getSpecificData(), new SpecificData());
+
+ FastSerializer fastGenericSerializer = new FastGenericSerializerGenerator(
+ schema, classesDir, classLoader, null, genericData)
+ .generateSerializer();
+
+ FastSerializer fastSpecificSerializer = new FastSpecificSerializerGenerator(
+ schema, classesDir, classLoader, null, specificData)
+ .generateSerializer();
+
+ FastSerdeCache.FastSerializerWithAvroGenericImpl fastSerializerWithAvroGeneric =
+ new FastSerdeCache.FastSerializerWithAvroGenericImpl<>(schema, genericData);
+
+ FastSerdeCache.FastSerializerWithAvroSpecificImpl fastSerializerWithAvroSpecific =
+ new FastSerdeCache.FastSerializerWithAvroSpecificImpl<>(schema, specificData);
+
+ GenericDatumWriter genericDatumWriter = new GenericDatumWriter<>(
+ schema, genericData);
+
+ SpecificDatumWriter specificDatumWriter = new SpecificDatumWriter<>(
+ schema, specificData);
+
+ fixConversionsIfAvro19(data.getSpecificData());
+
+ // when
+ byte[] fastGenericBytes = serialize(fastGenericSerializer, data);
+ byte[] fastSpecificBytes = serialize(fastSpecificSerializer, data);
+ byte[] fastGenericWithAvroBytes = serialize(fastSerializerWithAvroGeneric, data);
+ byte[] fastSpecificWithAvroBytes = serialize(fastSerializerWithAvroSpecific, data);
+ byte[] genericBytes = serialize(genericDatumWriter, data);
+ byte[] specificBytes = serialize(specificDatumWriter, data);
+ byte[] defaultBytes = toByteBuffer.apply(data).array();
+
+ // then all 7 serializing methods should return the same array of bytes
+ Assert.assertEquals(fastGenericBytes, defaultBytes);
+ Assert.assertEquals(fastSpecificBytes, defaultBytes);
+ Assert.assertEquals(fastGenericWithAvroBytes, defaultBytes);
+ Assert.assertEquals(fastSpecificWithAvroBytes, defaultBytes);
+ Assert.assertEquals(genericBytes, defaultBytes);
+ Assert.assertEquals(specificBytes, defaultBytes);
+
+ return defaultBytes;
+ }
+
+ protected T verifyDeserializers(byte[] bytesWithHeader,
+ FunctionThrowingIOException fromByteBuffer) throws IOException {
+ // given
+ T data = fromByteBuffer.apply(ByteBuffer.wrap(bytesWithHeader));
+ byte[] bytes = dropV1Header(bytesWithHeader);
+ Schema schema = data.getSchema();
+ GenericData genericData = copyConversions(data.getSpecificData(), new GenericData());
+ SpecificData specificData = copyConversions(data.getSpecificData(), new SpecificData());
+ Supplier decoderSupplier = () -> DecoderFactory.get().binaryDecoder(bytes, null);
+
+ FastDeserializer fastGenericDeserializer = new FastGenericDeserializerGenerator(
+ schema, schema, classesDir, classLoader, null, genericData)
+ .generateDeserializer();
+
+ FastDeserializer fastSpecificDeserializer = new FastSpecificDeserializerGenerator(
+ schema, schema, classesDir, classLoader, null, specificData)
+ .generateDeserializer();
+
+ FastSerdeCache.FastDeserializerWithAvroGenericImpl fastDeserializerWithAvroGeneric =
+ new FastSerdeCache.FastDeserializerWithAvroGenericImpl<>(schema, schema, genericData);
+
+ FastSerdeCache.FastDeserializerWithAvroSpecificImpl fastDeserializerWithAvroSpecific =
+ new FastSerdeCache.FastDeserializerWithAvroSpecificImpl<>(schema, schema, specificData);
+
+ GenericDatumReader genericDatumReader = new GenericDatumReader<>(schema, schema, genericData);
+
+ SpecificDatumReader specificDatumReader = new SpecificDatumReader<>(schema, schema, specificData);
+
+ ColdGenericDatumReader coldGenericDatumReader = ColdGenericDatumReader.of(schema, schema, genericData);
+
+ ColdSpecificDatumReader coldSpecificDatumReader = ColdSpecificDatumReader.of(schema, schema, specificData);
+
+ // when deserializing with different serializers/writers
+ GenericData.Record deserializedWithFastGeneric = fastGenericDeserializer.deserialize(decoderSupplier.get());
+ T deserializedWithFastSpecific = fastSpecificDeserializer.deserialize(decoderSupplier.get());
+
+ GenericData.Record deserializedWithFastWithAvroGeneric = fastDeserializerWithAvroGeneric.deserialize(decoderSupplier.get());
+ T deserializedWithFastWithAvroSpecific = fastDeserializerWithAvroSpecific.deserialize(decoderSupplier.get());
+
+ GenericData.Record deserializedWithGenericReader = genericDatumReader.read(null, decoderSupplier.get());
+ T deserializedWithSpecificReader = specificDatumReader.read(null, decoderSupplier.get());
+
+ GenericData.Record deserializedWithColdGenericReader = coldGenericDatumReader.read(null, decoderSupplier.get());
+ T deserializedWithColdSpecificReader = coldSpecificDatumReader.read(null, decoderSupplier.get());
+
+ // then
+ Assert.assertEquals(deserializedWithFastSpecific, data);
+ Assert.assertEquals(deserializedWithFastWithAvroSpecific, data);
+ Assert.assertEquals(deserializedWithSpecificReader, data);
+ Assert.assertEquals(deserializedWithColdSpecificReader, data);
+
+ assertEquals(deserializedWithFastGeneric, data);
+ assertEquals(deserializedWithFastWithAvroGeneric, data);
+ assertEquals(deserializedWithGenericReader, data);
+ assertEquals(deserializedWithColdGenericReader, data);
+
+ return deserializedWithFastSpecific;
+ }
+
+ protected void assertEquals(GenericData.Record actual, T expected) throws IOException {
+ Assert.assertEquals(actual.toString(), expected.toString());
+
+ GenericDatumWriter genericDatumWriter = new GenericDatumWriter<>(
+ actual.getSchema(), copyConversions(expected.getSpecificData(), new GenericData()));
+
+ SpecificDatumWriter specificDatumWriter = new SpecificDatumWriter<>(
+ expected.getSchema(), copyConversions(expected.getSpecificData(), new SpecificData()));
+
+ byte[] genericBytes = serialize(genericDatumWriter, actual);
+ byte[] expectedBytes = serialize(specificDatumWriter, expected);
+
+ Assert.assertEquals(genericBytes, expectedBytes);
+ }
+
+ protected byte[] dropV1Header(byte[] bytesWithHeader) {
+ final int v1HeaderLength = 10;
+ return Arrays.copyOfRange(bytesWithHeader, v1HeaderLength, bytesWithHeader.length);
+ }
+}
diff --git a/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/UndefinedLogicalTypesFastSerdeTest.java b/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/UndefinedLogicalTypesFastSerdeTest.java
new file mode 100644
index 000000000..19c1843e3
--- /dev/null
+++ b/fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/logical/types/UndefinedLogicalTypesFastSerdeTest.java
@@ -0,0 +1,117 @@
+package com.linkedin.avro.fastserde.logical.types;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalTime;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import org.testng.collections.Lists;
+
+import com.linkedin.avro.fastserde.generated.avro.FastSerdeLogicalTypesDefined;
+import com.linkedin.avro.fastserde.generated.avro.FastSerdeLogicalTypesUndefined;
+
+public class UndefinedLogicalTypesFastSerdeTest extends LogicalTypesTestBase {
+
+ private final LocalTime localTime = LocalTime.now();
+ private final LocalDate localDate = LocalDate.now();
+
+ @Test(groups = "serializationTest")
+ void definedAndUndefinedLogicalTypesShouldSerializeToTheSameBytes() throws IOException {
+ // given
+ FastSerdeLogicalTypesDefined logicalTypesDefined = createFastSerdeLogicalTypesDefined();
+ FastSerdeLogicalTypesUndefined logicalTypesUndefined = createFastSerdeLogicalTypesUndefined();
+
+ // when
+ byte[] bytesFromDefinedLogicalTypes = dropV1Header(verifySerializers(
+ logicalTypesDefined, FastSerdeLogicalTypesDefined::toByteBuffer));
+
+ // and
+ byte[] bytesFromUndefinedLogicalTypes = dropV1Header(verifySerializers(
+ logicalTypesUndefined, FastSerdeLogicalTypesUndefined::toByteBuffer));
+
+ // then
+ Assert.assertEquals(bytesFromDefinedLogicalTypes, bytesFromUndefinedLogicalTypes);
+ }
+
+ @Test(groups = "deserializationTest")
+ void deserializationToDefinedAndUndefinedLogicalTypesShouldBeEquivalent() throws IOException {
+ // given
+ byte[] bytesWithHeaderAndLogicalTypesDefined = verifySerializers(createFastSerdeLogicalTypesDefined(),
+ FastSerdeLogicalTypesDefined::toByteBuffer);
+ byte[] bytesWithHeaderAndLogicalTypesUndefined = verifySerializers(createFastSerdeLogicalTypesUndefined(),
+ FastSerdeLogicalTypesUndefined::toByteBuffer);
+
+ // assume
+ Assert.assertEquals(dropV1Header(bytesWithHeaderAndLogicalTypesDefined), dropV1Header(bytesWithHeaderAndLogicalTypesUndefined));
+
+ // when
+ FastSerdeLogicalTypesDefined logicalTypesDefined = verifyDeserializers(bytesWithHeaderAndLogicalTypesDefined,
+ FastSerdeLogicalTypesDefined::fromByteBuffer);
+
+ // and
+ FastSerdeLogicalTypesUndefined logicalTypesUndefined = verifyDeserializers(bytesWithHeaderAndLogicalTypesUndefined,
+ FastSerdeLogicalTypesUndefined::fromByteBuffer);
+
+ // then
+ Assert.assertEquals(toInt(logicalTypesDefined.getTimeMillisField()), logicalTypesUndefined.getTimeMillisField());
+ Assert.assertEquals(toInt(logicalTypesDefined.getDateField()), logicalTypesUndefined.getDateField());
+ assertArraysOfUnionsAreEquivalent(logicalTypesDefined.getArrayOfUnionOfDateAndTimestampMillis(),
+ logicalTypesUndefined.getArrayOfUnionOfDateAndTimestampMillis());
+ }
+
+ private void assertArraysOfUnionsAreEquivalent(List